"""Telegram-интеграция (п.4.2, 4.3 ТЗ).

api_id/api_hash берутся из настроек (введены в UI, НЕ из env). Сессия
Telethon сохраняется в файловой системе — повторная авторизация не нужна.
Вход: по телефону (+2FA) или по QR-ссылке. Сообщения из каналов с
включённым мониторингом кладутся в очередь pipeline, откуда их разбирает
фоновый воркер (стоп-фразы → ML → ИИ). В БД оседает только прошедшее
фильтры; исходное сообщение карточки хранит msg_id для «открыть исходник».
"""
from __future__ import annotations

import asyncio
import logging
import math
import random
import threading
import time

from telethon import TelegramClient, events, utils
from telethon.errors import SessionPasswordNeededError, PhoneCodeInvalidError, PhoneCodeExpiredError
from telethon.errors.rpcerrorlist import FloodWaitError
from telethon.tl import functions

from .. import config
from ..constants import DIALOG_HUES
from ..crypto import decrypt_text
from ..db import store
from ..sse import broker
from . import ban_guard
from . import pipeline

log = logging.getLogger("leadradar.tg")


BACKFILL_PER_MESSAGE = (1.5, 3.0)  # секунды, чтобы не попасть под бан
BACKFILL_PER_DIALOG = (3.0, 6.0)


def _keys() -> dict:
    raw = store.get_setting("tgKeys") or {}
    api_hash = str(raw.get("apiHash", ""))
    return {
        "apiId": str(raw.get("apiId", "")).strip(),
        "apiHash": (decrypt_text(api_hash) if api_hash.startswith("enc:") else api_hash),
    }


# Главный event loop приложения: на нём живут Telethon-клиент и фоновые
# задачи. Синхронные роутеры FastAPI исполняются в threadpool (без running
# loop), поэтому корутины нужно планировать на этот loop через
# call_soon_threadsafe — Telethon не переносит клиент между циклами.
_MAIN_LOOP: asyncio.AbstractEventLoop | None = None


def register_main_loop(loop: asyncio.AbstractEventLoop) -> None:
    global _MAIN_LOOP
    _MAIN_LOOP = loop


def _spawn(coro) -> None:
    """Запустить корутину из любого контекста: running loop, главный loop или поток.

    set_monitor/set_monitor_all вызываются из синхронных роутеров FastAPI, где
    running loop отсутствует, поэтому прямой asyncio.create_task падает с
    RuntimeError (отсюда Internal Server Error при включении канала).
    """
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = _MAIN_LOOP
        if loop is not None and loop.is_running():
            loop.call_soon_threadsafe(loop.create_task, coro)
            return
        threading.Thread(
            target=lambda: asyncio.new_event_loop().run_until_complete(coro),
            daemon=True,
        ).start()
    else:
        loop.create_task(coro)


