"""server.py — tradein-browser service main process. Этот модуль запускается как точка входа контейнера ``tradein-browser``. Он запускает camoufox **локально** внутри контейнера (AsyncCamoufox) и экспонирует простой HTTP API на базе aiohttp: GET /health → {"status": "ok", "browsers": {"avito": bool, ...}} POST /fetch → {"url": "..."} → {"html": "..."} POST /fetch-json → {"url","method","headers","body","origin"} → {"status","body"} POST /login → {"url": "...", "email": "...", "password": "...", ...} → {"cookies": [...]} Такой подход выбран потому, что ``camoufox.server.launch_server`` (Playwright WS-сервер) несовместим с современными версиями playwright (1.45–1.60): ``browserServerImpl.js`` отсутствует в дистрибутиве → контейнер краш-лупится с MODULE_NOT_FOUND / RuntimeError при запуске через ``launch_server``. Локальный запуск ``AsyncCamoufox`` работает стабильно — проверено. Per-provider модель (#1793): Вместо одного глобального браузера + одного egress-прокси держим ОТДЕЛЬНЫЙ инстанс camoufox на каждого поставщика (avito/cian/yandex/generic). Поставщик определяется по host URL запроса (или явному полю body["provider"]/["source"]). Каждый инстанс ходит через свой http-прокси (BROWSER_PROXY_AVITO/CIAN/YANDEX) и имеет geoip=True — фингерпринт привязан к гео конкретного прокси, поэтому смешивать источники в одном браузере нельзя. Конкурентность: * МЕЖДУ поставщиками — параллельно (avito/cian/yandex гоняются одновременно); * ВНУТРИ поставщика — строго ≤1 одновременный fetch (per-provider lock). Это решает проблему, когда источники клинили друг друга через единственный egress-прокси / единственный сериализующий лок. Переменные окружения: BROWSER_PORT — TCP-порт HTTP-сервера (default: 3000) BROWSER_RECYCLE_PAGES — страниц в одном сеансе браузера до перезапуска (default: 15) BROWSER_NAV_TIMEOUT_MS — таймаут page.goto в мс (default: 60000) BROWSER_WAIT_MS — ожидание гидрации listings после DOMContentLoaded, мс (default: 6000) BROWSER_BLOCK_RESOURCE_TYPES — CSV типов ресурсов Playwright, которые abort'ить через page.route при навигации (default: "font,media"). Снижает число одновременных под-коннектов на страницу → мобильный прокси не ловит NS_ERROR_PROXY_TOO_MANY_REQUESTS. image уже глушит camoufox block_images. Пустая строка = НЕ блокировать ничего. КРИТИЧНО: НЕ добавлять document/script/stylesheet/xhr/fetch — данные парсятся из HTML/JS-state (INITIAL_STATE/__preloadedState__), JS нужен для гидрации, CSS проверяется anti-bot fingerprint. (legacy alias BROWSER_BLOCKED_RESOURCE_TYPES всё ещё читается как fallback, если новая переменная не задана.) BROWSER_MIN_PAGE_INTERVAL_S — минимальный интервал (сек) между последовательными page.goto ОДНОГО провайдера (default: 2.0). Даёт под- коннектам предыдущей страницы дренироваться, прежде чем открыть новую → меньше пиковый параллелизм на прокси. 0 = без пейсинга. BROWSER_MIN_PAGE_INTERVAL_S_{PROVIDER} — per-provider override интервала пейсинга. PROVIDER в upper-case: AVITO, CIAN, YANDEX, GENERIC. Например: BROWSER_MIN_PAGE_INTERVAL_S_CIAN=18 задаёт 18с только для cian, не затрагивая другие провайдеры. Если env не задан или содержит невалидное значение — фолбэк на глобальный BROWSER_MIN_PAGE_INTERVAL_S. Backward-compat: при незаданных per-provider env поведение идентично предыдущему (чисто аддитивный override). BROWSER_PROXY_AVITO — http-прокси для avito (fallback AVITO_PROXY_URL). BROWSER_PROXY_CIAN — http-прокси для cian (fallback CIAN_PROXY_URL). BROWSER_PROXY_YANDEX — http-прокси для yandex (fallback YANDEX_PROXY_URL). SCRAPER_PROXY_URL — http-прокси для generic (fallback AVITO_PROXY_URL). ВАЖНО: для браузера используем именно BROWSER_PROXY_* (http-прокси). Legacy CIAN_PROXY_URL/YANDEX_PROXY_URL могут быть socks5 с авторизацией — playwright их не умеет, поэтому они лишь fallback (если кто-то положил http туда же). Контракт /login (провалидировано вживую 2026-05-31, Cian email+пароль без SMS): pre_click_selectors — список селекторов для последовательного клика до формы; каждый клик non-fatal (пропускается при отсутствии элемента). Двухшаговый submit: после первого сабмита Cian может показать «Введите пароль» повторно. Эндпоинт /fetch доступен из Docker-сети как ``http://tradein-browser:3000/fetch``. """ import asyncio import base64 import logging import os from collections.abc import Mapping from urllib.parse import urlparse from aiohttp import web logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") logger = logging.getLogger(__name__) # ── конфигурация из env ──────────────────────────────────────────────────────── BROWSER_PORT: int = int(os.environ.get("BROWSER_PORT", "3000")) BROWSER_RECYCLE_PAGES: int = int(os.environ.get("BROWSER_RECYCLE_PAGES", "15")) BROWSER_NAV_TIMEOUT_MS: int = int(os.environ.get("BROWSER_NAV_TIMEOUT_MS", "60000")) # 6000 (не 2500): avito гидрирует listings client-side ПОСЛЕ domcontentloaded; # на 2.5с в HTML генерик-шелл без объявлений (0 listings), на 5-6с — полная # выдача (~50 карточек, 3.2МБ). Подтверждено прод-дебагом 2026-05-31. BROWSER_WAIT_MS: int = int(os.environ.get("BROWSER_WAIT_MS", "6000")) # /fetch-json settle после goto(origin) перед in-page fetch (#1917). 500мс мало: # первый XHR иногда ловит `NetworkError when attempting to fetch resource` (anti-bot/ # сетевой стек страницы ещё не готов). Лечился внешним retry (re-navigation ~30-45с/дом). # Поднимаем дефолт до 1200мс (меньше first-fail) + добавляем дешёвый in-page retry # самого fetch() (см. FETCH_JSON_INPAGE_RETRIES) — повтор внутри страницы стоит ~доли # секунды против полной ре-навигации. НЕ трогаем /fetch (SERP): там свой BROWSER_WAIT_MS. FETCH_JSON_SETTLE_MS: int = int(os.environ.get("FETCH_JSON_SETTLE_MS", "1200")) # Сколько раз повторить in-page fetch() при сетевом throw (TypeError/NetworkError), # НЕ при HTTP-статусе (4xx/5xx возвращаются как есть — их решает caller). 0 = выкл. FETCH_JSON_INPAGE_RETRIES: int = int(os.environ.get("FETCH_JSON_INPAGE_RETRIES", "1")) # Пауза между in-page попытками fetch(), мс. FETCH_JSON_RETRY_DELAY_MS: int = int(os.environ.get("FETCH_JSON_RETRY_DELAY_MS", "800")) # Известные поставщики. "generic" — фолбэк для всех прочих хостов (один общий # инстанс на неузнанные домены). Порядок задаёт детерминированный health-вывод. PROVIDERS: tuple[str, ...] = ("avito", "cian", "yandex", "generic") def _parse_bool(raw: str | None, default: bool) -> bool: """Парсит булев env-флаг ("true"/"1"/"yes"/"on" → True), fallback на default.""" if raw is None: return default return raw.strip().lower() in ("true", "1", "yes", "on") def _parse_block_types(raw: str | None) -> frozenset[str]: """Парсит CSV типов ресурсов для блокировки → frozenset (lower/strip). None (env не задана) → дефолт "font,media". Пустая строка → пустой набор (ничего не блокируем). Не задаём document/script/stylesheet/xhr/fetch здесь — дефолт намеренно узкий (font+media), эти типы не нужны для extraction/fingerprint. """ if raw is None: raw = "font,media" return frozenset(t.strip().lower() for t in raw.split(",") if t.strip()) # Типы под-ресурсов, которые abort'им через page.route на каждой странице. Каждый # page.goto тянет десятки sub-requests (js/css/xhr/font/media) через прокси; мобильный # прокси держит ~5 параллельных коннектов → 2-3 быстрых страницы подряд на одном # аккаунте упираются в NS_ERROR_PROXY_TOO_MANY_REQUESTS/500. Глуша font+media (тяжёлые, # не нужны ни для extraction, ни для fingerprint), снижаем пиковый fan-out. image уже # глушит camoufox block_images. КРИТИЧНО: document/script/stylesheet/xhr/fetch НЕ # блокируем — данные в HTML/JS-state (INITIAL_STATE/__preloadedState__), JS гидрирует # listings, CSS проверяет anti-bot fingerprint. Legacy alias BROWSER_BLOCKED_RESOURCE_TYPES # читается fallback'ом, если новая BROWSER_BLOCK_RESOURCE_TYPES не задана. _BLOCKED_TYPES: frozenset[str] = _parse_block_types( os.environ.get("BROWSER_BLOCK_RESOURCE_TYPES") or os.environ.get("BROWSER_BLOCKED_RESOURCE_TYPES") ) # Минимальный интервал (сек) между последовательными page.goto одного провайдера — # даёт под-коннектам предыдущей страницы дренироваться (прокси освобождает слоты), # прежде чем открыть новую страницу. 0 = без пейсинга. BROWSER_MIN_PAGE_INTERVAL_S: float = float(os.environ.get("BROWSER_MIN_PAGE_INTERVAL_S", "2.0")) def _resolve_min_interval( provider: str, environ: Mapping[str, str] | None = None, ) -> float: """Возвращает эффективный интервал пейсинга для провайдера. Читает per-provider override из ``BROWSER_MIN_PAGE_INTERVAL_S_{PROVIDER}`` (PROVIDER в upper-case). Невалидное или отсутствующее значение → фолбэк на глобальный ``BROWSER_MIN_PAGE_INTERVAL_S``. Не кидает исключений. Args: provider: имя провайдера ("avito", "cian", "yandex", "generic"). environ: env-словарь; None → ``os.environ`` (тестируемость без патча модуля). Returns: Эффективный интервал в секундах (≥0). """ env = environ if environ is not None else os.environ key = f"BROWSER_MIN_PAGE_INTERVAL_S_{provider.upper()}" raw = env.get(key) if raw is not None: try: return float(raw) except ValueError: logger.warning( "tradein-browser: %s=%r невалидно (не float), используем глобал %.1f", key, raw, BROWSER_MIN_PAGE_INTERVAL_S, ) return BROWSER_MIN_PAGE_INTERVAL_S # Per-provider эффективные интервалы пейсинга, вычисляются на module-load из os.environ. # Ключи: все известные провайдеры из PROVIDERS. Логируются на старте (_on_startup). _MIN_PAGE_INTERVAL_BY_PROVIDER: dict[str, float] = { p: _resolve_min_interval(p) for p in ("avito", "cian", "yandex", "generic") } def _provider_for_url(url: str) -> str: """Определяет поставщика по host URL. avito/cian/yandex → одноимённый провайдер; всё прочее → "generic". """ host = (urlparse(url).hostname or "").lower() if "avito" in host: return "avito" if "cian" in host: return "cian" if "yandex" in host: return "yandex" return "generic" def _provider_proxy(provider: str) -> str | None: """Возвращает http-прокси для поставщика из env (с фолбэком), иначе None. Используем BROWSER_PROXY_* как primary — это http-прокси для браузера. Legacy *_PROXY_URL — лишь fallback: playwright НЕ умеет socks5 с авторизацией, поэтому socks5-варианты CIAN/YANDEX как primary не годятся. """ env_map: dict[str, tuple[str, str]] = { "avito": ("BROWSER_PROXY_AVITO", "AVITO_PROXY_URL"), "cian": ("BROWSER_PROXY_CIAN", "CIAN_PROXY_URL"), "yandex": ("BROWSER_PROXY_YANDEX", "YANDEX_PROXY_URL"), "generic": ("SCRAPER_PROXY_URL", "AVITO_PROXY_URL"), } primary, fallback = env_map.get(provider, env_map["generic"]) return os.environ.get(primary) or os.environ.get(fallback) def _parse_proxy(proxy_url: str | None) -> dict[str, str] | None: """Парсит proxy URL → camoufox proxy dict. Формат входного URL: ``http://user:pass@host:port`` или ``http://host:port``. Возвращает None если proxy_url пуст. Returns: ``{"server": "http://host:port", "username": "...", "password": "..."}`` или None. """ if not proxy_url: return None parsed = urlparse(proxy_url) port_part = f":{parsed.port}" if parsed.port else "" server = f"{parsed.scheme}://{parsed.hostname}{port_part}" result: dict[str, str] = {"server": server} if parsed.username: result["username"] = parsed.username if parsed.password: result["password"] = parsed.password return result # ── per-provider состояние браузеров ─────────────────────────────────────────── # Каждый словарь keyed по provider (str). Заполняется ЛЕНИВО: инстанс конкретного # поставщика поднимается на первом /fetch (или /login) к этому поставщику. # # _locks[provider] сериализует ВЕСЬ доступ к инстансу этого поставщика # (launch + relaunch + fetch + login) → внутри поставщика строго ≤1 операция # одновременно. Разные поставщики держат РАЗНЫЕ локи → работают параллельно. _browsers: dict[str, object] = {} # provider → Browser _browser_cms: dict[str, object] = {} # provider → AsyncCamoufox CM (для __aexit__) _page_counters: dict[str, int] = {} # provider → страниц с момента launch'а _locks: dict[str, asyncio.Lock] = {} # provider → Lock (весь доступ к инстансу) _retry_tasks: dict[str, asyncio.Task[None]] = {} # provider → фоновая retry-задача _last_goto_at: dict[str, float] = {} # provider → loop-time последнего page.goto (пейсинг) # Guard на ленивое создание per-provider локов: setdefault на обычном dict из # разных корутин гонок не даёт (нет await между read-modify-write), но держим # явный guard на случай будущей сложной инициализации. Создаётся в _on_startup. _locks_guard: asyncio.Lock | None = None # Resilience (#crash-storm): если прокси (mproxy.site) лежит, camoufox делает # eager public_ip(proxy) для geoip-fingerprint и бросает InvalidProxy. Раньше это # валило app.startup → exit(1) → docker restart → ∞ (RestartCount 3000+). Теперь # фейл launch'а НЕ фатален: контейнер остаётся Up, а фоновая задача retry'ит # launch с backoff пока прокси не вернётся. proxy+geoip СОХРАНЕНЫ (анти-бан). _BROWSER_RETRY_MIN_S: float = 30.0 # стартовый интервал retry launch'а _BROWSER_RETRY_MAX_S: float = 60.0 # потолок backoff'а async def _lock_for(provider: str) -> asyncio.Lock: """Возвращает (создавая лениво под guard'ом) per-provider лок. Лок сериализует launch+fetch+login на инстансе поставщика: один поставщик = один браузер = ≤1 операция за раз. Guard защищает создание лока от гонки параллельных корутин на первом запросе к новому поставщику. """ existing = _locks.get(provider) if existing is not None: return existing assert _locks_guard is not None, "_locks_guard not initialised" async with _locks_guard: # setdefault идемпотентен — вторая корутина под guard получит уже созданный лок. return _locks.setdefault(provider, asyncio.Lock()) async def _launch_browser(provider: str) -> None: """Запускает AsyncCamoufox для поставщика и кладёт в *_…[provider]. Использует _provider_proxy(provider) + geoip=True (фингерпринт привязан к гео IP конкретного прокси — поэтому отдельный инстанс на каждого поставщика). """ from camoufox.async_api import AsyncCamoufox proxy = _parse_proxy(_provider_proxy(provider)) kwargs: dict[str, object] = { "headless": True, "os": "windows", "locale": "ru-RU", "geoip": True, "humanize": True, # Скорость: не грузим картинки — нам нужен HTML/JSON, рендер быстрее # (трафик безлимитный, выигрыш именно по времени загрузки). "block_images": True, # Анти-leak: WebRTC может слить реальный IP контейнера мимо прокси → деанон. "block_webrtc": True, # Форсим таймзону +3 МСК. geoip берёт tz одного IP на старте, а БэкКоннект # ротирует IP на каждый запрос (видели Ульяновск UTC+4 при выходе СПб/МСК) → # геолик tz↔IP. Жёсткий Europe/Moscow убирает рассинхрон для РФ-мобильного. "config": {"timezone": "Europe/Moscow"}, } if proxy is not None: kwargs["proxy"] = proxy logger.info( "tradein-browser[%s]: запуск AsyncCamoufox (proxy=%s, recycle_pages=%d)", provider, proxy is not None, BROWSER_RECYCLE_PAGES, ) cm = AsyncCamoufox(**kwargs) # type: ignore[arg-type] browser = await cm.__aenter__() _browser_cms[provider] = cm _browsers[provider] = browser _page_counters[provider] = 0 logger.info("tradein-browser[%s]: браузер запущен", provider) async def _close_browser(provider: str) -> None: """Закрывает per-provider браузер и чистит состояние для поставщика.""" cm = _browser_cms.get(provider) if cm is not None: try: await cm.__aexit__(None, None, None) # type: ignore[attr-defined] logger.info("tradein-browser[%s]: браузер закрыт", provider) except Exception as exc: logger.warning( "tradein-browser[%s]: ошибка при закрытии браузера: %s", provider, type(exc).__name__, ) _browsers.pop(provider, None) _browser_cms.pop(provider, None) _page_counters[provider] = 0 _last_goto_at.pop(provider, None) async def _relaunch_browser(provider: str) -> None: """Закрывает и заново запускает per-provider браузер (recycle / crash-recovery). Caller держит _locks[provider], поэтому параллельных страниц на этом инстансе нет — teardown безопасен. """ logger.info("tradein-browser[%s]: перезапуск браузера", provider) await _close_browser(provider) await _launch_browser(provider) async def _try_launch_browser(provider: str) -> bool: """Одна попытка launch'а поставщика с санитизированным логом фейла. НЕ пробрасывает исключение наружу — возвращает True/False. Критично для resilience: фейл launch'а (прокси лёг → InvalidProxy) не должен валить startup или фоновую задачу. Креды прокси НЕ логируются: camoufox светит ``user:pass@host`` в тексте своего исключения, поэтому мы логируем только тип, без ``str(exc)``. При частичном фейле (CM создан, но __aenter__ не довёл) чистит состояние через _close_browser, чтобы следующий retry стартовал с чистого листа. """ try: await _launch_browser(provider) return True except Exception as exc: logger.warning( "tradein-browser[%s]: browser launch failed: proxy unreachable " "(exc_type=%s) — контейнер остаётся Up, retry на следующем запросе/в фоне", provider, type(exc).__name__, ) await _close_browser(provider) return False async def _ensure_browser(provider: str) -> bool: """Гарантирует поднятый инстанс поставщика; одна lazy-попытка launch'а. Caller держит _locks[provider] (весь доступ к инстансу сериализован), поэтому здесь своей сериализации нет — двух одновременных launch'ей на один provider быть не может. Returns: True если браузер доступен (был или поднялся), False если launch не удался (прокси лежит) — caller отдаёт 503. При неудачном lazy-launch (прокси лёг) запускает фоновую retry-задачу для ЭТОГО поставщика: при eager-startup её ставил _on_startup, в lazy-режиме — первый обратившийся /fetch. Поставщики независимы — фоновый retry поднимается только для тех, к кому был запрос, а не для всех сразу. """ if _browsers.get(provider) is not None: return True await _try_launch_browser(provider) if _browsers.get(provider) is not None: return True _start_retry_task(provider) return False async def _browser_retry_loop(provider: str) -> None: """Фоновый retry launch'а поставщика с backoff, пока прокси не вернётся. Идемпотентна: если инстанс уже поднят — сразу выходит. Иначе ретраит _try_launch_browser под _locks[provider] с возрастающим интервалом (_BROWSER_RETRY_MIN_S → _BROWSER_RETRY_MAX_S). Завершается после первого успешного launch'а. Отменяется в _on_cleanup (CancelledError пробрасывается). """ if _browsers.get(provider) is not None: return delay = _BROWSER_RETRY_MIN_S while _browsers.get(provider) is None: await asyncio.sleep(delay) lock = await _lock_for(provider) async with lock: if _browsers.get(provider) is not None: return # handler уже поднял инстанс — задача отработала logger.info("tradein-browser[%s]: фоновый retry launch'а браузера", provider) launched = await _try_launch_browser(provider) if launched: logger.info( "tradein-browser[%s]: браузер поднят после retry, задача завершена", provider ) return delay = min(delay * 1.5, _BROWSER_RETRY_MAX_S) def _start_retry_task(provider: str) -> None: """Запускает (если ещё нет) фоновую retry-задачу для поставщика.""" existing = _retry_tasks.get(provider) if existing is not None and not existing.done(): return # задача уже бежит — не плодим дубли _retry_tasks[provider] = asyncio.create_task(_browser_retry_loop(provider)) # ── aiohttp lifecycle hooks ──────────────────────────────────────────────────── async def _on_startup(app: web.Application) -> None: # Startup ТОЛЬКО инициализирует guard для ленивого создания локов — браузеры # НЕ поднимаются здесь. Lazy-launch: инстанс конкретного поставщика стартует на # первом /fetch (или /login) к нему через _ensure_browser под _locks[provider]. # # Eager-launch всех 4 на старте давал бы 4 camoufox-инстанса сразу → риск OOM: # до 4 camoufox-инстансов (avito/cian/yandex/generic), lazy; ~0.5-1ГБ RSS # каждый, mem_limit контейнера 5g. Resilience (фоновый retry при недоступном # прокси) запускается при первом обращении к поставщику, не для всех сразу. global _locks_guard _locks_guard = asyncio.Lock() logger.info( "tradein-browser: per-provider concurrency (lazy launch) — " "параллель между %s, ≤1 на поставщика; браузеры стартуют по требованию", list(PROVIDERS), ) logger.info("tradein-browser: page-intervals %s", _MIN_PAGE_INTERVAL_BY_PROVIDER) async def _on_cleanup(app: web.Application) -> None: # Отменяем все фоновые retry-задачи. for provider, task in list(_retry_tasks.items()): if task is not None and not task.done(): task.cancel() try: await task except asyncio.CancelledError: pass except Exception as exc: logger.warning( "tradein-browser[%s]: ошибка при отмене retry-задачи: %s", provider, type(exc).__name__, ) _retry_tasks.clear() # Закрываем все поднятые инстансы. for provider in list(_browsers.keys()): await _close_browser(provider) # ── handlers ─────────────────────────────────────────────────────────────────── def _resolve_provider(body: dict, url: str) -> str: """Выбирает поставщика: явный body["provider"]/body["source"] (если валиден) иначе host. Caller (browser_fetcher.py) шлёт {"url","source"} — поэтому explicit-значение берём из "provider" ИЛИ "source", но применяем ТОЛЬКО если оно в PROVIDERS; иначе host-detect; иначе generic. # domclick (source='domklik') не имеет выделенного BROWSER_PROXY_DOMCLICK в prod # → generic. Намеренно. ('domklik' нет в PROVIDERS → fallthrough на host-detect # domclick.ru → generic.) """ explicit = body.get("provider") or body.get("source") if isinstance(explicit, str) and explicit in PROVIDERS: return explicit return _provider_for_url(url) async def health_handler(request: web.Request) -> web.Response: # Всегда 200 — иначе compose healthcheck (если появится) убьёт контейнер, # хотя процесс жив и ретраит браузеры. browsers — per-provider readiness для # observability: видно какой поставщик лежит, не теряя контейнер. Текущий prod # compose НЕ имеет healthcheck на browser (depends_on: service_started). browsers = {p: _browsers.get(p) is not None for p in PROVIDERS} return web.json_response({"status": "ok", "browsers": browsers}) async def fetch_handler(request: web.Request) -> web.Response: """POST /fetch {"url": "..."} → {"html": "..."} Выполняет навигацию в браузере поставщика и возвращает полный HTML страницы. Поставщик определяется по host URL (или явному body["provider"]/["source"]). Берётся ТОЛЬКО лок этого поставщика → разные поставщики работают параллельно, внутри поставщика — строго ≤1 fetch. """ try: body = await request.json() except Exception: return web.json_response({"error": "invalid JSON body"}, status=400) url: str | None = body.get("url") if not url: return web.json_response({"error": "missing 'url' field"}, status=400) provider = _resolve_provider(body, url) lock = await _lock_for(provider) async with lock: # Браузер мог не подняться на старте (прокси лежал). Одна lazy-попытка — # вдруг прокси уже вернулся. НИКОГДА не фетчим без прокси/браузера: если # всё ещё None → 503 (caller словит как HTTPStatusError → failed fetch, не # crash). if not await _ensure_browser(provider): logger.warning( "tradein-browser[%s]: /fetch 503 — браузер недоступен (proxy may be down)", provider, ) return web.json_response( {"error": "browser unavailable (proxy may be down)"}, status=503 ) try: html = await _do_fetch(provider, url) except Exception as exc: logger.error( "tradein-browser[%s]: fetch error url=%r: %s: %s", provider, url, type(exc).__name__, exc, ) return web.json_response({"error": f"{type(exc).__name__}: {exc}"}, status=500) return web.json_response({"html": html}) async def fetch_json_handler(request: web.Request) -> web.Response: """POST /fetch-json {"url","method","headers","body","origin","source"} → {"status","body"} Выполняет in-page ``fetch()`` ИЗ ТЁПЛОЙ страницы браузера поставщика: камуфокс сначала переходит на ``origin`` (same-origin якорь), затем дёргает ``fetch(url)`` с реальным фингерпринтом + контекст-cookies этого инстанса + per-provider прокси. Возвращает JSON/текст ответа как ``{"status": int, "body": str}``. Чисто аддитивный путь (#915 Stage 2) — никто из прод-флоу пока не вызывает. Делит тот же per-provider браузер/лок что и /fetch (поставщик по host URL или явному body["provider"]/["source"]). Берётся ТОЛЬКО лок этого поставщика → разные поставщики работают параллельно, внутри поставщика — строго ≤1 операция. """ try: body = await request.json() except Exception: return web.json_response({"error": "invalid JSON body"}, status=400) url: str | None = body.get("url") if not url: return web.json_response({"error": "missing 'url' field"}, status=400) method: str = body.get("method") or "GET" headers: dict = body.get("headers") or {} req_body = body.get("body") # None | str (caller сериализует через json.dumps) # origin — same-origin страница, на которую перейдёт камуфокс перед fetch'ем. # Явный body["origin"] имеет приоритет; иначе выводим f"{scheme}://{host}/" из url. origin: str | None = body.get("origin") if not origin: parsed = urlparse(url) origin = f"{parsed.scheme}://{parsed.netloc}/" provider = _resolve_provider(body, url) lock = await _lock_for(provider) async with lock: # Та же resilience что и в /fetch: браузер мог не подняться (прокси лежал). # Lazy-попытка, иначе 503 — без прокси/браузера не фетчим. if not await _ensure_browser(provider): logger.warning( "tradein-browser[%s]: /fetch-json 503 — браузер недоступен (proxy may be down)", provider, ) return web.json_response( {"error": "browser unavailable (proxy may be down)"}, status=503 ) try: result = await _do_fetch_json( provider, url, method=method, headers=headers, body=req_body, origin=origin ) except Exception as exc: logger.error( "tradein-browser[%s]: fetch-json error url=%r: %s: %s", provider, url, type(exc).__name__, exc, ) return web.json_response({"error": f"{type(exc).__name__}: {exc}"}, status=500) return web.json_response({"status": result["status"], "body": result["body"]}) async def _apply_resource_block(page: object) -> None: """Навешивает route-interception, блокирующую тяжёлые под-ресурсы (font/media). No-op при пустом _BLOCKED_TYPES (BROWSER_BLOCK_RESOURCE_TYPES=""). Route живёт на странице и снимается автоматически при page.close(). Блокируем ТОЛЬКО _BLOCKED_TYPES (по умолчанию font/media) — document/script/stylesheet/xhr/fetch проходят, чтобы гидрация listings работала и stealth не палился «браузером без ресурсов». Handler обёрнут в try/except: любой сбой abort/continue (например, route уже обработан гонкой) деградирует в continue_(), чтобы один битый ресурс не ронял навигацию. Сам continue_() тоже под try — если страница уже закрыта, молча выходим. """ if not _BLOCKED_TYPES: return async def _route_block(route: object) -> None: # type: ignore[no-untyped-def] try: if route.request.resource_type in _BLOCKED_TYPES: # type: ignore[attr-defined] await route.abort() # type: ignore[attr-defined] else: await route.continue_() # type: ignore[attr-defined] except Exception: # Fallback: не блокируем спорный ресурс, пропускаем его — лучше лишний # байт, чем сорванная навигация. continue_() под своим try на случай # уже-закрытой страницы / повторно обработанного route. try: await route.continue_() # type: ignore[attr-defined] except Exception: pass await page.route("**/*", _route_block) # type: ignore[attr-defined] async def _pace_provider(provider: str) -> None: """Притормаживает перед page.goto, если с прошлого goto этого провайдера прошло меньше эффективного интервала — даёт под-коннектам прошлой страницы дренироваться, снижая пиковый параллелизм на прокси. Caller держит _locks[provider] → timestamp per-provider читается/пишется без гонки. Записывает момент ПОСЛЕ возможного sleep (т.е. фактический момент следующего goto), чтобы интервал считался между реальными навигациями, а не между входами в функцию. No-op при эффективном интервале провайдера <= 0. """ interval = _MIN_PAGE_INTERVAL_BY_PROVIDER.get(provider, BROWSER_MIN_PAGE_INTERVAL_S) if interval <= 0: return now = asyncio.get_event_loop().time() last = _last_goto_at.get(provider) if last is not None: elapsed = now - last remaining = interval - elapsed if remaining > 0: logger.debug( "tradein-browser[%s]: pacing — ждём %.2fс перед goto", provider, remaining ) await asyncio.sleep(remaining) _last_goto_at[provider] = asyncio.get_event_loop().time() async def _do_fetch(provider: str, url: str) -> str: """Одна попытка навигации; при краше браузера — relaunch и один retry. Caller держит _locks[provider] (нет параллельных страниц на этом инстансе), поэтому relaunch безопасен — никого не рвёт из-под живых страниц. """ try: return await _fetch_once(provider, url) except Exception as exc: if _is_browser_crash(exc): logger.warning( "tradein-browser[%s]: краш браузера (%s), перезапуск + retry: %s", provider, type(exc).__name__, url, ) await _relaunch_browser(provider) if _browsers.get(provider) is None: raise return await _fetch_once(provider, url) raise async def _fetch_once(provider: str, url: str) -> str: """Открывает СОБСТВЕННУЮ страницу, переходит по URL, ждёт JS, возвращает HTML. Caller держит _locks[provider], поэтому страницы на этом инстансе не параллелятся — recycle через _relaunch_browser безопасен прямо здесь. """ browser = _browsers.get(provider) assert browser is not None, "browser not launched" page = await browser.new_page() # type: ignore[attr-defined] try: await _apply_resource_block(page) await _pace_provider(provider) await page.goto(url, timeout=BROWSER_NAV_TIMEOUT_MS, wait_until="domcontentloaded") # type: ignore[attr-defined] if BROWSER_WAIT_MS > 0: await page.wait_for_timeout(BROWSER_WAIT_MS) # type: ignore[attr-defined] html: str = await page.content() # type: ignore[attr-defined] finally: await page.close() # type: ignore[attr-defined] _page_counters[provider] = _page_counters.get(provider, 0) + 1 logger.debug( "tradein-browser[%s]: fetch OK url=%r pages_since_launch=%d", provider, url, _page_counters[provider], ) if _page_counters[provider] >= BROWSER_RECYCLE_PAGES: logger.info( "tradein-browser[%s]: recycle threshold (%d) достигнут, перезапуск браузера", provider, BROWSER_RECYCLE_PAGES, ) await _relaunch_browser(provider) return html async def _do_fetch_json( provider: str, url: str, *, method: str, headers: dict, body: object, origin: str, ) -> dict: """Одна попытка in-page fetch'а; при краше браузера — relaunch и один retry. Зеркалит _do_fetch: caller держит _locks[provider] (нет параллельных страниц на этом инстансе), поэтому relaunch безопасен — никого не рвёт из-под живых страниц. """ try: return await _fetch_json_once( provider, url, method=method, headers=headers, body=body, origin=origin ) except Exception as exc: if _is_browser_crash(exc): logger.warning( "tradein-browser[%s]: краш браузера (%s), перезапуск + retry fetch-json: %s", provider, type(exc).__name__, url, ) await _relaunch_browser(provider) if _browsers.get(provider) is None: raise return await _fetch_json_once( provider, url, method=method, headers=headers, body=body, origin=origin ) raise async def _fetch_json_once( provider: str, url: str, *, method: str, headers: dict, body: object, origin: str, ) -> dict: """Переходит на origin (same-origin якорь) и выполняет in-page fetch(url). Зеркалит _fetch_once по жизненному циклу страницы. Навигация идёт на ``origin`` (а не на ``url``), чтобы in-page fetch был same-origin и нёс контекст-cookies + реальный фингерпринт инстанса. Возвращает {"status": int, "body": str}. Caller держит _locks[provider], поэтому страницы на этом инстансе не параллелятся — recycle через _relaunch_browser безопасен прямо здесь. """ browser = _browsers.get(provider) assert browser is not None, "browser not launched" page = await browser.new_page() # type: ignore[attr-defined] try: await _apply_resource_block(page) await _pace_provider(provider) await page.goto(origin, timeout=BROWSER_NAV_TIMEOUT_MS, wait_until="domcontentloaded") # type: ignore[attr-defined] # БЕЗ полного BROWSER_WAIT_MS: нам нужен лишь origin-контекст (cookies + # same-origin scope для fetch), а не отрендеренные listings. Settle-паузы # (#1917, FETCH_JSON_SETTLE_MS) хватает, чтобы страница инициализировалась # перед in-page fetch'ем. await page.wait_for_timeout(FETCH_JSON_SETTLE_MS) # type: ignore[attr-defined] # In-page retry самого fetch() при СЕТЕВОМ throw (TypeError/"NetworkError"), # #1917: первый XHR после навигации иногда падает до готовности стека. Повтор # внутри страницы (~retryDelayMs) дешевле полной ре-навигации (~30-45с/дом). # HTTP-статусы (4xx/5xx) — это успешный resp, НЕ ретраим (решает caller). result: dict = await page.evaluate( # type: ignore[attr-defined] """async ({url, method, headers, body, retries, retryDelayMs}) => { let lastErr; for (let attempt = 0; attempt <= retries; attempt++) { try { const resp = await fetch(url, { method: method || 'GET', headers: headers || {}, body: (body !== null && body !== undefined) ? body : undefined, credentials: 'include', }); const text = await resp.text(); return { status: resp.status, body: text, attempts: attempt + 1 }; } catch (e) { lastErr = e; if (attempt < retries) { await new Promise(r => setTimeout(r, retryDelayMs)); } } } throw lastErr; }""", { "url": url, "method": method, "headers": headers or {}, "body": body, "retries": FETCH_JSON_INPAGE_RETRIES, "retryDelayMs": FETCH_JSON_RETRY_DELAY_MS, }, ) finally: await page.close() # type: ignore[attr-defined] _page_counters[provider] = _page_counters.get(provider, 0) + 1 logger.debug( "tradein-browser[%s]: fetch-json OK url=%r status=%s pages_since_launch=%d", provider, url, result.get("status"), _page_counters[provider], ) if _page_counters[provider] >= BROWSER_RECYCLE_PAGES: logger.info( "tradein-browser[%s]: recycle threshold (%d) достигнут, перезапуск браузера", provider, BROWSER_RECYCLE_PAGES, ) await _relaunch_browser(provider) return result def _is_browser_crash(exc: BaseException) -> bool: """Проверяет является ли исключение признаком краша / разрыва браузера.""" cls_name = type(exc).__name__ if cls_name in ("TargetClosedError", "BrowserClosedError", "ConnectionClosedError"): return True msg = str(exc).lower() return ( "browser has been closed" in msg or "target closed" in msg or "connection closed" in msg or "browser disconnected" in msg ) # ── login handler ────────────────────────────────────────────────────────────── class LoginError(Exception): """Исключение с диагностическим payload для /login эндпоинта.""" def __init__(self, payload: dict) -> None: super().__init__(str(payload)) self.payload = payload async def pacing_get_handler(request: web.Request) -> web.Response: """GET /pacing → текущие live-интервалы + env-дефолты для всех провайдеров. interval_s — живое значение из _MIN_PAGE_INTERVAL_BY_PROVIDER (изменяется PUT). env_default_s — что вернёт _resolve_min_interval(p) из os.environ (reset при рестарте). Всегда 200; не требует browser-инстансов (memory-only). """ providers = [] for p in PROVIDERS: providers.append({ "source": p, "interval_s": _MIN_PAGE_INTERVAL_BY_PROVIDER.get(p, BROWSER_MIN_PAGE_INTERVAL_S), "env_default_s": _resolve_min_interval(p), }) return web.json_response({"providers": providers}) async def pacing_put_handler(request: web.Request) -> web.Response: """PUT /pacing {"source":

