gendesign/tradein-mvp/browser/server.py

"""server.py — tradein-browser service main process.

Этот модуль запускается как точка входа контейнера ``tradein-browser``.
Он запускает camoufox **локально** внутри контейнера (AsyncCamoufox) и
экспонирует простой HTTP API на базе aiohttp:

    GET  /health     → {"status": "ok", "browsers": {"avito": bool, ...}}
    POST /fetch      → {"url","origin"(опц.),"cookies"(опц.)} → {"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"]). Каждый
    инстанс имеет geoip=True — фингерпринт привязан к гео прокси, поэтому смешивать
    источники в одном браузере нельзя. Egress-прокси сейчас общий для всех
    поставщиков (SCRAPER_PROXY_URL, #2616 шаг 2) — раздельные instance'ы остаются
    ради geoip-изоляции fingerprint'а и per-provider concurrency (ниже), не прокси.

    Конкурентность:
      * МЕЖДУ поставщиками — параллельно (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_CHALLENGE_WAIT_MS — бюджет ожидания QRATOR PoW-челленджа Авито (#3045),
                              мс (default: 30000). Челлендж-страница сама считает
                              proof-of-work в JS, ставит куку pow_solved и через
                              setTimeout(3000) делает window.location = location.href
                              (self-reload, URL не меняется). /fetch опрашивает
                              page.content() пока маркеры челленджа не исчезнут; по
                              истечении бюджета — ChallengeTimeoutError вместо тихой
                              отдачи заглушки. Действует ТОЛЬКО при обнаружении
                              маркеров челленджа в разметке — прочие провайдеры этот
                              путь никогда не задевают.
    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_BLOCK_RESOURCES (булев выключатель до #1812)
                              НЕ читается — см. _RETIRED_ENV; в проде он всё ещё
                              выставлен во всех трёх контейнерах и ни на что не влияет.
    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).
    SCRAPER_PROXY_URL       — http-прокси, ОБЩИЙ для всех поставщиков (avito/cian/
                              yandex/generic). #2616 шаг 2: per-provider
                              BROWSER_PROXY_AVITO/CIAN/YANDEX и legacy
                              AVITO_PROXY_URL/CIAN_PROXY_URL/YANDEX_PROXY_URL сняты —
                              все указывали на закрытые mobileproxy-аккаунты
                              (407/connection refused, проверено вживую #2613).

    ENVIRONMENT             — "production" в прод-контейнерах, иначе "dev" (дефолт).
                              #2616 шаг 1: прод + нет НИ override в теле, НИ
                              SCRAPER_PROXY_URL → отказ (503, явная причина), а НЕ
                              launch camoufox без proxy (= прямое подключение с IP
                              сервера). В dev отсутствие прокси легитимно (см.
                              _no_live_proxy).

Контракт /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 ────────────────────────────────────────────────────────

# #2616 шаг 1: признак окружения. "production" в прод-контейнерах (ENV: ENVIRONMENT,
# см. app.core.config.Settings.environment в основном backend). Дефолт "dev" — как и
# везде в проекте (Settings.environment: str = "dev").
ENVIRONMENT: str = os.environ.get("ENVIRONMENT", "dev")
IS_PROD: bool = ENVIRONMENT == "production"

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"))

# 30000: цепочка PoW-челленджа Авито — startPow() в JS, затем setTimeout(3000) на
# self-reload, затем повторная гидрация страницы. 3с таймера самой площадки — это
# ТОЛЬКО задержка перед reload, не бюджет на сам расчёт PoW: под headless-браузером
# и egress-прокси решение может занять заметно дольше, чем в обычном браузере
# пользователя. Живой замер 2026-08-21 (#3045): без ожидания челленджа 4 из 6
# карточек с органической навигацией отдавали 7891-байтную челлендж-страницу вместо
# контента — не бан (403/429 не было), просто уходили раньше, чем страница себя
# перезагрузила. 30с — запас с кратным резервом на решение + reload + догидрацию,
# не превращающий единичный фетч в минуту ожидания при реальном бане/сетевой пробе.
BROWSER_CHALLENGE_WAIT_MS: int = int(os.environ.get("BROWSER_CHALLENGE_WAIT_MS", "30000"))

# /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"))
# Сколько ждать события `load` на ПОВТОРЕ после гонки «execution context destroyed»
# (#2676). Только на повторе: happy-path остаётся на дешёвом FETCH_JSON_SETTLE_MS,
# иначе бесконечно дозагружающаяся страница удлиняла бы КАЖДЫЙ запрос. Ожидание
# best-effort — по таймауту всё равно пробуем evaluate.
FETCH_JSON_LOAD_WAIT_MS: int = int(os.environ.get("FETCH_JSON_LOAD_WAIT_MS", "15000"))

# Известные поставщики. "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.

    #2616 шаг 2: per-provider BROWSER_PROXY_AVITO/CIAN/YANDEX и legacy
    AVITO_PROXY_URL/CIAN_PROXY_URL/YANDEX_PROXY_URL сняты — все указывали на
    закрытые mobileproxy-аккаунты (#2613). SCRAPER_PROXY_URL (ASocks, http-прокси,
    playwright-совместимый) — единственный источник, общий для ВСЕХ поставщиков
    (включая generic). `provider` оставлен параметром сигнатуры (вызывающие
    передают его) — на случай возврата к per-provider egress в будущем.
    """
    return os.environ.get("SCRAPER_PROXY_URL")


