feat(publisher): implement interval-based paced queue publisher

services/publisher.py

@@ -0,0 +1,102 @@
+import os
+import asyncio
+import logging
+from datetime import datetime, timezone
+from typing import Optional
+from telethon import TelegramClient
+from db.repository import Repository
+from core.metrics import POSTS_PUBLISHED_TOTAL, QUEUE_POSTS_GAUGE
+
+logger = logging.getLogger(__name__)
+
+class PublisherService:
+    def __init__(
+        self,
+        repo: Repository,
+        api_id: Optional[int] = None,
+        api_hash: Optional[str] = None,
+        session_name: str = "/projects/telegram-bots/copykar/sessions/publisher.session",
+        bot_token: Optional[str] = None,
+    ):
+        self.repo = repo
+        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")
+        self.session_name = session_name
+        self.client = TelegramClient(self.session_name, self.api_id, self.api_hash)
+        self._running = False
+        self._task: Optional[asyncio.Task] = None
+
+    async def start(self):
+        os.makedirs(os.path.dirname(self.session_name), exist_ok=True)
+        logger.info("Starting Publisher Service...")
+        if self.bot_token:
+            await self.client.start(bot_token=self.bot_token)
+        else:
+            await self.client.start()
+        logger.info("Publisher Service connected successfully.")
+
+        self._running = True
+        self._task = asyncio.create_task(self._publisher_loop())
+
+    async def _publisher_loop(self):
+        while self._running:
+            try:
+                await self._process_pending_queues()
+            except Exception as e:
+                logger.error(f"Error in publisher loop: {e}", exc_info=True)
+            await asyncio.sleep(15)  # Check queue every 15 seconds
+
+    async def _process_pending_queues(self):
+        targets = await self.repo.get_active_targets()
+        now = datetime.now(timezone.utc)
+
+        # Update gauge metrics
+        pending_count = len(await self.repo.get_posts_by_status("approved", limit=5000))
+        QUEUE_POSTS_GAUGE.labels(status="approved").set(pending_count)
+
+        for target in targets:
+            # Check interval cooldown
+            if target.last_post_time:
+                last_post = target.last_post_time
+                if last_post.tzinfo is None:
+                    last_post = last_post.replace(tzinfo=timezone.utc)
+                diff_minutes = (now - last_post).total_seconds() / 60.0
+                if diff_minutes < target.post_interval_min:
+                    continue  # Cooldown not elapsed yet
+
+            # Fetch next approved post for this target
+            post = await self.repo.get_next_approved_post_for_target(target.id)
+            if not post:
+                continue
+
+            try:
+                publish_text = post.ai_text or post.raw_text or ""
+                if post.media_path and os.path.exists(post.media_path):
+                    await self.client.send_file(
+                        target.channel_id,
+                        file=post.media_path,
+                        caption=publish_text,
+                        parse_mode="markdown"
+                    )
+                else:
+                    await self.client.send_message(
+                        target.channel_id,
+                        publish_text,
+                        parse_mode="markdown"
+                    )
+
+                await self.repo.mark_post_published(post.id)
+                await self.repo.update_target_last_post(target.id)
+                POSTS_PUBLISHED_TOTAL.labels(target_channel_id=str(target.channel_id)).inc()
+                logger.info(f"Successfully published post {post.id} to target channel {target.title} ({target.channel_id})")
+            except Exception as e:
+                logger.error(f"Failed to publish post {post.id} to target {target.channel_id}: {e}", exc_info=True)
+
+    async def stop(self):
+        self._running = False
+        if self._task:
+            self._task.cancel()
+        if self.client.is_connected():
+            await self.client.disconnect()
+            logger.info("Publisher Service disconnected.")