, "interval_s": } → обновляет in-memory интервал. Валидация: source ∈ {avito,cian,yandex,generic}, interval_s ≥ 0. Сброс к env-дефолту происходит при рестарте контейнера (by design). """ try: body = await request.json() except Exception: return web.json_response({"error": "invalid JSON body"}, status=400) source = body.get("source") if not isinstance(source, str) or source not in PROVIDERS: return web.json_response( {"error": f"source must be one of {list(PROVIDERS)}"}, status=400, ) interval_s = body.get("interval_s") if not isinstance(interval_s, (int, float)): return web.json_response( {"error": "interval_s must be a non-negative number"}, status=422, ) interval_s = float(interval_s) if interval_s < 0: return web.json_response( {"error": "interval_s must be >= 0"}, status=422, ) _MIN_PAGE_INTERVAL_BY_PROVIDER[source] = interval_s logger.info( "tradein-browser: pacing updated source=%s interval_s=%.2f (resets on restart)", source, interval_s, ) return web.json_response({"ok": True, "source": source, "interval_s": interval_s}) async def login_handler(request: web.Request) -> web.Response: """POST /login {...} → {"cookies": [...]} Выполняет вход на указанную страницу (email + пароль), ждёт появления success_cookie, возвращает полный список cookies браузерного контекста. Делит тот же per-provider браузер/лок что и /fetch (поставщик по host URL или явному body["provider"]/["source"]). """ try: body = await request.json() except Exception: return web.json_response({"error": "invalid JSON body"}, status=400) missing = [f for f in ("url", "email", "password", "email_selector", "password_selector", "submit_selector") if not body.get(f)] if missing: return web.json_response({"error": f"missing required fields: {missing}"}, status=400) provider = _resolve_provider(body, body["url"]) lock = await _lock_for(provider) async with lock: # Та же resilience что и в /fetch: браузер мог не подняться (прокси лежал). # Lazy-попытка, иначе 503 — без прокси не логинимся. if not await _ensure_browser(provider): logger.warning( "tradein-browser[%s]: /login 503 — браузер недоступен (proxy may be down)", provider, ) return web.json_response( {"error": "browser unavailable (proxy may be down)"}, status=503 ) try: cookies = await _do_login(provider, body) except LoginError as exc: return web.json_response(exc.payload, status=502) except Exception as exc: logger.error( "tradein-browser[%s]: login error url=%r: %s: %s", provider, body.get("url"), type(exc).__name__, exc, ) return web.json_response({"error": f"{type(exc).__name__}: {exc}"}, status=502) return web.json_response({"cookies": cookies}) async def _do_login(provider: str, params: dict) -> list[dict]: """Одна попытка логина; при краше браузера — relaunch и повтор. Caller держит _locks[provider], поэтому relaunch безопасен. """ try: return await _login_once(provider, params) except LoginError: raise # диагностический payload идёт напрямую в хендлер except Exception as exc: if _is_browser_crash(exc): logger.warning( "tradein-browser[%s]: краш браузера (%s) при логине, перезапуск + retry", provider, type(exc).__name__, ) await _relaunch_browser(provider) if _browsers.get(provider) is None: raise return await _login_once(provider, params) raise async def _wait_cookie(page: object, name: str, tries: int, interval_ms: int) -> bool: """Ожидает появления cookie с заданным именем в контексте страницы. Returns: True если cookie появился в течение tries × interval_ms мс, иначе False. """ for _ in range(tries): cookies = await page.context.cookies() # type: ignore[attr-defined] if any(c["name"] == name for c in cookies): return True await page.wait_for_timeout(interval_ms) # type: ignore[attr-defined] return False async def _login_once(provider: str, params: dict) -> list[dict]: """Открывает страницу, заполняет форму, нажимает submit, ждёт success_cookie. Поддерживает список pre_click_selectors (клик по очереди, каждый non-fatal) и 2-шаговый submit Cian: после первого сабмита появляется экран «Введите пароль» — дозаполняем password и кликаем повторно (макс. 2 попытки). """ browser = _browsers.get(provider) assert browser is not None, "browser not launched" url: str = params["url"] email: str = params["email"] password: str = params["password"] email_selector: str = params["email_selector"] password_selector: str = params["password_selector"] submit_selector: str = params["submit_selector"] pre_click_selectors: list[str] = params.get("pre_click_selectors") or [] success_cookie: str = params.get("success_cookie", "") wait_ms: int = params.get("wait_ms") or BROWSER_WAIT_MS # Страница закрывается в finally: asyncio.CancelledError (BaseException, НЕ # Exception) обходит except-ветку ниже — без finally страница утекала бы при # отмене таска (например, на shutdown). _fetch_once делает то же. page = await browser.new_page() # type: ignore[attr-defined] try: try: await page.goto( # type: ignore[attr-defined] url, timeout=BROWSER_NAV_TIMEOUT_MS, wait_until="domcontentloaded" ) await page.wait_for_timeout(wait_ms) # type: ignore[attr-defined] # Pre-click sequence: каждый клик non-fatal (AnotherAccountBtn на fresh # headless-сессии отсутствует — молча пропускается). for sel in pre_click_selectors: try: await page.click(sel, timeout=12000) # type: ignore[attr-defined] await page.wait_for_timeout(1200) # type: ignore[attr-defined] except Exception: logger.info("login: pre-click skipped: %s", sel) await page.fill(email_selector, email, timeout=15000) # type: ignore[attr-defined] await page.fill(password_selector, password, timeout=15000) # type: ignore[attr-defined] # 2-шаговый submit: Cian после первого сабмита показывает экран # «Введите пароль» с тем же password_selector — дозаполняем и повторяем. cookie_found = False for attempt in range(2): await page.click(submit_selector, timeout=15000) # type: ignore[attr-defined] if success_cookie: cookie_found = await _wait_cookie( page, success_cookie, tries=20, interval_ms=500 ) if cookie_found: break # 2-step: если поле пароля ещё видимо — дозаполнить и повторить still_visible = False try: still_visible = await page.is_visible(password_selector) # type: ignore[attr-defined] except Exception: pass if attempt == 0 and still_visible: await page.fill(password_selector, password, timeout=15000) # type: ignore[attr-defined] await page.wait_for_timeout(800) # type: ignore[attr-defined] continue break logger.info( "tradein-browser[%s]: login success_cookie=%r present=%s", provider, success_cookie, cookie_found, ) cookies: list[dict] = await page.context.cookies() # type: ignore[attr-defined] logger.info( "tradein-browser[%s]: login OK url=%r email_selector=%r cookie_count=%d", provider, url, email_selector, len(cookies), ) except Exception as exc: screenshot_b64 = "" try: screenshot_bytes: bytes = await page.screenshot() # type: ignore[attr-defined] screenshot_b64 = base64.b64encode(screenshot_bytes).decode() except Exception: pass page_url = "" try: page_url = page.url # type: ignore[attr-defined] except Exception: pass raise LoginError({ "error": f"{type(exc).__name__}: {exc}", "page_url": page_url, "screenshot_b64": screenshot_b64, }) from exc finally: await page.close() # type: ignore[attr-defined] _page_counters[provider] = _page_counters.get(provider, 0) + 1 if _page_counters[provider] >= BROWSER_RECYCLE_PAGES: logger.info( "tradein-browser[%s]: recycle threshold (%d) достигнут после login, перезапуск", provider, BROWSER_RECYCLE_PAGES, ) await _relaunch_browser(provider) return cookies # ── entrypoint ───────────────────────────────────────────────────────────────── def build_app() -> web.Application: app = web.Application() app.on_startup.append(_on_startup) app.on_cleanup.append(_on_cleanup) app.router.add_get("/health", health_handler) app.router.add_post("/fetch", fetch_handler) app.router.add_post("/fetch-json", fetch_json_handler) app.router.add_post("/login", login_handler) app.router.add_get("/pacing", pacing_get_handler) app.router.add_put("/pacing", pacing_put_handler) return app if __name__ == "__main__": logger.info("tradein-browser: старт HTTP-сервера на порту %d", BROWSER_PORT) web.run_app(build_app(), host="0.0.0.0", port=BROWSER_PORT)