def _no_live_proxy(provider: str, proxy_override: str | None) -> bool:
    """True если нет НИ explicit proxy в теле запроса, НИ SCRAPER_PROXY_URL.

    #2616 шаг 1: используется ТОЛЬКО в prod (IS_PROD) — в dev отсутствие прокси
    легитимно (прямое подключение для локальной разработки/тестов, см. #2616 issue
    пункт 2). Отдельная функция от `_provider_proxy`, потому что "нет живого прокси"
    здесь означает буквально "нет НИКАКОГО прокси" (ни override, ни env) — единственный
    случай, который код МОЖЕТ проверить без сетевой пробы. Env-переменная, что задана
    но мертва (407/connection refused), этот guard не триггерит — за живость значения
    отвечает мониторинг прокси-пула, не этот код.
    """
    return proxy_override is None and _provider_proxy(provider) is None


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 (пейсинг)
# #2164 P4: proxy-url, с которым СЕЙЧАС запущен инстанс провайдера (env или динамический
# из пула, переданный в теле /fetch). Нужен для политики «relaunch ТОЛЬКО при реальной
# смене прокси» — camoufox берёт proxy на launch, релонч дорогой, поэтому не релончим,
# если запрошенный proxy совпадает с текущим. None = запущен без прокси (прямое).
_launched_proxy: dict[str, str | None] = {}

# 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, proxy_override: str | None = None) -> None:
    """Запускает AsyncCamoufox для поставщика и кладёт в *_…[provider].

    proxy_override (#2164 P4): если задан — запускаем camoufox с ЭТИМ прокси (динамический
    из пула, переданный scraper-стороной в теле /fetch). None → env-прокси
    _provider_proxy(provider) как раньше. geoip=True (фингерпринт привязан к гео IP
    конкретного прокси — поэтому отдельный инстанс на каждого поставщика).

    Эффективный proxy-url сохраняется в _launched_proxy[provider] для политики
    «relaunch только при реальной смене» (см. _ensure_browser).
    """
    from camoufox.async_api import AsyncCamoufox

    proxy_url = proxy_override if proxy_override is not None else _provider_proxy(provider)
    proxy = _parse_proxy(proxy_url)
    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
    _launched_proxy[provider] = proxy_url
    logger.info(
        "tradein-browser[%s]: браузер запущен (proxy_override=%s)",
        provider,
        proxy_override is not None,
    )


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)
    _launched_proxy.pop(provider, None)


async def _relaunch_browser(provider: str, proxy_override: str | None = None) -> None:
    """Закрывает и заново запускает per-provider браузер (recycle / crash-recovery).

    proxy_override (#2164 P4): прокинуть конкретный прокси в новый инстанс. None →
    СОХРАНИТЬ текущий прокси инстанса (_launched_proxy) — recycle/crash-recovery не должны
    терять динамический прокси пула и не должны его менять (это НЕ смена прокси). Если
    инстанс ещё не запускался (_launched_proxy пуст) — None → env-прокси в _launch_browser.
    Caller держит _locks[provider], поэтому параллельных страниц на этом инстансе нет —
    teardown безопасен.
    """
    effective = proxy_override if proxy_override is not None else _launched_proxy.get(provider)
    logger.info("tradein-browser[%s]: перезапуск браузера", provider)
    await _close_browser(provider)
    await _launch_browser(provider, proxy_override=effective)


