Rustam Khalimov
9e07568ddd
Первый коммит: модульный монолит ядра (.NET 10) и gRPC-сервисы ai/ml/telegram, фронтенд Vue 3/Vite/Tailwind, документация (ТЗ, инструкция пользователя, техдокументация, код-стайл), бэклог, скрипты развёртывания и архив прототипа LeadRadar.
884 lines
42 KiB
Python
884 lines
42 KiB
Python
"""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()
|