feat: add source channel context history configuration and original sent timestamp

RELEASE_NOTES.md

@@ -2,12 +2,12 @@
 
 👤 <b>ویرایش‌کننده:</b> <code>mamad</code>
 
-🔹 <b>ارتقای دکمه تست ارائه‌دهنده و پردازش تصویر (Vision Test Upgrade):</b>
+🔹 <b>ارسال پیام‌های زمینه کانال مبدا به هوش مصنوعی (Source Context History):</b>
-• اتصال خودکار تصویر تستی (<code>CODE: COPYKAR-TEST-7799</code>) در صورت فعال بودن گزینه پردازش تصویر (Vision) هنگام زدن دکمه <b>🧪 تست اختصاصی این Provider</b>.
+• افزودن گزینه تنظیم تعداد پیام‌های قبلی کانال مبدا (۰ تا ۱۰ پیام) در منوی مدیریت کانال‌های مبدا در ربات تلگرام.
-• بررسی و نمایش خودکار تحلیل و متن تصویر در خروجی پاسخ ارائه‌دهنده در ربات تلگرام.
+• استخراج خودکار آخرین پیام‌های معتبر هر کانال و تزریق آن‌ها به عنوان «زمینه و خط داستانی» به پرامپت هوش مصنوعی تا پست جدید با آگاهی از سیر رویدادهای قبلی بازنویسی شود.
-• پشتیبانی از کدگشایی فرمت مالتیمدال base64 در بریج محلی.
+📁 <i>فایل‌های تغییریافته:</i> <code>services/ai_processor.py</code>, <code>services/admin_bot.py</code>, <code>db/models.py</code>, <code>db/repository.py</code>, <code>tests/test_context_and_source_time.py</code>
-📁 <i>فایل‌های تغییریافته:</i> <code>services/admin_bot.py</code>, <code>agy_bridge.py</code>
 
-🔹 <b>اصلاح باکت‌های هیستوگرام تاخیر هوش مصنوعی (Histogram Latency Buckets):</b>
+🔹 <b>ثبت زمان واقعی ارسال پست در مبدا (Source Sent Timestamp):</b>
-• تعریف باکت‌های اختصاصی تا ۱۸۰ ثانیه برای پرومتئوس جهت جلوگیری از محدود شدن اشتباه صدک ۹۵ روی ۱۰ ثانیه در نمودار زمان پاسخ‌دهی هوش مصنوعی در گرافانا.
+• افزودن فیلد <code>source_created_at</code> به پایگاه‌داده و ذخیره مستقیم زمان واقعی ارسال پست در تلگرام (به جای صرفاً زمان دریافت محلی).
-📁 <i>فایل‌های تغییریافته:</i> <code>core/metrics.py</code>
+• مرتب‌سازی دقیق و زمانی پیام‌های اخیر بر اساس تاریخچه انتشار واقعی در کانال.
+📁 <i>فایل‌های تغییریافته:</i> <code>db/database.py</code>, <code>db/models.py</code>, <code>services/collector.py</code>

db/database.py

@@ -14,6 +14,7 @@ CREATE TABLE IF NOT EXISTS sources (
     channel_id BIGINT UNIQUE NOT NULL,
     username VARCHAR(255),
     title VARCHAR(255),
+    context_message_count INT DEFAULT 0,
     is_active BOOLEAN DEFAULT TRUE,
     created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
 );
@@ -59,6 +60,7 @@ CREATE TABLE IF NOT EXISTS posts (
     review_message_id BIGINT,
     scheduled_at TIMESTAMPTZ,
     published_at TIMESTAMPTZ,
+    source_created_at TIMESTAMPTZ,
     created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP,
     CONSTRAINT unique_source_message UNIQUE (source_channel_id, source_message_id)
 );
@@ -150,6 +152,9 @@ ALTER TABLE targets ADD COLUMN IF NOT EXISTS is_sleep_enabled BOOLEAN DEFAULT FA
 ALTER TABLE targets ADD COLUMN IF NOT EXISTS auto_source_ids BIGINT[] DEFAULT '{}';
 ALTER TABLE targets ADD COLUMN IF NOT EXISTS language VARCHAR(32) DEFAULT 'fa';
 ALTER TABLE targets ADD COLUMN IF NOT EXISTS custom_prompt TEXT DEFAULT '';
