feat(workflow): implement target channel personalities, raw review cards, on-demand AI rewrites and multi-target dispatch

2026-08-27 22:11:33 +03:30
parent 4f5570850e
commit 26cd91a48f
10 changed files with 695 additions and 699 deletions
@@ -1,5 +1,6 @@
 import asyncpg
 import os
+import json
 from typing import Optional

 DATABASE_URL = os.getenv(
@@ -23,6 +24,8 @@ CREATE TABLE IF NOT EXISTS targets (
    title VARCHAR(255),
    username VARCHAR(255),
    post_interval_min INT DEFAULT 30,
+    personality TEXT DEFAULT '',
+    custom_footer TEXT DEFAULT '',
    last_post_time TIMESTAMPTZ,
    is_active BOOLEAN DEFAULT TRUE,
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
@@ -44,7 +47,8 @@ CREATE TABLE IF NOT EXISTS posts (
    ai_text TEXT,
    suggested_target_id INT REFERENCES targets(id) ON DELETE SET NULL,
    target_channel_id INT REFERENCES targets(id) ON DELETE SET NULL,
-    status VARCHAR(32) DEFAULT 'pending_ai',
+    status VARCHAR(32) DEFAULT 'pending_review',
+    published_to JSONB DEFAULT '[]'::jsonb,
    review_message_id BIGINT,
    scheduled_at TIMESTAMPTZ,
    published_at TIMESTAMPTZ,
@@ -63,6 +67,11 @@ CREATE TABLE IF NOT EXISTS settings (
    value TEXT NOT NULL,
    description TEXT
 );
+
+-- Migration safety for existing tables
+ALTER TABLE targets ADD COLUMN IF NOT EXISTS personality TEXT DEFAULT '';
+ALTER TABLE targets ADD COLUMN IF NOT EXISTS custom_footer TEXT DEFAULT '';
+ALTER TABLE posts ADD COLUMN IF NOT EXISTS published_to JSONB DEFAULT '[]'::jsonb;
 """

 _pool: Optional[asyncpg.Pool] = None
@@ -1,5 +1,5 @@
 from dataclasses import dataclass, field
-from typing import Optional, List
+from typing import Optional, List, Dict, Any

@dataclass
 class SourceChannel:
@@ -17,6 +17,8 @@ class TargetChannel:
    title: Optional[str]
    username: Optional[str]
    post_interval_min: int = 30
+    personality: str = ""
+    custom_footer: str = ""
    last_post_time: Optional[str] = None
    is_active: bool = True
    created_at: Optional[str] = None
@@ -38,7 +40,8 @@ class Post:
    ai_text: Optional[str] = None
    suggested_target_id: Optional[int] = None
    target_channel_id: Optional[int] = None
-    status: str = "pending_ai"  # pending_ai, pending_review, approved, scheduled, published, rejected
+    status: str = "pending_review"  # pending_review, published, rejected
+    published_to: List[Dict[str, Any]] = field(default_factory=list)
    review_message_id: Optional[int] = None
    scheduled_at: Optional[str] = None
    published_at: Optional[str] = None
@@ -1,8 +1,20 @@
+import json
 import asyncpg
-from typing import List, Optional
+from typing import List, Optional, Dict, Any
 from db.models import SourceChannel, TargetChannel, Post, Setting
 from db.database import get_db_pool

+def _parse_post_row(row: asyncpg.Record) -> Post:
+    data = dict(row)
+    if isinstance(data.get("published_to"), str):
+        try:
+            data["published_to"] = json.loads(data["published_to"])
+        except Exception:
+            data["published_to"] = []
+    elif data.get("published_to") is None:
+        data["published_to"] = []
+    return Post(**data)
+
 class Repository:
    def __init__(self, dsn: Optional[str] = None):
        self.dsn = dsn
@@ -33,7 +45,7 @@ class Repository:
    async def get_active_sources(self) -> List[SourceChannel]:
        pool = await self._get_pool()
        async with pool.acquire() as conn:
-            rows = await conn.fetch("SELECT * FROM sources WHERE is_active = TRUE;")
+            rows = await conn.fetch("SELECT * FROM sources WHERE is_active = TRUE ORDER BY id ASC;")
            return [SourceChannel(**dict(r)) for r in rows]

    async def get_source_by_channel_id(self, channel_id: int) -> Optional[SourceChannel]:
@@ -43,13 +55,21 @@ class Repository:
            return SourceChannel(**dict(row)) if row else None

    # --- Target Channels ---
-    async def add_target(self, channel_id: int, title: Optional[str] = None, username: Optional[str] = None, post_interval_min: int = 30) -> int:
+    async def add_target(
+        self,
+        channel_id: int,
+        title: Optional[str] = None,
+        username: Optional[str] = None,
+        post_interval_min: int = 30,
+        personality: str = "",
+        custom_footer: str = ""
+    ) -> int:
        pool = await self._get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                """
-                INSERT INTO targets (channel_id, title, username, post_interval_min)
-                VALUES ($1, $2, $3, $4)
+                INSERT INTO targets (channel_id, title, username, post_interval_min, personality, custom_footer)
+                VALUES ($1, $2, $3, $4, $5, $6)
                ON CONFLICT(channel_id) DO UPDATE SET
                    title = EXCLUDED.title,
                    username = EXCLUDED.username,
@@ -57,14 +77,36 @@ class Repository:
                    is_active = TRUE
                RETURNING id;
                """,
-                channel_id, title, username, post_interval_min,
+                channel_id, title, username, post_interval_min, personality, custom_footer,
            )
            return row["id"]

+    async def update_target_personality(self, target_id: int, personality: str, custom_footer: Optional[str] = None) -> None:
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            if custom_footer is not None:
+                await conn.execute(
+                    "UPDATE targets SET personality = $1, custom_footer = $2 WHERE id = $3;",
+                    personality, custom_footer, target_id
+                )
+            else:
+                await conn.execute(
+                    "UPDATE targets SET personality = $1 WHERE id = $2;",
+                    personality, target_id
+                )
+
+    async def update_target_footer(self, target_id: int, custom_footer: str) -> None:
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            await conn.execute(
+                "UPDATE targets SET custom_footer = $1 WHERE id = $2;",
+                custom_footer, target_id
+            )
+
    async def get_active_targets(self) -> List[TargetChannel]:
        pool = await self._get_pool()
        async with pool.acquire() as conn:
-            rows = await conn.fetch("SELECT * FROM targets WHERE is_active = TRUE;")
+            rows = await conn.fetch("SELECT * FROM targets WHERE is_active = TRUE ORDER BY id ASC;")
            return [TargetChannel(**dict(r)) for r in rows]

    async def get_target_by_id(self, target_id: int) -> Optional[TargetChannel]:
@@ -81,34 +123,7 @@ class Repository:
                target_id,
            )

-    # --- Posts & Deduplication ---
-    async def find_duplicate_post_by_hash(self, content_hash: str) -> Optional[Post]:
-        if not content_hash:
-            return None
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            row = await conn.fetchrow(
-                "SELECT * FROM posts WHERE content_hash = $1 ORDER BY id ASC LIMIT 1;",
-                content_hash,
-            )
-            return Post(**dict(row)) if row else None
-
-    async def find_candidate_posts_by_tags(self, tags: List[str], exclude_post_id: Optional[int] = None, hours_lookback: int = 72, limit: int = 5) -> List[Post]:
-        if not tags:
-            return []
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            query = """
-                SELECT * FROM posts
-                WHERE tags && $1::text[]
-                AND created_at >= NOW() - ($2 || ' hours')::interval
-                AND ($3::bigint IS NULL OR id != $3::bigint)
-                ORDER BY created_at DESC
-                LIMIT $4;
-            """
-            rows = await conn.fetch(query, tags, str(hours_lookback), exclude_post_id, limit)
-            return [Post(**dict(r)) for r in rows]
-
+    # --- Posts & Multi-Channel Dispatch ---
    async def create_raw_post(
        self,
        source_channel_id: int,
@@ -130,7 +145,7 @@ class Repository:
                        source_channel_id, source_message_id, raw_text, media_path,
                        media_type, content_hash, is_duplicate, duplicate_of_id, similarity_reason, status
                    )
-                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'pending_ai')
+                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, 'pending_review')
                    RETURNING id;
                    """,
                    source_channel_id,
@@ -147,6 +162,12 @@ class Repository:
            except asyncpg.UniqueViolationError:
                return None

+    async def get_post_by_id(self, post_id: int) -> Optional[Post]:
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            row = await conn.fetchrow("SELECT * FROM posts WHERE id = $1;", post_id)
+            return _parse_post_row(row) if row else None
+
    async def get_posts_by_status(self, status: str, limit: int = 20) -> List[Post]:
        pool = await self._get_pool()
        async with pool.acquire() as conn:
@@ -154,63 +175,7 @@ class Repository:
                "SELECT * FROM posts WHERE status = $1 ORDER BY id ASC LIMIT $2;",
                status, limit,
            )
-            return [Post(**dict(r)) for r in rows]
-
-    async def get_post_by_id(self, post_id: int) -> Optional[Post]:
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            row = await conn.fetchrow("SELECT * FROM posts WHERE id = $1;", post_id)
-            return Post(**dict(row)) if row else None
-
-    async def update_post_tags(self, post_id: int, tags: List[str], subject: str) -> None:
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            await conn.execute(
-                "UPDATE posts SET tags = $1, subject = $2 WHERE id = $3;",
-                tags, subject, post_id,
-            )
-
-    async def update_post_duplicate_status(
-        self,
-        post_id: int,
-        is_duplicate: bool,
-        duplicate_of_id: Optional[int] = None,
-        similarity_reason: Optional[str] = None,
-    ) -> None:
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            await conn.execute(
-                """
-                UPDATE posts
-                SET is_duplicate = $1, duplicate_of_id = $2, similarity_reason = $3
-                WHERE id = $4;
-                """,
-                is_duplicate, duplicate_of_id, similarity_reason, post_id,
-            )
-
-    async def update_ai_result(
-        self,
-        post_id: int,
-        subject: str,
-        ai_text: str,
-        tags: List[str],
-        suggested_target_id: Optional[int] = None,
-        is_duplicate: bool = False,
-        duplicate_of_id: Optional[int] = None,
-        similarity_reason: Optional[str] = None,
-    ) -> None:
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            await conn.execute(
-                """
-                UPDATE posts
-                SET subject = $1, ai_text = $2, tags = $3, suggested_target_id = $4,
-                    is_duplicate = $5, duplicate_of_id = $6, similarity_reason = $7,
-                    status = 'pending_review'
-                WHERE id = $8;
-                """,
-                subject, ai_text, tags, suggested_target_id, is_duplicate, duplicate_of_id, similarity_reason, post_id,
-            )
+            return [_parse_post_row(r) for r in rows]

    async def update_review_message_id(self, post_id: int, review_message_id: int) -> None:
        pool = await self._get_pool()
@@ -220,62 +185,27 @@ class Repository:
                review_message_id, post_id,
            )

-    async def approve_post(self, post_id: int, target_channel_id: int) -> None:
+    async def record_post_published_to_target(self, post_id: int, target_id: int, target_title: str) -> None:
        pool = await self._get_pool()
        async with pool.acquire() as conn:
+            record_item = json.dumps({
+                "target_id": target_id,
+                "target_title": target_title,
+                "published_at": str(asyncpg.types.Type)
+            })
            await conn.execute(
                """
                UPDATE posts
-                SET target_channel_id = $1, status = 'approved'
+                SET published_to = published_to || $1::jsonb,
+                    status = 'published',
+                    published_at = CURRENT_TIMESTAMP
                WHERE id = $2;
                """,
-                target_channel_id, post_id,
+                f'[{{"target_id": {target_id}, "target_title": "{target_title}", "published_at": "{asyncpg.types.Type}"}}]',
+                post_id,
            )

    async def reject_post(self, post_id: int) -> None:
        pool = await self._get_pool()
        async with pool.acquire() as conn:
            await conn.execute("UPDATE posts SET status = 'rejected' WHERE id = $1;", post_id)
-
-    async def mark_post_published(self, post_id: int) -> None:
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            await conn.execute(
-                "UPDATE posts SET status = 'published', published_at = CURRENT_TIMESTAMP WHERE id = $1;",
-                post_id,
-            )
-
-    async def get_next_approved_post_for_target(self, target_id: int) -> Optional[Post]:
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            row = await conn.fetchrow(
-                """
-                SELECT * FROM posts
-                WHERE target_channel_id = $1 AND status = 'approved'
-                ORDER BY id ASC
-                LIMIT 1;
-                """,
-                target_id,
-            )
-            return Post(**dict(row)) if row else None
-
-    # --- Settings ---
-    async def get_setting(self, key: str, default: Optional[str] = None) -> Optional[str]:
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            row = await conn.fetchrow("SELECT value FROM settings WHERE key = $1;", key)
-            return row["value"] if row else default
-
-    async def set_setting(self, key: str, value: str, description: Optional[str] = None) -> None:
-        pool = await self._get_pool()
-        async with pool.acquire() as conn:
-            await conn.execute(
-                """
-                INSERT INTO settings (key, value, description)
-                VALUES ($1, $2, $3)
-                ON CONFLICT(key) DO UPDATE SET
-                    value = EXCLUDED.value,
-                    description = COALESCE(EXCLUDED.description, settings.description);
-                """,
-                key, value, description,
-            )