""" main.py ─────── Telegram Channel Monitor Bot — production entry-point. Architecture overview ───────────────────── ┌──────────────────────────────────────────────────────────────┐ │ TelegramMonitor │ │ ┌───────────────┐ new_message ┌─────────────────────┐ │ │ │ Telethon │ ──────────────► │ MessageHandler │ │ │ │ UserClient │ │ (per source chan.) │ │ │ └───────────────┘ └──────┬──────────────┘ │ │ │ │ │ ┌──────▼──────────────┐ │ │ │ MediaDownloader │ │ │ │ (temp_dir) │ │ │ └──────┬──────────────┘ │ │ │ │ │ ┌──────▼──────────────┐ │ │ │ ForwardDispatcher │ │ │ │ (destination chan.) │ │ │ └──────┬──────────────┘ │ │ │ │ │ ┌──────▼──────────────┐ │ │ │ ProcessedMsgDB │ │ │ │ (SQLite) │ │ │ └─────────────────────┘ │ └──────────────────────────────────────────────────────────────┘ Extension points ──────────────── Each public method in :class:`ForwardDispatcher` is a clean hook for future features such as AI translation, Twitter/X formatting, or content filters. """ from __future__ import annotations import asyncio import logging import os import sqlite3 import time from pathlib import Path from typing import Dict, List, Optional, Tuple try: from aiohttp import web as aiohttp_web _AIOHTTP_AVAILABLE = True except ImportError: _AIOHTTP_AVAILABLE = False from telethon import TelegramClient, events from telethon.errors import ( ChannelPrivateError, ChatAdminRequiredError, FloodWaitError, RPCError, ) from telethon.tl.types import ( Document, DocumentAttributeAnimated, DocumentAttributeAudio, DocumentAttributeSticker, DocumentAttributeVideo, MessageEntityBold, MessageMediaDocument, MessageMediaGeo, MessageMediaPhoto, MessageMediaPoll, Photo, ) from config import Config, configure_logging, get_config # Optional AI layer (imported lazily to allow running without google-genai installed). try: from ai_processor import AIProcessor _AI_AVAILABLE = True except ImportError: _AI_AVAILABLE = False # Optional Hugging Face persistent store (replaces SQLite on ephemeral hosts). try: from hf_store import HFProcessedMsgStore _HF_AVAILABLE = True except ImportError: _HF_AVAILABLE = False # ───────────────────────────────────────────────────────────────────────────── # Module-level logger (reconfigured in main() once we have the log level) # ───────────────────────────────────────────────────────────────────────────── log = logging.getLogger("telegram_monitor") # ═════════════════════════════════════════════════════════════════════════════ # Duplicate-detection database # ═════════════════════════════════════════════════════════════════════════════ class ProcessedMsgDB: """Lightweight SQLite store that persists processed (channel_id, msg_id) pairs across restarts to avoid duplicate forwarding. Thread-safety: SQLite connections must stay on the thread/loop that created them. We use a single connection owned by the async event loop. """ def __init__(self, db_path: Path) -> None: self._path = db_path self._conn: Optional[sqlite3.Connection] = None # ── Lifecycle ───────────────────────────────────────────────────────────── def open(self) -> None: self._conn = sqlite3.connect(self._path, check_same_thread=False) self._conn.execute( """ CREATE TABLE IF NOT EXISTS processed_messages ( channel_id TEXT NOT NULL, message_id INTEGER NOT NULL, forwarded_at INTEGER NOT NULL, PRIMARY KEY (channel_id, message_id) ) """ ) self._conn.execute( """ CREATE INDEX IF NOT EXISTS idx_forwarded_at ON processed_messages (forwarded_at) """ ) self._conn.commit() log.debug("ProcessedMsgDB opened: %s", self._path) def close(self) -> None: if self._conn: self._conn.close() self._conn = None # ── Public API ──────────────────────────────────────────────────────────── def is_processed(self, channel_id: str, message_id: int) -> bool: """Return True if this (channel, message) pair was already forwarded.""" row = self._conn.execute( "SELECT 1 FROM processed_messages WHERE channel_id=? AND message_id=?", (channel_id, message_id), ).fetchone() return row is not None def mark_processed(self, channel_id: str, message_id: int) -> None: """Persist a forwarded message so it is never re-sent after a restart.""" try: self._conn.execute( """ INSERT OR IGNORE INTO processed_messages (channel_id, message_id, forwarded_at) VALUES (?, ?, ?) """, (channel_id, message_id, int(time.time())), ) self._conn.commit() except sqlite3.Error as exc: log.error("DB write failed for (%s, %d): %s", channel_id, message_id, exc) def prune(self, keep_days: int = 30) -> int: """Delete records older than *keep_days* days. Returns number of rows removed.""" cutoff = int(time.time()) - keep_days * 86_400 cur = self._conn.execute( "DELETE FROM processed_messages WHERE forwarded_at < ?", (cutoff,) ) self._conn.commit() return cur.rowcount # ═════════════════════════════════════════════════════════════════════════════ # Media helpers # ═════════════════════════════════════════════════════════════════════════════ def _detect_media_type(message) -> str: """Return a human-readable string describing the media type of *message*.""" media = getattr(message, "media", None) if media is None: return "text" if isinstance(media, MessageMediaPhoto): return "photo" if isinstance(media, MessageMediaDocument): doc: Document = media.document if doc and doc.attributes: for attr in doc.attributes: if isinstance(attr, DocumentAttributeSticker): return "sticker" if isinstance(attr, DocumentAttributeAnimated): return "gif" if isinstance(attr, DocumentAttributeAudio): return "voice" if getattr(attr, "voice", False) else "audio" if isinstance(attr, DocumentAttributeVideo): return ( "round_video" if getattr(attr, "round_message", False) else "video" ) return "document" if isinstance(media, MessageMediaGeo): return "location" if isinstance(media, MessageMediaPoll): return "poll" return "other_media" async def _download_media_to_temp( client: TelegramClient, message, temp_dir: Path, ) -> Optional[Path]: """Download *message* media into *temp_dir* and return the local path. Returns ``None`` if the message has no downloadable media or if the download fails. """ media = getattr(message, "media", None) if media is None: return None try: dest = await client.download_media(message, file=str(temp_dir) + "/") if dest: return Path(dest) except Exception as exc: log.warning("Media download failed (msg_id=%d): %s", message.id, exc) return None # ═════════════════════════════════════════════════════════════════════════════ # Forward dispatcher # ═════════════════════════════════════════════════════════════════════════════ class ForwardDispatcher: """Responsible for building the forwarded message and sending it to the destination channel. Extension hooks ─────────────── • ``pre_process(message, channel_name)`` — override for AI translation, content filtering, or metadata enrichment. • ``format_caption(original_caption, channel_name)`` — override to change how the source attribution is prepended. • ``post_send(message_id, channel_name)`` — override for analytics / downstream posting (Twitter/X, etc.). """ def __init__( self, client: TelegramClient, destination: str, temp_dir: Path, db: ProcessedMsgDB, ) -> None: self._client = client self._destination = destination self._temp_dir = temp_dir self._db = db # Cache resolved destination entity to avoid repeated API calls. self._dest_entity = None # ── Initialisation ──────────────────────────────────────────────────────── async def resolve_destination(self) -> None: """Resolve the destination channel entity once at startup.""" try: self._dest_entity = await self._client.get_entity(self._destination) log.info( "Destination channel resolved: %s (id=%s)", getattr(self._dest_entity, "title", self._destination), getattr(self._dest_entity, "id", "?"), ) except Exception as exc: log.critical( "Cannot resolve destination channel %r: %s", self._destination, exc ) raise # ── Public dispatch entry-point ─────────────────────────────────────────── async def dispatch(self, message, channel_name: str) -> bool: """Forward *message* from *channel_name* to the destination. Returns True on success, False on any non-fatal failure. """ media_type = _detect_media_type(message) log.info( "↳ Dispatching | channel=%-30s | type=%-12s | msg_id=%d | date=%s", channel_name, media_type, message.id, message.date.strftime("%Y-%m-%d %H:%M:%S UTC") if message.date else "?", ) # ── Extension hook: pre-processing ──────────────────────────────────── await self.pre_process(message, channel_name) caption = self.format_caption( getattr(message, "text", "") or getattr(message, "message", "") or "", channel_name, ) success = False local_file: Optional[Path] = None try: if media_type == "text": success = await self._send_text(caption) elif media_type in {"photo", "gif", "video", "audio", "document", "sticker", "voice", "round_video"}: local_file = await _download_media_to_temp( self._client, message, self._temp_dir ) if local_file: success = await self._send_file(local_file, caption) else: # Fallback: try native Telethon forward (no caption modification). success = await self._forward_native(message, channel_name) else: # Polls, locations, and unknown media — forward natively. success = await self._forward_native(message, channel_name) except FloodWaitError as exc: wait = exc.seconds log.warning("FloodWaitError: sleeping %d s before retry…", wait) await asyncio.sleep(wait) # Retry once after flood wait. success = await self.dispatch(message, channel_name) except (ChannelPrivateError, ChatAdminRequiredError) as exc: log.error("Permission error forwarding to destination: %s", exc) success = False except RPCError as exc: log.error("Telegram RPC error (msg_id=%d): %s", message.id, exc) success = False except Exception as exc: log.exception("Unexpected error dispatching msg_id=%d: %s", message.id, exc) success = False finally: if local_file and local_file.exists(): try: local_file.unlink() log.debug("Temp file removed: %s", local_file) except OSError: pass status_icon = "✓" if success else "✗" log.info( "%s %s | channel=%-30s | msg_id=%d", status_icon, "Forwarded" if success else "FAILED ", channel_name, message.id, ) # ── Extension hook: post-send ───────────────────────────────────────── if success: await self.post_send(message.id, channel_name) return success # ── Sending primitives ──────────────────────────────────────────────────── async def _send_text(self, text: str) -> bool: if not text.strip(): return True # nothing to send await self._client.send_message(self._dest_entity, text) return True async def _send_file(self, file_path: Path, caption: str) -> bool: await self._client.send_file( self._dest_entity, str(file_path), caption=caption or None, # Preserve video as video, not document. supports_streaming=True, ) return True async def _forward_native(self, message, channel_name: str) -> bool: """Use Telethon's built-in forward_messages as a last-resort fallback. Native forwarding preserves all metadata but loses caption prefix. """ log.debug( "Using native forward for msg_id=%d from %s", message.id, channel_name ) await self._client.forward_messages( self._dest_entity, messages=message, ) return True # ── Extension hooks (override in subclasses) ────────────────────────────── async def pre_process(self, message, channel_name: str) -> None: """Hook called before any processing. Override for AI translation, etc.""" pass # noqa: WPS420 def format_caption(self, original_caption: str, channel_name: str) -> str: """Build the final caption string. Override this to change how source attribution is formatted, e.g. for Twitter/X character limits or custom markdown templates. """ header = f"📡 **{channel_name}**" if original_caption: return f"{header}\n\n{original_caption}" return header async def post_send(self, message_id: int, channel_name: str) -> None: """Hook called after a successful send. Override for analytics, etc.""" pass # noqa: WPS420 # ════════════════════════════════════════════════════════════════════════════════ # AI-powered dispatcher (Gemini Flash) # ════════════════════════════════════════════════════════════════════════════════ class AIForwardDispatcher(ForwardDispatcher): """ForwardDispatcher subclass that adds Gemini AI capabilities: 1. **Semantic deduplication** — if Gemini reports the incoming news is the same story as a recent one, ``dispatch`` returns True (success) without forwarding anything, effectively silencing the duplicate. 2. **Twitter/X translation** — if the news is unique, Gemini rewrites it as a punchy English tweet (≤ 240 chars + hashtags + emoji). Both operations happen in a **single Gemini API call** per message so we stay well within the free-tier limit (1 500 RPD). """ def __init__( self, client: TelegramClient, destination: str, temp_dir: Path, db: ProcessedMsgDB, ai: "AIProcessor", ) -> None: super().__init__(client, destination, temp_dir, db) self._ai = ai # Stores the tweet text produced by Gemini for the current message. # Set by dispatch() before format_caption() is called. self._pending_tweet: Optional[str] = None # ── Override dispatch to gate on AI deduplication ──────────────────── async def dispatch(self, message, channel_name: str) -> bool: text = ( getattr(message, "text", "") or getattr(message, "message", "") or "" ).strip() if text: is_dup, tweet_text = await self._ai.process(text) if is_dup: log.info( "❌ Duplicate suppressed | channel=%-30s | msg_id=%d", channel_name, message.id, ) # Return True so the caller marks it as processed and doesn't # retry, but we deliberately do NOT forward anything. return True self._pending_tweet = tweet_text else: # Media-only message — no text to analyse, pass through as-is. self._pending_tweet = None return await super().dispatch(message, channel_name) # ── Override format_caption to inject the AI-generated tweet ────────── def format_caption(self, original_caption: str, channel_name: str) -> str: header = f"📡 **{channel_name}**" tweet = self._pending_tweet if tweet: # Primary: AI-translated English tweet return f"{header}\n\n{tweet}" if original_caption: # Fallback: original text (AI unavailable or media-only) return f"{header}\n\n{original_caption}" return header # ═════════════════════════════════════════════════════════════════════════════ # Album (media group) collector # ═════════════════════════════════════════════════════════════════════════════ class AlbumCollector: """Groups messages that belong to the same media-group (album) and fires a callback only after the last message of the group has been received. Telethon delivers album messages individually in rapid succession. We wait a short window (``wait_seconds``) after the first message in a group before processing the whole album together. """ def __init__( self, client: TelegramClient, dispatcher: ForwardDispatcher, db: ProcessedMsgDB, wait_seconds: float = 1.5, ) -> None: self._client = client self._dispatcher = dispatcher self._db = db self._wait = wait_seconds # grouped_id → (channel_name, [messages], asyncio.Task) self._groups: Dict[int, Tuple[str, List, asyncio.Task]] = {} async def handle(self, message, channel_name: str) -> None: """Accept *message* and schedule group processing.""" gid = getattr(message, "grouped_id", None) if gid is None: # Not part of an album — process immediately. await self._process_single(message, channel_name) return if gid in self._groups: # Add to existing group and reset the timer. _cname, msgs, task = self._groups[gid] msgs.append(message) task.cancel() else: msgs = [message] task = asyncio.ensure_future( self._delayed_album_flush(gid, channel_name, msgs) ) self._groups[gid] = (channel_name, msgs, task) async def _delayed_album_flush( self, grouped_id: int, channel_name: str, msgs: List ) -> None: """Wait for the collection window, then process all messages.""" try: await asyncio.sleep(self._wait) except asyncio.CancelledError: return # A new message restarted the timer. self._groups.pop(grouped_id, None) # Sort by message ID to preserve original order. msgs.sort(key=lambda m: m.id) log.info( "Album detected | channel=%-30s | group_id=%d | parts=%d", channel_name, grouped_id, len(msgs), ) await self._process_album(msgs, channel_name) async def _process_single(self, message, channel_name: str) -> None: channel_id = str(getattr(message.chat, "id", channel_name)) if self._db.is_processed(channel_id, message.id): log.debug("Skipping duplicate msg_id=%d from %s", message.id, channel_name) return success = await self._dispatcher.dispatch(message, channel_name) if success: self._db.mark_processed(channel_id, message.id) async def _process_album(self, msgs: List, channel_name: str) -> None: """Download all album media and send as a single album to destination.""" channel_id = str(getattr(msgs[0].chat, "id", channel_name)) # Dedup: skip if the first message was already processed. if self._db.is_processed(channel_id, msgs[0].id): log.debug( "Skipping duplicate album (first msg_id=%d) from %s", msgs[0].id, channel_name, ) return local_files: List[Path] = [] caption = "" for msg in msgs: if not caption: raw = getattr(msg, "text", "") or getattr(msg, "message", "") or "" if raw: caption = self._dispatcher.format_caption(raw, channel_name) lf = await _download_media_to_temp( self._client, msg, self._dispatcher._temp_dir ) if lf: local_files.append(lf) if not caption: caption = self._dispatcher.format_caption("", channel_name) success = False try: if local_files: await self._client.send_file( self._dispatcher._dest_entity, local_files, # Telethon sends a list as an album caption=caption or None, ) success = True else: # Nothing downloaded — fall back to native forwarding. await self._client.forward_messages( self._dispatcher._dest_entity, messages=msgs, ) success = True except Exception as exc: log.error("Album send failed for group from %s: %s", channel_name, exc) finally: for lf in local_files: try: lf.unlink() except OSError: pass if success: for msg in msgs: self._db.mark_processed(channel_id, msg.id) log.info("✓ Album forwarded | channel=%-30s | parts=%d", channel_name, len(msgs)) # ═════════════════════════════════════════════════════════════════════════════ # Main monitor class # ═════════════════════════════════════════════════════════════════════════════ class TelegramMonitor: """Orchestrates the Telethon client, event registration, and reconnection.""" # How long (seconds) to wait before retrying after a connection failure. _RECONNECT_DELAYS = [5, 10, 30, 60, 120] def __init__(self, config: Config) -> None: self._cfg = config # ── Choose persistent store: HF Dataset or local SQLite ─────────────── if config.hf_token and config.hf_dataset_repo and _HF_AVAILABLE: self._db = HFProcessedMsgStore( hf_token=config.hf_token, repo_id=config.hf_dataset_repo, push_every=config.hf_push_every, ) log.info( "Persistent store: Hugging Face Dataset (%s)", config.hf_dataset_repo, ) else: self._db = ProcessedMsgDB(config.db_path) log.info("Persistent store: SQLite (%s)", config.db_path) # ── Choose session: String (cloud) or file (local) ──────────────────── if config.session_string: from telethon.sessions import StringSession session = StringSession(config.session_string) log.info("Session mode: StringSession (cloud/Render)") else: session = config.session_name log.info("Session mode: file-based (%s.session)", config.session_name) self._client = TelegramClient( session, config.api_id, config.api_hash, # Connection settings tuned for stability with 50+ channels. connection_retries=None, # infinite retries retry_delay=5, auto_reconnect=True, request_retries=5, ) self._dispatcher: Optional[ForwardDispatcher] = None self._collector: Optional[AlbumCollector] = None self._ai: Optional["AIProcessor"] = None # Map channel entity ID → human-readable name (populated at startup). self._channel_names: Dict[int, str] = {} # ── Startup ─────────────────────────────────────────────────────────────── async def start(self) -> None: self._db.open() log.info("Connecting to Telegram…") await self._client.start(phone=self._cfg.phone_number) log.info("✓ Connected to Telegram") # ── Initialise AI layer ───────────────────────────────────────────── if self._cfg.ai_enabled and _AI_AVAILABLE and self._cfg.groq_api_key: self._ai = AIProcessor( api_key=self._cfg.groq_api_key, similarity_threshold=self._cfg.similarity_threshold, window_size=self._cfg.recent_news_window, ) log.info( "AI enabled (Groq llama-3.1-8b-instant) | window=%d | threshold=%.0f%%", self._cfg.recent_news_window, self._cfg.similarity_threshold * 100, ) self._dispatcher = AIForwardDispatcher( client=self._client, destination=self._cfg.destination_channel, temp_dir=self._cfg.temp_dir, db=self._db, ai=self._ai, ) else: if self._cfg.ai_enabled and not _AI_AVAILABLE: log.warning( "AI is enabled in config but google-genai is not installed. " "Run: pip install google-genai" ) self._dispatcher = ForwardDispatcher( client=self._client, destination=self._cfg.destination_channel, temp_dir=self._cfg.temp_dir, db=self._db, ) await self._dispatcher.resolve_destination() self._collector = AlbumCollector( client=self._client, dispatcher=self._dispatcher, db=self._db, ) await self._resolve_source_channels() self._register_handlers() log.info( "Monitoring %d channel(s). Forwarding → %s", len(self._channel_names), self._cfg.destination_channel, ) async def _resolve_source_channels(self) -> None: """Resolve each SOURCE_CHANNEL identifier to a name and numeric ID.""" for raw in self._cfg.source_channels: try: entity = await self._client.get_entity(raw) cid = entity.id name = getattr(entity, "title", raw) or raw self._channel_names[cid] = name log.info(" ✓ Source: %-40s (id=%d)", name, cid) except ChannelPrivateError: log.error(" ✗ Cannot access (private): %s", raw) except Exception as exc: log.error(" ✗ Cannot resolve channel %r: %s", raw, exc) def _register_handlers(self) -> None: """Register a single NewMessage event handler for all source channels.""" watched_ids = list(self._channel_names.keys()) @self._client.on(events.NewMessage(chats=watched_ids)) async def _on_new_message(event: events.NewMessage.Event) -> None: message = event.message chat_id = event.chat_id channel_name = self._channel_names.get(chat_id, str(chat_id)) log.debug( "← Received | channel=%-30s | msg_id=%-8d | grouped=%s", channel_name, message.id, getattr(message, "grouped_id", None), ) await self._collector.handle(message, channel_name) log.debug("Event handler registered for %d channel(s).", len(watched_ids)) # ── Run loop ────────────────────────────────────────────────────────────── async def run_forever(self) -> None: """Block forever, handling disconnections with exponential back-off.""" attempt = 0 while True: try: await self.start() attempt = 0 # reset on successful connection # Prune old DB entries once at startup. removed = self._db.prune(keep_days=30) if removed: log.info("Pruned %d old processed-message records.", removed) log.info("Bot is running. Press Ctrl+C to stop.") await self._client.run_until_disconnected() except KeyboardInterrupt: log.info("Keyboard interrupt — shutting down.") break except Exception as exc: delay = self._RECONNECT_DELAYS[ min(attempt, len(self._RECONNECT_DELAYS) - 1) ] log.error( "Connection lost (attempt %d): %s — reconnecting in %d s…", attempt + 1, exc, delay, ) attempt += 1 await asyncio.sleep(delay) finally: await self._cleanup() async def _cleanup(self) -> None: """Gracefully disconnect and close resources.""" try: if self._client.is_connected(): await self._client.disconnect() except Exception: pass # DB is closed only on final shutdown (not between reconnects). # ═════════════════════════════════════════════════════════════════════════════ # Health-check web server (for UptimeRobot / HF Spaces keep-alive) # ═════════════════════════════════════════════════════════════════════════════ _HEALTH_HTML = """\ PepBielsa Bot — Status

