gendesign/backend/app/services/scrapers/stealth.py

"""Stealth-fetch utilities for наш.дом.рф scraping.

Pure-httpx hits ServicePipe WAF JS-challenge even with Playwright-exported
cookies (different TLS-fingerprint = blocked). The proven pattern from
data/sql/30_scrape_domrf.py is to issue `fetch()` calls *inside* the live
Chromium page — same JA3 fingerprint as a real browser, cookies/CSP/CORS
all handled natively.

BrowserSession keeps the browser+page open for the lifetime of a sweep and
exposes get_json() that calls page.evaluate(fetch_in_browser).
"""

from __future__ import annotations

import asyncio
import json
import logging
import random
from typing import Any
from urllib.parse import urlencode, urlsplit

from playwright.async_api import Browser, BrowserContext, Page, async_playwright

logger = logging.getLogger(__name__)

BASE_URL = "https://xn--80az8a.xn--d1aqf.xn--p1ai"

# Real Chrome 147 fingerprint (Win10 x64). Keep in sync with data/sql/30_scrape_domrf.py.
USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36"
)

# Максимум одновременных in-page fetch() на одну BrowserSession;
# поднимали 3→6→8 под asyncio.gather fan-out, на наш.дом.рф ServicePipe WAF-tolerant.
# Это МОДУЛЬНЫЙ ДЕФОЛТ — используется nspd/прочими скраперами вне /сервисы/* path
# family. KN-sweep с 2026-05 ловит volume-ban WAF (#1945) и переопределяет это через
# BrowserSession(concurrency=...) на settings.scrape_kn_browser_concurrency=2; после
# WAF hard-ban 2026-05-24 (#2443) domrf_catalog.py / domrf_catalog_object.py — тот же
# /сервисы/* path family — тоже переопределяют на throttled настройки (#2445 D2).
_BROWSER_CONCURRENCY = 8

# Дефолтное окно inter-request jitter (мс). KN-sweep ширит его через
# BrowserSession(jitter_min_ms=, jitter_max_ms=) против rate-бана (#1945).
_DEFAULT_JITTER_MIN_MS = 600
_DEFAULT_JITTER_MAX_MS = 1500


def parse_proxy_url(proxy_url: str | None) -> dict[str, str] | None:
    """Распарсить ``http://user:pass@host:port`` в Playwright proxy-dict.

    Playwright ждёт ``{"server": "http://host:port", "username": ..., "password": ...}``
    (creds — отдельные ключи, НЕ в server-URL). Возвращает None для пустого/None
    входа → вызывающий код запускает Chromium без прокси (поведение по умолчанию).

    Тот же URL-формат, что использует tradein-стек (SCRAPER_PROXY_URL).
    """
    if not proxy_url:
        return None
    parts = urlsplit(proxy_url)
    if not parts.hostname:
        # НЕ эхо-им сырой proxy_url — он содержит пароль, а это исключение
        # всплывает в kn_scrape_runs.error / log_progress / Sentry (#1945 sec-review).
        raise ValueError(f"proxy URL без host (scheme={parts.scheme!r})")
    scheme = parts.scheme or "http"
    server = f"{scheme}://{parts.hostname}"
    if parts.port:
        server += f":{parts.port}"
    out: dict[str, str] = {"server": server}
    if parts.username:
        out["username"] = parts.username
    if parts.password:
        out["password"] = parts.password
    return out


# Маппинг region_code → URL-сегмент города для реалистичного Referer.
# Не обязан быть исчерпывающим — fallback на /новостройки/строящиеся/.
REGION_LANDING_PATH = {
    66: "/новостройки/строящиеся/екатеринбург/",
    77: "/новостройки/строящиеся/москва/",
    78: "/новостройки/строящиеся/санкт-петербург/",
    50: "/новостройки/строящиеся/московская-область/",
    47: "/новостройки/строящиеся/ленинградская-область/",
}


def make_referer(region_code: int) -> str:
    path = REGION_LANDING_PATH.get(region_code, "/новостройки/строящиеся/")
    return BASE_URL + path


async def jitter_sleep(min_ms: int = 600, max_ms: int = 1500) -> None:
    delay = random.uniform(min_ms / 1000, max_ms / 1000)
    await asyncio.sleep(delay)


# JS executed inside the live page. Returns {ok, status, body, contentType}.
# Auth header is hardcoded `Basic MTpxd2U=` (decodes to `1:qwe`, the public
# debug auth shipped in the site's frontend bundle for the kn API).
_FETCH_JS = """
async ({url, auth}) => {
  const headers = {'Accept': 'application/json, text/plain, */*'};
  if (auth) headers['Authorization'] = auth;
  try {
    const r = await fetch(url, {credentials: 'include', headers});
    const ctype = r.headers.get('content-type') || '';
    const body = await r.text();
    return {ok: r.ok, status: r.status, body, contentType: ctype};
  } catch (e) {
    return {ok: false, status: 0, body: String(e), contentType: ''};
  }
}
"""


