main.py

"""
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 = """\
<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>PepBielsa Bot — Status</title>
  <style>
    *{{margin:0;padding:0;box-sizing:border-box}}
    body{{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;
          background:#0f172a;color:#e2e8f0;display:flex;align-items:center;
          justify-content:center;min-height:100vh}}
    .card{{background:#1e293b;border:1px solid #334155;border-radius:16px;
           padding:40px 48px;text-align:center;max-width:420px;width:90%}}
    .dot{{width:14px;height:14px;background:#22c55e;border-radius:50%;
          display:inline-block;margin-right:8px;
          box-shadow:0 0 0 4px rgba(34,197,94,.25);animation:pulse 2s infinite}}
    @keyframes pulse{{0%,100%{{box-shadow:0 0 0 4px rgba(34,197,94,.25)}}
                      50%{{box-shadow:0 0 0 8px rgba(34,197,94,.1)}}}}
    h1{{font-size:1.6rem;font-weight:700;margin-bottom:8px}}
    p{{color:#94a3b8;font-size:.9rem;margin-top:12px}}
    .badge{{display:inline-block;background:#22c55e22;color:#4ade80;
            border:1px solid #22c55e55;border-radius:999px;
            padding:4px 14px;font-size:.8rem;margin-top:20px}}
  </style>
</head>
<body>
  <div class="card">
    <h1><span class="dot"></span>PepBielsa Bot</h1>
    <p>Telegram channel monitor is active and forwarding.</p>
    <div class="badge">&#x2713;&nbsp;Online</div>
  </div>
</body>
</html>
"""


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()