PepBielsa Bot

Telegram channel monitor is active and forwarding.

✓ Online
""" async def _health_handler(request) -> "aiohttp_web.Response": """Return an HTML health page that satisfies UptimeRobot keyword monitors.""" return aiohttp_web.Response( text=_HEALTH_HTML, content_type="text/html", charset="utf-8", headers={ # Mimic a real browser server so UptimeRobot treats us as a web page. "Server": "nginx/1.25.3", "X-Content-Type-Options": "nosniff", "Cache-Control": "no-cache, no-store, must-revalidate", }, ) async def run_health_server(port: int = 7860) -> None: """Start the aiohttp health-check server on *port*. HF Spaces expects the app to listen on port 7860 by default. Override via the PORT environment variable. """ if not _AIOHTTP_AVAILABLE: log.warning( "aiohttp not installed — health-check server disabled. " "Add 'aiohttp>=3.9.0' to requirements.txt." ) return app = aiohttp_web.Application() app.router.add_get("/", _health_handler) app.router.add_get("/health", _health_handler) runner = aiohttp_web.AppRunner(app) await runner.setup() site = aiohttp_web.TCPSite(runner, "0.0.0.0", port) await site.start() log.info("Health-check server listening on http://0.0.0.0:%d/health", port) # Keep running forever (cancelled when the event loop shuts down). try: await asyncio.Event().wait() except asyncio.CancelledError: pass finally: await runner.cleanup() # ═════════════════════════════════════════════════════════════════════════════ # Entry-point # ═════════════════════════════════════════════════════════════════════════════ def main() -> None: import threading # ── Step 1: Start health server immediately (before config check) ────────── # Runs in a daemon thread with its own event loop so /health always responds # on port 7860, even if env vars are missing or Telegram is unreachable. health_port = int(os.getenv("PORT", "7860")) def _health_thread() -> None: loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: loop.run_until_complete(run_health_server(health_port)) finally: loop.close() t = threading.Thread(target=_health_thread, daemon=True, name="health-server") t.start() print(f"[BOOT] Health server starting on :{health_port} …") # ── Step 2: Load and validate bot configuration ──────────────────────────── try: cfg = get_config() except ValueError as exc: print(f"[FATAL] {exc}") print("[FATAL] Bot is stopped due to config errors — health server stays up.") t.join() # block forever so the container stays alive for UptimeRobot return # Set up logging now that we have the desired level. global log log = configure_logging(cfg.log_level) log.info("=" * 60) log.info(" Telegram Channel Monitor Bot") log.info("=" * 60) log.info("Session : %s", cfg.session_name) log.info("Temp dir : %s", cfg.temp_dir) log.info("DB path : %s", cfg.db_path) log.info("Log level: %s", cfg.log_level) log.info("Sources : %d channel(s)", len(cfg.source_channels)) log.info("=" * 60) monitor = TelegramMonitor(cfg) # ── Step 3: Run bot in the main event loop ───────────────────────────────── try: asyncio.run(monitor.run_forever()) except KeyboardInterrupt: pass finally: monitor._db.close() log.info("Bot stopped. Goodbye.") if __name__ == "__main__": main()