feat(observability): implement global database error logging, Prometheus error metrics, and max 3-panel Grafana dashboard

2026-08-27 22:42:19 +03:30
parent f7c80b68d2
commit 8eb5ca0a3f
8 changed files with 268 additions and 123 deletions
@@ -10,6 +10,7 @@ from db.repository import Repository
 from core.queue import RedisQueue
 from core.metrics import ADMIN_ACTIONS_TOTAL, TARGET_ACTIVITY_TOTAL
 from core.proxy import get_telegram_proxy
+from core.error_logger import log_exception

 logger = logging.getLogger(__name__)

@@ -5,6 +5,7 @@ from db.models import Post, TargetChannel
 from db.repository import Repository
 from core.llm import LLMClient
 from core.metrics import DUPLICATES_DETECTED_TOTAL
+from core.error_logger import log_exception

 logger = logging.getLogger(__name__)

@@ -57,7 +58,7 @@ class AIProcessor:
            if rewritten:
                return rewritten.strip()
        except Exception as e:
-            logger.error(f"Failed to rewrite post for target {target.id} ({target.title}): {e}")
+            await log_exception("ai_processor.rewrite", e, {"target_id": target.id, "target_title": target.title})

        # Fallback if AI fails: clean basic @mentions and append footer
        fallback = raw_text
@@ -9,6 +9,7 @@ from core.dedup import compute_content_hash, compute_file_hash
 from core.queue import RedisQueue
 from core.metrics import COLLECTED_POSTS_TOTAL, SOURCE_ACTIVITY_TOTAL
 from core.proxy import get_telegram_proxy
+from core.error_logger import log_exception

 logger = logging.getLogger(__name__)

@@ -41,29 +42,33 @@ class CollectorService:

    async def start(self, notify_fn: Optional[Callable[[str], Awaitable[None]]] = None):
        logger.info("Initializing Collector Userbot client...")
-        await self.client.connect()
+        try:
+            await self.client.connect()

-        if await self.client.is_user_authorized():
-            me = await self.client.get_me()
-            logger.info(f"Collector Userbot is authorized as: {me.first_name} (@{me.username})")
-            self._register_handlers()
-            return True
+            if await self.client.is_user_authorized():
+                me = await self.client.get_me()
+                logger.info(f"Collector Userbot is authorized as: {me.first_name} (@{me.username})")
+                self._register_handlers()
+                return True

-        logger.warning("Collector Userbot is not authorized. Requesting login code...")
-        if self.phone and notify_fn:
-            try:
-                sent = await self.client.send_code_request(self.phone)
-                self.phone_code_hash = sent.phone_code_hash
-                await notify_fn(
-                    f"🔐 <b>نیاز به ورود ربات جمع‌آوری‌کننده</b>\n\n"
-                    f"کد تایید تلگرام به شماره <code>{self.phone}</code> ارسال شد.\n\n"
-                    f"لطفا با دستور زیر پاسخ دهید:\n"
-                    f"<code>/code 12345</code>"
-                )
-            except Exception as e:
-                logger.error(f"Failed to send login code request: {e}")
-                await notify_fn(f"❌ خطا در ارسال کد ورود: {e}")
-        return False
+            logger.warning("Collector Userbot is not authorized. Requesting login code...")
+            if self.phone and notify_fn:
+                try:
+                    sent = await self.client.send_code_request(self.phone)
+                    self.phone_code_hash = sent.phone_code_hash
+                    await notify_fn(
+                        f"🔐 <b>نیاز به ورود ربات جمع‌آوری‌کننده</b>\n\n"
+                        f"کد تایید تلگرام به شماره <code>{self.phone}</code> ارسال شد.\n\n"
+                        f"لطفا با دستور زیر پاسخ دهید:\n"
+                        f"<code>/code 12345</code>"
+                    )
+                except Exception as e:
+                    await log_exception("collector.login", e, {"phone": self.phone})
+                    await notify_fn(f"❌ خطا در ارسال کد ورود: {e}")
+            return False
+        except Exception as e:
+            await log_exception("collector.start", e)
+            return False

    async def submit_code(self, code: str) -> str:
        if not self.phone or not self.phone_code_hash:
@@ -78,6 +83,7 @@ class CollectorService:
        except SessionPasswordNeededError:
            return "🔐 <b>رمز دو مرحله‌ای فعال است.</b> لطفا با این دستور رمز را وارد کنید: <code>/password رمز_عبور</code>"
        except Exception as e:
+            await log_exception("collector.submit_code", e)
            return f"❌ خطا در ورود: {e}"

    async def submit_password(self, password: str) -> str:
@@ -87,6 +93,7 @@ class CollectorService:
            self._register_handlers()
            return f"✅ تایید دو مرحله‌ای موفق بود! حساب فعال: <b>{me.first_name}</b>."
        except Exception as e:
+            await log_exception("collector.submit_password", e)
            return f"❌ خطا در تایید رمز دو مرحله‌ای: {e}"

    def _register_handlers(self):
@@ -100,6 +107,33 @@ class CollectorService:
        self._handlers_registered = True
        logger.info("Collector real-time event handlers registered.")

+    async def _resolve_channel_entity(self, channel_id: int, username: Optional[str] = None):
+        """Robustly resolve channel entity even if not yet cached in local Telethon session."""
+        if username:
+            try:
+                clean_user = username.replace("@", "").strip()
+                return await self.client.get_entity(clean_user)
+            except Exception:
+                pass
+
+        try:
+            return await self.client.get_entity(channel_id)
+        except Exception:
+            pass
+
+        try:
+            # Refresh dialogs cache
+            await self.client.get_dialogs(limit=50)
+            return await self.client.get_entity(channel_id)
+        except Exception:
+            pass
+
+        try:
+            raw_str = str(channel_id).replace("-100", "").replace("-", "")
+            return await self.client.get_entity(int(raw_str))
+        except Exception as e:
+            raise ValueError(f"Could not resolve entity for channel {channel_id} (@{username}): {e}")
+
    async def _handle_message(self, event: events.NewMessage.Event):
        try:
            chat_id = event.chat_id
@@ -146,7 +180,7 @@ class CollectorService:
                if self.on_post_received:
                    await self.on_post_received(post_id)
        except Exception as e:
-            logger.error(f"Error handling message from {event.chat_id}: {e}", exc_info=True)
+            await log_exception("collector.handle_message", e, {"chat_id": event.chat_id, "msg_id": getattr(event.message, "id", None)})

    async def scrape_channel_history(
        self,
@@ -162,11 +196,12 @@ class CollectorService:

        source = await self.repo.get_source_by_channel_id(channel_id)
        source_title = source.title if source else str(channel_id)
+        username = source.username if source else None
        collected_count = 0
        skipped_count = 0

        try:
-            entity = await self.client.get_input_entity(channel_id)
+            entity = await self._resolve_channel_entity(channel_id, username)
            messages = []
            async for msg in self.client.iter_messages(entity, limit=limit):
                messages.append(msg)
@@ -225,7 +260,7 @@ class CollectorService:
                )
            return collected_count
        except Exception as e:
-            logger.error(f"Error scraping history from {channel_id}: {e}", exc_info=True)
+            await log_exception("collector.scrape_history", e, {"channel_id": channel_id, "limit": limit})
            if progress_callback:
                await progress_callback(f"❌ خطا در دریافت پست‌های کانال <code>{channel_id}</code>: {e}")
            return collected_count
@@ -9,6 +9,7 @@ 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.proxy import get_telegram_proxy
+from core.error_logger import log_exception

 logger = logging.getLogger(__name__)

@@ -129,7 +130,7 @@ 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:
-                logger.error(f"Failed to publish queued post {post_id} to target {target.channel_id}: {e}", exc_info=True)
+                await log_exception("publisher.publish", e, {"post_id": post_id, "target_id": target.id, "channel_id": target.channel_id})

        REDIS_QUEUE_SIZE_GAUGE.set(total_queued)