+ALTER TABLE sources ADD COLUMN IF NOT EXISTS context_message_count INT DEFAULT 0;
+ALTER TABLE posts ADD COLUMN IF NOT EXISTS source_created_at TIMESTAMPTZ;
+CREATE INDEX IF NOT EXISTS idx_posts_source_created ON posts(source_channel_id, source_created_at DESC);
 ALTER TABLE posts ADD COLUMN IF NOT EXISTS published_to JSONB DEFAULT '[]'::jsonb;
 ALTER TABLE posts ADD COLUMN IF NOT EXISTS is_deleted BOOLEAN DEFAULT FALSE;
 ALTER TABLE posts ADD COLUMN IF NOT EXISTS rejection_reason TEXT DEFAULT '';

db/models.py

@@ -16,6 +16,7 @@ class SourceChannel:
     username: Optional[str]
     title: Optional[str]
     category_id: Optional[int] = None
+    context_message_count: int = 0
     is_active: bool = True
     created_at: Optional[str] = None
 
@@ -65,6 +66,7 @@ class Post:
     review_message_id: Optional[int] = None
     scheduled_at: Optional[str] = None
     published_at: Optional[str] = None
+    source_created_at: Optional[str] = None
     created_at: Optional[str] = None
 
 @dataclass

db/repository.py

@@ -62,6 +62,11 @@ class Repository:
             row = await conn.fetchrow("SELECT * FROM sources WHERE id = $1;", source_id)
             return SourceChannel(**dict(row)) if row else None
 
+    async def update_source_context_count(self, source_id: int, count: int) -> None:
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            await conn.execute("UPDATE sources SET context_message_count = $1 WHERE id = $2;", max(0, count), source_id)
+
     async def delete_source(self, source_id: int) -> None:
         pool = await self._get_pool()
         async with pool.acquire() as conn:
@@ -236,6 +241,7 @@ class Repository:
         is_duplicate: bool = False,
         duplicate_of_id: Optional[int] = None,
         similarity_reason: Optional[str] = None,
+        source_created_at: Optional[datetime] = None,
     ) -> Optional[int]:
         pool = await self._get_pool()
         async with pool.acquire() as conn:
@@ -244,9 +250,9 @@ class Repository:
                     """
                     INSERT INTO posts (
                         source_channel_id, source_message_id, raw_text, media_path,
-                        media_type, content_hash, tags, subject, is_duplicate, duplicate_of_id, similarity_reason, status
+                        media_type, content_hash, tags, subject, is_duplicate, duplicate_of_id, similarity_reason, source_created_at, status
                     )
-                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 'pending_review')
+                    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, 'pending_review')
                     RETURNING id;
                     """,
                     source_channel_id,
@@ -260,6 +266,7 @@ class Repository:
                     is_duplicate,
                     duplicate_of_id,
                     similarity_reason,
+                    source_created_at,
                 )
                 return row["id"] if row else None
             except asyncpg.UniqueViolationError:
@@ -271,6 +278,33 @@ class Repository:
             row = await conn.fetchrow("SELECT * FROM posts WHERE id = $1;", post_id)
             return _parse_post_row(row) if row else None
 
+    async def get_recent_source_posts(
+        self,
+        source_channel_id: int,
+        limit: int = 5,
+        exclude_post_id: Optional[int] = None
+    ) -> List[Post]:
+        """Fetch the most recent posts from this source channel for narrative context."""
+        if limit <= 0:
+            return []
+        pool = await self._get_pool()
+        async with pool.acquire() as conn:
+            rows = await conn.fetch(
+                """
+                SELECT * FROM posts
+                WHERE source_channel_id = $1
+                  AND is_deleted = FALSE
+                  AND ($2::bigint IS NULL OR id <> $2)
+                ORDER BY COALESCE(source_created_at, created_at) DESC, id DESC
+                LIMIT $3;
+                """,
+                source_channel_id, exclude_post_id, limit
+            )
+            # Return in chronological order so the AI sees the natural progression (earliest to latest)
+            posts = [_parse_post_row(r) for r in rows]
+            posts.reverse()
+            return posts
+
     async def find_candidate_posts_by_tags(
         self,
         tags: List[str],

services/admin_bot.py

@@ -62,6 +62,9 @@ def get_source_fetch_buttons(channel_id: int, source_id: Optional[int] = None):
         ]
     ]
     if source_id is not None:
+        rows.append([
+            Button.inline("📜 تنظیم پیام‌های زمینه (Context)", data=f"src_ctx_menu:{source_id}"),
+        ])
         rows.append([
             Button.inline("📁 تعیین دسته‌بندی", data=f"src_cat:{source_id}"),
             Button.inline("🗑 حذف این کانال مبدا", data=f"del_src:{source_id}")
@@ -852,20 +855,56 @@ class AdminBotService:
             if auto_targets else "🛑 <b>ارسال خودکار:</b> غیرفعال"
         )
         collected = await self.repo.count_posts_from_source(source.channel_id)
+        ctx_count = getattr(source, "context_message_count", 0)
+        ctx_label = f"🟢 {ctx_count} پیام اخیر" if ctx_count > 0 else "⚪️ غیرفعال (فقط پست جاری)"
 
         card = (
             f"📢 <b>{source.title or 'کانال مبدا'}</b>\n\n"
             f"• 🆔 <b>شناسه کانال:</b> <code>{source.channel_id}</code>\n"
             f"• 🔗 <b>یوزرنیم:</b> @{source.username or 'ندارد'}\n"
             f"• 📁 <b>دسته‌بندی:</b> <b>{cat_name}</b>\n"
+            f"• 📜 <b>پیام‌های زمینه (Context):</b> <b>{ctx_label}</b>\n"
             f"• 📥 <b>پست‌های دریافت‌شده:</b> <b>{collected}</b>\n"
             f"• {auto_line}\n\n"
-            "<i>👇 برای دریافت پست‌های گذشته یکی از گزینه‌ها را انتخاب کنید:</i>"
+            "<i>👇 برای تنظیمات زمینه یا دریافت پست‌های گذشته یکی از گزینه‌ها را انتخاب کنید:</i>"
         )
         buttons = get_source_fetch_buttons(source.channel_id, source.id)
         buttons.append([Button.inline("🔙 بازگشت به لیست کانال‌های مبدا", data="list_src")])
         return card, buttons
 
+    async def _render_source_context_menu(self, source_id: int):
+        source = await self.repo.get_source_by_id(source_id)
+        if not source:
+            return "❌ کانال مبدا یافت نشد.", []
+
+        current_cnt = getattr(source, "context_message_count", 0)
+        status_text = f"<b>{current_cnt} پیام اخیر</b>" if current_cnt > 0 else "<b>غیرفعال (۰ پیام)</b>"
+
+        text = (
+            f"📜 <b>تنظیم تعداد پیام‌های زمینه (Context History)</b>\n"
+            f"📢 کانال مبدا: <b>{source.title or source.channel_id}</b>\n\n"
+            f"با فعال‌سازی این قابلیت، هنگام بازنویسی هر پست ورودی توسط هوش مصنوعی، تعداد مشخصی از پیام‌های قبلی این کانال نیز جهت درک پیوستگی موضوعی و خط داستانی در اختیار مدل قرار می‌گیرد.\n\n"
+            f"• وضعیت کنونی: {status_text}\n\n"
+            f"<i>تعداد پیام مورد نظر را انتخاب کنید:</i>"
+        )
+
+        buttons = [
+            [
+                Button.inline("0️⃣ خاموش (۰)", data=f"src_ctx_set:{source.id}:0"),
+                Button.inline("1️⃣ ۱ پیام", data=f"src_ctx_set:{source.id}:1"),
+                Button.inline("2️⃣ ۲ پیام", data=f"src_ctx_set:{source.id}:2"),
+            ],
+            [
+                Button.inline("3️⃣ ۳ پیام", data=f"src_ctx_set:{source.id}:3"),
+                Button.inline("5️⃣ ۵ پیام", data=f"src_ctx_set:{source.id}:5"),
+                Button.inline("🔟 ۱۰ پیام", data=f"src_ctx_set:{source.id}:10"),
+            ],
+            [
+                Button.inline("🔙 بازگشت به کانال مبدا", data=f"src_view:{source.id}")
+            ]
+        ]
+        return text, buttons
+
 
     async def _render_auto_sources(self, target_id: int):
         """Toggle screen listing every source with its on/off state for this target."""
@@ -1012,6 +1051,15 @@ class AdminBotService:
         if not targets:
             return 0
 
+        source = await self.repo.get_source_by_channel_id(post.source_channel_id)
+        context_posts = []
+        if source and getattr(source, "context_message_count", 0) > 0:
+            context_posts = await self.repo.get_recent_source_posts(
+                source_channel_id=post.source_channel_id,
+                limit=source.context_message_count,
+                exclude_post_id=post.id
+            )
+
         routed = 0
         ai_rejected_reasons = []
         for target in targets:
@@ -1019,7 +1067,11 @@ class AdminBotService:
                 text = post.raw_text or ""
                 if self.ai_processor:
                     rewrite_res = await self.ai_processor.rewrite_for_target(
-                        text, target, has_media=bool(post.media_path), image_path=post.media_path
+                        text,
+                        target,
+                        has_media=bool(post.media_path),
+                        image_path=post.media_path,
+                        context_posts=context_posts or None
                     )
                     if getattr(rewrite_res, "is_rejected", False):
                         reason = getattr(rewrite_res, "rejection_reason", "رد شده توسط هوش مصنوعی")
@@ -2202,6 +2254,21 @@ class AdminBotService:
             await event.edit(card, parse_mode="html", buttons=buttons or None)
             await event.answer()
 
+        elif data.startswith("src_ctx_menu:"):
+            src_id = int(data.split(":")[1])
+            text, buttons = await self._render_source_context_menu(src_id)
+            await event.edit(text, parse_mode="html", buttons=buttons)
+            await event.answer()
+
+        elif data.startswith("src_ctx_set:"):
+            _, src_id_str, cnt_str = data.split(":")
+            src_id = int(src_id_str)
+            cnt = int(cnt_str)
+            await self.repo.update_source_context_count(src_id, cnt)
+            card, buttons = await self._render_source_config(src_id)
+            await event.edit(card, parse_mode="html", buttons=buttons or None)
+            await event.answer(f"✅ تعداد پیام‌های زمینه روی {cnt} تنظیم شد.")
+
         elif data.startswith("trg_view:"):
             target_id = int(data.split(":")[1])
             card, buttons = await self._render_target_config(target_id)
@@ -2387,8 +2454,21 @@ class AdminBotService:
             except Exception:
                 pass
 
+            source = await self.repo.get_source_by_channel_id(post.source_channel_id)
+            context_posts = []
+            if source and getattr(source, "context_message_count", 0) > 0:
+                context_posts = await self.repo.get_recent_source_posts(
+                    source_channel_id=post.source_channel_id,
+                    limit=source.context_message_count,
+                    exclude_post_id=post.id
+                )
+
             rewrite_res = await self.ai_processor.rewrite_for_target(
-                post.raw_text or "", target, has_media=bool(post.media_path), image_path=post.media_path
+                post.raw_text or "",
+                target,
+                has_media=bool(post.media_path),
+                image_path=post.media_path,
+                context_posts=context_posts or None
             )
             rewritten_text = str(rewrite_res)
             cache_key = f"{post_id}:{target_id}"

services/ai_processor.py

@@ -78,11 +78,12 @@ CRITICAL RULES:
 1. Completely REMOVE all original channel usernames (e.g. @source_channel), sponsor tags, author watermarks, and source links.
 2. __LANGUAGE_INSTRUCTION__
 3. Rewrite and format the post to fully embody the target personality with appropriate emojis and clear paragraph spacing.
-4. If a custom footer/tag is provided below, append it cleanly at the very end of the post:
+4. If recent channel context posts are provided in the user prompt, use them to understand recent narrative flow, story progression, and context. Focus your rewritten output specifically on transforming the NEW incoming post.
+5. If a custom footer/tag is provided below, append it cleanly at the very end of the post:
 __CUSTOM_FOOTER__
-5. TELEGRAM CHARACTER LIMIT CONSTRAINT:
+6. TELEGRAM CHARACTER LIMIT CONSTRAINT:
 __LENGTH_LIMIT_RULE__
-6. POST EVALUATION & REJECTION CRITERIA:
+7. POST EVALUATION & REJECTION CRITERIA:
 - Default decision is "accept". You should adapt and rewrite normal posts even if their original tone or subject is diverse.
 - You should ONLY reject a post if it is:
   * Pure spam, unrelated scam/gambling/phishing ads
@@ -249,8 +250,9 @@ class AIProcessor:
         target: TargetChannel,
         has_media: bool = False,
         image_path: Optional[str] = None,
+        context_posts: Optional[List[Post]] = None,
     ) -> RewriteResult:
-        """Rewrite raw text according to target channel's language, personality, custom prompt commands, length limits, and optional image."""
+        """Rewrite raw text according to target channel's language, personality, custom prompt commands, length limits, context history, and optional image."""
         if not raw_text and not image_path:
             return RewriteResult(decision="reject", rejection_reason="متن پیام و تصویر هر دو خالی هستند", rewritten_text="")
 
@@ -269,7 +271,33 @@ class AIProcessor:
             has_media=has_media,
         )
 
-        user_prompt_text = f"متن اصلی پست برای بازنویسی و تبدیل به لحن و استایل کانال مقصد:\n\n{raw_text}" if raw_text else "لطفاً با توجه به تصویر پیوست، یک متن جذاب و مناسب برای کانال بنویسید."
+        context_block = ""
+        if context_posts:
+            context_entries = []
+            for idx, cp in enumerate(context_posts, start=1):
+                post_date = cp.source_created_at or cp.created_at or "اخیر"
+                clean_snippet = (cp.raw_text or "").strip()
+                if len(clean_snippet) > 300:
+                    clean_snippet = clean_snippet[:300] + "..."
+                context_entries.append(f"[{idx}] (تاریخ/زمان: {post_date}):\n{clean_snippet}")
+
+            context_block = (
+                "📜 پیام‌ها و پست‌های قبلی/اخیر این کانال (جهت اطلاع از خط داستانی و پیوستگی موضوع):\n"
+                + "\n\n".join(context_entries)
+                + "\n\n━━━━━━━━━━━━━━━━━━━━\n"
+            )
+
+        if raw_text:
+            user_prompt_text = (
+                f"{context_block}"
+                f"📥 پست جدید ورودی برای بازنویسی و انتشار در کانال مقصد:\n\n{raw_text}\n\n"
+                f"راهنما: با توجه به پیام‌های اخیر فوق، پست جدید را طوری بازنویسی کنید که با روند موضوعات همخوانی داشته و لحن کانال مقصد را به بهترین شکل بازتاب دهد."
+            )
+        else:
+            user_prompt_text = (
+                f"{context_block}"
+                f"📥 پست جدید ورودی شامل تصویر است. لطفاً با توجه به تصویر پیوست و زمینه پیام‌های اخیر، یک متن جذاب و متناسب برای کانال مقصد بنویسید."
+            )
 
         try:
             res = await self.llm.generate_json(

services/collector.py

@@ -256,6 +256,7 @@ class CollectorService:
                 is_duplicate=is_duplicate,
                 duplicate_of_id=duplicate_of_id,
                 similarity_reason=similarity_reason,
+                source_created_at=getattr(event.message, "date", None),
             )
 
             if post_id:
@@ -371,6 +372,7 @@ class CollectorService:
                     is_duplicate=is_duplicate,
                     duplicate_of_id=duplicate_of_id,
                     similarity_reason=similarity_reason,
+                    source_created_at=getattr(message, "date", None),
                 )
 
                 if post_id:

tests/test_context_and_source_time.py

@@ -0,0 +1,110 @@
+import asyncio
+import time
+from datetime import datetime, timezone, timedelta
+from unittest.mock import AsyncMock, patch, MagicMock
+from db.database import init_db
+from db.repository import Repository
+from db.models import TargetChannel, Post, AIProviderProfile
+from services.ai_processor import AIProcessor
+from core.llm import LLMClient
+
+
+async def test_source_created_at_and_context_count():
+    await init_db()
+    repo = Repository()
+
+    unique_channel_id = -10099887700 - int(time.time() % 100000)
+    src_id = await repo.add_source(unique_channel_id, "Context Test Source", "ctx_test")
+    assert src_id is not None
+
+    # Check default context_message_count is 0
+    src = await repo.get_source_by_id(src_id)
+    assert src.context_message_count == 0
+
+    # Update context_message_count to 5
+    await repo.update_source_context_count(src_id, 5)
+    src_updated = await repo.get_source_by_id(src_id)
+    assert src_updated.context_message_count == 5
+
+    # Insert posts with explicit source_created_at timestamps
+    t0 = datetime(2026, 8, 28, 10, 0, 0, tzinfo=timezone.utc)
+    t1 = datetime(2026, 8, 28, 11, 0, 0, tzinfo=timezone.utc)
+    t2 = datetime(2026, 8, 28, 12, 0, 0, tzinfo=timezone.utc)
+
+    p0_id = await repo.create_raw_post(
+        source_channel_id=unique_channel_id,
+        source_message_id=101,
+        raw_text="خبر اول: مذاکرات آغاز شد.",
+        source_created_at=t0
+    )
+    p1_id = await repo.create_raw_post(
+        source_channel_id=unique_channel_id,
+        source_message_id=102,
+        raw_text="خبر دوم: توافقات اولیه حاصل گردید.",
+        source_created_at=t1
+    )
+    p2_id = await repo.create_raw_post(
+        source_channel_id=unique_channel_id,
+        source_message_id=103,
+        raw_text="خبر سوم: بیانیه مشترک امضا شد.",
+        source_created_at=t2
+    )
+
+    assert p0_id is not None
+    assert p1_id is not None
+    assert p2_id is not None
+
+    # Check source_created_at persisted
+    p2_loaded = await repo.get_post_by_id(p2_id)
+    assert p2_loaded.source_created_at is not None
+
+    # Fetch recent 2 posts excluding p2_id -> should return p0 and p1 in chronological order
+    recent_posts = await repo.get_recent_source_posts(unique_channel_id, limit=2, exclude_post_id=p2_id)
+    assert len(recent_posts) == 2
+    assert recent_posts[0].id == p0_id
+    assert recent_posts[1].id == p1_id
+
+    # Clean up
+    await repo.delete_source(src_id)
+
+
+async def test_ai_processor_context_injection():
+    repo_mock = AsyncMock()
+    llm_mock = MagicMock()
+    captured_payload = {}
+
+    async def fake_generate_json(prompt, system_prompt, action_name, image_path=None):
+        captured_payload["prompt"] = prompt
+        captured_payload["system_prompt"] = system_prompt
+        return {"decision": "accept", "rewritten_text": "پست بازنویسی‌شده با در نظر گرفتن پیوستگی زمینه"}
+
+    llm_mock.generate_json = AsyncMock(side_effect=fake_generate_json)
+
+    processor = AIProcessor(repo=repo_mock, llm=llm_mock, double_check=False)
+    target = TargetChannel(id=1, channel_id=-1001234, title="Target News", username="trg_news", personality="رسمی و خبری")
+
+    ctx_p1 = Post(id=10, source_channel_id=-100, source_message_id=1, raw_text="پست زمینه ۱: مرحله اول آغاز شد.", source_created_at="2026-08-28 10:00:00")
+    ctx_p2 = Post(id=11, source_channel_id=-100, source_message_id=2, raw_text="پست زمینه ۲: مرحله دوم با موفقیت انجام شد.", source_created_at="2026-08-28 11:00:00")
+
+    result = await processor.rewrite_for_target(
+        raw_text="پست جدید ۳: نتایج نهایی اعلام شد.",
+        target=target,
+        context_posts=[ctx_p1, ctx_p2]
+    )
+
+    assert result.is_rejected is False
+    prompt_sent = captured_payload["prompt"]
+    assert "پیام‌ها و پست‌های قبلی/اخیر این کانال" in prompt_sent
+    assert "پست زمینه ۱" in prompt_sent
+    assert "پست زمینه ۲" in prompt_sent
+    assert "پست جدید ۳: نتایج نهایی اعلام شد." in prompt_sent
+
+
+async def main():
+    await test_source_created_at_and_context_count()
+    await test_ai_processor_context_injection()
+    print("All source context history and original timestamp tests passed successfully!")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())