async def _try_launch_browser(provider: str, proxy_override: str | None = None) -> bool:
    """Одна попытка launch'а поставщика с санитизированным логом фейла.

    НЕ пробрасывает исключение наружу — возвращает True/False. Критично для
    resilience: фейл launch'а (прокси лёг → InvalidProxy) не должен валить
    startup или фоновую задачу. Креды прокси НЕ логируются: camoufox светит
    ``user:pass@host`` в тексте своего исключения, поэтому мы логируем только
    тип, без ``str(exc)``.

    При частичном фейле (CM создан, но __aenter__ не довёл) чистит состояние
    через _close_browser, чтобы следующий retry стартовал с чистого листа.
    """
    try:
        await _launch_browser(provider, proxy_override=proxy_override)
        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, proxy_override: str | None = None) -> bool:
    """Гарантирует поднятый инстанс поставщика с нужным прокси; lazy launch / relaunch.

    proxy_override (#2164 P4): желаемый динамический прокси из пула (тело /fetch). None →
    желаемый прокси = env (_provider_proxy). Политика «relaunch ТОЛЬКО при реальной смене»:
      - инстанса нет → launch с желаемым прокси;
      - инстанс есть, но _launched_proxy[provider] != желаемого → close + relaunch;
      - инстанс есть и прокси совпадает → no-op (НЕ релончим — camoufox дорого поднимать).
    При выключенном флаге на scraper-стороне proxy_override всегда None → желаемый=env=
    текущий → релонча нет, поведение идентично прежнему (golden-parity).

    Caller держит _locks[provider] (весь доступ к инстансу сериализован), поэтому здесь
    своей сериализации нет — двух одновременных launch'ей на один provider быть не может.

    Returns:
        True если браузер доступен (был или поднялся), False если launch не удался
        (прокси лежит) — caller отдаёт 503.

    При неудачном launch (прокси лёг) запускает фоновую retry-задачу для ЭТОГО
    поставщика (env-прокси). Поставщики независимы — фоновый retry поднимается только
    для тех, к кому был запрос, а не для всех сразу.
    """
    desired_proxy = proxy_override if proxy_override is not None else _provider_proxy(provider)
    if _browsers.get(provider) is not None:
        if _launched_proxy.get(provider) == desired_proxy:
            return True
        # Реальная смена прокси → пересобрать инстанс с новым прокси.
        logger.info(
            "tradein-browser[%s]: прокси изменился (override=%s) — relaunch перед fetch",
            provider,
            proxy_override is not None,
        )
        await _close_browser(provider)

    await _try_launch_browser(provider, proxy_override=proxy_override)
    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)
    _warn_retired_env()


# Переменные, которые окружение всё ещё задаёт, а код больше не читает. Ставим сюда
# ровно те, что реально видели в проде — молчаливо игнорируемая ручка опаснее
# отсутствующей: оператор ставит BROWSER_BLOCK_RESOURCES=false, чтобы посмотреть
# страницу с ресурсами, ничего не меняется, и он делает вывод не о переменной,
# а о блокировке.
_RETIRED_ENV: dict[str, str] = {
    # #1812 заменил булев выключатель на список типов. Блокировка при этом НЕ
    # ослабла: image глушится camoufox block_images (_launch_browser), font/media —
    # дефолтом BROWSER_BLOCK_RESOURCE_TYPES. Мёртв только сам выключатель.
    "BROWSER_BLOCK_RESOURCES": (
        "BROWSER_BLOCK_RESOURCE_TYPES (CSV типов; пустая строка = не блокировать)"
    ),
}


