From 76a19131a684fa5f10006ab00a7ff0300f1190c1 Mon Sep 17 00:00:00 2001 From: mamad Date: Thu, 27 Aug 2026 20:42:23 +0330 Subject: [PATCH] fix(services): ensure session directory exists and handle container paths --- services/admin_bot.py | 9 +++++---- services/collector.py | 23 ++++++++++++----------- services/publisher.py | 15 +++++++-------- 3 files changed, 24 insertions(+), 23 deletions(-) diff --git a/services/admin_bot.py b/services/admin_bot.py index cac67f4..cc7f2c7 100644 --- a/services/admin_bot.py +++ b/services/admin_bot.py @@ -9,6 +9,8 @@ from core.metrics import ADMIN_ACTIONS_TOTAL logger = logging.getLogger(__name__) +SESSION_DIR = os.getenv("SESSION_DIR", "/app/sessions" if os.path.exists("/app") else "/projects/telegram-bots/copykar/sessions") + class AdminBotService: def __init__( self, @@ -18,7 +20,7 @@ class AdminBotService: api_hash: Optional[str] = None, review_channel_id: Optional[int] = None, admin_user_ids: Optional[List[int]] = None, - session_name: str = "/projects/telegram-bots/copykar/sessions/admin_bot.session", + session_name: Optional[str] = None, ): self.repo = repo self.bot_token = bot_token or os.getenv("BOT_TOKEN", "") @@ -27,18 +29,17 @@ class AdminBotService: self.review_channel_id = review_channel_id or int(os.getenv("REVIEW_CHANNEL_ID", "0")) raw_admins = os.getenv("ADMIN_USER_IDS", "") self.admin_user_ids = admin_user_ids or [int(x.strip()) for x in raw_admins.split(",") if x.strip()] - self.session_name = session_name + self.session_name = session_name or os.path.join(SESSION_DIR, "admin_bot.session") + os.makedirs(os.path.dirname(self.session_name), exist_ok=True) self.client = TelegramClient(self.session_name, self.api_id, self.api_hash) def is_admin(self, user_id: int) -> bool: return not self.admin_user_ids or user_id in self.admin_user_ids async def start(self): - os.makedirs(os.path.dirname(self.session_name), exist_ok=True) logger.info("Starting Admin Review Bot...") await self.client.start(bot_token=self.bot_token) logger.info("Admin Review Bot connected successfully.") - self._register_handlers() def _register_handlers(self): diff --git a/services/collector.py b/services/collector.py index ec5c4d8..098728d 100644 --- a/services/collector.py +++ b/services/collector.py @@ -10,7 +10,8 @@ from core.metrics import COLLECTED_POSTS_TOTAL logger = logging.getLogger(__name__) -MEDIA_DIR = os.getenv("MEDIA_DIR", "/projects/telegram-bots/copykar/data/media") +SESSION_DIR = os.getenv("SESSION_DIR", "/app/sessions" if os.path.exists("/app") else "/projects/telegram-bots/copykar/sessions") +MEDIA_DIR = os.getenv("MEDIA_DIR", "/app/data/media" if os.path.exists("/app") else "/projects/telegram-bots/copykar/data/media") class CollectorService: def __init__( @@ -19,21 +20,25 @@ class CollectorService: ai_processor: AIProcessor, api_id: Optional[int] = None, api_hash: Optional[str] = None, - session_name: str = "/projects/telegram-bots/copykar/sessions/collector.session", + phone: Optional[str] = None, + session_name: Optional[str] = None, ): self.repo = repo self.ai_processor = ai_processor self.api_id = api_id or int(os.getenv("API_ID", "0")) self.api_hash = api_hash or os.getenv("API_HASH", "") - self.session_name = session_name + self.phone = phone or os.getenv("PHONE") + self.session_name = session_name or os.path.join(SESSION_DIR, "collector.session") + os.makedirs(os.path.dirname(self.session_name), exist_ok=True) + os.makedirs(MEDIA_DIR, exist_ok=True) self.client = TelegramClient(self.session_name, self.api_id, self.api_hash) async def start(self): - os.makedirs(os.path.dirname(self.session_name), exist_ok=True) - os.makedirs(MEDIA_DIR, exist_ok=True) - logger.info("Starting Collector Userbot...") - await self.client.start() + if self.phone: + await self.client.start(phone=self.phone) + else: + await self.client.start() logger.info("Collector Userbot connected successfully.") @self.client.on(events.NewMessage) @@ -42,7 +47,6 @@ class CollectorService: async def _handle_message(self, event: events.NewMessage.Event): try: - # Check if source channel is in our monitored sources chat_id = event.chat_id source = await self.repo.get_source_by_channel_id(chat_id) if not source or not source.is_active: @@ -53,7 +57,6 @@ class CollectorService: media_type = None media_hash = None - # Download media if present if event.message.media: if isinstance(event.message.media, MessageMediaPhoto): media_type = "photo" @@ -71,7 +74,6 @@ class CollectorService: content_hash = compute_content_hash(raw_text, media_hash) - # Store raw post in DB post_id = await self.repo.create_raw_post( source_channel_id=chat_id, source_message_id=event.message.id, @@ -84,7 +86,6 @@ class CollectorService: if post_id: COLLECTED_POSTS_TOTAL.labels(source_channel_id=str(chat_id)).inc() logger.info(f"Collected new post ID {post_id} from channel {chat_id}") - # Trigger AI Processor pipeline await self.ai_processor.process_post(post_id) except Exception as e: logger.error(f"Error handling message from {event.chat_id}: {e}", exc_info=True) diff --git a/services/publisher.py b/services/publisher.py index d1353d7..59255dc 100644 --- a/services/publisher.py +++ b/services/publisher.py @@ -9,26 +9,28 @@ from core.metrics import POSTS_PUBLISHED_TOTAL, QUEUE_POSTS_GAUGE logger = logging.getLogger(__name__) +SESSION_DIR = os.getenv("SESSION_DIR", "/app/sessions" if os.path.exists("/app") else "/projects/telegram-bots/copykar/sessions") + 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", + session_name: Optional[str] = None, 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.session_name = session_name or os.path.join(SESSION_DIR, "publisher.session") + os.makedirs(os.path.dirname(self.session_name), exist_ok=True) 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) @@ -45,27 +47,24 @@ class PublisherService: 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 + await asyncio.sleep(15) 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 + continue - # Fetch next approved post for this target post = await self.repo.get_next_approved_post_for_target(target.id) if not post: continue