fix(scrapers): throttle + optional proxy levers for KN flats loader (#1945)

DOM.РФ WAF банит на full sweep (~17k req @ concurrency=8): сессия,
тронувшая gated /сервисы/api/object/* extras, получает 403 → cookies
сессии флагуются → последующие запросы (incl. flats) тоже 403.

Эмпирически (prod worker, throttle@concurrency=2, без прокси):
  • flats-only, 100 объектов / 104 req — 0 WAF-бан;
  • mixed flats+extras — extras 403 сразу, затем травит сессию (flats→403).
Значит throttle снижает rate, но НЕ устраняет session-poisoning от extras;
прокси (mobile-proxy, как в tradein) — следующий рычаг, off-by-default.

- config: scrape_kn_browser_concurrency=2, scrape_kn_request_jitter_{min,max}_ms,
  scrape_kn_proxy_url=None (прокси опционален, дефолт — прямое подключение).
- stealth.BrowserSession: per-instance concurrency / jitter / proxy_url;
  parse_proxy_url(http://user:pass@host:port → Playwright proxy-dict).
  Модульный _BROWSER_CONCURRENCY=8 не тронут — catalog/nspd без изменений.
- domrf_kn.run_region_sweep + workers/tasks/scrape_kn: прокидывают новые
  settings ТОЛЬКО в KN-sweep BrowserSession.
- tests: parse_proxy_url, threading concurrency/jitter/proxy, defaults unchanged.

backend/app/core/config.py

@@ -114,6 +114,33 @@ class Settings(BaseSettings):
     # Empty string = endpoint disabled.
     scrape_admin_token: str = ""
 
+    # ── #1945 KN-loader anti-ban (throttle + optional proxy) ──────────────────
+    # DOM.РФ WAF банит IP по volume/rate (HTTP 403 «Доступ заблокирован», БЕЗ
+    # captcha — подтверждено): per-region sweep гонит ~1548 объектов × 11
+    # endpoint'ов ≈ 17k запросов через один браузер. На старой concurrency=8
+    # WAF банил VPS-IP mid-sweep. Лечим двумя рычагами.
+    #
+    # Рычаг 1 — throttle. Ограничивает число одновременных in-page fetch()
+    # ТОЛЬКО для KN-sweep BrowserSession (другие скраперы — nspd/catalog —
+    # продолжают использовать модульный дефолт _BROWSER_CONCURRENCY=8 без
+    # изменений). 2 — эмпирически безопасный потолок против volume-бана.
+    # ENV: SCRAPE_KN_BROWSER_CONCURRENCY.
+    scrape_kn_browser_concurrency: int = 2
+    # Окно случайной паузы (мс) между запросами KN-sweep. Шире дефолта
+    # (600–1500), чтобы размазать запросы во времени и не триггерить rate-ban.
+    # min < max обязателен (иначе random.uniform отдаст границу). При throttle
+    # ширим до 1200–3000. ENV: SCRAPE_KN_REQUEST_JITTER_MIN_MS / _MAX_MS.
+    scrape_kn_request_jitter_min_ms: int = 1200
+    scrape_kn_request_jitter_max_ms: int = 3000
+    # Рычаг 2 — прокси (ОПЦИОНАЛЬНО, default None → прямое подключение, поведение
+    # без изменений). Когда задан — KN-sweep BrowserSession запускает Chromium
+    # через этот прокси (формат http://user:pass@host:port; парсится в
+    # Playwright proxy={server,username,password}). Сильнейший рычаг против
+    # IP-бана: переиспользует тот же mobile-proxy паттерн, что tradein-стек
+    # (SCRAPER_PROXY_URL). Заводится ТОЛЬКО через окружение прод-контейнера.
+    # ENV: SCRAPE_KN_PROXY_URL.
+    scrape_kn_proxy_url: str | None = None
+
     # NSPD-scraper (Playwright) УДАЛЁН 2026-05-11. Сменён на bulk geo-fetcher
     # через rosreestr2coord — запускается вручную через /admin/scrape/geo.
     # Settings оставлены deprecated на случай отката (можно удалить позже).

backend/app/services/scrapers/domrf_kn.py

@@ -1142,7 +1142,7 @@ async def fetch_obj_checks(sess: BrowserSession, obj_id: int) -> tuple[Any, str]
 # ── _fetch_*_safe wrappers for asyncio.gather in Phase B/C ───────────────────
 # Каждый wrapper возвращает (kind, full_url, result_or_exception).
 # Exceptions НЕ raise — помещаются в возвращаемый tuple.
-# BrowserSession._sem (Semaphore(_BROWSER_CONCURRENCY)=8) bounds concurrency per-request.
+# BrowserSession._sem (size = per-instance concurrency, #1945 KN=2) bounds concurrency.
 
 
 async def _fetch_flats_safe(
@@ -1660,9 +1660,18 @@ async def run_region_sweep(
     photos_dir: Path | str | None = None,
     place_override: str | None = None,
     resume_from_run_id: int | None = None,
+    browser_concurrency: int | None = None,
+    request_jitter_min_ms: int | None = None,
+    request_jitter_max_ms: int | None = None,
+    proxy_url: str | None = None,
 ) -> dict[str, Any]:
     """Full sweep for one region: bootstrap → fetch objects → per-object processing.
 
+    #1945 anti-ban: ``browser_concurrency`` / ``request_jitter_*_ms`` / ``proxy_url``
+    прокидываются в BrowserSession sweep'а, чтобы троттлить/проксировать ТОЛЬКО KN-
+    прогон (другие скраперы используют модульные дефолты stealth.py без изменений).
+    None у каждого = дефолт stealth.py (concurrency=8, jitter 600–1500, без прокси).
+
     Resume-aware: если `resume_from_run_id` задан, читаем objects_snapshot и
     progress_obj_index из исходного run и продолжаем с этого индекса. Новый
     run_id создаётся (с resumed_from_run_id ссылкой), исходный остаётся
@@ -1802,6 +1811,10 @@ async def run_region_sweep(
             headed=headed,
             load_state=load_state,
             save_state=save_state,
+            concurrency=browser_concurrency,
+            jitter_min_ms=request_jitter_min_ms,
+            jitter_max_ms=request_jitter_max_ms,
+            proxy_url=proxy_url,
         ) as sess:
             # Warm-up: visit /сервисы/каталог-новостроек/ to obtain WAF cookies
             # (___dmpkit___, domain_sid). Required since 2026-05-24 — without these
@@ -1881,7 +1894,8 @@ async def run_region_sweep(
 
             # ── Phase B/C — per-object processing (resumable, parallel per-object) ─
             # Все endpoint'ы одного obj_id запускаются параллельно через asyncio.gather.
-            # BrowserSession._sem (Semaphore(_BROWSER_CONCURRENCY)=8) ограничивает запросы.
+            # BrowserSession._sem (size = browser_concurrency, #1945 троттлит KN до 2)
+            # ограничивает запросы.
             # DB upserts выполняются последовательно после gather — один db Session
             # не thread-safe для параллельной записи.
             pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT

backend/app/services/scrapers/stealth.py

@@ -17,7 +17,7 @@ import json
 import logging
 import random
 from typing import Any
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlsplit
 
 from playwright.async_api import Browser, BrowserContext, Page, async_playwright
 
@@ -33,8 +33,43 @@ USER_AGENT = (
 
 # Максимум одновременных in-page fetch() на одну BrowserSession;
 # поднимали 3→6→8 под asyncio.gather fan-out, на наш.дом.рф ServicePipe WAF-tolerant.
+# Это МОДУЛЬНЫЙ ДЕФОЛТ — используется catalog/nspd/прочими скраперами. KN-sweep с
+# 2026-05 ловит volume-ban WAF (#1945) и переопределяет это через
+# BrowserSession(concurrency=...) на settings.scrape_kn_browser_concurrency=2.
 _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:
+        raise ValueError(f"proxy URL без host: {proxy_url!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 = {
@@ -94,6 +129,10 @@ class BrowserSession:
         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
@@ -105,11 +144,20 @@ class BrowserSession:
         # 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(_BROWSER_CONCURRENCY)
+        self._sem = asyncio.Semaphore(self._concurrency)
         self._request_count = 0
         self._warmed_up = False
 
@@ -127,12 +175,19 @@ class BrowserSession:
         landing_url = make_referer(self.region_code)
         logger.info("bootstrap: opening %s", landing_url)
         self._pw = await async_playwright().start()
-        self._browser = await self._pw.chromium.launch(headless=not self.headed)
+        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)
@@ -193,7 +248,7 @@ class BrowserSession:
         last_err: Exception | None = None
         for attempt in range(5):
             async with self._sem:
-                await jitter_sleep()
+                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})

backend/app/workers/tasks/scrape_kn.py

@@ -210,6 +210,11 @@ def scrape_kn_region(
                 fetch_flats=fetch_flats,
                 extras=extras,
                 download_photos_binary=download_photos,
+                # #1945 anti-ban: throttle + optional proxy для KN-sweep.
+                browser_concurrency=settings.scrape_kn_browser_concurrency,
+                request_jitter_min_ms=settings.scrape_kn_request_jitter_min_ms,
+                request_jitter_max_ms=settings.scrape_kn_request_jitter_max_ms,
+                proxy_url=settings.scrape_kn_proxy_url,
             )
         )
 
@@ -289,5 +294,10 @@ def resume_kn_run(self: Any, run_id: int) -> dict[str, Any]:
                 developers=developers,
                 load_state=state_path,
                 resume_from_run_id=run_id,
+                # #1945 anti-ban: throttle + optional proxy для KN-sweep.
+                browser_concurrency=settings.scrape_kn_browser_concurrency,
+                request_jitter_min_ms=settings.scrape_kn_request_jitter_min_ms,
+                request_jitter_max_ms=settings.scrape_kn_request_jitter_max_ms,
+                proxy_url=settings.scrape_kn_proxy_url,
             )
         )

backend/tests/services/scrapers/test_stealth_throttle_proxy.py

@@ -0,0 +1,114 @@
+"""Unit-тесты для KN-loader anti-ban рычагов в scrapers/stealth (#1945).
+
+DOM.РФ WAF банит VPS-IP по volume/rate на full sweep (~17k запросов при
+concurrency=8). Лечим: (1) per-instance throttle (concurrency + шире jitter)
+для KN-sweep, НЕ трогая модульный дефолт остальных скраперов; (2) опциональный
+прокси (default off → поведение без изменений).
+
+Тесты проверяют:
+  • parse_proxy_url корректно разбирает http://user:pass@host:port;
+  • BrowserSession прокидывает concurrency/jitter/proxy в self-поля;
+  • дефолтное (None) поведение НЕ меняется (concurrency=8, jitter 600–1500, без прокси).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.services.scrapers.stealth import (
+    _BROWSER_CONCURRENCY,
+    _DEFAULT_JITTER_MAX_MS,
+    _DEFAULT_JITTER_MIN_MS,
+    BrowserSession,
+    parse_proxy_url,
+)
+
+
+class TestParseProxyUrl:
+    """parse_proxy_url → Playwright proxy-dict {server, username?, password?}."""
+
+    def test_none_returns_none(self) -> None:
+        assert parse_proxy_url(None) is None
+
+    def test_empty_string_returns_none(self) -> None:
+        assert parse_proxy_url("") is None
+
+    def test_full_creds_url(self) -> None:
+        got = parse_proxy_url("http://user:pass@host.example:8080")
+        assert got == {
+            "server": "http://host.example:8080",
+            "username": "user",
+            "password": "pass",
+        }
+
+    def test_no_creds(self) -> None:
+        got = parse_proxy_url("http://host.example:3128")
+        assert got == {"server": "http://host.example:3128"}
+
+    def test_creds_not_leaked_into_server(self) -> None:
+        # Playwright требует creds ОТДЕЛЬНЫМИ ключами, не в server-URL.
+        # Используем уникальные токены, которых нет в host/scheme.
+        got = parse_proxy_url("http://secretuser:secretpass@1.2.3.4:9999")
+        assert got is not None
+        assert got["server"] == "http://1.2.3.4:9999"
+        assert "secretuser" not in got["server"]
+        assert "secretpass" not in got["server"]
+        assert "@" not in got["server"]
+
+    def test_scheme_preserved(self) -> None:
+        got = parse_proxy_url("https://u:p@host:443")
+        assert got is not None
+        assert got["server"] == "https://host:443"
+
+    def test_default_scheme_when_missing(self) -> None:
+        # urlsplit без схемы кладёт всё в path → host не резолвится → ValueError.
+        with pytest.raises(ValueError):
+            parse_proxy_url("host.example:8080")
+
+    def test_host_required(self) -> None:
+        with pytest.raises(ValueError):
+            parse_proxy_url("http://:8080")
+
+
+class TestBrowserSessionThrottleThreading:
+    """concurrency/jitter/proxy прокидываются в инстанс-поля BrowserSession."""
+
+    def test_concurrency_threaded(self) -> None:
+        sess = BrowserSession(concurrency=2)
+        assert sess._concurrency == 2
+        assert sess._sem._value == 2
+
+    def test_jitter_threaded(self) -> None:
+        sess = BrowserSession(jitter_min_ms=1200, jitter_max_ms=3000)
+        assert sess._jitter_min_ms == 1200
+        assert sess._jitter_max_ms == 3000
+
+    def test_proxy_threaded_and_parsed(self) -> None:
+        sess = BrowserSession(proxy_url="http://u:p@proxy.local:8000")
+        assert sess._proxy == {
+            "server": "http://proxy.local:8000",
+            "username": "u",
+            "password": "p",
+        }
+
+
+class TestBrowserSessionDefaultsUnchanged:
+    """None у каждого рычага = старое поведение (другие скраперы НЕ затронуты)."""
+
+    def test_default_concurrency_is_module_default(self) -> None:
+        sess = BrowserSession()
+        assert sess._concurrency == _BROWSER_CONCURRENCY == 8
+        assert sess._sem._value == 8
+
+    def test_default_jitter_is_module_default(self) -> None:
+        sess = BrowserSession()
+        assert sess._jitter_min_ms == _DEFAULT_JITTER_MIN_MS == 600
+        assert sess._jitter_max_ms == _DEFAULT_JITTER_MAX_MS == 1500
+
+    def test_default_proxy_is_none(self) -> None:
+        sess = BrowserSession()
+        assert sess._proxy is None
+
+    def test_explicit_none_proxy_is_none(self) -> None:
+        sess = BrowserSession(proxy_url=None)
+        assert sess._proxy is None