"""Панель OZPAY: API девайсов + статика мини-аппа. Отдельный процесс от notify_server.py. Проверки (check_balance / check_turnover / check_cards / full_check) и выпуск карты (add_card) идут в thread pool, чтобы не блокировать event loop на ADB. """ from __future__ import annotations import asyncio import json import queue import re import sqlite3 import sys import threading import time from collections import deque from pathlib import Path from typing import Optional _ROOT = Path(__file__).resolve().parent.parent if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) from fastapi import APIRouter, FastAPI, HTTPException, Query, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from server.auth import current_user_from_request from server.config import ADMIN_TG_IDS, DEVICE_CHAT_MAP, SSL_CERTFILE, SSL_KEYFILE from api.db_api import ( add_ip_ban, add_panel_worker, adjust_usd_balance, bind_device_owner, claim_free_device, count_free_slots, create_device, delete_device, ensure_client_lease, find_device_by_ip_port, get_app_user, get_device, get_slot_price, get_sms_chat_id, get_watch_settings, ip_covers_client, is_ip_banned, list_app_users, list_devices, list_devices_by_tg, list_ip_bans, list_lock_logout_due, list_panel_workers, parse_ip_ban_inputs, parse_sms_chat_id, refund_client_lease, release_device_owner, remove_ip_ban, remove_panel_worker, rename_device, save_watch_settings, set_slot_price, set_sms_chat_id, start_client_lease, update_card_flags, update_device, update_password, ) from server.main import ActionCancelled, add_card, add_device, cancel_login, capture_screen, check_balance, check_cards, check_login_state, check_turnover, full_check, logout_lk, press_device_back, probe_adb, reboot_device, screencap_png, set_action_cancel_hook from server.userbot_service import runtime as userbot_runtime from server.tg_send import SMS_CHAT_ERROR, SMS_TEST_TEXT, send_bot_message from utils.ssh_cmd import ( DEFAULT_PORT as SSH_DEFAULT_PORT, DEFAULT_USER as SSH_DEFAULT_USER, delete_host, get_host, rename_host, run_ssh_commands, upsert_host, ) WEBAPP_DIR = _ROOT / "app" PORT = 5001 CHECKERS = { "balance": check_balance, "turnover": check_turnover, "cards": check_cards, "all": full_check, } app = FastAPI(title="OZPAY Panel") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) api = APIRouter(prefix="/api") _device_locks: dict[str, asyncio.Lock] = {} _checking: set[str] = set() _cancel_flags: dict[str, threading.Event] = {} _tls = threading.local() _orig_sleep = time.sleep _log_lock = threading.Lock() _log_lines: deque[dict] = deque(maxlen=5000) _log_seq = 0 _log_partial = "" LOCK_LOGOUT_SECONDS = 8 * 3600 _lock_logout_task: Optional[asyncio.Task] = None def _skip_captured_log(line: str) -> bool: if " /api/logs" in line: return True if "/screen" in line and " /api/devices/" in line: return True return False def _ingest_log(text: str) -> None: global _log_partial, _log_seq if not text: return chunk = _log_partial + text parts = chunk.split("\n") _log_partial = parts.pop() if not parts: return with _log_lock: for line in parts: if _skip_captured_log(line): continue _log_seq += 1 _log_lines.append({"id": _log_seq, "text": line}) class _StdTee: def __init__(self, original): self._original = original def write(self, data): if data is None: return 0 if isinstance(data, bytes): try: text = data.decode("utf-8", "replace") except Exception: text = repr(data) else: text = str(data) try: self._original.write(text) except Exception: pass _ingest_log(text) return len(data) if not isinstance(data, int) else data def flush(self): try: self._original.flush() except Exception: pass def isatty(self): try: return self._original.isatty() except Exception: return False def __getattr__(self, name): return getattr(self._original, name) if not isinstance(sys.stdout, _StdTee): sys.stdout = _StdTee(sys.stdout) if not isinstance(sys.stderr, _StdTee): sys.stderr = _StdTee(sys.stderr) def _action_cancel_hook() -> None: device_id = getattr(_tls, "device_id", None) if not device_id: return flag = _cancel_flags.get(device_id) if flag is None or not flag.is_set(): return if not getattr(_tls, "cancel_logged", False): _tls.cancel_logged = True print(f"действие отменено ({device_id})") raise ActionCancelled("Действие отменено") def _interruptible_sleep(seconds) -> None: device_id = getattr(_tls, "device_id", None) if not device_id: _orig_sleep(seconds) return end = time.monotonic() + max(0.0, float(seconds or 0)) while True: _action_cancel_hook() remaining = end - time.monotonic() if remaining <= 0: return _orig_sleep(min(0.2, remaining)) time.sleep = _interruptible_sleep set_action_cancel_hook(_action_cancel_hook) def _begin_action(device_id: str) -> None: _checking.add(device_id) _cancel_flags[device_id] = threading.Event() def _end_action(device_id: str) -> None: _checking.discard(device_id) _cancel_flags.pop(device_id, None) def _call_device_action(device_id: str, fn, *args): _tls.device_id = device_id _tls.cancel_logged = False try: return fn(*args) finally: _tls.device_id = None # Активные сессии входа в ЛК (device_id -> LoginSession). _login_sessions: dict[str, "LoginSession"] = {} _ACTIVE_LOGIN_STATES = {"running", "awaiting_code", "verifying", "cancelling"} CODE_WAIT_TIMEOUT = 300.0 class LoginCancelled(Exception): """Пользователь отменил вход из панели.""" class LoginSession: """Интерактивная сессия входа: add_device выполняется в отдельном потоке и на шаге ввода кода блокируется, ожидая действие из мини-аппа (код или повторную отправку). Пока ждём код, поток сам опрашивает состояние кнопки 'Получить новый код' на устройстве и кладёт его в `resend_available`, чтобы кнопка в панели была активна ровно тогда же, когда она активна в Ozon.""" def __init__(self, device_id: str, number: str, password: str, user_id: Optional[int] = None): self.device_id = device_id self.number = number self.password = password self.user_id = user_id self.status = "running" # running | awaiting_code | verifying | done | error self.method: Optional[str] = None self.target: Optional[str] = None self.error: Optional[str] = None self.device: Optional[dict] = None self.resend_available = False self.cancel_requested = False self._device = None # ppadb device, выдаётся add_device на шаге кода self._action_q: "queue.Queue[tuple]" = queue.Queue() self.thread: Optional[threading.Thread] = None def _update_resend_available(self): if self._device is None: return try: from server.main import _get_new_code_button_enabled state = _get_new_code_button_enabled(self._device) if state is not None: self.resend_available = bool(state) except Exception: pass def _perform_resend(self): if self._device is None: return try: from server.main import detect_code_screen, dismiss_permission_dialog, wait_and_tap_get_new_code self.resend_available = False wait_and_tap_get_new_code(self._device, timeout=10.0) dismiss_permission_dialog(self._device) time.sleep(1.0) hint = detect_code_screen(self._device) or {} self.method = hint.get("method") self.target = hint.get("target") except Exception: pass def code_provider(self, ctx=None): ctx = ctx or {} self._device = ctx.get("device") hint = ctx.get("hint") if hint is None and ("method" in ctx or "target" in ctx): hint = ctx # совместимость: ctx уже является хинтом hint = hint or {} self.method = hint.get("method") self.target = hint.get("target") if self.cancel_requested: raise LoginCancelled() self.status = "awaiting_code" deadline = time.time() + CODE_WAIT_TIMEOUT while time.time() < deadline: self._update_resend_available() try: action, payload = self._action_q.get(timeout=2.0) except queue.Empty: continue if action == "cancel": raise LoginCancelled() if action == "code": self.status = "verifying" return payload if action == "resend": self._perform_resend() deadline = time.time() + CODE_WAIT_TIMEOUT raise RuntimeError("Код не был введён вовремя") def submit_code(self, code: str): self._action_q.put(("code", code)) def request_resend(self): self._action_q.put(("resend", None)) def request_cancel(self): """Пометить сессию как отменяемую и разбудить поток, ждущий код.""" self.cancel_requested = True self.status = "cancelling" self._action_q.put(("cancel", None)) def _end_failed_login(session: "LoginSession") -> None: if session.user_id: refund_client_lease(session.device_id) release_device_owner(session.device_id) def _run_login(session: LoginSession): try: add_device( session.device_id, session.number, session.password, code_provider=session.code_provider, press_get_new_code=False, ) session.device = serialize_device(_require_device(session.device_id), is_admin=False) session.status = "done" except LoginCancelled: session.status = "cancelling" try: cancel_login(session.device_id) except Exception as exc: # noqa: BLE001 — навигация назад не должна ронять поток print(f"cancel_login({session.device_id}) failed: {exc}") session.status = "cancelled" _end_failed_login(session) except Exception as exc: # noqa: BLE001 — прокидываем текст ошибки в UI if session.cancel_requested: try: cancel_login(session.device_id) except Exception as nav_exc: # noqa: BLE001 print(f"cancel_login({session.device_id}) failed: {nav_exc}") session.status = "cancelled" _end_failed_login(session) else: session.error = str(exc) session.status = "error" _end_failed_login(session) finally: _checking.discard(session.device_id) def _start_login_session(device_id: str, number: str, password: str, user_id: Optional[int] = None) -> "LoginSession": existing = _login_sessions.get(device_id) if existing and existing.status in _ACTIVE_LOGIN_STATES: raise HTTPException(status_code=409, detail="Вход уже выполняется") if device_id in _checking: raise HTTPException(status_code=409, detail="Девайс занят проверкой") session = LoginSession(device_id, number, password, user_id=user_id) _login_sessions[device_id] = session _checking.add(device_id) session.thread = threading.Thread(target=_run_login, args=(session,), daemon=True) session.thread.start() return session def _lock_for(device_id: str) -> asyncio.Lock: lock = _device_locks.get(device_id) if lock is None: lock = asyncio.Lock() _device_locks[device_id] = lock return lock def _to_number(value) -> float: if value is None or value == "": return 0.0 if isinstance(value, (int, float)): return float(value) text = str(value).replace("\xa0", " ").replace(" ", "").replace(",", ".") text = re.sub(r"[^\d.]", "", text) try: return float(text) if text else 0.0 except ValueError: return 0.0 def _format_card_number(number: str) -> str: digits = re.sub(r"\D", "", number or "") if not digits: return number or "" return " ".join(digits[i:i + 4] for i in range(0, len(digits), 4)) def _format_expiry(raw: str) -> str: text = (raw or "").strip() if not text: return "" if re.fullmatch(r"(0[1-9]|1[0-2])/\d{2}(?:\d{2})?", text): return text[:5] if len(text) > 5 else text digits = re.sub(r"\D", "", text) if len(digits) == 4: month = int(digits[:2]) if digits[:2].isdigit() else 0 if 1 <= month <= 12: return f"{digits[:2]}/{digits[2:]}" return "" def parse_card_flags(raw) -> dict: if not raw: return {} if isinstance(raw, dict): data = raw else: try: data = json.loads(raw) except (TypeError, ValueError, json.JSONDecodeError): return {} if not isinstance(data, dict): return {} flags = {} for key, value in data.items(): digits = re.sub(r"\D", "", str(key)) if not digits or not isinstance(value, dict): continue flags[digits] = { "beeline": bool(value.get("beeline")), "yapay": bool(value.get("yapay")), "mts": bool(value.get("mts")), } return flags def parse_cards(raw: Optional[str], flags_raw=None) -> list[dict]: """cards в БД: number/expiry/cvv, карты через ':'. Старый формат: number/last4/cvv.""" flags_map = parse_card_flags(flags_raw) if not raw: return [] cards = [] for chunk in str(raw).split(":"): chunk = chunk.strip() if not chunk: continue parts = chunk.split("/") number = parts[0] if parts else "" middle = parts[1] if len(parts) > 1 else "" cvv = parts[2] if len(parts) > 2 else "" number_digits = re.sub(r"\D", "", number) middle_digits = re.sub(r"\D", "", middle) if number_digits and middle_digits == number_digits[-4:]: expiry = "" else: expiry = _format_expiry(middle) flags = flags_map.get(number_digits) or {} cards.append({ "number": _format_card_number(number), "expiry": expiry, "cvv": cvv, "beeline": bool(flags.get("beeline")), "yapay": bool(flags.get("yapay")), "mts": bool(flags.get("mts")), }) return cards def serialize_device(row: dict, *, is_admin: bool = False, owners: Optional[dict] = None) -> dict: device_id = row.get("device") or "" ip = row.get("ip") port = row.get("port") linked = bool(row.get("number")) blocked = bool(row.get("blocked")) tg_id = row.get("tg_id") try: tg_id = int(tg_id) if tg_id not in (None, "", 0, "0") else None except (TypeError, ValueError): tg_id = None if device_id in _checking: status = "busy" elif not linked: status = "new" elif blocked: status = "blocked" elif ip and port: status = "online" else: status = "offline" host = get_host(device_id) or {} ssh_user = str(host.get("username") or "").strip() or SSH_DEFAULT_USER try: ssh_port = int(host.get("port") or SSH_DEFAULT_PORT) except (TypeError, ValueError): ssh_port = SSH_DEFAULT_PORT payload = { "id": device_id, "name": (row.get("name") or "").strip(), "number": row.get("number") or "", "ip": ip or "", "port": port if port not in (None, "") else "", "status": status, "linked": linked, "blocked": blocked, "checking": device_id in _checking, "balance": _to_number(row.get("balance")), "income": _to_number(row.get("income")), "outcome": _to_number(row.get("outcome")), "cards": parse_cards(row.get("cards"), row.get("card_flags")), "tg_id": tg_id, "free": (not linked) and tg_id is None, "ssh_user": ssh_user, "ssh_port": ssh_port, "leased": bool(row.get("leased_at")), "blocked_at": None, "lock_logout_at": None, } try: blocked_at = float(row.get("blocked_at")) if row.get("blocked_at") not in (None, "") else None except (TypeError, ValueError): blocked_at = None payload["blocked_at"] = blocked_at if blocked and blocked_at and row.get("leased_at"): payload["lock_logout_at"] = blocked_at + LOCK_LOGOUT_SECONDS login = _login_sessions.get(device_id) login_status = login.status if login is not None else None payload["login_status"] = login_status if login_status in _ACTIVE_LOGIN_STATES else None if is_admin and tg_id: if owners is None: owners = _owners_by_id() payload["owner"] = owners.get(tg_id) or { "id": tg_id, "username": "", "first_name": "", "is_admin": False, "is_worker": False, "balance_usd": 0, } if not is_admin: payload["ip"] = "" payload["port"] = "" payload["ssh_user"] = "" payload["ssh_port"] = "" return payload _CORS_HEADERS = "Accept, Content-Type, X-Telegram-Init-Data" def _pays_for_lease(user: dict) -> bool: return not bool(user.get("is_admin")) and not bool(user.get("is_worker")) def _user(request: Request) -> dict: user = getattr(request.state, "user", None) if not user: raise HTTPException(status_code=401, detail="Откройте панель из Telegram") return user def _require_admin(request: Request) -> dict: user = _user(request) if not user.get("is_admin"): raise HTTPException(status_code=403, detail="Только администратор") return user def _request_ip(request: Request) -> str: host = (request.client.host if request.client else "") or "" host = host.strip() if host.startswith("::ffff:"): host = host[7:] return host def _ip_bans_payload(request: Request) -> dict: return { "bans": list_ip_bans(), "client_ip": _request_ip(request), "ok": True, } def _is_staff(user: dict) -> bool: return bool(user.get("is_admin") or user.get("is_worker")) def _owns(row: dict, user: dict) -> bool: tg = row.get("tg_id") try: return tg not in (None, "", 0, "0") and int(tg) == int(user["id"]) except (TypeError, ValueError): return False def _is_free_slot(row: dict) -> bool: return (not row.get("number")) and row.get("tg_id") in (None, "", 0, "0") def _require_device_access(request: Request, device_id: str): user = _user(request) row = _require_device(device_id) if user.get("is_admin") or user.get("is_worker") or _owns(row, user): return user, row raise HTTPException(status_code=403, detail="Нет доступа к этому девайсу") def _require_owner_or_admin(request: Request, device_id: str): user = _user(request) row = _require_device(device_id) if user.get("is_admin") or _owns(row, user): return user, row raise HTTPException(status_code=403, detail="Нет доступа к этому девайсу") def _require_login_access(request: Request, device_id: str) -> dict: user = _user(request) session = _login_sessions.get(device_id) if session is not None and session.user_id and int(session.user_id) == int(user["id"]): return user _require_owner_or_admin(request, device_id) return user def _bind_if_free(device_id: str, user: dict) -> dict: row = _require_device(device_id) if _owns(row, user) or user.get("is_admin"): if _is_free_slot(row): bind_device_owner(device_id, user["id"]) row = _require_device(device_id) return row if _is_free_slot(row): bind_device_owner(device_id, user["id"]) return _require_device(device_id) raise HTTPException(status_code=403, detail="Слот уже занят") def _ser(row: dict, request: Request) -> dict: return serialize_device(row, is_admin=bool(_user(request).get("is_admin"))) def _me_payload(user: dict) -> dict: admin = bool(user.get("is_admin")) worker = bool(user.get("is_worker")) role = "admin" if admin else ("worker" if worker else "user") return { "id": user["id"], "is_admin": admin, "is_worker": worker, "role": role, "username": user.get("username") or "", "first_name": user.get("first_name") or "", "free_slots": count_free_slots(), "balance_usd": _to_number((get_app_user(user["id"]) or {}).get("usd_balance")), "slot_price_usd": get_slot_price(), } _DEVICE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$") SSH_COMMAND_TIMEOUT = 60.0 def _require_device(device_id: str) -> dict: row = get_device(device_id) if not row: raise HTTPException(status_code=404, detail=f"Устройство '{device_id}' не найдено") return row @app.middleware("http") async def log_requests(request: Request, call_next): origin = request.headers.get("origin") or "*" cors = { "Access-Control-Allow-Origin": origin, "Access-Control-Allow-Methods": "GET, POST, DELETE, OPTIONS", "Access-Control-Allow-Headers": _CORS_HEADERS, "Access-Control-Max-Age": "86400", "Vary": "Origin", } if request.method == "OPTIONS": return Response(status_code=204, headers=cors) path = request.url.path client_ip = _request_ip(request) if path != "/api/health" and client_ip and is_ip_banned(client_ip): return JSONResponse({"detail": "Доступ запрещён"}, status_code=403, headers=cors) if path.startswith("/api/") and path != "/api/health": try: request.state.user = current_user_from_request(request) except HTTPException as exc: return JSONResponse({"detail": exc.detail}, status_code=exc.status_code, headers=cors) response = await call_next(request) for key, value in cors.items(): response.headers[key] = value if path != "/api/logs" and not path.endswith("/screen"): print(f"{request.method} {path} -> {response.status_code}") return response @api.get("/health") async def health() -> dict: return {"status": "ok"} @api.get("/me") async def api_me(request: Request) -> dict: return {"me": _me_payload(_user(request))} def _sms_settings_payload(user: dict) -> dict: personal = int(user["id"]) custom = get_sms_chat_id(personal) return { "personal_id": personal, "sms_chat_id": custom, "effective_chat_id": custom if custom is not None else personal, } @api.get("/settings") async def api_settings(request: Request) -> dict: return _sms_settings_payload(_user(request)) @api.post("/settings/sms-chat") async def api_set_sms_chat(request: Request) -> dict: user = _user(request) personal = int(user["id"]) try: body = await request.json() except Exception: body = {} raw = body.get("sms_chat_id") if isinstance(body, dict) else None if raw in (None, ""): raw = body.get("chat_id") if isinstance(body, dict) else None text = str(raw or "").strip() if not text: set_sms_chat_id(personal, None) return {"ok": True, **_sms_settings_payload(user)} parsed = parse_sms_chat_id(text) if parsed is None: raise HTTPException(status_code=400, detail="Некорректный ID чата") if parsed == personal: set_sms_chat_id(personal, None) return {"ok": True, **_sms_settings_payload(user)} try: send_bot_message(parsed, SMS_TEST_TEXT) except ValueError: set_sms_chat_id(personal, None) raise HTTPException(status_code=400, detail=SMS_CHAT_ERROR) set_sms_chat_id(personal, parsed) return {"ok": True, **_sms_settings_payload(user)} def _rent_user_payload(row: dict, worker_ids: set[int] | None = None) -> dict: tg_id = int(row.get("tg_id") or 0) workers = worker_ids or set() is_admin = bool(row.get("is_admin")) return { "id": tg_id, "username": row.get("username") or "", "first_name": row.get("first_name") or "", "is_admin": is_admin, "is_worker": (not is_admin) and tg_id in workers, "balance_usd": _to_number(row.get("usd_balance")), } def _owners_by_id() -> dict[int, dict]: worker_ids = {int(item) for item in list_panel_workers()} owners: dict[int, dict] = {} for row in list_app_users(): try: tg_id = int(row.get("tg_id") or 0) except (TypeError, ValueError): continue if tg_id: owners[tg_id] = _rent_user_payload(row, worker_ids) return owners @api.get("/rent") async def api_rent_settings(request: Request) -> dict: user = _user(request) payload = {"slot_price_usd": get_slot_price(), "balance_usd": _to_number((get_app_user(user["id"]) or {}).get("usd_balance"))} if user.get("is_admin"): worker_ids = {int(item) for item in list_panel_workers()} payload["users"] = [_rent_user_payload(row, worker_ids) for row in list_app_users()] return payload @api.post("/rent/price") async def api_set_rent_price(request: Request) -> dict: _require_admin(request) try: body = await request.json() except Exception: body = {} raw = body.get("slot_price_usd") if isinstance(body, dict) else None if raw in (None, ""): raw = body.get("price") if isinstance(body, dict) else None try: price = float(str(raw).replace(",", ".").replace(" ", "")) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Укажите цену слота") try: saved = set_slot_price(price) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc return {"ok": True, "slot_price_usd": saved} @api.post("/rent/topup") async def api_rent_topup(request: Request) -> dict: admin = _require_admin(request) try: body = await request.json() except Exception: body = {} raw_id = body.get("tg_id") if isinstance(body, dict) else None if raw_id in (None, ""): raw_id = body.get("id") if isinstance(body, dict) else None try: tg_id = int(raw_id) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Укажите Telegram ID") if tg_id <= 0: raise HTTPException(status_code=400, detail="Укажите Telegram ID") raw_amount = body.get("amount") if isinstance(body, dict) else None try: amount = round(float(str(raw_amount).replace(",", ".").replace(" ", "")), 2) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Укажите сумму пополнения") if amount <= 0: raise HTTPException(status_code=400, detail="Сумма должна быть больше нуля") new_balance = adjust_usd_balance(tg_id, amount, reason="topup", admin_id=int(admin["id"])) worker_ids = {int(item) for item in list_panel_workers()} return { "ok": True, "tg_id": tg_id, "balance_usd": new_balance, "user": _rent_user_payload(get_app_user(tg_id) or {"tg_id": tg_id, "usd_balance": new_balance}, worker_ids), } @api.get("/workers") async def api_list_workers(request: Request) -> dict: _require_admin(request) return {"workers": list_panel_workers()} @api.post("/workers") async def api_add_worker(request: Request) -> dict: _require_admin(request) body = await request.json() raw = body.get("tg_id") if isinstance(body, dict) else None if raw in (None, ""): raw = body.get("id") if isinstance(body, dict) else None try: tg_id = int(raw) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Укажите Telegram ID") if tg_id <= 0: raise HTTPException(status_code=400, detail="Укажите Telegram ID") if tg_id in set(ADMIN_TG_IDS or ()): raise HTTPException(status_code=400, detail="Админ и так имеет доступ") if not add_panel_worker(tg_id): raise HTTPException(status_code=400, detail="Некорректный Telegram ID") return {"workers": list_panel_workers(), "ok": True} @api.delete("/workers/{tg_id}") async def api_delete_worker(tg_id: int, request: Request) -> dict: _require_admin(request) if not remove_panel_worker(tg_id): raise HTTPException(status_code=404, detail="Работник не найден") return {"workers": list_panel_workers(), "ok": True} def _userbot_http_error(exc: BaseException) -> HTTPException: if isinstance(exc, ValueError): return HTTPException(status_code=400, detail=str(exc)) return HTTPException(status_code=400, detail=str(exc) or "Ошибка userbot") @api.get("/userbot") async def api_userbot_status(request: Request) -> dict: _require_admin(request) return userbot_runtime.snapshot() @api.post("/userbot/login") async def api_userbot_login(request: Request) -> dict: _require_admin(request) body = await request.json() try: return await userbot_runtime.request_code(str(body.get("phone") or "")) except (ValueError, RuntimeError) as exc: raise _userbot_http_error(exc) from exc @api.post("/userbot/code") async def api_userbot_code(request: Request) -> dict: _require_admin(request) body = await request.json() try: return await userbot_runtime.submit_code(str(body.get("code") or "")) except (ValueError, RuntimeError) as exc: raise _userbot_http_error(exc) from exc @api.post("/userbot/password") async def api_userbot_password(request: Request) -> dict: _require_admin(request) body = await request.json() try: return await userbot_runtime.submit_password(str(body.get("password") or "")) except (ValueError, RuntimeError) as exc: raise _userbot_http_error(exc) from exc @api.post("/userbot/cancel") async def api_userbot_cancel(request: Request) -> dict: _require_admin(request) return await userbot_runtime.cancel_login() @api.post("/userbot/logout") async def api_userbot_logout(request: Request) -> dict: _require_admin(request) try: return await userbot_runtime.logout() except RuntimeError as exc: raise _userbot_http_error(exc) from exc def _watch_payload() -> dict: data = get_watch_settings() if data.get("alert_chat_id") is None: for value in DEVICE_CHAT_MAP.values(): if value is not None: try: data["alert_chat_id"] = int(value) break except (TypeError, ValueError): continue return data @api.get("/userbot/watch") async def api_userbot_watch(request: Request) -> dict: _require_admin(request) return _watch_payload() @api.post("/userbot/watch") async def api_userbot_watch_save(request: Request) -> dict: _require_admin(request) body = await request.json() if not isinstance(body, dict): raise HTTPException(status_code=400, detail="Некорректные настройки") save_watch_settings(body) return _watch_payload() @api.get("/userbot/dialogs") async def api_userbot_dialogs(request: Request) -> dict: _require_admin(request) try: dialogs = await userbot_runtime.list_dialogs() except RuntimeError as exc: raise _userbot_http_error(exc) from exc return {"dialogs": dialogs} @api.get("/devices") async def api_list_devices(request: Request) -> dict: user = _user(request) try: rows = list_devices() if _is_staff(user) else list_devices_by_tg(user["id"]) except Exception as exc: raise HTTPException(status_code=500, detail=f"Ошибка БД: {exc}") from exc owners = _owners_by_id() if user.get("is_admin") else None return { "devices": [serialize_device(row, is_admin=user.get("is_admin"), owners=owners) for row in rows], "me": _me_payload(user), } @api.post("/devices") async def api_create_device(request: Request) -> dict: _require_admin(request) body = await request.json() device_id = (body.get("id") or body.get("device") or body.get("name") or "").strip() if not device_id: raise HTTPException(status_code=400, detail="Укажите имя девайса") if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}", device_id): raise HTTPException(status_code=400, detail="Имя девайса: латиница, цифры, . _ -") ip = (str(body.get("ip") or "")).replace(",", ".").strip() if not ip: raise HTTPException(status_code=400, detail="Укажите IP") port_raw = body.get("port") if port_raw in (None, ""): raise HTTPException(status_code=400, detail="Укажите порт девайса") try: port = int(str(port_raw).strip()) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Порт девайса должен быть числом") password = body.get("password") if password is None: password = body.get("ssh_password") if password is None or str(password) == "": raise HTTPException(status_code=400, detail="Укажите пароль сервера") password = str(password) server_login = (str(body.get("server_login") or body.get("ssh_user") or "")).strip() or SSH_DEFAULT_USER server_port_raw = body.get("server_port", body.get("ssh_port")) if server_port_raw in (None, ""): server_port = SSH_DEFAULT_PORT else: try: server_port = int(str(server_port_raw).strip()) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Порт сервера должен быть числом") if not 1 <= server_port <= 65535: raise HTTPException(status_code=400, detail="Порт сервера: 1–65535") duplicate = find_device_by_ip_port(ip, port) if duplicate: raise HTTPException( status_code=409, detail=f"IP {ip}:{port} уже занят девайсом '{duplicate.get('device')}'", ) try: await asyncio.to_thread(probe_adb, ip, port) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc try: create_device(device_id, ip=ip, port=port) except sqlite3.IntegrityError: raise HTTPException(status_code=409, detail=f"Девайс '{device_id}' уже есть") except Exception as exc: raise HTTPException(status_code=500, detail=f"Не удалось создать девайс: {exc}") from exc try: upsert_host(device_id, ip, password, username=server_login, port=server_port) except Exception as exc: delete_device(device_id) raise HTTPException(status_code=500, detail=f"Не удалось сохранить SSH: {exc}") from exc return {"device": _ser(_require_device(device_id), request)} @api.post("/devices/{device_id}/update") async def api_update_device(device_id: str, request: Request) -> dict: _require_admin(request) row = _require_device(device_id) if device_id in _checking: raise HTTPException(status_code=409, detail="Дождитесь окончания проверки") body = await request.json() new_id = str(body.get("id") or body.get("device") or device_id).strip() if not new_id: raise HTTPException(status_code=400, detail="Укажите имя девайса") if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}", new_id): raise HTTPException(status_code=400, detail="Имя девайса: латиница, цифры, . _ -") ip = (str(body.get("ip") if "ip" in body else row.get("ip") or "")).replace(",", ".").strip() if not ip: raise HTTPException(status_code=400, detail="Укажите IP") port_raw = body.get("port") if "port" in body else None if port_raw in (None, ""): try: port = int(row.get("port")) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Укажите порт девайса") else: try: port = int(str(port_raw).strip()) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Порт девайса должен быть числом") password = body.get("password") if password is None: password = body.get("ssh_password") password = "" if password is None else str(password) host = get_host(device_id) server_login = (str(body.get("server_login") or body.get("ssh_user") or "")).strip() if not server_login: server_login = str((host or {}).get("username") or "").strip() or SSH_DEFAULT_USER server_port_raw = body.get("server_port", body.get("ssh_port")) if server_port_raw in (None, ""): try: server_port = int((host or {}).get("port") or SSH_DEFAULT_PORT) except (TypeError, ValueError): server_port = SSH_DEFAULT_PORT else: try: server_port = int(str(server_port_raw).strip()) except (TypeError, ValueError): raise HTTPException(status_code=400, detail="Порт сервера должен быть числом") if not 1 <= server_port <= 65535: raise HTTPException(status_code=400, detail="Порт сервера: 1–65535") if new_id != device_id and get_device(new_id): raise HTTPException(status_code=409, detail=f"Девайс '{new_id}' уже есть") duplicate = find_device_by_ip_port(ip, port) if duplicate and duplicate.get("device") not in {device_id, new_id}: raise HTTPException( status_code=409, detail=f"IP {ip}:{port} уже занят девайсом '{duplicate.get('device')}'", ) old_ip = str(row.get("ip") or "") try: old_port = int(row.get("port")) except (TypeError, ValueError): old_port = None if old_ip != ip or old_port != port: try: await asyncio.to_thread(probe_adb, ip, port) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc current_id = device_id if new_id != device_id: if not rename_device(device_id, new_id): raise HTTPException(status_code=500, detail="Не удалось переименовать девайс") try: rename_host(device_id, new_id) except ValueError as exc: rename_device(new_id, device_id) raise HTTPException(status_code=409, detail=str(exc)) from exc except Exception as exc: rename_device(new_id, device_id) raise HTTPException(status_code=500, detail=f"Не удалось переименовать SSH: {exc}") from exc if device_id in _device_locks: _device_locks[new_id] = _device_locks.pop(device_id) if device_id in _checking: _checking.discard(device_id) _checking.add(new_id) if device_id in _cancel_flags: _cancel_flags[new_id] = _cancel_flags.pop(device_id) current_id = new_id host = get_host(current_id) or host update_device(current_id, {"ip": ip, "port": port}) if not get_device(current_id): raise HTTPException(status_code=500, detail="Не удалось сохранить девайс") if host or password: try: upsert_host( current_id, ip, password if password else None, username=server_login, port=server_port, ) except Exception as exc: raise HTTPException(status_code=500, detail=f"Не удалось сохранить SSH: {exc}") from exc return {"device": _ser(_require_device(current_id), request)} @api.get("/devices/{device_id}") async def api_get_device(device_id: str, request: Request) -> dict: user, row = _require_device_access(request, device_id) return {"device": serialize_device(row, is_admin=user.get("is_admin"))} @api.delete("/devices/{device_id}") async def api_delete_device(device_id: str, request: Request) -> dict: _require_admin(request) _require_device(device_id) if device_id in _checking: raise HTTPException(status_code=409, detail="Дождитесь окончания проверки") if not delete_device(device_id): raise HTTPException(status_code=500, detail="Не удалось удалить девайс") try: delete_host(device_id) except Exception as exc: # noqa: BLE001 — отсутствие SSH-записи не должно ломать удаление print(f"delete_host({device_id}) failed: {exc}") return {"ok": True, "id": device_id} @api.post("/devices/{device_id}/logout") async def api_logout_device(device_id: str, request: Request) -> dict: user, row = _require_owner_or_admin(request, device_id) if device_id in _checking: raise HTTPException(status_code=409, detail="Девайс уже проверяется") if not row.get("number"): raise HTTPException(status_code=400, detail="ЛК не привязан") _begin_action(device_id) lock = _lock_for(device_id) cancelled = False try: async with lock: await asyncio.to_thread(_call_device_action, device_id, logout_lk, device_id) except ActionCancelled: cancelled = True except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc finally: _end_action(device_id) payload = {"device": serialize_device(_require_device(device_id), is_admin=user.get("is_admin"))} if cancelled: payload["cancelled"] = True return payload @api.post("/devices/{device_id}/kick") async def api_kick_session(device_id: str, request: Request) -> dict: _require_admin(request) row = _require_device(device_id) owners = _owners_by_id() session = _login_sessions.get(device_id) if session is not None and session.status in _ACTIVE_LOGIN_STATES: session.request_cancel() return { "ok": True, "action": "cancel", "device": serialize_device(_require_device(device_id), is_admin=True, owners=owners), } if row.get("number"): if device_id in _checking: raise HTTPException(status_code=409, detail="Девайс уже проверяется") _begin_action(device_id) lock = _lock_for(device_id) cancelled = False try: async with lock: await asyncio.to_thread(_call_device_action, device_id, logout_lk, device_id) except ActionCancelled: cancelled = True except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc finally: _end_action(device_id) payload = { "ok": True, "action": "logout", "device": serialize_device(_require_device(device_id), is_admin=True, owners=owners), } if cancelled: payload["cancelled"] = True return payload try: owner_id = int(row.get("tg_id") or 0) except (TypeError, ValueError): owner_id = 0 if owner_id: refund_client_lease(device_id) release_device_owner(device_id) return { "ok": True, "action": "release", "device": serialize_device(_require_device(device_id), is_admin=True, owners=owners), } raise HTTPException(status_code=400, detail="Нет активной сессии") @api.post("/devices/{device_id}/cards/add") async def api_add_card(device_id: str, request: Request) -> dict: user, _row = _require_owner_or_admin(request, device_id) if device_id in _checking: raise HTTPException(status_code=409, detail="Девайс уже проверяется") _begin_action(device_id) lock = _lock_for(device_id) result = None cancelled = False try: async with lock: result = await asyncio.to_thread(_call_device_action, device_id, add_card, device_id) except ActionCancelled: cancelled = True except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc finally: _end_action(device_id) device = serialize_device(_require_device(device_id), is_admin=user.get("is_admin")) if cancelled: return {"device": device, "cancelled": True} if not result or not result.get("success"): raise HTTPException(status_code=500, detail=(result or {}).get("message") or "Не удалось выпустить карту") return {"device": device, "message": result.get("message")} @api.post("/devices/{device_id}/cards/flags") async def api_update_card_flags(device_id: str, request: Request) -> dict: user, _row = _require_device_access(request, device_id) body = await request.json() number = re.sub(r"\D", "", str(body.get("number") or "")) if not number: raise HTTPException(status_code=400, detail="Укажите номер карты") row = get_device(device_id) flags = parse_card_flags(row.get("card_flags") if row else None) current = flags.get(number) or {"beeline": False, "yapay": False, "mts": False} if "beeline" in body: current["beeline"] = bool(body.get("beeline")) if "yapay" in body: current["yapay"] = bool(body.get("yapay")) if "mts" in body: current["mts"] = bool(body.get("mts")) if current.get("beeline") or current.get("yapay") or current.get("mts"): flags[number] = current else: flags.pop(number, None) if not update_card_flags(device_id, json.dumps(flags, ensure_ascii=False)): raise HTTPException(status_code=500, detail="Не удалось сохранить флаги карты") return {"device": serialize_device(_require_device(device_id), is_admin=user.get("is_admin"))} async def run_device_check(device_id: str, kind: str, *, is_admin: bool = True) -> dict: if kind not in CHECKERS: raise HTTPException(status_code=400, detail="Неизвестный тип проверки") if device_id in _checking: raise HTTPException(status_code=409, detail="Девайс уже проверяется") _begin_action(device_id) lock = _lock_for(device_id) cancelled = False try: async with lock: await asyncio.to_thread(_call_device_action, device_id, CHECKERS[kind], device_id) except ActionCancelled: cancelled = True except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc finally: _end_action(device_id) payload = {"device": serialize_device(_require_device(device_id), is_admin=is_admin)} if cancelled: payload["cancelled"] = True return payload @api.post("/devices/{device_id}/check/{kind}") async def api_check_device(device_id: str, kind: str, request: Request) -> dict: user, _row = _require_device_access(request, device_id) if user.get("is_worker") and not user.get("is_admin") and kind not in {"balance", "turnover"}: raise HTTPException(status_code=403, detail="Недостаточно прав") return await run_device_check(device_id, kind, is_admin=bool(user.get("is_admin"))) def _parse_login_credentials(body: dict) -> tuple[str, str]: number = re.sub(r"\D", "", str(body.get("number") or "")) password = re.sub(r"\D", "", str(body.get("password") or "")) if not number: raise HTTPException(status_code=400, detail="Укажите номер телефона") if not password: raise HTTPException(status_code=400, detail="Укажите пароль (код-пароль)") return number, password @api.post("/lk/login") async def api_lk_login(request: Request) -> dict: """Пользователь оплачивает слот (если нужно), занимает его и начинает вход в ЛК.""" user = _user(request) if user.get("is_worker") and not user.get("is_admin"): raise HTTPException(status_code=403, detail="Недостаточно прав") body = await request.json() number, password = _parse_login_credentials(body) paid = _pays_for_lease(user) if paid: try: row = start_client_lease(user["id"]) except ValueError as exc: status = 409 if "слот" in str(exc).lower() else 400 raise HTTPException(status_code=status, detail=str(exc)) from exc else: row = claim_free_device(user["id"]) if not row: raise HTTPException(status_code=409, detail="Нет свободных слотов") device_id = row["device"] try: session = _start_login_session(device_id, number, password, user_id=user["id"]) except HTTPException: if paid: refund_client_lease(device_id) release_device_owner(device_id) raise return { "status": session.status, "device_id": device_id, "device": serialize_device(_require_device(device_id), is_admin=user.get("is_admin")), "me": _me_payload(user), } @api.post("/devices/{device_id}/login") async def api_login_start(device_id: str, request: Request) -> dict: user = _user(request) if user.get("is_worker") and not user.get("is_admin"): raise HTTPException(status_code=403, detail="Недостаточно прав") if _pays_for_lease(user): try: ensure_client_lease(user["id"], device_id) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc _bind_if_free(device_id, user) body = await request.json() number, password = _parse_login_credentials(body) try: session = _start_login_session(device_id, number, password, user_id=user["id"]) except HTTPException: if _pays_for_lease(user): refund_client_lease(device_id) raise return {"status": session.status, "device_id": device_id, "me": _me_payload(user)} @api.get("/devices/{device_id}/login") async def api_login_status(device_id: str, request: Request) -> dict: _require_device_access(request, device_id) session = _login_sessions.get(device_id) if not session: return {"status": "idle"} payload = { "status": session.status, "method": session.method, "target": session.target, "resend_available": session.resend_available, "device_id": device_id, } if session.status == "error": payload["error"] = session.error if session.status == "done" and session.device: payload["device"] = session.device return payload @api.post("/devices/{device_id}/login/code") async def api_login_code(device_id: str, request: Request) -> dict: _require_device_access(request, device_id) session = _login_sessions.get(device_id) if not session or session.status != "awaiting_code": raise HTTPException(status_code=409, detail="Сейчас код не ожидается") body = await request.json() code = re.sub(r"\D", "", str(body.get("code") or "")) if len(code) != 6: raise HTTPException(status_code=400, detail="Код должен состоять из 6 цифр") session.submit_code(code) return {"status": "verifying"} @api.post("/devices/{device_id}/login/resend") async def api_login_resend(device_id: str, request: Request) -> dict: _require_device_access(request, device_id) session = _login_sessions.get(device_id) if not session or session.status != "awaiting_code": raise HTTPException(status_code=409, detail="Сейчас код не ожидается") if not session.resend_available: raise HTTPException(status_code=409, detail="Кнопка ещё не активна") session.request_resend() return {"status": "awaiting_code"} @api.post("/devices/{device_id}/login/cancel") async def api_login_cancel(device_id: str, request: Request) -> dict: _require_device_access(request, device_id) session = _login_sessions.get(device_id) if not session or session.status not in _ACTIVE_LOGIN_STATES: return {"status": "idle"} session.request_cancel() return {"status": session.status} def _normalize_lk_number(raw) -> str: number = re.sub(r"\D", "", str(raw or "")) if len(number) == 11 and number[0] in ("7", "8"): number = number[1:] return number @api.post("/devices/{device_id}/login/refresh") async def api_login_refresh(device_id: str, request: Request) -> dict: _require_device_access(request, device_id) if device_id in _checking: raise HTTPException(status_code=409, detail="Девайс занят") try: state = await asyncio.to_thread(check_login_state, device_id) except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc return state @api.post("/devices/{device_id}/login/claim") async def api_login_claim(device_id: str, request: Request) -> dict: """Сохранить номер и код-пароль, если вход в ЛК на устройстве уже есть.""" user = _user(request) row = _bind_if_free(device_id, user) if row.get("number"): raise HTTPException(status_code=409, detail="ЛК уже привязан") if device_id in _checking: raise HTTPException(status_code=409, detail="Девайс занят") body = await request.json() number = _normalize_lk_number(body.get("number")) password = re.sub(r"\D", "", str(body.get("password") or "")) if len(number) != 10: raise HTTPException(status_code=400, detail="Укажите номер телефона") if len(password) < 4: raise HTTPException(status_code=400, detail="Укажите пароль (код-пароль)") if not update_device(device_id, {"number": number, "password": password, "tg_id": user["id"]}): raise HTTPException(status_code=500, detail="Не удалось сохранить ЛК") return {"device": serialize_device(_require_device(device_id), is_admin=user.get("is_admin"))} @api.get("/devices/{device_id}/screen") async def api_device_screen(device_id: str, request: Request) -> Response: """Снять PNG-скриншот текущего экрана устройства. Во время активного входа переиспользуем живое ADB-соединение из сессии (если оно уже получено на шаге ввода кода), иначе подключаемся заново. Эндпоинт не блокируется на `_checking`, чтобы экран можно было смотреть прямо во время входа и проверок. """ _require_device_access(request, device_id) session = _login_sessions.get(device_id) try: if session is not None and session._device is not None and session.status in _ACTIVE_LOGIN_STATES: png = await asyncio.to_thread(screencap_png, session._device) else: png = await asyncio.to_thread(capture_screen, device_id) except Exception as exc: raise HTTPException(status_code=502, detail=f"Не удалось снять экран: {exc}") from exc return Response( content=png, media_type="image/png", headers={"Cache-Control": "no-store, max-age=0"}, ) @api.get("/logs") async def api_logs(request: Request, after: int = 0) -> dict: _require_admin(request) with _log_lock: items = [item for item in _log_lines if item["id"] > after] nxt = _log_lines[-1]["id"] if _log_lines else after return {"lines": [item["text"] for item in items], "next": nxt} @api.post("/devices/{device_id}/back") async def api_device_back(device_id: str, request: Request) -> dict: _require_device_access(request, device_id) if device_id in _checking: raise HTTPException(status_code=409, detail="Недоступно во время действия") try: await asyncio.to_thread(press_device_back, device_id) except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc return {"ok": True} @api.post("/devices/{device_id}/cancel") async def api_cancel_action(device_id: str, request: Request) -> dict: _require_device_access(request, device_id) session = _login_sessions.get(device_id) if session is not None and session.status in _ACTIVE_LOGIN_STATES: session.request_cancel() return {"ok": True, "kind": "login"} if device_id not in _checking: raise HTTPException(status_code=409, detail="Нет активного действия") flag = _cancel_flags.get(device_id) if flag is not None: flag.set() try: await asyncio.to_thread(press_device_back, device_id) except Exception as exc: print(f"cancel: press_back failed: {exc}") return {"ok": True, "kind": "action"} def _parse_ssh_commands(body: dict) -> list[str]: raw = body.get("commands") if raw is None: raw = body.get("command") if isinstance(raw, str): items = raw.splitlines() elif isinstance(raw, list): items = raw else: items = [] return [str(item).strip() for item in items if str(item).strip()] @api.post("/devices/{device_id}/ssh") async def api_ssh_command(device_id: str, request: Request) -> dict: _require_admin(request) _require_device(device_id) body = await request.json() commands = _parse_ssh_commands(body) if not commands: raise HTTPException(status_code=400, detail="Укажите команду") timeout = min(300.0, SSH_COMMAND_TIMEOUT * max(1, len(commands))) try: results = await asyncio.to_thread( run_ssh_commands, device_id, commands, timeout, False, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except RuntimeError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except Exception as exc: raise HTTPException(status_code=502, detail=str(exc)) from exc payload = {"results": results} if len(results) == 1: payload["stdout"] = results[0].get("stdout") or "" payload["stderr"] = results[0].get("stderr") or "" payload["exit_code"] = results[0].get("exit_code") return payload @api.post("/devices/{device_id}/reboot") async def api_reboot_device(device_id: str, request: Request) -> dict: user, _row = _require_device_access(request, device_id) if device_id in _checking: raise HTTPException(status_code=409, detail="Девайс уже проверяется") _begin_action(device_id) lock = _lock_for(device_id) cancelled = False try: async with lock: await asyncio.to_thread(_call_device_action, device_id, reboot_device, device_id) except ActionCancelled: cancelled = True except HTTPException: raise except Exception as exc: raise HTTPException(status_code=500, detail=str(exc)) from exc finally: _end_action(device_id) payload = {"device": serialize_device(_require_device(device_id), is_admin=user.get("is_admin")), "ok": True} if cancelled: payload["cancelled"] = True return payload @api.post("/devices/{device_id}/password") async def api_update_password(device_id: str, request: Request) -> dict: _require_owner_or_admin(request, device_id) body = await request.json() password = re.sub(r"\D", "", str(body.get("password") or "")) if len(password) < 4: raise HTTPException(status_code=400, detail="Код-пароль: минимум 4 цифры") if not update_password(device_id, password): raise HTTPException(status_code=500, detail="Не удалось сохранить пароль") return {"ok": True, "id": device_id} app.include_router(api) async def _auto_lock_logout(device_id: str) -> None: if not device_id or device_id in _checking: return _begin_action(device_id) lock = _lock_for(device_id) try: async with lock: await asyncio.to_thread(_call_device_action, device_id, logout_lk, device_id) print(f"auto lock-logout done: {device_id}") except Exception as exc: # noqa: BLE001 print(f"auto lock-logout {device_id}: {exc}") finally: _end_action(device_id) async def _lock_logout_loop() -> None: while True: await asyncio.sleep(60) try: due = await asyncio.to_thread(list_lock_logout_due, LOCK_LOGOUT_SECONDS) for row in due: await _auto_lock_logout(str(row.get("device") or "")) except asyncio.CancelledError: raise except Exception as exc: # noqa: BLE001 print(f"lock logout loop: {exc}") @app.on_event("startup") async def _userbot_startup() -> None: global _lock_logout_task await userbot_runtime.boot() _lock_logout_task = asyncio.create_task(_lock_logout_loop()) @app.on_event("shutdown") async def _userbot_shutdown() -> None: global _lock_logout_task task = _lock_logout_task _lock_logout_task = None if task: task.cancel() try: await task except asyncio.CancelledError: pass await userbot_runtime.shutdown() @app.get("/health") async def root_health() -> dict: return {"status": "ok"} @app.get("/") async def index() -> FileResponse: return FileResponse(WEBAPP_DIR / "index.html") app.mount("/", StaticFiles(directory=WEBAPP_DIR), name="webapp") if __name__ == "__main__": import uvicorn kwargs = {"host": "0.0.0.0", "port": PORT, "reload": False} cert = Path(SSL_CERTFILE) if SSL_CERTFILE else None key = Path(SSL_KEYFILE) if SSL_KEYFILE else None if cert and key and cert.exists() and key.exists(): kwargs["ssl_certfile"] = str(cert) kwargs["ssl_keyfile"] = str(key) uvicorn.run("server.panel_server:app", **kwargs)