diff --git a/services/publisher.py b/services/publisher.py new file mode 100644 index 0000000..d1353d7 --- /dev/null +++ b/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.")