def _warn_retired_env() -> None:
    """Предупреждает про заданные env, которые код больше не читает (#2674)."""
    for name, replacement in _RETIRED_ENV.items():
        if os.environ.get(name) is not None:
            logger.warning(
                "tradein-browser: %s=%r задана, но НЕ читается кодом с #1812 — "
                "значение ни на что не влияет. Замена: %s. Уберите переменную из "
                "окружения, чтобы она не выглядела рабочим выключателем",
                name,
                os.environ[name],
                replacement,
            )


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') нет в PROVIDERS → fallthrough на host-detect
    # domclick.ru → generic. Намеренно (все поставщики делят один SCRAPER_PROXY_URL
    # с #2616 шаг 2, generic-путь эквивалентен).
    """
    explicit = body.get("provider") or body.get("source")
    if isinstance(explicit, str) and explicit in PROVIDERS:
        return explicit
    return _provider_for_url(url)


def _resolve_proxy_override(body: dict, provider: str) -> str | None:
    """Динамический прокси из тела /fetch (#2164 P4). None → env-прокси (текущее поведение).

    Scraper-сторона (BrowserFetcher за флагом use_proxy_pool_browser) кладёт в тело
    ``{"proxy": "http://user:pass@host:port", "proxy_kind": "http"}`` — прокси, взятый из
    пула через acquire(source). Если поля нет / пусто → None → инстанс поставщика берёт
    общий SCRAPER_PROXY_URL (#2616 шаг 2), поведение не меняется (ship-dark parity).

    camoufox НЕ умеет socks5 с авторизацией → принимаем ТОЛЬКО http/https-прокси; иной
    proxy_kind игнорируем (→ None, env-fallback) с предупреждением. Отсутствие proxy_kind
    трактуем как http (в пуле kind=http).
    """
    raw = body.get("proxy")
    if not isinstance(raw, str) or not raw:
        return None
    kind = body.get("proxy_kind")
    if isinstance(kind, str) and kind and kind.lower() not in ("http", "https"):
        logger.warning(
            "tradein-browser[%s]: proxy_kind=%r не поддерживается camoufox — env-прокси",
            provider,
            kind,
        )
        return None
    return raw


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)

    # origin (опционально) — same-site якорь, на который камуфокс зайдёт ПЕРЕД url
    # (органическая навигация: реальные cookies/Referer вместо холодного goto,
    # зеркалит /fetch-json #1917). Явного auto-derive из url здесь НЕТ (в отличие
    # от fetch-json) — caller не передал → остаётся None, origin-goto не выполняется,
    # поведение идентично прежнему (avito/cian/yandex его никогда не передают).
    origin: str | None = body.get("origin")

    # cookies (опционально) — dict name→value для инъекции в контекст страницы ПЕРЕД
    # навигацией (обход QRATOR-блока DomClick при валидной test-аккаунт сессии,
    # эмпирически подтверждено вживую 2026-07-04). Явного auto-derive нет — caller не
    # передал → остаётся None, инъекции нет, поведение идентично прежнему (avito/cian/
    # yandex его никогда не передают).
    cookies: dict | None = body.get("cookies")

    provider = _resolve_provider(body, url)
    proxy_override = _resolve_proxy_override(body, provider)

    # #2616 шаг 1: прод + нет НИ override, НИ env-прокси → явный отказ ДО лока/launch'а,
    # а не тихий заход camoufox напрямую с IP сервера (dev это легитимно, см. _no_live_proxy).
    if IS_PROD and _no_live_proxy(provider, proxy_override):
        logger.warning(
            "tradein-browser[%s]: /fetch отказ — нет прокси в теле и нет env-прокси "
            "(prod) — не подключаемся напрямую с IP сервера (#2616)",
            provider,
        )
        return web.json_response(
            {"error": "no proxy configured — refusing direct connection (prod)"}, status=503
        )

    lock = await _lock_for(provider)
    async with lock:
        # Браузер мог не подняться на старте (прокси лежал). Одна lazy-попытка —
        # вдруг прокси уже вернулся. НИКОГДА не фетчим без прокси/браузера: если
        # всё ещё None → 503 (caller словит как HTTPStatusError → failed fetch, не
        # crash). proxy_override (#2164 P4): relaunch camoufox с прокси из пула, если
        # он отличается от текущего launch-прокси инстанса (иначе no-op).
        if not await _ensure_browser(provider, proxy_override=proxy_override):
            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, origin=origin, cookies=cookies)
        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)
    proxy_override = _resolve_proxy_override(body, provider)

    # #2616 шаг 1: см. fetch_handler — прод + нет прокси вообще → отказ, не direct-IP.
    if IS_PROD and _no_live_proxy(provider, proxy_override):
        logger.warning(
            "tradein-browser[%s]: /fetch-json отказ — нет прокси в теле и нет env-прокси "
            "(prod) — не подключаемся напрямую с IP сервера (#2616)",
            provider,
        )
        return web.json_response(
            {"error": "no proxy configured — refusing direct connection (prod)"}, status=503
        )

    lock = await _lock_for(provider)
    async with lock:
        # Та же resilience что и в /fetch: браузер мог не подняться (прокси лежал).
        # Lazy-попытка, иначе 503 — без прокси/браузера не фетчим. proxy_override
        # (#2164 P4): relaunch с прокси из пула при реальной смене.
        if not await _ensure_browser(provider, proxy_override=proxy_override):
            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,
    *,
    origin: str | None = None,
    cookies: dict | None = None,
) -> str:
    """Одна попытка навигации; при краше браузера — relaunch и один retry.

    Caller держит _locks[provider] (нет параллельных страниц на этом инстансе),
    поэтому relaunch безопасен. Crash-relaunch сохраняет текущий прокси инстанса
    (_relaunch_browser reuse _launched_proxy) — динамический прокси пула не теряется.

    origin/cookies — см. _fetch_once. None (дефолт) не меняет поведение.
    """
    try:
        return await _fetch_once(provider, url, origin=origin, cookies=cookies)
    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, origin=origin, cookies=cookies)
        raise


# ── QRATOR PoW-челлендж Авито (#3045) ────────────────────────────────────────────
# Маркеры сняты живьём с challenge-страницы Авито 2026-08-21 (замер: 6 карточек,
# органическая навигация из выдачи, 4/6 ушли с челленджа раньше времени). Разметка
# площадки может поменяться со временем — при протухании маркеров переснять их
# заново вживую, а не гадать по памяти. Форма провайдер-агностична: детектор просто
# ищет строки в HTML, другие площадки (cian/yandex/generic) их никогда не отдают,
# поэтому ветка ожидания для них не включается.

# Признак самого челленджа: JS-функция startPow(), которую страница вызывает в
# DOMContentLoaded (см. хвост challenge-скрипта в #3045), либо заголовок блока
# «Доступ ограничен: проверка безопасности» — оба встречались на снятых страницах.
_CHALLENGE_MARKERS: tuple[str, ...] = (
    "startpow",
    "доступ ограничен: проверка безопасности",
)

# Признак БАН-страницы (не челлендж): «Доступ ограничен: проблема с IP» — статика
# без PoW-скрипта, приходит с 403/429 и заметно меньше challenge-страницы весом.
# Ждать тут бессмысленно — адрес заблокирован, а не временно проверяется.
_BAN_MARKERS: tuple[str, ...] = ("доступ ограничен: проблема с ip",)


class ChallengeTimeoutError(Exception):
    """PoW-челлендж не снялся за BROWSER_CHALLENGE_WAIT_MS.

    Caller должен трактовать как временный отказ (retry/backoff), НЕ как валидный
    контент — раньше caller получал 7891-байтную challenge-страницу и парсер либо
    падал на ней, либо молча ничего не находил (#3045).
    """


class BanPageDetectedError(Exception):
    """Площадка отдала бан-страницу («проблема с IP») вместо контента/челленджа.

    В отличие от ChallengeTimeoutError ждать здесь бессмысленно: адрес забанен, а
    не проходит временную проверку — поднимается сразу, без траты
    BROWSER_CHALLENGE_WAIT_MS.
    """


def _is_pow_challenge(html: str) -> bool:
    """True, если HTML — QRATOR PoW-челлендж Авито (см. _CHALLENGE_MARKERS)."""
    lower = html.lower()
    return any(marker in lower for marker in _CHALLENGE_MARKERS)


def _is_ban_page(html: str) -> bool:
    """True, если HTML — бан-страница «проблема с IP» (см. _BAN_MARKERS)."""
    lower = html.lower()
    return any(marker in lower for marker in _BAN_MARKERS)


async def _wait_out_pow_challenge(page: object, provider: str, url: str) -> str:
    """Опрашивает page.content() пока не исчезнут маркеры PoW-челленджа.

    Страница перезагружает СЕБЯ САМА (`window.location = location.href`) после
    решения PoW — URL не меняется, поэтому page.wait_for_url тут не годится;
    опрашиваем контент с шагом ~1с вместо этого. По истечении
    BROWSER_CHALLENGE_WAIT_MS — ChallengeTimeoutError, а не тихая отдача
    challenge-страницы как будто это валидный контент.

    После снятия челленджа даём странице догидрироваться тем же BROWSER_WAIT_MS,
    каким ждём обычную навигацию (второй таймаут не изобретаем).
    """
    poll_interval_ms = 1000
    elapsed_ms = 0
    html: str = await page.content()  # type: ignore[attr-defined]
    while _is_pow_challenge(html) and elapsed_ms < BROWSER_CHALLENGE_WAIT_MS:
        await page.wait_for_timeout(poll_interval_ms)  # type: ignore[attr-defined]
        elapsed_ms += poll_interval_ms
        html = await page.content()  # type: ignore[attr-defined]

    if _is_pow_challenge(html):
        raise ChallengeTimeoutError(
            f"tradein-browser[{provider}]: PoW-челлендж не снялся за "
            f"{BROWSER_CHALLENGE_WAIT_MS}мс url={url!r}"
        )

    logger.info(
        "tradein-browser[%s]: PoW-челлендж снят за ~%dмс, догидрация url=%r",
        provider,
        elapsed_ms,
        url,
    )
    if BROWSER_WAIT_MS > 0:
        await page.wait_for_timeout(BROWSER_WAIT_MS)  # type: ignore[attr-defined]
    return await page.content()  # type: ignore[attr-defined]


async def _fetch_once(
    provider: str,
    url: str,
    *,
    origin: str | None = None,
    cookies: dict | None = None,
) -> str:
    """Открывает СОБСТВЕННУЮ страницу, переходит по URL, ждёт JS, возвращает HTML.

    Caller держит _locks[provider], поэтому страницы на этом инстансе не
    параллелятся — recycle через _relaunch_browser безопасен прямо здесь.

    origin (опционально) — same-site якорь (например SERP), на который камуфокс
    заходит ПЕРЕД целевым url: органическая навигация с реальными cookies/Referer
    вместо холодного прямого захода (зеркалит _fetch_json_once, #1917 — DomClick
    card-fetch, эмпирически подтверждено вживую 2026-07-04). None (дефолт) → поведение
    не меняется, ровно один goto(url) как раньше (avito/cian/yandex не передают origin).

    cookies (опционально) — dict cookie_name→value для инъекции в контекст страницы
    ДО любой навигации (обход QRATOR-блока DomClick при валидной test-аккаунт сессии,
    эмпирически подтверждено вживую 2026-07-04). Провайдер-агностично: домен НЕ
    захардкожен, а выводится из hostname целевого url (с ведущей точкой — покрывает
    поддомены, зеркалит реальную DomClick cookie-scope ".domclick.ru" для
    ekaterinburg.domclick.ru/spb.domclick.ru/etc, но работает для любого хоста, не
    только DomClick) — механизм пригоден для будущих caller'ов (Cian/Avito/Yandex).
    None (дефолт) → без инъекции, поведение не меняется (avito/cian/yandex сейчас
    cookies не передают).
    """
    browser = _browsers.get(provider)
    assert browser is not None, "browser not launched"

    page = await browser.new_page()  # type: ignore[attr-defined]
    try:
        if cookies:
            cookie_domain = f".{urlparse(url).hostname or ''}"
            await page.context.add_cookies(  # type: ignore[attr-defined]
                [
                    {"name": name, "value": value, "domain": cookie_domain, "path": "/"}
                    for name, value in cookies.items()
                ]
            )
        await _apply_resource_block(page)
        await _pace_provider(provider)
        if origin:
            await page.goto(  # type: ignore[attr-defined]
                origin, timeout=BROWSER_NAV_TIMEOUT_MS, wait_until="domcontentloaded"
            )
            await page.wait_for_timeout(FETCH_JSON_SETTLE_MS)  # type: ignore[attr-defined]
        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]

        # Бан-страница («проблема с IP») распознаётся и падает СРАЗУ, без траты
        # BROWSER_CHALLENGE_WAIT_MS — ждать там нечего, адрес заблокирован (#3045).
        if _is_ban_page(html):
            raise BanPageDetectedError(
                f"tradein-browser[{provider}]: бан-страница (проблема с IP) url={url!r}"
            )
        # PoW-челлендж (QRATOR) — в отличие от бана снимается сам по себе; ждём его
        # прохождения вместо того, чтобы вернуть 7891-байтную заглушку как контент.
        if _is_pow_challenge(html):
            html = await _wait_out_pow_challenge(page, provider, url)
    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 безопасен. Crash-relaunch сохраняет текущий прокси
    инстанса (_relaunch_browser reuse _launched_proxy).
    """
    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
            )
        # #2676 обрабатывается ВНУТРИ _fetch_json_once (повтор evaluate на той же
        # странице). Повторять здесь, на свежей странице, бесполезно: пересоздание
        # воспроизводит ту же первую навигацию origin'а — проверено на проде 06.08,
        # обе попытки упали одинаково.
        raise


# In-page retry самого fetch() при СЕТЕВОМ throw (TypeError/"NetworkError"), #1917:
# первый XHR после навигации иногда падает до готовности стека. Повтор внутри страницы
# (~retryDelayMs) дешевле полной ре-навигации (~30-45с/дом). HTTP-статусы (4xx/5xx) —
# это успешный resp, НЕ ретраим (решает caller).
_IN_PAGE_FETCH_JS = """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;
}"""


async def _wait_for_load_best_effort(page: object, provider: str) -> None:
    """Ждёт `load`, но не даёт ожиданию сорвать саму попытку (реклама/трекеры могут
    держать страницу «загружающейся» бесконечно). Таймаут логируется, не пробрасывается.
    """
    try:
        await page.wait_for_load_state("load", timeout=FETCH_JSON_LOAD_WAIT_MS)  # type: ignore[attr-defined]
    except Exception as exc:
        logger.info(
            "tradein-browser[%s]: load не дождались (%s), пробуем evaluate как есть",
            provider,
            type(exc).__name__,
        )


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]
        payload = {
            "url": url,
            "method": method,
            "headers": headers or {},
            "body": body,
            "retries": FETCH_JSON_INPAGE_RETRIES,
            "retryDelayMs": FETCH_JSON_RETRY_DELAY_MS,
        }
        try:
            result: dict = await page.evaluate(_IN_PAGE_FETCH_JS, payload)  # type: ignore[attr-defined]
        except Exception as exc:
            if not _is_page_context_lost(exc):
                raise
            # #2676, живой прод 06.08: страница origin уходит в клиентскую навигацию
            # ПОСЛЕ load, поэтому ни settle, ни ожидание load её не опережают, а повтор
            # с новой страницей воспроизводит ту же первую навигацию (проверено на
            # проде: две попытки подряд — та же ошибка). Навигация при этом ОДНА:
            # к моменту исключения она уже случилась и у страницы НОВЫЙ контекст.
            # Поэтому повторяем evaluate на ЭТОЙ же странице — стоит миллисекунды.
            logger.warning(
                "tradein-browser[%s]: контекст страницы пересоздан навигацией, "
                "повтор evaluate на той же странице: %s",
                provider,
                url,
            )
            await _wait_for_load_best_effort(page, provider)
            await page.wait_for_timeout(FETCH_JSON_SETTLE_MS)  # type: ignore[attr-defined]
            result = await page.evaluate(_IN_PAGE_FETCH_JS, payload)  # 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-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
    )


