fix(services): ensure session directory exists and handle container paths

2026-08-27 20:42:23 +03:30
parent 4e807bbb7e
commit 76a19131a6
3 changed files with 24 additions and 23 deletions
@@ -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):
@@ -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)
@@ -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