services: update ingestion collector, paced publisher, and deduplication

services/publisher.py

@@ -1,13 +1,20 @@
 import os
 import asyncio
 import logging
-from datetime import datetime, timezone
-from typing import Optional, List
+from datetime import datetime, timedelta, timezone
+from typing import Optional, List, Awaitable, Callable, Set, Tuple
 from telethon import TelegramClient
+from telethon.errors import (
+    ChannelPrivateError,
+    ChatAdminRequiredError,
+    ChatWriteForbiddenError,
+    PeerIdInvalidError,
+    UserBannedInChannelError,
+)
 from db.models import TargetChannel
 from db.repository import Repository
 from core.queue import RedisQueue
-from core.metrics import TARGET_ACTIVITY_TOTAL, QUEUE_POSTS_GAUGE, REDIS_QUEUE_SIZE_GAUGE
+from core.metrics import TARGET_ACTIVITY_TOTAL, QUEUE_POSTS_GAUGE
 from core.proxy import get_telegram_proxy
 from core.error_logger import log_exception
 
@@ -15,6 +22,19 @@ logger = logging.getLogger(__name__)
 
 SESSION_DIR = os.getenv("SESSION_DIR", "/app/sessions" if os.path.exists("/app") else "/projects/telegram-bots/copykar/sessions")
 
+# Sleep windows are configured in the operator's wall-clock time, not UTC.
+TIMEZONE_OFFSET_HOURS = float(os.getenv("TIMEZONE_OFFSET_HOURS", "3.5"))
+
+# Failures that will never resolve by retrying: the account simply cannot post there.
+# Re-queueing these would spin the same post through the loop forever.
+PERMANENT_DELIVERY_ERRORS = (
+    ChatAdminRequiredError,
+    ChatWriteForbiddenError,
+    ChannelPrivateError,
+    UserBannedInChannelError,
+    PeerIdInvalidError,
+)
+
 class PublisherService:
     def __init__(
         self,
@@ -25,10 +45,15 @@ class PublisherService:
         api_hash: Optional[str] = None,
         bot_token: Optional[str] = None,
         session_name: Optional[str] = None,
+        notify_fn: Optional[Callable[[str], Awaitable[None]]] = None,
     ):
         self.repo = repo
         self.queue = queue
         self.client = client
+        self.notify_fn = notify_fn
+        # (target_id, error type) pairs already reported, so a broken target is
+        # announced once instead of every polling cycle.
+        self._reported_failures: Set[Tuple[int, str]] = set()
         self.api_id = api_id or int(os.getenv("API_ID", "0"))
         self.api_hash = api_hash or os.getenv("API_HASH", "")
         self.bot_token = bot_token or os.getenv("BOT_TOKEN")
@@ -42,11 +67,13 @@ class PublisherService:
     async def start(self):
         logger.info("Starting Paced Target Publisher Service...")
         if not self.client.is_connected():
+            # Only owns the connection when it built its own client; a shared client
+            # (the admin bot's) is already connected by its owner.
             if self.bot_token:
                 await self.client.start(bot_token=self.bot_token)
             else:
                 await self.client.start()
-        logger.info("Paced Target Publisher Service connected.")
+        logger.info("Paced Target Publisher Service connected (delivering as the bot account).")
         self._running = True
         self._task = asyncio.create_task(self._publisher_loop())
 
@@ -66,7 +93,10 @@ class PublisherService:
     async def _publisher_loop(self):
         while self._running:
             try:
-                await self._process_all_target_queues()
+                if await self.repo.is_system_paused():
+                    logger.debug("[publisher] System is paused by admin. Skipping queue processing.")
+                else:
+                    await self._process_all_target_queues()
             except Exception as e:
                 logger.error(f"Error in target publisher loop: {e}", exc_info=True)
             await asyncio.sleep(15)
@@ -74,13 +104,11 @@ class PublisherService:
     async def _process_all_target_queues(self):
         targets = await self.repo.get_active_targets()
         now = datetime.now(timezone.utc)
-        current_hour_local = (now.hour + 3) % 24  # UTC+3:30 approx hour
-
-        total_queued = 0
+        local_now = now + timedelta(hours=TIMEZONE_OFFSET_HOURS)
+        current_hour_local = local_now.hour
 
         for target in targets:
             qsize = await self.queue.get_target_queue_size(target.id)
-            total_queued += qsize
             QUEUE_POSTS_GAUGE.labels(status=f"target_{target.id}").set(qsize)
 
             if qsize == 0:
@@ -130,9 +158,52 @@ class PublisherService:
                 TARGET_ACTIVITY_TOTAL.labels(channel_id=str(target.channel_id), title=target.title or '').inc()
                 logger.info(f"Published post ID {post_id} to Target {target.title} ({target.channel_id})")
             except Exception as e:
-                await log_exception("publisher.publish", e, {"post_id": post_id, "target_id": target.id, "channel_id": target.channel_id})
+                permanent = isinstance(e, PERMANENT_DELIVERY_ERRORS)
+                if permanent:
+                    # Dropping the payload is deliberate: the post stays 'pending_review'
+                    # so an admin can re-send it once the permission problem is resolved.
+                    await self._report_broken_target(target, e)
+                else:
+                    # The payload was already popped; putting it back keeps the post from
+                    # being silently lost on a transient Telegram failure.
+                    try:
+                        await self.queue.push_target_post(target.id, payload)
+                    except Exception as requeue_err:
+                        logger.critical(f"Failed to requeue post {post_id} for target {target.id}: {requeue_err}")
 
-        REDIS_QUEUE_SIZE_GAUGE.set(total_queued)
+                await log_exception("publisher.publish", e, {
+                    "post_id": post_id,
+                    "target_id": target.id,
+                    "channel_id": target.channel_id,
+                    "permanent": permanent,
+                })
+
+    async def _report_broken_target(self, target: TargetChannel, error: Exception) -> None:
+        """Tell the admins once that a target channel is unreachable for this account."""
+        key = (target.id, type(error).__name__)
+        if key in self._reported_failures:
+            return
+        self._reported_failures.add(key)
+
+        logger.error(
+            f"Target #{target.id} ({target.title}) rejected delivery permanently: "
+            f"{type(error).__name__}. Queue drained for this target until it is fixed."
+        )
+        if not self.notify_fn:
+            return
+
+        try:
+            await self.notify_fn(
+                f"⛔ <b>ارسال به کانال مقصد «{target.title}» ممکن نیست!</b>\n\n"
+                f"• 🆔 شناسه کانال: <code>{target.channel_id}</code>\n"
+                f"• ❗️ خطا: <code>{type(error).__name__}</code>\n\n"
+                "<b>ربات</b> در این کانال عضو یا ادمین با دسترسی ارسال پیام نیست.\n"
+                "لطفا ربات را در کانال <b>ادمین</b> کنید و دسترسی <b>ارسال پیام (Post Messages)</b> بدهید، "
+                "سپس پست را دوباره به صف بفرستید.\n\n"
+                "<i>تا رفع این مشکل، پست‌های این کانال ارسال نمی‌شوند و در وضعیت بررسی باقی می‌مانند.</i>"
+            )
+        except Exception as notify_err:
+            logger.error(f"Could not notify admins about broken target {target.id}: {notify_err}")
 
     async def stop(self):
         self._running = False