# Литерал playwright, а не строка из нашего лога: driver 1.60.0 (в образе сайдкара)
# бросает ровно «Execution context was destroyed» / «... , most likely because of a
# navigation.» — обе формы начинаются одинаково, поэтому хватает одного маркера.
# Проверено grep'ом по playwright/driver/package/lib/coreBundle.js в живом контейнере.
_PAGE_CONTEXT_LOST_MARKER = "execution context was destroyed"


def _is_page_context_lost(exc: BaseException) -> bool:
    """Страница потеряла JS-контекст (ушла в навигацию между goto и evaluate), #2676.

    НЕ краш браузера: инстанс жив, потеряна одна страница. Поэтому обрабатывается
    отдельно от _is_browser_crash — relaunch здесь стоил бы ~10-20с и тёплый профиль
    (cookies/фингерпринт инстанса) ради браузера, с которым всё в порядке.
    """
    return _PAGE_CONTEXT_LOST_MARKER in str(exc).lower()


# ── 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": <p>, "interval_s": <float>} → обновляет 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"])

    # #2616 шаг 1: см. fetch_handler — прод + нет env-прокси (login override не берёт) →
    # отказ, не direct-IP.
    if IS_PROD and _no_live_proxy(provider, None):
        logger.warning(
            "tradein-browser[%s]: /login отказ — нет env-прокси (prod) — "
            "не подключаемся напрямую с IP сервера (#2616)",
            provider,
        )
        return web.json_response(
            {"error": "no proxy configured — refusing direct connection (prod)"}, status=503
        )

    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)