from __future__ import annotations import asyncio import logging import os import re import time from pathlib import Path from playwright.async_api import BrowserContext, Page, Playwright, async_playwright import config logger = logging.getLogger(__name__) CHROME_ARGS = [ "--no-sandbox", "--disable-dev-shm-usage", "--disable-gpu", "--disable-blink-features=AutomationControlled", # На хосте IPv6 до Telegram часто мёртвый — не зависаем на AAAA. "--disable-ipv6", ] REPORT_URL_RE = re.compile(r"https://[^\s<>\"']+") LINK_PROMPT_RE = re.compile(r"Do you want to open|хотите открыть", re.I) QR_PAGE_RE = re.compile(r"Log in to Telegram by QR|Войти в Telegram по QR") PHONE_LOGIN_RE = re.compile( r"LOG IN BY PHONE NUMBER|Log in by phone number|Вход по номеру телефона", re.I, ) NEXT_BTN_RE = re.compile(r"^(Next|Далее)$", re.I) PLEASE_WAIT_RE = re.compile(r"PLEASE WAIT|Пожалуйста, подождите|Подождите", re.I) PHONE_HEADING_RE = re.compile( r"Your Phone Number|Enter your phone number|номер телефона|Phone Number", re.I, ) CODE_HEADING_RE = re.compile( r"Enter code|We have sent|код|Code", re.I, ) PASSWORD_HEADING_RE = re.compile( r"Two-Step Verification|Cloud Password|Облачный пароль|Введите пароль", re.I, ) REPORT_BUTTON_RE = re.compile(r"Открыть полный отчет") MODAL_CANCEL_RE = re.compile(r"^(Cancel|Отмена)$") DIRECTION_TEXT = "Выберите направление для поиска" REPORT_HINT_TEXT = "Открыть полный отчет" UPDATING_RE = re.compile(r"^updating\.?$|^Connecting|^Подключение", re.I) class TelegramClient: """Один браузер на процесс: профиль Chrome + скрины QR до логина.""" def __init__(self) -> None: self.last_error: str | None = None self._playwright: Playwright | None = None self._context: BrowserContext | None = None self._page: Page | None = None self._lock = asyncio.Lock() self._started = False self._logged_in = False self._login_step = "not_started" @property def started(self) -> bool: return self._started and self._page is not None @property def logged_in(self) -> bool: return self._logged_in @property def login_step(self) -> str: return self._login_step async def start(self) -> None: if self.started: return config.PROFILE_DIR.mkdir(parents=True, exist_ok=True) config.DATA_DIR.mkdir(parents=True, exist_ok=True) logger.info( "Запуск Chromium (headless=%s, proxy=%s, profile=%s)", config.HEADLESS, config.PROXY or "off", config.PROFILE_DIR, ) os.environ.setdefault("PLAYWRIGHT_CHROMIUM_USE_HEADLESS_SHELL", "0") self._playwright = await async_playwright().start() chrome_args = list(CHROME_ARGS) if config.HEADLESS: chrome_args.append("--headless=new") launch_kwargs: dict[str, object] = { "user_data_dir": str(config.PROFILE_DIR), "headless": config.HEADLESS, "args": chrome_args, "viewport": {"width": 1280, "height": 800}, "locale": "ru-RU", "timezone_id": "Europe/Moscow", } if config.PROXY: launch_kwargs["proxy"] = {"server": config.PROXY} self._context = await self._playwright.chromium.launch_persistent_context(**launch_kwargs) self._page = self._context.pages[0] if self._context.pages else await self._context.new_page() self._page.set_default_timeout(15_000) self.last_error = None self._started = True try: await self._page.goto( config.WEB_URL, wait_until="domcontentloaded", timeout=config.NAV_TIMEOUT_MS, ) await self._wait_auth_screen() except Exception as exc: self.last_error = str(exc) logger.exception("Не удалось открыть Telegram Web") await self._safe_screenshot("error.png") raise await self._refresh_login_state() await self._safe_screenshot("latest.png") if self._logged_in: await self._open_bot_chat() await self._save_storage_state() logger.info("Сессия Telegram уже активна") return try: await self._open_phone_form() await self._ensure_country_russia() if config.PHONE: await self._submit_phone_number(config.PHONE) if self._login_step == "waiting_password" and config.TWO_FA_PASSWORD: await self._submit_cloud_password(config.TWO_FA_PASSWORD) await self._refresh_login_state() await self._safe_screenshot("latest.png") except Exception as exc: self.last_error = str(exc) logger.exception("Не удалось открыть вход по номеру") await self._save_qr() logger.info("Ожидание входа Telegram, шаг=%s", self._login_step) async def stop(self) -> None: self._started = False self._logged_in = False if self._context is not None: try: await self._context.close() except Exception: logger.exception("Ошибка закрытия браузера") self._context = None self._page = None if self._playwright is not None: await self._playwright.stop() self._playwright = None async def watch_login(self) -> None: """Пока нет логина — обновляет скрин формы; после логина — контроль сессии.""" while True: try: if not self.started: await asyncio.sleep(2) continue async with self._lock: was_logged_in = self._logged_in await self._refresh_login_state() await self._screenshot("latest.png") if self._logged_in: if not was_logged_in: logger.info("Вход подтверждён, сессия сохранена") await self._save_storage_state() await self._open_bot_chat() await self._screenshot("latest.png") elif self._login_step == "waiting_qr": await self._save_qr() delay = ( config.LOGGED_IN_POLL_SECONDS if self._logged_in else config.QR_POLL_SECONDS ) await asyncio.sleep(delay) except asyncio.CancelledError: raise except Exception as exc: self.last_error = str(exc) logger.exception("Сбой цикла логина") await self._safe_screenshot("error.png") await asyncio.sleep(5) async def capture_screenshot(self, name: str = "latest.png") -> Path: async with self._lock: if not self.started or self._page is None: raise RuntimeError("Браузер ещё не запущен") return await self._screenshot(name) async def screenshot_path(self, name: str) -> Path: return config.DATA_DIR / name async def lookup(self, query: str, action: str = "inn") -> dict[str, object]: """ИНН в чат бота → направление → ссылка на полный отчёт.""" button_name = config.ACTION_BUTTONS.get(action) if button_name is None: known = ", ".join(sorted(config.ACTION_BUTTONS)) raise ValueError(f"Неизвестное действие {action!r}. Доступно: {known}") async with self._lock: if not self.started or self._page is None: raise RuntimeError("Браузер ещё не запущен") await self._refresh_login_state() if not self._logged_in: raise RuntimeError( "Нет сессии Telegram: войдите по номеру (POST /login/phone и /login/code)" ) try: await self._dismiss_link_modal() await self._ensure_bot_chat() await self._wait_connection_ready() before_bot_id = await self._newest_message_id(own=False) before_own_id = await self._newest_message_id(own=True) await self._ensure_query_delivered(query, before_own_id, before_bot_id) await self._screenshot("lookup-01-sent.png") direction_id = await self._wait_new_bot_message(DIRECTION_TEXT, before_bot_id) await self._screenshot("lookup-02-direction.png") await self._click_in_message(direction_id, button_name) report_id = await self._wait_new_bot_message(REPORT_HINT_TEXT, direction_id) await self._screenshot("lookup-03-report.png") last_text = await self._message_text(report_id) from_text = REPORT_URL_RE.search(last_text) report_url = from_text.group(0).rstrip(").,;") if from_text else "" if not report_url.startswith("http"): await self._scroll_message_into_view(report_id) await self._click_in_message(report_id, REPORT_BUTTON_RE) report_url = await self._grab_report_url() await self._screenshot("lookup-04-link.png") await self._dismiss_link_modal() await self._screenshot("latest.png") except Exception: await self._safe_screenshot("error.png") raise logger.info("Ссылка на отчёт получена") return { "query": query, "action": action, "button": button_name, "report_url": report_url, "preview_text": last_text, } async def _send_chat_text(self, text: str) -> None: if self._page is None: raise RuntimeError("Нет открытой страницы") box = self._page.locator("#editable-message-text") await box.wait_for(state="visible", timeout=20_000) await box.click() await self._page.keyboard.press("Control+A") await self._page.keyboard.press("Backspace") await box.press_sequentially(text, delay=30) await asyncio.sleep(0.4) send = self._page.locator("button.main-button") if await send.count() > 0: await send.click() else: await self._page.keyboard.press("Enter") async def _is_updating(self) -> bool: page = self._require_page() return await self._is_visible(page.get_by_text(UPDATING_RE)) async def _wait_connection_ready(self, timeout_s: float = 8) -> None: deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: if not await self._is_updating(): return await asyncio.sleep(0.5) logger.warning("Заголовок всё ещё updating — сообщения при этом могут уходить") async def _message_is_pending(self, message_id: str) -> bool: loc = await self._message_locator(message_id) # Галочки тоже живут в .MessageOutgoingStatus — его нельзя считать pending. if await self._is_visible( loc.locator( ".icon-message-succeeded, .icon-checks, .icon-message-read, " "[class*='succeeded'], [class*='message-read']" ) ): return False return await self._is_visible( loc.locator(".icon-message-pending, .icon-clock, [class*='message-pending']") ) async def _direction_message_id(self, before_bot_id: int) -> str | None: newest = await self._newest_message_id(own=False) if newest <= before_bot_id: return None body = await self._message_text(str(newest)) if DIRECTION_TEXT in body: return str(newest) return None async def _wait_own_delivered(self, message_id: int, before_bot_id: int = 0, timeout_s: float = 20) -> None: deadline = time.monotonic() + timeout_s while time.monotonic() < deadline: if not await self._message_is_pending(str(message_id)): return if await self._direction_message_id(before_bot_id) is not None: logger.info("Бот ответил — исходящее уже доставлено") return await asyncio.sleep(0.6) raise TimeoutError("Сообщение не ушло (часы у исходящего)") async def _ensure_query_delivered(self, query: str, before_own_id: int, before_bot_id: int = 0) -> int: await self._send_chat_text(query) msg_id = await self._wait_own_message(query, before_own_id) try: await self._wait_own_delivered(msg_id, before_bot_id) return msg_id except TimeoutError: if await self._direction_message_id(before_bot_id) is not None: logger.info("Таймаут часов, но бот уже прислал направление") return msg_id logger.warning("Исходящее зависло, перезагружаю страницу и повторяю отправку") await self._reload_auth() await self._ensure_bot_chat() await self._wait_connection_ready() newest = await self._newest_message_id(own=True) if newest > before_own_id: body = await self._message_text(str(newest)) if query in body and not await self._message_is_pending(str(newest)): return newest if await self._direction_message_id(before_bot_id) is not None: return msg_id await self._send_chat_text(query) msg_id = await self._wait_own_message(query, before_own_id) await self._wait_own_delivered(msg_id, before_bot_id) return msg_id async def _newest_message_id(self, *, own: bool) -> int: if self._page is None: return 0 selector = ".Message.own[data-message-id]" if own else ".Message[data-message-id]:not(.own)" loc = self._page.locator(selector) count = await loc.count() best = 0 for index in range(count): raw = await loc.nth(index).get_attribute("data-message-id") if raw and raw.isdigit(): best = max(best, int(raw)) return best async def _wait_own_message(self, text: str, prev_id: int) -> int: return await self._wait_newest_message(own=True, prev_id=prev_id, text_fragment=text) async def _wait_new_bot_message(self, text_fragment: str, prev_id: int | str | None) -> str: prev = int(prev_id) if str(prev_id or "").isdigit() else 0 newest = await self._wait_newest_message(own=False, prev_id=prev, text_fragment=text_fragment) return str(newest) async def _wait_newest_message(self, *, own: bool, prev_id: int, text_fragment: str) -> int: if self._page is None: raise RuntimeError("Нет открытой страницы") deadline = time.monotonic() + (config.LOOKUP_TIMEOUT_MS / 1000) last_seen = prev_id while time.monotonic() < deadline: newest = await self._newest_message_id(own=own) if newest > prev_id: last_seen = newest body = await self._message_text(str(newest)) if text_fragment in body: return newest await asyncio.sleep(0.4) kind = "своё" if own else "бота" raise TimeoutError( f"Не дождались {kind} сообщения с текстом {text_fragment!r} " f"(prev_id={prev_id}, last_seen={last_seen})" ) async def _message_locator(self, message_id: str): if self._page is None: raise RuntimeError("Нет открытой страницы") return self._page.locator(f'.Message[data-message-id="{message_id}"]') async def _message_text(self, message_id: str) -> str: loc = await self._message_locator(message_id) if await loc.count() == 0: return "" return (await loc.inner_text()).strip() async def _scroll_message_into_view(self, message_id: str) -> None: loc = await self._message_locator(message_id) try: await loc.evaluate("el => el.scrollIntoView({block: 'center', inline: 'nearest'})") except Exception: await loc.scroll_into_view_if_needed() await asyncio.sleep(0.3) async def _click_in_message(self, message_id: str, name: str | re.Pattern[str]) -> None: loc = await self._message_locator(message_id) await self._scroll_message_into_view(message_id) btn = loc.get_by_role("button", name=name) if await btn.count() == 0: btn = loc.get_by_text(name) if await btn.count() == 0: raise RuntimeError(f"В сообщении {message_id} нет кнопки {name!r}") target = btn.last try: await target.evaluate("el => el.scrollIntoView({block: 'center'})") except Exception: pass await target.click(timeout=15_000, force=True) async def _grab_report_url(self, fallback: str = "") -> str: """URL из диалога Telegram, без перехода на страницу отчёта (капча).""" page = self._require_page() prompt = page.get_by_text(LINK_PROMPT_RE) try: await prompt.first.wait_for(timeout=12_000) modal_text = await prompt.first.inner_text() match = REPORT_URL_RE.search(modal_text) if match is not None: return match.group(0).rstrip(").,;") parent = prompt.first.locator("xpath=ancestor::div[1]") if await parent.count() > 0: block = await parent.first.inner_text() match = REPORT_URL_RE.search(block) if match is not None: return match.group(0).rstrip(").,;") except Exception: logger.warning("Диалог ссылки не появился, ищем URL в чате") hrefs = await page.locator('a[href^="http"]').evaluate_all( "els => els.map(e => e.href)" ) for href in reversed(hrefs): if not isinstance(href, str): continue if "web.telegram.org" in href or "t.me/" in href: continue if href.startswith("http"): return href.rstrip(").,;") if fallback.startswith("http"): return fallback.rstrip(").,;") raise RuntimeError("Не удалось прочитать URL отчёта из диалога") async def _ensure_bot_chat(self) -> None: if self._page is None: return composer = self._page.locator("#editable-message-text") try: if await composer.count() > 0 and await composer.first.is_visible(): return except Exception: pass await self._open_bot_chat() async def _dismiss_link_modal(self) -> None: if self._page is None: return cancel = self._page.get_by_role("button", name=MODAL_CANCEL_RE) try: if await cancel.count() > 0 and await cancel.first.is_visible(): await cancel.first.click(timeout=3_000) except Exception: logger.debug("Модальное окно ссылки уже закрыто") async def refresh_login_state(self) -> bool: async with self._lock: if not self.started or self._page is None: return False await self._refresh_login_state() return self._logged_in async def submit_phone(self, phone: str) -> str: async with self._lock: self._require_page() await self._open_phone_form() await self._submit_phone_number(phone) await self._screenshot("latest.png") return self._login_step async def submit_code(self, code: str) -> str: async with self._lock: self._require_page() await self._refresh_login_state() if self._login_step != "waiting_code": await self._recover_code_screen() await self._submit_login_code(code) if self._login_step == "waiting_password" and config.TWO_FA_PASSWORD: await self._submit_cloud_password(config.TWO_FA_PASSWORD) if self._logged_in: await self._save_storage_state() await self._open_bot_chat() await self._screenshot("latest.png") return self._login_step async def submit_password(self, password: str) -> str: async with self._lock: self._require_page() await self._submit_cloud_password(password) if self._logged_in: await self._save_storage_state() await self._open_bot_chat() await self._screenshot("latest.png") return self._login_step def _require_page(self) -> Page: if not self.started or self._page is None: raise RuntimeError("Браузер ещё не запущен") return self._page async def _refresh_login_state(self) -> None: if self._page is None: self._logged_in = False self._login_step = "not_started" return try: self._login_step = await self._detect_login_step() self._logged_in = self._login_step == "logged_in" except Exception: self._logged_in = False self._login_step = "unknown" async def _detect_login_step(self) -> str: page = self._require_page() # Сначала чат: поля входа и слова «пароль»/телефон в переписке не должны # перекрывать уже открытую сессию. composer = page.locator( "#editable-message-text, #editable-message-input, " "#MiddleColumn [contenteditable='true']" ) if await self._is_visible(composer): return "logged_in" left = page.locator("#LeftColumn, #column-left, #MiddleColumn") if await self._is_visible(left): return "logged_in" if await self._is_visible(page.locator("#sign-in-password")): return "waiting_password" if await self._is_visible(page.get_by_text(PASSWORD_HEADING_RE)): return "waiting_password" if await self._is_visible(page.locator("#sign-in-code")): return "waiting_code" code_input = page.locator('input[autocomplete="one-time-code"]') if await self._is_visible(code_input) and not await self._is_visible( page.locator("#sign-in-phone-number") ): return "waiting_code" if await self._is_visible(page.locator("#sign-in-phone-number")): return "waiting_phone" if await self._is_visible(page.get_by_text(PHONE_HEADING_RE)) and not await self._is_visible( page.get_by_text(QR_PAGE_RE) ): return "waiting_phone" if await self._is_visible(page.get_by_text(QR_PAGE_RE)): return "waiting_qr" if await self._is_visible(page.get_by_text(PHONE_LOGIN_RE)): return "waiting_qr" return "unknown" async def _is_visible(self, locator) -> bool: try: count = await locator.count() for index in range(min(count, 8)): if await locator.nth(index).is_visible(): return True except Exception: return False return False async def _wait_auth_screen(self) -> None: page = self._require_page() deadline = time.monotonic() + 30 while time.monotonic() < deadline: await self._refresh_login_state() if self._login_step in { "waiting_qr", "waiting_phone", "waiting_code", "waiting_password", "logged_in", }: return if await self._phone_login_button() is not None: return await asyncio.sleep(0.4) async def _phone_login_button(self): page = self._require_page() locators = [ page.get_by_role("button", name=PHONE_LOGIN_RE), page.get_by_role("link", name=PHONE_LOGIN_RE), page.get_by_text(PHONE_LOGIN_RE), page.locator("button, a, div, span").filter(has_text=PHONE_LOGIN_RE), ] for loc in locators: try: count = await loc.count() for index in range(count): item = loc.nth(index) if await item.is_visible(): return item except Exception: continue return None async def _open_phone_form(self) -> None: await self._wait_auth_screen() await self._refresh_login_state() if self._login_step in {"waiting_phone", "waiting_code", "waiting_password", "logged_in"}: return # Ссылки под QR появляются чуть позже заголовка. deadline = time.monotonic() + 25 button = None while time.monotonic() < deadline: button = await self._phone_login_button() if button is not None: break await asyncio.sleep(0.4) if button is None: await self._screenshot("error.png") raise RuntimeError("На экране нет ссылки «Log in by phone number»") await button.click(force=True) deadline = time.monotonic() + 20 while time.monotonic() < deadline: await self._refresh_login_state() if self._login_step == "waiting_phone": self.last_error = None await self._ensure_country_russia() return await asyncio.sleep(0.3) await self._screenshot("error.png") raise RuntimeError("Форма номера телефона не открылась") def _normalize_phone(self, raw: str) -> str: value = re.sub(r"[^\d+]", "", raw.strip()) if value.startswith("8") and len(value) == 11: value = "+7" + value[1:] elif value.startswith("7") and len(value) == 11: value = "+" + value elif not value.startswith("+"): value = "+" + value return value async def _country_field_value(self) -> str: page = self._require_page() field = await self._first_visible( page.get_by_label(re.compile(r"^Country$|^Страна$", re.I)), page.locator("#sign-in-phone-code"), page.locator('input[autoComplete="off"]').nth(0), ) if field is None: return "" try: return (await field.input_value()).strip() except Exception: try: return (await field.inner_text()).strip() except Exception: return "" async def _ensure_country_russia(self) -> None: """Прокси даёт гео США; для российского номера нужна страна +7.""" page = self._require_page() current = await self._country_field_value() if re.search(r"Russia|Росси", current, re.I): return country = await self._first_visible( page.get_by_label(re.compile(r"^Country$|^Страна$", re.I)), page.locator("#sign-in-phone-code"), ) if country is None: raise RuntimeError("Поле страны не найдено") await country.click() await asyncio.sleep(0.3) await page.keyboard.press("Control+A") await page.keyboard.press("Backspace") await country.press_sequentially("Russia", delay=50) await asyncio.sleep(0.6) menu = page.locator(".Menu.CountryCodeInput, .CountryCodeInput") option = menu.get_by_text(re.compile(r"Russian Federation")) if await option.count() == 0: option = page.get_by_text(re.compile(r"^Russian Federation$")) if await option.count() == 0: await self._screenshot("error.png") raise RuntimeError("В списке стран нет Russian Federation") await option.first.click(force=True, timeout=8_000) await asyncio.sleep(0.5) current = await self._country_field_value() if not re.search(r"Russia|Росси", current, re.I): await page.keyboard.press("Enter") await asyncio.sleep(0.4) current = await self._country_field_value() if not re.search(r"Russia|Росси", current, re.I): await self._screenshot("error.png") raise RuntimeError(f"Не удалось выбрать Россию, сейчас: {current or 'пусто'}") logger.info("Страна входа: Russian Federation (+7)") await self._safe_screenshot("latest.png") async def _submit_phone_number(self, phone: str) -> None: page = self._require_page() normalized = self._normalize_phone(phone) await self._ensure_country_russia() field = await self._first_visible( page.locator("#sign-in-phone-number"), page.locator('input[type="tel"]'), page.locator('input[autocomplete="tel"]'), ) if field is None: raise RuntimeError("Поле номера телефона не найдено") await field.click() await page.keyboard.press("Control+A") await page.keyboard.press("Backspace") await field.press_sequentially(normalized, delay=40) await asyncio.sleep(0.4) await self._click_next_or_enter() await self._wait_phone_accepted() async def _wait_phone_accepted(self) -> None: started = time.monotonic() deadline = started + 90 reloaded = False while time.monotonic() < deadline: await self._refresh_login_state() if self._login_step in {"waiting_code", "waiting_password", "logged_in"}: self.last_error = None return error = await self._auth_error_text() if error and not await self._is_please_wait(): self.last_error = error raise RuntimeError(error) if ( not reloaded and await self._is_please_wait() and time.monotonic() - started >= 20 ): logger.warning("Telegram завис на PLEASE WAIT — перезагружаю страницу") await self._reload_auth() reloaded = True await asyncio.sleep(0.5) raise RuntimeError("Telegram не открыл поле кода (зависание после отправки номера)") async def _recover_code_screen(self) -> None: await self._reload_auth() await self._refresh_login_state() if self._login_step == "waiting_code": return if await self._is_please_wait(): await asyncio.sleep(8) await self._reload_auth() await self._refresh_login_state() if self._login_step != "waiting_code": raise RuntimeError( f"Поля кода нет (шаг {self._login_step}). " "Запросите код заново: POST /login/phone" ) async def _reload_auth(self) -> None: page = self._require_page() await page.reload(wait_until="domcontentloaded", timeout=config.NAV_TIMEOUT_MS) await self._wait_auth_screen() await self._refresh_login_state() await self._safe_screenshot("latest.png") async def _is_please_wait(self) -> bool: page = self._require_page() return await self._is_visible(page.get_by_text(PLEASE_WAIT_RE)) async def _submit_login_code(self, code: str) -> None: page = self._require_page() await self._refresh_login_state() if self._login_step != "waiting_code": raise RuntimeError(f"Сейчас шаг {self._login_step}, а не ввод кода") digits = re.sub(r"\D", "", code) field = await self._first_visible( page.locator("#sign-in-code"), page.locator('input[autocomplete="one-time-code"]'), page.locator('input[inputmode="numeric"]'), ) if field is None: raise RuntimeError("Поле кода не найдено") await field.click() await page.keyboard.press("Control+A") await page.keyboard.press("Backspace") await field.press_sequentially(digits, delay=50) await asyncio.sleep(0.8) if self._login_step == "waiting_code": await self._click_next_or_enter() deadline = time.monotonic() + 25 while time.monotonic() < deadline: await self._refresh_login_state() if self._login_step in {"waiting_password", "logged_in"}: self.last_error = None return error = await self._auth_error_text() if error: self.last_error = error raise RuntimeError(error) await asyncio.sleep(0.4) raise RuntimeError("Код не принят. Проверьте цифры или 2FA") async def _submit_cloud_password(self, password: str) -> None: page = self._require_page() await self._refresh_login_state() if self._login_step != "waiting_password": raise RuntimeError(f"Сейчас шаг {self._login_step}, а не пароль 2FA") field = await self._first_visible( page.locator("#sign-in-password"), page.locator('input[type="password"]'), ) if field is None: raise RuntimeError("Поле пароля 2FA не найдено") await field.click() await page.keyboard.press("Control+A") await page.keyboard.press("Backspace") await field.press_sequentially(password, delay=30) await asyncio.sleep(0.3) await self._click_next_or_enter() deadline = time.monotonic() + 25 while time.monotonic() < deadline: await self._refresh_login_state() if self._login_step == "logged_in": self.last_error = None return error = await self._auth_error_text() if error: self.last_error = error raise RuntimeError(error) await asyncio.sleep(0.4) raise RuntimeError("Пароль 2FA не принят") async def _click_next_or_enter(self) -> None: page = self._require_page() nxt = page.get_by_role("button", name=NEXT_BTN_RE) try: if await nxt.count() > 0 and await nxt.first.is_visible(): await nxt.first.click(timeout=5_000) return except Exception: logger.debug("Кнопка Next не нажата, пробуем Enter") await page.keyboard.press("Enter") async def _first_visible(self, *locators): for loc in locators: try: count = await loc.count() for index in range(count): item = loc.nth(index) if await item.is_visible(): return item except Exception: continue return None async def _auth_error_text(self) -> str | None: page = self._require_page() loc = page.locator(".error, .input-error, .FormError, [class*='error']") try: count = await loc.count() for index in range(min(count, 10)): item = loc.nth(index) if not await item.is_visible(): continue text = (await item.inner_text()).strip() if text: return text except Exception: return None return None async def _open_bot_chat(self) -> None: if self._page is None: return composer = self._page.locator("#editable-message-text") try: await self._page.goto( config.CHAT_URL, wait_until="domcontentloaded", timeout=config.NAV_TIMEOUT_MS, ) await composer.wait_for(state="visible", timeout=20_000) return except Exception as exc: logger.warning("Переход по URL чата не открыл поле ввода: %s", exc) try: await self._page.get_by_text(config.BOT_TITLE, exact=False).first.click(timeout=10_000) await composer.wait_for(state="visible", timeout=20_000) except Exception as exc: self.last_error = str(exc) logger.warning("Чат бота не открылся: %s", exc) async def _save_storage_state(self) -> None: if self._context is None: return config.STORAGE_STATE.parent.mkdir(parents=True, exist_ok=True) await self._context.storage_state(path=str(config.STORAGE_STATE)) async def _save_qr(self) -> None: if self._page is None: return qr_path = config.DATA_DIR / "qr.png" canvases = self._page.locator("canvas") count = await canvases.count() best_index = -1 best_area = 0.0 for index in range(count): box = await canvases.nth(index).bounding_box() if box is None: continue area = float(box["width"]) * float(box["height"]) if area > best_area: best_area = area best_index = index # Маленький canvas — логотип в центре QR, для скана нужен крупный. if best_index >= 0 and best_area >= 80 * 80: await canvases.nth(best_index).screenshot(path=str(qr_path)) return await self._screenshot("qr.png") async def _screenshot(self, name: str) -> Path: if self._page is None: raise RuntimeError("Нет открытой страницы") path = config.DATA_DIR / name config.DATA_DIR.mkdir(parents=True, exist_ok=True) await self._page.screenshot(path=str(path), full_page=False) return path async def _safe_screenshot(self, name: str) -> None: try: if self._page is not None: await self._screenshot(name) except Exception: logger.debug("Скрин %s не записан", name)