"""HTTP-сервис: сессия Telegram Web через Playwright.""" from __future__ import annotations import asyncio import logging from contextlib import asynccontextmanager from typing import AsyncIterator from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse from pydantic import BaseModel, Field import config from client import TelegramClient logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) logger = logging.getLogger(__name__) telegram = TelegramClient() _watch_task: asyncio.Task[None] | None = None @asynccontextmanager async def lifespan(_app: FastAPI) -> AsyncIterator[None]: global _watch_task async def start_and_watch() -> None: try: await telegram.start() except Exception: logger.exception("Playwright не стартовал") return await telegram.watch_login() _watch_task = asyncio.create_task(start_and_watch()) yield if _watch_task is not None: _watch_task.cancel() try: await _watch_task except asyncio.CancelledError: pass _watch_task = None await telegram.stop() app = FastAPI(title="svt13-telegram", version="0.2.0", lifespan=lifespan) class LookupRequest(BaseModel): query: str = Field(..., min_length=1) action: str = Field(default="inn", min_length=1) class LookupResponse(BaseModel): success: bool message: str data: dict | None = None class PhoneLoginRequest(BaseModel): phone: str = Field(..., min_length=8) class CodeLoginRequest(BaseModel): code: str = Field(..., min_length=4, max_length=8) class PasswordLoginRequest(BaseModel): password: str = Field(..., min_length=1) class LoginResponse(BaseModel): success: bool message: str login_step: str logged_in: bool last_error: str | None = None @app.get("/health") async def health() -> dict[str, str]: return { "status": "ok", "bot": config.BOT_USERNAME, } @app.get("/ready") async def ready() -> dict[str, object]: logged_in = await telegram.refresh_login_state() latest = config.DATA_DIR / "latest.png" qr = config.DATA_DIR / "qr.png" step = telegram.login_step status = "ok" if logged_in else step return { "status": status, "login_step": step, "headless": config.HEADLESS, "playwright_ready": telegram.started, "logged_in": logged_in, "phone_configured": bool(config.PHONE), "last_error": telegram.last_error, "profile_dir": str(config.PROFILE_DIR), "storage_state": str(config.STORAGE_STATE), "storage_state_exists": config.STORAGE_STATE.is_file(), "latest_screenshot": latest.is_file(), "qr_screenshot": qr.is_file(), "bot_id": config.BOT_ID, "bot_username": config.BOT_USERNAME, } @app.get("/debug/screenshot") async def debug_screenshot() -> FileResponse: try: path = await telegram.capture_screenshot("latest.png") except RuntimeError as exc: raise HTTPException(status_code=503, detail=str(exc)) from exc except Exception as exc: logger.exception("Скрин не снят") raise HTTPException(status_code=500, detail="Не удалось снять скрин") from exc return FileResponse(path, media_type="image/png", filename="latest.png") @app.get("/debug/qr") async def debug_qr() -> FileResponse: path = config.DATA_DIR / "qr.png" if not path.is_file(): if not telegram.started: raise HTTPException(status_code=503, detail="Браузер ещё не запущен") raise HTTPException( status_code=404, detail="QR ещё нет. Подождите несколько секунд или откройте /debug/screenshot", ) return FileResponse(path, media_type="image/png", filename="qr.png") def _login_message(step: str) -> str: messages = { "logged_in": "Вход выполнен", "waiting_phone": "Введите номер: POST /login/phone {\"phone\":\"+7...\"}", "waiting_code": "Код отправлен в Telegram. POST /login/code {\"code\":\"12345\"}", "waiting_password": "Нужен пароль 2FA: POST /login/password {\"password\":\"...\"}", "waiting_qr": "Открыт QR. Можно POST /login/phone, чтобы перейти к номеру", } return messages.get(step, f"Текущий шаг входа: {step}") @app.post("/login/phone", response_model=LoginResponse) async def login_phone(body: PhoneLoginRequest) -> LoginResponse: if not telegram.started: raise HTTPException(status_code=503, detail="Браузер ещё не запущен") try: step = await telegram.submit_phone(body.phone) except RuntimeError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: logger.exception("Вход по номеру не выполнен") raise HTTPException(status_code=500, detail=str(exc)) from exc return LoginResponse( success=step in {"waiting_code", "waiting_password", "logged_in"}, message=_login_message(step), login_step=step, logged_in=telegram.logged_in, last_error=telegram.last_error, ) @app.post("/login/code", response_model=LoginResponse) async def login_code(body: CodeLoginRequest) -> LoginResponse: if not telegram.started: raise HTTPException(status_code=503, detail="Браузер ещё не запущен") try: step = await telegram.submit_code(body.code) except RuntimeError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: logger.exception("Код входа не принят") raise HTTPException(status_code=500, detail=str(exc)) from exc return LoginResponse( success=step in {"waiting_password", "logged_in"}, message=_login_message(step), login_step=step, logged_in=telegram.logged_in, last_error=telegram.last_error, ) @app.post("/login/password", response_model=LoginResponse) async def login_password(body: PasswordLoginRequest) -> LoginResponse: if not telegram.started: raise HTTPException(status_code=503, detail="Браузер ещё не запущен") try: step = await telegram.submit_password(body.password) except RuntimeError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: logger.exception("Пароль 2FA не принят") raise HTTPException(status_code=500, detail=str(exc)) from exc return LoginResponse( success=step == "logged_in", message=_login_message(step), login_step=step, logged_in=telegram.logged_in, last_error=telegram.last_error, ) @app.post("/lookup", response_model=LookupResponse) async def lookup(body: LookupRequest) -> LookupResponse: if not telegram.started: return LookupResponse( success=False, message="Браузер ещё запускается. Повторите запрос через несколько секунд.", ) if not await telegram.refresh_login_state(): return LookupResponse( success=False, message="Нужна авторизация: POST /login/phone и POST /login/code " "(скрин: GET /debug/screenshot)", data={"query": body.query, "action": body.action}, ) try: data = await telegram.lookup(body.query, body.action) except ValueError as exc: return LookupResponse(success=False, message=str(exc), data={"query": body.query, "action": body.action}) except Exception as exc: logger.exception("Lookup не выполнен") return LookupResponse( success=False, message=f"Не удалось получить отчёт: {exc}. Смотрите telegram/data/error.png", data={"query": body.query, "action": body.action}, ) return LookupResponse( success=True, message="Ссылка на отчёт получена", data=data, )