class WafBlockedError(RuntimeError):
    """Server returned non-JSON (likely ServicePipe JS-challenge)."""


class BrowserSession:
    """Keeps a live Playwright Chromium page; issues kn-API fetches inside it.

    Use as async context manager. Concurrency is bounded by an asyncio.Semaphore
    of size ``_BROWSER_CONCURRENCY`` (the same page can serve multiple concurrent
    fetches via JS-await).
    """

    def __init__(
        self,
        region_code: int = 66,
        headed: bool = False,
        auth: str | None = "Basic MTpxd2U=",
        load_state: str | None = None,
        save_state: str | None = None,
        concurrency: int | None = None,
        jitter_min_ms: int | None = None,
        jitter_max_ms: int | None = None,
        proxy_url: str | None = None,
    ) -> None:
        self.region_code = region_code
        self.headed = headed
        self.auth = auth
        # Path to a Playwright storage_state JSON (cookies + localStorage).
        # If load_state is set, the context is created with these cookies preloaded —
        # ServicePipe sees a returning user, no fresh JS-challenge.
        # If save_state is set, the post-bootstrap state is dumped to that path
        # so it can be committed and reused on a server.
        self.load_state = load_state
        self.save_state = save_state
        # Per-instance concurrency (#1945): KN-sweep throttles to 2 against the
        # volume-ban; catalog/nspd callers pass nothing → keep the module default
        # _BROWSER_CONCURRENCY=8 (их поведение НЕ меняется).
        self._concurrency = concurrency if concurrency is not None else _BROWSER_CONCURRENCY
        # Per-instance inter-request jitter window (#1945). None → module default.
        self._jitter_min_ms = jitter_min_ms if jitter_min_ms is not None else _DEFAULT_JITTER_MIN_MS
        self._jitter_max_ms = jitter_max_ms if jitter_max_ms is not None else _DEFAULT_JITTER_MAX_MS
        # Optional rotating proxy (#1945). None → Chromium launches direct (default).
        self._proxy = parse_proxy_url(proxy_url)
        self._pw: Any = None
        self._browser: Browser | None = None
        self._context: BrowserContext | None = None
        self._page: Page | None = None
        self._sem = asyncio.Semaphore(self._concurrency)
        self._request_count = 0
        self._warmed_up = False

    async def __aenter__(self) -> BrowserSession:
        await self._bootstrap()
        return self

    async def __aexit__(self, *exc: Any) -> None:
        if self._browser is not None:
            await self._browser.close()
        if self._pw is not None:
            await self._pw.stop()

    async def _bootstrap(self) -> None:
        landing_url = make_referer(self.region_code)
        logger.info("bootstrap: opening %s", landing_url)
        self._pw = await async_playwright().start()
        launch_kwargs: dict[str, Any] = {"headless": not self.headed}
        ctx_kwargs: dict[str, Any] = {
            "user_agent": USER_AGENT,
            "locale": "ru-RU",
            "viewport": {"width": 1920, "height": 1080},
        }
        if self._proxy is not None:
            # Прокси задаётся и на launch, и на context (Playwright требует на
            # browser-level для Chromium; context-level дублирует для надёжности).
            launch_kwargs["proxy"] = self._proxy
            ctx_kwargs["proxy"] = self._proxy
            logger.info("bootstrap: routing Chromium through proxy %s", self._proxy["server"])
        self._browser = await self._pw.chromium.launch(**launch_kwargs)
        if self.load_state:
            ctx_kwargs["storage_state"] = self.load_state
            logger.info("bootstrap: loading saved storage_state from %s", self.load_state)
        self._context = await self._browser.new_context(**ctx_kwargs)
        self._page = await self._context.new_page()
        await self._page.goto(landing_url, wait_until="domcontentloaded", timeout=30_000)
        try:
            await self._page.wait_for_load_state("networkidle", timeout=10_000)
        except Exception:
            pass
        if self.save_state:
            await self._context.storage_state(path=self.save_state)
            logger.info("bootstrap: storage_state saved to %s", self.save_state)
        logger.info("bootstrap: page ready, WAF challenge passed")

    async def warm_up(self, force: bool = False) -> None:
        """Visit catalog listing to obtain WAF cookies (___dmpkit___, domain_sid).

        DOM.РФ WAF (накатан между 2026-05-17 и 2026-05-24) требует session cookies для
        любого запроса на /сервисы/api/object/*/*. Cookies выдаются JS-челленджем при
        visit на /сервисы/каталог-новостроек/. Один warm-up на BrowserSession достаточен —
        cookies валидны для всего домена.

        Idempotent: повторные вызовы без force=True — no-op после первого успешного warm_up.
        """
        if self._warmed_up and not force:
            return
        if self._context is None:
            raise RuntimeError("BrowserSession not bootstrapped — call inside `async with` block")
        page = await self._context.new_page()
        try:
            await page.goto(
                f"{BASE_URL}/сервисы/каталог-новостроек/",
                wait_until="domcontentloaded",
                timeout=30_000,
            )
            await page.wait_for_timeout(2000)  # JS challenge ставит cookies async
            cookies = await self._context.cookies()
            names = {c["name"] for c in cookies}
            critical = {"___dmpkit___", "domain_sid"}
            got = names & critical
            if not got:
                logger.warning(
                    "warm_up: WAF cookies missing after catalog visit (have: %s)",
                    sorted(names),
                )
            else:
                logger.info("warm_up: got WAF cookies %s (total=%d)", sorted(got), len(cookies))
            self._warmed_up = True
        finally:
            await page.close()

    async def get_json(self, path: str, params: dict[str, Any]) -> dict[str, Any]:
        """Fetch JSON via fetch() in browser. Retries with backoff on transient errors."""
        if self._page is None:
            raise RuntimeError("BrowserSession not bootstrapped")
        url = BASE_URL + path + "?" + urlencode(params)
        last_err: Exception | None = None
        for attempt in range(5):
            async with self._sem:
                await jitter_sleep(self._jitter_min_ms, self._jitter_max_ms)
                try:
                    self._request_count += 1
                    result = await self._page.evaluate(_FETCH_JS, {"url": url, "auth": self.auth})
                except Exception as e:
                    last_err = e
                    logger.warning("evaluate err attempt=%d: %r", attempt, e)
                    await asyncio.sleep(2**attempt)
                    continue
            status = result.get("status", 0)
            body = result.get("body", "")
            ctype = result.get("contentType", "")
            if status in (429,) or status >= 500 or status == 0:
                last_err = RuntimeError(f"transient status={status}")
                logger.warning("transient status=%d attempt=%d, backing off", status, attempt)
                await asyncio.sleep(2**attempt)
                continue
            if "application/json" not in ctype:
                # WAF returned the JS challenge HTML; re-bootstrap could help, but for now fail.
                raise WafBlockedError(
                    f"non-JSON response: status={status} ctype={ctype} body[:120]={body[:120]!r}"
                )
            if status != 200:
                raise RuntimeError(f"http {status}: {body[:200]}")
            return json.loads(body)
        raise RuntimeError(f"max retries exhausted: {last_err!r}")

    @property
    def request_count(self) -> int:
        return self._request_count

    async def download_binary(self, url: str) -> bytes:
        """Download a binary asset (e.g. photo PNG) reusing browser cookies/auth.

        Uses Playwright APIRequest which goes through the browser context — same
        cookies, same TLS fingerprint as the page itself.

        Ретраи с backoff на транзиентных ответах (429 / 5xx / 0) — так же, как в
        get_json выше (#2464). Раньше их не было: один 429 под тем же WAF, под
        которым get_json переживает до пяти попыток, ронял загрузку картинки
        насовсем, и вызывающий (download_plan_image, download_photos) записывал
        это в лог как «не удалось» — неотличимо от «файла нет».

        Непереходные коды (403, 404) поднимаются сразу, без ожидания: повтор их
        не изменит, а под WAF лишний стук вредит.
        """
        if self._context is None:
            raise RuntimeError("BrowserSession not bootstrapped")
        last_err: Exception | None = None
        for attempt in range(5):
            async with self._sem:
                await jitter_sleep(200, 500)  # Lighter throttle for static assets.
                self._request_count += 1
                try:
                    resp = await self._context.request.get(
                        url,
                        headers={"Authorization": self.auth} if self.auth else {},
                    )
                except Exception as e:
                    last_err = e
                    logger.warning("download_binary err attempt=%d url=%s: %r", attempt, url, e)
                    await asyncio.sleep(2**attempt)
                    continue
                status = resp.status
                if status == 200:
                    return await resp.body()
                body = await resp.text()
            # Разбор статуса — ВНЕ семафора: sleep не должен держать слот.
            if status in (429,) or status >= 500:
                last_err = RuntimeError(f"binary transient status={status}")
                logger.warning(
                    "download_binary transient status=%d attempt=%d url=%s, backing off",
                    status,
                    attempt,
                    url,
                )
                await asyncio.sleep(2**attempt)
                continue
            raise RuntimeError(f"binary http {status}: {body[:200]}")
        raise RuntimeError(f"binary max retries exhausted: {last_err!r}")