"""Несколько юзербот-сессий: один Telegram-аккаунт ловца на каждого пользователя бота.""" from __future__ import annotations import asyncio import re import sqlite3 import time from pathlib import Path from typing import Optional from telethon import TelegramClient from telethon.errors import ( FloodWaitError, PasswordHashInvalidError, PhoneCodeExpiredError, PhoneCodeInvalidError, PhoneNumberInvalidError, SessionPasswordNeededError, ) from db import ( delete_session_meta, find_owner_by_userbot_id, list_session_owners, save_session_meta, session_path, ) from watch import register_watch_handlers from config import AVATARS_DIR def normalize_tg_phone(raw: str) -> str: digits = re.sub(r"\D", "", str(raw or "")) if len(digits) == 11 and digits[0] in ("7", "8"): digits = "7" + digits[1:] if len(digits) == 10: digits = "7" + digits if len(digits) != 11 or not digits.startswith("7"): raise ValueError("Укажите номер в формате +7…") return "+" + digits def _me_payload(me) -> Optional[dict]: if not me: return None return { "id": me.id, "first_name": me.first_name or "", "last_name": me.last_name or "", "username": me.username or "", "phone": me.phone or "", } def _friendly_error(exc: BaseException) -> str: if isinstance(exc, FloodWaitError): return f"Telegram просит подождать {exc.seconds} сек." if isinstance(exc, PhoneNumberInvalidError): return "Неверный номер телефона" if isinstance(exc, PhoneCodeInvalidError): return "Неверный код" if isinstance(exc, PhoneCodeExpiredError): return "Код устарел, запросите новый" if isinstance(exc, PasswordHashInvalidError): return "Неверный пароль 2FA" text = str(exc) or exc.__class__.__name__ if "database is locked" in text.lower(): return "Сессия занята. Перезапустите бота ловца." return text[:240] def _message_preview(message) -> str: if message is None: return "" text = (getattr(message, "message", None) or "").strip() if text: return " ".join(text.split())[:90] if getattr(message, "sticker", None): return "Стикер" if getattr(message, "photo", None): return "Фото" if getattr(message, "video", None) or getattr(message, "video_note", None): return "Видео" if getattr(message, "voice", None) or getattr(message, "audio", None): return "Аудио" if getattr(message, "document", None): return "Файл" return "" class OwnerRuntime: def __init__(self, owner_tg_id: int, api_id: int, api_hash: str) -> None: self.owner_tg_id = int(owner_tg_id) self.api_id = api_id self.api_hash = api_hash self.client: Optional[TelegramClient] = None self.status = "offline" self.phone: Optional[str] = None self.phone_code_hash: Optional[str] = None self.me: Optional[dict] = None self.error: Optional[str] = None self._lock = asyncio.Lock() self._watch_seen: set[tuple[int, int]] = set() self._task: Optional[asyncio.Task] = None def snapshot(self) -> dict: return { "status": self.status, "phone": self.phone, "error": self.error, "me": self.me, } def _persist(self) -> None: save_session_meta( self.owner_tg_id, phone=self.phone or (self.me or {}).get("phone"), me=self.me, status=self.status, ) async def _new_client(self) -> TelegramClient: if not self.api_id or not self.api_hash: raise RuntimeError("Не заданы API_ID / API_HASH в catcher/config.py") client = TelegramClient(session_path(self.owner_tg_id), self.api_id, self.api_hash) register_watch_handlers(client, self.owner_tg_id, self._watch_seen) return client async def ensure_client(self) -> TelegramClient: if self.client is None: self.client = await self._new_client() if not self.client.is_connected(): await self.client.connect() register_watch_handlers(self.client, self.owner_tg_id, self._watch_seen) return self.client async def _drop_client(self) -> None: if self.client is None: return try: if self.client.is_connected() and await self.client.is_user_authorized(): await self.client.log_out() elif self.client.is_connected(): await self.client.disconnect() except Exception: try: await self.client.disconnect() except Exception: pass self.client = None async def _ensure_unique_account(self) -> None: if not self.me or not self.me.get("id"): return other = find_owner_by_userbot_id(int(self.me["id"])) if other is not None and other != self.owner_tg_id: await self._drop_client() self.me = None self.phone = None self.phone_code_hash = None self.status = "logged_out" raise RuntimeError("Этот Telegram-аккаунт уже подключён у другого пользователя") async def boot(self) -> dict: async with self._lock: try: client = await self.ensure_client() if await client.is_user_authorized(): self.me = _me_payload(await client.get_me()) await self._ensure_unique_account() self.status = "online" self.error = None self.phone = None self.phone_code_hash = None else: self.status = "logged_out" self.me = None self.error = None except Exception as exc: self.status = "error" self.error = _friendly_error(exc) self.me = None self._persist() return self.snapshot() async def shutdown(self) -> None: async with self._lock: if self.client is not None: try: await self.client.disconnect() except Exception: pass self.client = None if self.status == "online": self.status = "offline" self._persist() async def request_code(self, phone_raw: str) -> dict: phone = normalize_tg_phone(phone_raw) async with self._lock: try: client = await self.ensure_client() if await client.is_user_authorized(): self.me = _me_payload(await client.get_me()) self.status = "online" self.error = None self._persist() return self.snapshot() sent = await client.send_code_request(phone) self.phone = phone self.phone_code_hash = sent.phone_code_hash self.status = "awaiting_code" self.error = None self.me = None except Exception as exc: self.status = "logged_out" self.error = _friendly_error(exc) self._persist() raise RuntimeError(self.error) from exc self._persist() return self.snapshot() async def submit_code(self, code_raw: str) -> dict: code = re.sub(r"\D", "", str(code_raw or "")) if len(code) < 5: raise ValueError("Введите код из Telegram") async with self._lock: if self.status not in ("awaiting_code", "awaiting_password"): raise RuntimeError("Сначала запросите код") if not self.phone: raise RuntimeError("Номер не задан") try: client = await self.ensure_client() await client.sign_in( phone=self.phone, code=code, phone_code_hash=self.phone_code_hash, ) self.me = _me_payload(await client.get_me()) await self._ensure_unique_account() self.status = "online" self.error = None self.phone_code_hash = None except SessionPasswordNeededError: self.status = "awaiting_password" self.error = None except Exception as exc: self.error = _friendly_error(exc) self._persist() raise RuntimeError(self.error) from exc self._persist() return self.snapshot() async def submit_password(self, password: str) -> dict: password = str(password or "") if not password: raise ValueError("Введите пароль 2FA") async with self._lock: if self.status != "awaiting_password": raise RuntimeError("Сейчас пароль не нужен") try: client = await self.ensure_client() await client.sign_in(password=password) self.me = _me_payload(await client.get_me()) await self._ensure_unique_account() self.status = "online" self.error = None self.phone_code_hash = None except Exception as exc: self.error = _friendly_error(exc) self._persist() raise RuntimeError(self.error) from exc self._persist() return self.snapshot() async def cancel_login(self) -> dict: async with self._lock: self.phone = None self.phone_code_hash = None self.error = None try: client = await self.ensure_client() if await client.is_user_authorized(): self.me = _me_payload(await client.get_me()) self.status = "online" else: self.me = None self.status = "logged_out" except Exception: self.me = None self.status = "logged_out" self._persist() return self.snapshot() async def logout(self) -> dict: async with self._lock: if self.client is not None: try: if self.client.is_connected() and await self.client.is_user_authorized(): await self.client.log_out() elif self.client.is_connected(): await self.client.disconnect() except sqlite3.OperationalError as exc: self.error = _friendly_error(exc) raise RuntimeError(self.error) from exc except Exception: try: await self.client.disconnect() except Exception: pass self.client = None self.me = None self.phone = None self.phone_code_hash = None self.error = None self.status = "logged_out" self._watch_seen.clear() delete_session_meta(self.owner_tg_id) path = Path(session_path(self.owner_tg_id) + ".session") journal = Path(session_path(self.owner_tg_id) + ".session-journal") for item in (path, journal): try: item.unlink(missing_ok=True) except TypeError: if item.exists(): item.unlink() except Exception: pass try: await self.ensure_client() except Exception as exc: self.status = "error" self.error = _friendly_error(exc) self._persist() return self.snapshot() async def list_dialogs(self, limit: int = 200) -> list[dict]: client = self.client if client is None or not client.is_connected() or self.status != "online": raise RuntimeError("Юзербот не в сети") items: list[dict] = [] async for dialog in client.iter_dialogs(limit=limit): if dialog.is_user: kind = "user" elif dialog.is_group: kind = "group" else: kind = "channel" entity = dialog.entity photo = getattr(entity, "photo", None) photo_id = "" has_photo = bool(photo) and "Empty" not in type(photo).__name__ if has_photo: photo_id = str(getattr(photo, "photo_id", None) or getattr(photo, "id", None) or "") items.append({ "id": dialog.id, "title": (dialog.name or "").strip() or str(dialog.id), "kind": kind, "username": (getattr(entity, "username", None) or "").strip(), "unread": int(getattr(dialog, "unread_count", 0) or 0), "preview": _message_preview(getattr(dialog, "message", None)), "has_photo": has_photo, "photo_id": photo_id, }) return items async def dialog_photo(self, chat_id: int) -> Optional[bytes]: client = self.client if client is None or not client.is_connected() or self.status != "online": raise RuntimeError("Юзербот не в сети") AVATARS_DIR.mkdir(parents=True, exist_ok=True) cache = AVATARS_DIR / f"{self.owner_tg_id}_{int(chat_id)}.jpg" if cache.is_file() and cache.stat().st_size > 0: age = time.time() - cache.stat().st_mtime if age < 15 * 60: return cache.read_bytes() try: data = await client.download_profile_photo(int(chat_id), file=bytes) except Exception: data = None if not data: if cache.is_file() and cache.stat().st_size > 0: return cache.read_bytes() return None try: cache.write_bytes(data) except Exception: pass return data async def self_photo(self) -> Optional[bytes]: if not self.me or not self.me.get("id"): raise RuntimeError("Юзербот не в сети") return await self.dialog_photo(int(self.me["id"])) class CatcherPool: def __init__(self, api_id: int, api_hash: str) -> None: self.api_id = api_id self.api_hash = api_hash self._sessions: dict[int, OwnerRuntime] = {} self._lock = asyncio.Lock() async def get(self, owner_tg_id: int) -> OwnerRuntime: owner_tg_id = int(owner_tg_id) async with self._lock: runtime = self._sessions.get(owner_tg_id) if runtime is None: runtime = OwnerRuntime(owner_tg_id, self.api_id, self.api_hash) self._sessions[owner_tg_id] = runtime return runtime async def boot_all(self) -> None: owners = list_session_owners() print(f"catcher: поднимаю {len(owners)} юзербот-сессий") for owner_id in owners: runtime = await self.get(owner_id) try: await runtime.boot() print(f"catcher: owner {owner_id} -> {runtime.status}") except Exception as exc: print(f"catcher: owner {owner_id} boot failed: {exc}") await asyncio.sleep(0.15) async def shutdown_all(self) -> None: async with self._lock: items = list(self._sessions.values()) await asyncio.gather(*[item.shutdown() for item in items], return_exceptions=True)