class TelegramManager:
    def __init__(self) -> None:
        self.client: TelegramClient | None = None
        self.phase = "idle"  # idle | phone | code | password | qr | ready
        self.phone: str = ""
        self._code_hash: str = ""
        self._auth_lock = asyncio.Lock()
        self._listener_task: asyncio.Task | None = None
        self._monitored: set[str] = set()
        self.error: str | None = None
        self._disconnect_hook = None
        # QR
        self.qr_url: str = ""
        self._qr_task: asyncio.Task | None = None
        # сердцебиение статуса
        self._last_connected: bool | None = None
        # backfill каналов при первом подключении
        self._backfilling: set[str] = set()

    # ── состояние для UI ──────────────────────────────────────────────────

    def status(self) -> dict:
        row = store.query_one(
            "SELECT count(*) AS d FROM dialogs WHERE monitor = TRUE"
        )
        acc = store.get_setting("tgAccount") or ""
        connected = bool(self.client and self.client.is_connected())
        listener_alive = bool(self._listener_task and not self._listener_task.done())
        return {
            "phase": self.phase,
            "connected": connected,
            "listener": listener_alive,
            "account": acc,
            "monitored": int(row["d"]) if row else 0,
            "keysSet": bool(_keys().get("apiId") and _keys().get("apiHash")),
            "error": self.error,
            "qrUrl": self.qr_url if self.phase == "qr" else None,
        }

    def _client(self) -> TelegramClient:
        keys = _keys()
        api_id = str(keys.get("apiId") or "").strip()
        api_hash = str(keys.get("apiHash") or "").strip()
        if not api_id or not api_hash:
            raise ValueError("Сначала сохраните Telegram api_id и api_hash в настройках")
        if self.client is None:
            session = str(config.SESSIONS_DIR / config.SESSION_PREFIX)
            self.client = TelegramClient(session, int(api_id), api_hash)
        return self.client

    # ── веб-авторизация ───────────────────────────────────────────────────

    async def start_phone(self, phone: str) -> None:
        async with self._auth_lock:
            self.phone = phone
            self.error = None
            try:
                client = self._client()
                await client.connect()
                sent = await client.send_code_request(phone)
                self._code_hash = sent.phone_code_hash
                self.phase = "code"
            except Exception as exc:  # noqa: BLE001
                self.phase = "idle"
                self.error = str(exc)
                raise

    async def submit_code(self, code: str) -> None:
        async with self._auth_lock:
            self.error = None
            client = self._client()
            try:
                await client.sign_in(self.phone, code, phone_code_hash=self._code_hash)
                await self._finalize()
            except SessionPasswordNeededError:
                self.phase = "password"
            except PhoneCodeInvalidError:
                self.error = "Неверный код"
                raise ValueError("Неверный код")
            except PhoneCodeExpiredError:
                self.error = "Код истёк — запросите новый"
                raise ValueError("Код истёк — запросите новый")
            except Exception as exc:  # noqa: BLE001
                self.error = str(exc)
                raise

    async def submit_password(self, password: str) -> None:
        async with self._auth_lock:
            self.error = None
            try:
                await self.client.sign_in(password=password)
                await self._finalize()
            except Exception as exc:  # noqa: BLE001
                self.error = "Неверный облачный пароль"
                raise ValueError("Неверный облачный пароль") from exc

    async def _finalize(self, notify: bool = True) -> None:
        me = await self.client.get_me()
        store.set_setting("tgAccount", f"@{me.username or 'user'}")
        self.phase = "ready"
        self._start_listener()
        await self.refresh_dialogs()
        self._schedule_first_backfill()
        if notify:
            await broker.publish_toast("Telegram подключён, сессия сохранена", "send")
        await self._publish_status()

    async def disconnect(self) -> None:
        async with self._auth_lock:
            if self._qr_task:
                self._qr_task.cancel()
                self._qr_task = None
            if self._listener_task:
                self._listener_task.cancel()
                self._listener_task = None
            if self.client:
                try:
                    await self.client.disconnect()
                except Exception:  # noqa: BLE001
                    pass
            self.phase = "idle"
            self._monitored.clear()
            self.qr_url = ""
            store.set_setting("tgAccount", "")
            await broker.publish_toast("Telegram отключён", "logout")
            await self._publish_status()

    async def auto_resume(self) -> None:
        """При старте сервера: если сессия сохранена — подключиться автоматически."""
        try:
            keys = _keys()
            if not (keys.get("apiId") and keys.get("apiHash")):
                return
            client = self._client()
            await client.connect()
            if await client.is_user_authorized():
                await self._finalize(notify=False)
        except Exception as exc:  # noqa: BLE001
            log.info("auto_resume skipped: %s", exc)
            self.phase = "idle"

    # ── прослушивание ─────────────────────────────────────────────────────

    def _start_listener(self) -> None:
        if self._listener_task and not self._listener_task.done():
            return
        self._reload_monitored()
        self.client.add_event_handler(self._on_message, events.NewMessage())
        self._listener_task = asyncio.create_task(self.client.run_until_disconnected())

        def _on_done(task: asyncio.Task) -> None:
            if task.cancelled():
                log.info("listener stopped (cancelled)")
                return
            exc = task.exception()
            if exc:
                log.error("listener crashed: %s", exc)
            else:
                log.info("listener stopped")

        self._listener_task.add_done_callback(_on_done)

    def _reload_monitored(self) -> None:
        rows = store.query("SELECT id FROM dialogs WHERE monitor = TRUE")
        self._monitored = {r["id"] for r in rows}

    def _dialog_id(self, message) -> str:
        try:
            chat_id = message.chat_id
        except Exception:  # noqa: BLE001
            chat_id = message.peer_id
        return str(chat_id)

    async def _on_message(self, event) -> None:
        """Каждое сообщение мониторящегося диалога кладём в очередь pipeline."""
        try:
            msg = event.message
            if msg is None or msg.text is None or not msg.text.strip():
                return
            dialog_id = self._dialog_id(event.message)
            if dialog_id not in self._monitored:
                return
            chat = await event.get_chat()
            ch_name = getattr(chat, "title", None) or getattr(chat, "first_name", "") or dialog_id
            ch_handle = getattr(chat, "username", "") or ""
            hue = dialog_hue(dialog_id, ch_name)
            now = time.time_ns() // 1_000_000
            ts = int(msg.date.timestamp() * 1000) if getattr(msg, "date", None) else now
            store.execute(
                "UPDATE dialogs SET last_text = ?, last_at = ?, updated_at = ? WHERE id = ?",
                [msg.text.strip()[:200], now, now, dialog_id],
            )
            pipeline.enqueue(dialog_id, ch_name, ch_handle, hue, msg.id, msg.text, ts)
            # помечаем сообщение прочитанным в Telegram, чтобы оно не висело
            # «новым» в других клиентах/устройствах
            try:
                await self.client.send_read_acknowledge(chat)
            except Exception:  # noqa: BLE001
                log.debug("read ack failed", exc_info=True)
        except Exception as exc:  # noqa: BLE001
            log.exception("on_message failed: %s", exc)

    # ── QR-вход ───────────────────────────────────────────────────────────

    async def qr_start(self) -> str:
        async with self._auth_lock:
            self.error = None
            client = self._client()
            await client.connect()
            if await client.is_user_authorized():
                await self._finalize(notify=False)
                return ""
            if self._qr_task and not self._qr_task.done():
                return self.qr_url
            qr = await client.qr_login()
            self.qr_url = qr.url
            self.phase = "qr"
            self._qr_task = asyncio.create_task(self._wait_qr(qr))
            return self.qr_url

    async def _wait_qr(self, qr) -> None:
        try:
            await qr.wait()
            await self._finalize(notify=True)
        except Exception as exc:  # noqa: BLE001
            log.warning("qr wait error: %s", exc)
            self.error = str(exc)
            self.phase = "idle"
        finally:
            self._qr_task = None

    # ── статус (сердцебиение) ─────────────────────────────────────────────

    async def _publish_status(self) -> None:
        await broker.publish("system_status", self.status())

    async def heartbeat(self) -> None:
        """Периодический вызов из планировщика: уведомляем, если Telegram «уснул»."""
        connected = bool(self.client and self.client.is_connected())
        if self.phase == "ready" and connected != self._last_connected:
            self._last_connected = connected
            await self._publish_status()
            if not connected:
                await broker.publish_toast("Telegram отключён — переподключение при следующей проверке", "bell")
        if self.phase == "ready" and connected:
            self._last_connected = True

    # ── backfill: при первом подключении по 10 последних сообщений ────────

    def _schedule_first_backfill(self) -> None:
        rows = store.query("SELECT id FROM dialogs WHERE monitor = TRUE AND backfilled = FALSE")
        ids = [r["id"] for r in rows]
        if ids:
            asyncio.create_task(self._backfill_dialogs(ids))

    async def _backfill_dialogs(self, dialog_ids: list[str], force: bool = False) -> None:
        for dialog_id in dialog_ids:
            if dialog_id in self._backfilling:
                continue
            try:
                processed = await self.backfill_dialog(dialog_id, force=force)
                if processed:
                    log.info("backfill %s: %d сообщений в очередь", dialog_id, processed)
            except Exception as exc:  # noqa: BLE001
                log.warning("backfill %s failed: %s", dialog_id, exc)
            await asyncio.sleep(random.uniform(*BACKFILL_PER_DIALOG))

    async def backfill_dialog(self, dialog_id: str, force: bool = False) -> int:
        """Последние 10 сообщений канала: разбираем с паузами (анти-бан).

        force=True — «Перечитать» по кнопке: даже если канал уже разобран.
        Повторы карточек не создаются (защита dedup по нормализованному тексту).
        """
        if dialog_id in self._backfilling:
            return 0
        client = self.client
        if not client or not client.is_connected():
            return 0
        row = store.query_one("SELECT backfilled FROM dialogs WHERE id = ?", [dialog_id])
        if not row or (not force and bool(row["backfilled"])):
            return 0
        self._backfilling.add(dialog_id)
        processed = 0
        try:
            entity = await client.get_entity(int(dialog_id))
            chat = await client.get_entity(int(dialog_id))
            name = getattr(chat, "title", None) or getattr(chat, "first_name", "") or dialog_id
            handle = getattr(chat, "username", "") or ""
            hue = dialog_hue(dialog_id, name)
            msgs = await client.get_messages(entity, limit=10)
            for m in reversed(msgs):  # от старых к новым, как реальный поток
                if m.text is None or not m.text.strip():
                    continue
                now = time.time_ns() // 1_000_000
                ts = int(m.date.timestamp() * 1000) if m.date else now
                # в очередь уходит всё; отсев сделает воркер, в БД осядет только
                # то, что прошло фильтры
                pipeline.enqueue(dialog_id, name, handle, hue, m.id, m.text, ts)
                processed += 1
                await asyncio.sleep(random.uniform(*BACKFILL_PER_MESSAGE))
            # «Перечитать» — вручную вытащили сообщения: снимаем «новое» в Telegram
            try:
                await client.send_read_acknowledge(entity)
            except Exception:  # noqa: BLE001
                log.debug("backfill read ack failed", exc_info=True)
            store.execute("UPDATE dialogs SET backfilled = TRUE WHERE id = ?", [dialog_id])
        finally:
            self._backfilling.discard(dialog_id)
        return processed

    async def realtime_sweep(self) -> None:
        """Страховка realtime: если событие потеряно (рестарт/разрыв), раз в 30 с
        докачиваем непрочитанные сообщения включённых каналов, кладём в очередь
        и помечаем прочитанными."""
        client = self.client
        if not client or not client.is_connected():
            return
        self._reload_monitored()
        if not self._monitored:
            return
        try:
            dialogs = await client.get_dialogs(limit=500)
        except Exception as exc:  # noqa: BLE001
            log.debug("realtime sweep: get_dialogs fail: %s", exc)
            return
        # синхронизация списка: новые каналы/группы появляются и включаются
        # автоматически, удалённые исчезают (см. _persist_dialogs)
        try:
            entries = []
            for dlg in dialogs:
                ent = dlg.entity
                name = dlg.name or ""
                entries.append(
                    (
                        str(dlg.id),
                        name,
                        getattr(ent, "username", "") or "",
                        self._kind_of(ent),
                        dialog_hue(str(dlg.id), name),
                    )
                )
            if entries:
                self._persist_dialogs(entries)
        except Exception as exc:  # noqa: BLE001
            log.debug("realtime sweep: sync dialogs fail: %s", exc)
        for dlg in dialogs:
            did = str(dlg.id)
            if did not in self._monitored:
                continue
            unread = int(getattr(dlg, "unread_count", 0) or 0)
            if unread <= 0:
                continue
            try:
                msgs = await client.get_messages(dlg.entity, limit=min(unread + 2, 10))
                added = 0
                for m in reversed(msgs):  # от старых к новым
                    if m.text is None or not m.text.strip():
                        continue
                    if store.scalar(
                        "SELECT 1 FROM pipeline_msg WHERE dialog_id = ? AND msg_id = ?", [did, m.id]
                    ):
                        continue
                    ent = dlg.entity
                    name = getattr(ent, "title", None) or getattr(ent, "first_name", "") or did
                    handle = getattr(ent, "username", "") or ""
                    hue = dialog_hue(did, name)
                    now = time.time_ns() // 1_000_000
                    ts = int(m.date.timestamp() * 1000) if getattr(m, "date", None) else now
                    pipeline.enqueue(did, name, handle, hue, m.id, m.text, ts)
                    added += 1
                if added:
                    log.info("realtime sweep %s: +%d в очередь (потерянные события)", did, added)
                await client.send_read_acknowledge(dlg.entity)
            except Exception as exc:  # noqa: BLE001
                log.debug("realtime sweep dialog %s fail: %s", did, exc)

    # ── диалоги/каналы ────────────────────────────────────────────────────

    @staticmethod
    def _kind_of(entity) -> str:
        if getattr(entity, "broadcast", False):
            return "канал"
        if getattr(entity, "megagroup", False) or getattr(entity, "gigagroup", False) or getattr(entity, "group", False):
            return "группа"
        return "чат"

    def _persist_dialogs(self, entries: list[tuple]) -> int:
        """Синхронизация списка диалогов с Telegram.

        - новые чаты/каналы добавляются; авто-мониторинг новых управляется
          настройкой autoMonitorNew (вкл — любой появившийся чат мониторится,
          выкл — появляется отключённым);
        - переименования/смена типа обновляются (monitor пользователя не трогаем);
        - диалоги, которых больше нет в Telegram (вышел/удалил), удаляются.
        """
        if not entries:
            return 0
        now = time.time_ns() // 1_000_000
        auto_new = bool(store.get_setting("autoMonitorNew"))
        seen: set[str] = set()
        for dlg_id, name, handle, kind, hue in entries:
            seen.add(dlg_id)
            store.execute(
                "INSERT INTO dialogs(id, name, handle, kind, hue, monitor, updated_at) "
                "VALUES (?, ?, ?, ?, ?, ?, ?) "
                "ON CONFLICT(id) DO UPDATE SET name = excluded.name, handle = excluded.handle, "
                "kind = excluded.kind, hue = excluded.hue, updated_at = excluded.updated_at",
                [dlg_id, name, handle, kind, hue, auto_new, now],
            )
        stale = store.query(
            "SELECT id FROM dialogs WHERE id NOT IN (" + ",".join(["?"] * len(seen)) + ")",
            list(seen),
        )
        if stale:
            stale_ids = [r["id"] for r in stale]
            store.execute(
                "DELETE FROM dialogs WHERE id IN (" + ",".join(["?"] * len(stale_ids)) + ")",
                stale_ids,
            )
            log.info("dialogs sync: удалено устаревших источников: %d", len(stale_ids))
        self._reload_monitored()
        return len(entries)

    async def refresh_dialogs(self) -> int:
        client = self._client()
        if not client.is_connected():
            await client.connect()
        entries: list[tuple] = []
        async for dialog in client.iter_dialogs(limit=500):
            entity = dialog.entity
            name = dialog.name or ""
            handle = getattr(entity, "username", "") or ""
            kind = self._kind_of(entity)
            hue = dialog_hue(str(dialog.id), name)
            entries.append((str(dialog.id), name, handle, kind, hue))
        if entries:
            self._persist_dialogs(entries)
        return len(entries)

    def list_dialogs(self) -> list[dict]:
        rows = store.query("SELECT * FROM dialogs ORDER BY monitor DESC, name")
        return [
            {
                "id": r["id"],
                "name": r["name"],
                "handle": r["handle"],
                "type": r["kind"],
                "hue": r["hue"],
                "on": bool(r["monitor"]),
                "last": {"text": r["last_text"], "time": r["last_at"]},
            }
            for r in rows
        ]

    def set_monitor(self, dialog_id: str, enabled: bool) -> None:
        store.execute(
            "UPDATE dialogs SET monitor = ?, updated_at = ? WHERE id = ?",
            [enabled, time.time_ns() // 1_000_000, dialog_id],
        )
        self._reload_monitored()
        if enabled:
            row = store.query_one("SELECT backfilled FROM dialogs WHERE id = ?", [dialog_id])
            if row and not bool(row["backfilled"]):
                # первое включение мониторинга: разбираем последние 10 сообщений с паузами
                _spawn(self.backfill_dialog(dialog_id))

    def set_monitor_all(self, enabled: bool) -> int:
        """Включить/выключить мониторинг сразу для всех диалогов.

        Первое включение каждого канала разбирается последовательно (с паузами),
        чтобы не попасть под бан Telegram.
        """
        now = time.time_ns() // 1_000_000
        rows = store.query("SELECT id, backfilled FROM dialogs")
        to_backfill: list[str] = []
        for r in rows:
            store.execute(
                "UPDATE dialogs SET monitor = ?, updated_at = ? WHERE id = ?",
                [enabled, now, r["id"]],
            )
            if enabled and not bool(r["backfilled"]):
                to_backfill.append(r["id"])
        self._reload_monitored()
        if enabled and to_backfill:
            _spawn(self._backfill_dialogs(to_backfill))
        return len(rows)

    def backfill_monitored(self) -> int:
        """Кнопка «Перечитать»: последние 10 сообщений всех включённых каналов.

        Разбор идёт в фоне последовательно с паузами (анти-бан), даже если
        канал уже разобран (force). Включённые каналы продолжают ловить новые
        сообщения в реальном времени — это ручная догонялка.
        """
        rows = store.query("SELECT id FROM dialogs WHERE monitor = TRUE")
        ids = [r["id"] for r in rows]
        if not ids:
            return 0
        _spawn(self._backfill_dialogs(ids, force=True))
        return len(ids)

    async def dialog_messages(self, dialog_id: str, limit: int = 24) -> list[dict]:
        """Последние сообщения диалога: свежие берём из Telegram, старые — из БД."""
        out: list[dict] = []
        client = self.client
        try:
            if client and client.is_connected():
                entity = await client.get_entity(int(dialog_id))
                msgs = await client.get_messages(entity, limit=min(limit, 30))
                now = time.time_ns() // 1_000_000
                for m in msgs:
                    if m.text:
                        row = store.query_one(
                            "SELECT id FROM messages WHERE id = ?",
                            [f"m_{dialog_id}_{m.id}"],
                        )
                        lead_id = None
                        if row:
                            lead_id = row.get("lead_id") or None
                        if not row:
                            store.execute(
                                "INSERT OR IGNORE INTO messages(id, dialog_id, text, msg_at) VALUES (?, ?, ?, ?)",
                                [f"m_{dialog_id}_{m.id}", dialog_id, m.text[:4000], now],
                            )
                        out.append({"id": m.id, "text": m.text, "time": m.date.timestamp() * 1000, "lead": bool(lead_id)})
                # вручную вытащили сообщения — снимаем «новое» в Telegram
                try:
                    await client.send_read_acknowledge(entity)
                except Exception:  # noqa: BLE001
                    log.debug("dialog_messages read ack failed", exc_info=True)
        except Exception as exc:  # noqa: BLE001
            log.warning("dialog_messages tg fail: %s", exc)
        if not out:
            rows = store.query(
                "SELECT * FROM messages WHERE dialog_id = ? ORDER BY msg_at DESC LIMIT ?",
                [dialog_id, limit],
            )
            out = [{"id": r["id"], "text": r["text"], "time": r["msg_at"], "lead": bool(r["lead_id"])} for r in rows]
        return out

    # ── discovery: поиск каналов и действия (Task 4) ──────────────────────

    async def discovery_search(self, q: str, limit: int = 30) -> list[dict]:
        """Глобальный поиск каналов/групп по ключу (contacts.search).

        id возвращается подписанным (как в dialogs: каналы -100…, группы -id,
        люди +id). Найденные entity кэшируются в сессию Telethon — тогда
        discovery_info/read смогут получить участников/историю по id даже без
        вступления (публичные источники).
        """
        client = self.client
        if not client or not client.is_connected():
            raise RuntimeError("Telegram не подключён")
        found = await client(functions.contacts.SearchRequest(q=q, limit=limit))
        # пауза между поисковыми запросами (анти-бан, BanGuard)
        await asyncio.sleep(ban_guard.search_pause())
        try:
            client.session.process_entities(found)
        except Exception:  # noqa: BLE001
            log.debug("discovery_search: cache entities failed", exc_info=True)
        out: list[dict] = []
        seen: set[str] = set()
        for ent in (*found.chats, *found.users):
            try:
                dialog_id = str(utils.get_peer_id(ent))
            except (TypeError, ValueError):
                continue
            if dialog_id in seen:
                continue
            seen.add(dialog_id)
            name = utils.get_display_name(ent) or dialog_id
            out.append(
                {
                    "id": dialog_id,
                    "name": name,
                    "username": getattr(ent, "username", "") or "",
                    "kind": self._kind_of(ent),
                    "hue": dialog_hue(dialog_id, name),
                }
            )
            if len(out) >= limit:
                break
        return out

    async def discovery_info(self, dialog_id: str) -> dict:
        """Инфо об источнике: тип, username, участники, признак форума.

        participants берётся из full_chat (channels.getFullChannel для
        каналов/супергрупп, messages.getFullChat для базовых групп); если
        определить не удалось (нет членства/приватный/ошибка) — None,
        исключение наружу не бросаем.
        """
        result = {
            "id": str(dialog_id),
            "name": str(dialog_id),
            "username": "",
            "kind": "",
            "hue": dialog_hue(str(dialog_id), str(dialog_id)),
            "participants": None,
            "is_forum": False,
        }
        client = self.client
        if not client or not client.is_connected():
            return result
        try:
            entity = await client.get_entity(int(dialog_id))
        except Exception as exc:  # noqa: BLE001
            log.warning("discovery_info %s: entity not resolved: %s", dialog_id, exc)
            return result
        name = utils.get_display_name(entity) or str(dialog_id)
        kind = self._kind_of(entity)
        result.update(
            name=name,
            username=getattr(entity, "username", "") or "",
            kind=kind,
            hue=dialog_hue(str(dialog_id), name),
            is_forum=bool(getattr(entity, "forum", False)),
        )
        try:
            if kind in ("канал", "группа"):
                full = await client(functions.channels.GetFullChannelRequest(entity))
                full_chat = full.full_chat
                participants = int(getattr(full_chat, "participants_count", 0) or 0)
                result["participants"] = participants or None
            elif getattr(entity, "title", None):
                # базовая группа (legacy): полный чат приносит список участников
                full = await client(functions.messages.GetFullChatRequest(entity.id))
                members = getattr(
                    getattr(full.full_chat, "participants", None), "participants", None
                )
                if members:
                    result["participants"] = len(members)
        except Exception as exc:  # noqa: BLE001
            log.debug("discovery_info %s: participants unavailable: %s", dialog_id, exc)
        return result

    async def discovery_read(self, dialog_id: str, limit: int) -> dict:
        """Последние сообщения источника для оценки (без создания карточек).

        Форум (entity.forum): читаем выборку по активным темам
        (channels.getForumTopics + по каждой теме get_messages(reply_to=topic_id))
        и возвращаем плоский список с topic_id/topic_title. Для обычных
        источников topic_id/topic_title = None. История недоступна
        (приватный/закрытый источник без членства) — ok=False,
        error="no_history". Исключения наружу не бросаем: ошибка в темах —
        безопасный fallback на обычное чтение ленты.
        """
        client = self.client
        limit = max(int(limit or 0), 0)
        if limit <= 0:
            return {"ok": True, "error": None, "messages": []}
        if not client or not client.is_connected():
            return {"ok": False, "error": "no_history", "messages": []}
        try:
            entity = await client.get_entity(int(dialog_id))
        except Exception as exc:  # noqa: BLE001
            log.warning("discovery_read %s: entity not resolved: %s", dialog_id, exc)
            return {"ok": False, "error": "no_history", "messages": []}
        if getattr(entity, "forum", False):
            try:
                topic_msgs = await self._read_forum_topics(client, entity, limit)
            except Exception as exc:  # noqa: BLE001
                log.debug("discovery_read %s: forum topics fail, fallback to feed: %s", dialog_id, exc)
                topic_msgs = []
            if topic_msgs:
                return {"ok": True, "error": None, "messages": topic_msgs}
        # обычная лента (каналы/группы; для форума — General-тема)
        try:
            msgs = await client.get_messages(entity, limit=limit)
        except Exception as exc:  # noqa: BLE001
            log.warning("discovery_read %s: history unavailable: %s", dialog_id, exc)
            return {"ok": False, "error": "no_history", "messages": []}
        now = time.time_ns() // 1_000_000
        out: list[dict] = []
        for m in msgs:
            item = self._discovery_message_item(m, None, None, now)
            if item:
                out.append(item)
        return {"ok": True, "error": None, "messages": out}

    async def _read_forum_topics(self, client, entity, limit: int) -> list[dict]:
        """Выборка сообщений по активным темам форума.

        Возвращает плоский список {id, text, date_ms, topic_id, topic_title}.
        Исключения не бросает: темы, которые не прочитались, пропускаются,
        вызывающий решает, делать ли fallback на обычную ленту.
        """
        out: list[dict] = []
        try:
            res = await client(
                functions.channels.GetForumTopicsRequest(
                    channel=entity, offset_date=0, offset_id=0, offset_topic=0, limit=5
                )
            )
            topics = list(getattr(res, "topics", None) or [])
        except Exception as exc:  # noqa: BLE001
            log.debug("discovery_read: getForumTopics fail: %s", exc)
            return out
        if not topics:
            return out
        # на тему минимум 3 сообщения (иначе тема почти всегда отсеется как
        # «мало подходящих»), cap 10; суммарно выборка может слегка превысить limit
        per_topic = min(max(3, math.ceil(limit / len(topics))), 10)
        now = time.time_ns() // 1_000_000
        for topic in topics:
            topic_id = int(getattr(topic, "id", 0) or 0)
            topic_title = getattr(topic, "title", "") or ""
            if not topic_id:
                continue
            try:
                msgs = await client.get_messages(entity, limit=per_topic, reply_to=topic_id)
            except Exception as exc:  # noqa: BLE001
                log.debug("discovery_read: topic %s read fail: %s", topic_id, exc)
                continue
            for m in msgs:
                item = self._discovery_message_item(m, topic_id, topic_title, now)
                if item:
                    out.append(item)
        return out

    @staticmethod
    def _discovery_message_item(m, topic_id, topic_title, now: int) -> dict | None:
        """Одно сообщение discovery_read: только непустой текст (как в
        backfill/dialog_messages), поля {id, text, date_ms, topic_id, topic_title}."""
        text = getattr(m, "text", None)
        if not text or not text.strip():
            return None
        date = getattr(m, "date", None)
        return {
            "id": m.id,
            "text": text,
            "date_ms": int(date.timestamp() * 1000) if date else now,
            "topic_id": topic_id,
            "topic_title": topic_title,
        }

    async def discovery_join(self, username: str) -> None:
        """Вступить в канал/группу по @username (channels.JoinChannelRequest).

        Ручной join из API — вне квот, без пауз; паузу перед авто-вступлением
        делает воркер (ban_guard.wait_join_delay). FloodWaitError фиксируется
        в BanGuard (стоп авто-вступлений до конца суток) и пробрасывается
        вызывающему.
        """
        username = (username or "").strip().lstrip("@")
        if not username:
            raise ValueError("Не указан username для вступления")
        client = self.client
        if not client or not client.is_connected():
            raise RuntimeError("Telegram не подключён")
        try:
            entity = await client.get_entity(username)
            await client(functions.channels.JoinChannelRequest(entity))
        except FloodWaitError:
            ban_guard.note_flood()
            log.warning("discovery_join %s: flood — авто-вступления стоп до конца суток", username)
            raise
        log.info("discovery_join: вступили в @%s", username)

    async def discovery_leave(self, dialog_id: str) -> None:
        """Выйти из канала/группы (channels.LeaveChannelRequest)."""
        client = self.client
        if not client or not client.is_connected():
            raise RuntimeError("Telegram не подключён")
        entity = await client.get_entity(int(dialog_id))
        await client(functions.channels.LeaveChannelRequest(entity))
        log.info("discovery_leave: вышли из %s", dialog_id)

    def add_dialog_monitored(self, dialog_id, name, username, kind, hue) -> None:
        """Добавить источник в dialogs с monitor=TRUE (после вступления).

        INSERT/UPDATE без авто-логики (как set_monitor, но без запуска
        backfill): backfilled=FALSE — разбор последних сообщений подхватит
        обычный механизм при первом подключении/перечитывании.
        """
        now = time.time_ns() // 1_000_000
        store.execute(
            "INSERT INTO dialogs(id, name, handle, kind, hue, monitor, backfilled, updated_at) "
            "VALUES (?, ?, ?, ?, ?, TRUE, FALSE, ?) "
            "ON CONFLICT(id) DO UPDATE SET name = excluded.name, handle = excluded.handle, "
            "kind = excluded.kind, hue = excluded.hue, monitor = TRUE, backfilled = FALSE, "
            "updated_at = excluded.updated_at",
            [
                str(dialog_id),
                name or str(dialog_id),
                username or "",
                kind or "",
                hue or "#666",
                now,
            ],
        )
        self._reload_monitored()


def dialog_hue(dialog_id: str, name: str = "") -> str:
    h = 0
    for ch in (dialog_id + name):
        h = (h * 31 + ord(ch)) % len(DIALOG_HUES)
    return DIALOG_HUES[h]


tg = TelegramManager()