backend/app/core/config.py

@@ -114,6 +114,46 @@ 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
+    # Рычаг 3 (ГЛАВНЫЙ unblock #1945) — изоляция flats от extras.
+    # Эмпирически (prod, 2026-06-27): flats endpoint /portal-kn/api/sales/portal/table
+    # на concurrency=2 = 0 WAF-бан на 100+ объектах; extras /сервисы/api/object/{id}/*
+    # отдают 403 СРАЗУ (volume-independent, мертвы с 2026-06-03) И ТРАВЯТ cookies
+    # сессии → последующие flats на той же сессии тоже 403. flats_count (метрика
+    # #1945) рухнул 3670→9 именно из-за этого.
+    #
+    # True (дефолт, NEW): flats тянутся в ЧИСТОЙ flats-only сессии (extras на ней
+    # НЕ дёргаются НИКОГДА) → flats_count восстанавливается; extras идут отдельным
+    # best-effort проходом с recycle сессии на каждый WAF-403 (яд не накапливается
+    # и НЕ касается flats). False: старое поведение (flats+extras в одной сессии,
+    # poison-prone). ENV: SCRAPE_KN_EXTRAS_ISOLATED.
+    scrape_kn_extras_isolated: bool = True
+
     # NSPD-scraper (Playwright) УДАЛЁН 2026-05-11. Сменён на bulk geo-fetcher
     # через rosreestr2coord — запускается вручную через /admin/scrape/geo.
     # Settings оставлены deprecated на случай отката (можно удалить позже).

backend/app/services/scrapers/domrf_kn.py

@@ -33,7 +33,7 @@ from app.services.scrapers.documents import extract_documents, upsert_documents
 # obj_checks import temporarily disabled — endpoint /checks returns 404 (run #19).
 # Re-enable with _fetch_obj_checks_safe when endpoint is found (see TODO in Phase B/C).
 # from app.services.scrapers.obj_checks import extract_obj_checks, upsert_obj_checks
-from app.services.scrapers.stealth import BASE_URL, BrowserSession
+from app.services.scrapers.stealth import BASE_URL, BrowserSession, WafBlockedError
 
 logger = logging.getLogger(__name__)
 
@@ -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(
@@ -1450,6 +1450,247 @@ async def download_photos(
     return locals_out, thumbs_out
 
 
+# ── result processing (shared by combined + isolated #1945 passes) ────────────
+
+# #1945 isolated extras pass: после стольких подряд полностью-WAF объектов
+# обрываем пас (extras заблокированы целиком — нет смысла жечь ~1500 recycle).
+_EXTRAS_ABORT_AFTER_CONSEC_WAF = 5
+
+# 5 doc-endpoint kinds, accumulated into a single per-object documents upsert.
+_DOC_KINDS = frozenset(
+    (
+        "doc_rpd",
+        "doc_developer_report",
+        "doc_project_documentation",
+        "doc_documentation_other",
+        "doc_permits",
+    )
+)
+
+
+def _process_flats_result(
+    db: Session,
+    run_id: int,
+    obj_id: int,
+    full_url: str,
+    result: list[dict[str, Any]] | Exception,
+    snapshot_date: date,
+    region_code: int,
+) -> int:
+    """Upsert flats from a _fetch_flats_safe result. Returns flats inserted.
+
+    Errors → kn_scrape_failures (via _classify_and_log), 0 returned. Вынесено из
+    Phase-B/C loop чтобы isolated-flats-pass (#1945) и combined-loop делили один код.
+    """
+    if isinstance(result, Exception):
+        _classify_and_log(db, run_id, obj_id, "flats", full_url, result)
+        return 0
+    if result:
+        return upsert_flats(db, result, snapshot_date, region_code)
+    return 0
+
+
+async def _process_extras_result(
+    sess: BrowserSession,
+    db: Session,
+    run_id: int,
+    obj_id: int,
+    kind_tag: str,
+    full_url: str,
+    result: Any,
+    snapshot_date: date,
+    extras_counts: dict[str, int],
+    all_docs: list[dict[str, Any]],
+    *,
+    download_photos_binary: bool,
+    pdir: Path,
+) -> None:
+    """Handle one extras endpoint result (sale_graph/sales_agg/infra/photos/docs).
+
+    Mutates extras_counts / all_docs in place. Errors → kn_scrape_failures.
+    Doc-kinds накапливаются в all_docs (единый upsert делает вызывающий код).
+    Вынесено из loop чтобы combined- и isolated-extras-pass (#1945) делили код.
+    """
+    if isinstance(result, Exception):
+        _classify_and_log(db, run_id, obj_id, kind_tag, full_url, result)
+        return
+
+    if kind_tag in ("sale_graph_apartments", "sale_graph_parking"):
+        sg_type = kind_tag.replace("sale_graph_", "")
+        rows_sg, _ = result
+        extras_counts["sale_graph_rows"] += upsert_sale_graph(
+            db, obj_id, sg_type, rows_sg, snapshot_date
+        )
+    elif kind_tag == "sales_agg":
+        agg_data, _ = result
+        extras_counts["sales_agg_rows"] += upsert_sales_agg(db, obj_id, agg_data, snapshot_date)
+    elif kind_tag == "infrastructure":
+        pois_data, _ = result
+        extras_counts["infra_rows"] += upsert_infrastructure(db, obj_id, pois_data, snapshot_date)
+    elif kind_tag == "photos":
+        photos_data, _ = result
+        local_paths: dict[str, str] = {}
+        thumb_paths: dict[str, str] = {}
+        if download_photos_binary and photos_data:
+            local_paths, thumb_paths = await download_photos(sess, obj_id, photos_data, pdir)
+            extras_counts["photos_downloaded"] += len(local_paths)
+        extras_counts["photos_rows"] += upsert_photos(
+            db, obj_id, photos_data, local_paths, thumb_paths
+        )
+    elif kind_tag in _DOC_KINDS:
+        doc_items, _ = result
+        all_docs.extend(extract_documents(doc_items or []))
+
+
+def _extras_coros(sess: BrowserSession, obj_id: int) -> list[Any]:
+    """The 10 extras-endpoint coroutines for one object (flats NOT included)."""
+    return [
+        _fetch_sale_graph_safe(sess, obj_id, "apartments"),
+        _fetch_sale_graph_safe(sess, obj_id, "parking"),
+        _fetch_sales_agg_safe(sess, obj_id),
+        _fetch_infrastructure_safe(sess, obj_id),
+        _fetch_photos_safe(sess, obj_id),
+        _fetch_doc_rpd_safe(sess, obj_id),
+        _fetch_doc_developer_report_safe(sess, obj_id),
+        _fetch_doc_project_documentation_safe(sess, obj_id),
+        _fetch_doc_documentation_other_safe(sess, obj_id),
+        _fetch_doc_permits_safe(sess, obj_id),
+    ]
+
+
+def _is_waf_poisoned(results: list[tuple[str, str, Any]]) -> bool:
+    """Любой WafBlockedError среди результатов = сессия отравлена (cookies флагнуты).
+
+    После этого ВСЕ запросы на этой сессии (incl. следующего объекта) пойдут в 403,
+    поэтому вызывающий код должен пересоздать BrowserSession перед продолжением.
+    """
+    return any(isinstance(r, WafBlockedError) for _kind, _url, r in results)
+
+
+async def _run_extras_pass(
+    *,
+    db: Session,
+    run_id: int,
+    region_code: int,
+    all_objects: list[dict[str, Any]],
+    start_index: int,
+    snapshot_date: date,
+    extras_counts: dict[str, int],
+    pdir: Path,
+    download_photos_binary: bool,
+    load_state: str | None,
+    headed: bool,
+    browser_concurrency: int | None,
+    request_jitter_min_ms: int | None,
+    request_jitter_max_ms: int | None,
+    proxy_url: str | None,
+) -> int:
+    """#1945 isolated extras pass: extras в ОТДЕЛЬНОЙ BrowserSession от flats.
+
+    Ключ изоляции: на ЛЮБОЙ WafBlockedError (extras-403 травит cookies сессии)
+    закрываем сессию и открываем НОВУЮ (свежие cookies/контекст) перед следующим
+    объектом — яд не накапливается и НИКОГДА не касается flats (flats-сессия уже
+    закрыта). Best-effort: каждый сбой → kn_scrape_failures, run не падает.
+
+    Возвращает суммарный request_count всех extras-сессий (для метрики run'а).
+    """
+    total = len(all_objects)
+    total_requests = 0
+    log_progress(
+        db, run_id, f"extras-пас (isolated): старт {total - start_index} объектов", stage="extras"
+    )
+
+    def _new_session() -> BrowserSession:
+        return BrowserSession(
+            region_code=region_code,
+            headed=headed,
+            load_state=load_state,
+            concurrency=browser_concurrency,
+            jitter_min_ms=request_jitter_min_ms,
+            jitter_max_ms=request_jitter_max_ms,
+            proxy_url=proxy_url,
+        )
+
+    # extras сейчас мертвы (403 с 2026-06-03). Recycle на КАЖДЫЙ объект (полный
+    # перезапуск браузера ~1500×) бессмыслен, если extras заблокированы целиком.
+    # Поэтому считаем подряд идущие полностью-WAF объекты и обрываем пас после
+    # порога — extras остаются best-effort opt-in, но не жгут час впустую.
+    consecutive_all_waf = 0
+    aborted = False
+    sess = _new_session()
+    await sess.__aenter__()
+    try:
+        await sess.warm_up()
+        for i in range(start_index, total):
+            obj = all_objects[i]
+            obj_id = obj.get("objId") or obj.get("obj_id")
+            if not obj_id:
+                continue
+
+            results = await asyncio.gather(*_extras_coros(sess, obj_id), return_exceptions=False)
+            all_docs: list[dict[str, Any]] = []
+            for kind_tag, full_url, result in results:
+                await _process_extras_result(
+                    sess,
+                    db,
+                    run_id,
+                    obj_id,
+                    kind_tag,
+                    full_url,
+                    result,
+                    snapshot_date,
+                    extras_counts,
+                    all_docs,
+                    download_photos_binary=download_photos_binary,
+                    pdir=pdir,
+                )
+            if all_docs:
+                ins, _skip = upsert_documents(db, obj_id, all_docs)
+                extras_counts["documents_rows"] += ins
+
+            poisoned = _is_waf_poisoned(results)
+            # Полностью-WAF объект = ни один extras не отдал JSON (все 403).
+            all_waf = all(isinstance(r, WafBlockedError) for _k, _u, r in results)
+            consecutive_all_waf = consecutive_all_waf + 1 if all_waf else 0
+            if consecutive_all_waf >= _EXTRAS_ABORT_AFTER_CONSEC_WAF:
+                aborted = True
+                log_progress(
+                    db,
+                    run_id,
+                    f"extras-пас оборван: {consecutive_all_waf} объектов подряд"
+                    " полностью WAF-403 (extras заблокированы целиком)",
+                    level="warn",
+                    stage="extras",
+                )
+                break
+
+            # RECYCLE: если сессия отравлена WAF-403 — пересоздаём перед next obj.
+            if poisoned:
+                total_requests += sess.request_count
+                await sess.__aexit__(None, None, None)
+                sess = _new_session()
+                await sess.__aenter__()
+                await sess.warm_up()
+
+            if (i + 1) % 10 == 0:
+                _checkpoint(db, run_id, i + 1)
+    finally:
+        total_requests += sess.request_count
+        await sess.__aexit__(None, None, None)
+    if aborted:
+        logger.info("extras isolated pass aborted early — extras WAF-blocked wholesale")
+
+    log_progress(
+        db,
+        run_id,
+        f"extras-пас done: sale_graph={extras_counts['sale_graph_rows']}"
+        f" agg={extras_counts['sales_agg_rows']} infra={extras_counts['infra_rows']}"
+        f" photos={extras_counts['photos_rows']} docs={extras_counts['documents_rows']}",
+        stage="extras",
+    )
+    return total_requests
+
+
 # ── orchestrator ─────────────────────────────────────────────────────────────
 
 
@@ -1660,9 +1901,25 @@ 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,
+    extras_isolated: bool = True,
 ) -> 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, без прокси).
+
+    ``extras_isolated`` (#1945 ГЛАВНЫЙ unblock): True (дефолт) → flats тянутся в
+    ЧИСТОЙ flats-only сессии (extras на ней НЕ дёргаются НИКОГДА → 403-яд extras не
+    травит cookies flats-сессии → flats_count восстанавливается), а extras идут
+    ОТДЕЛЬНЫМ best-effort проходом с recycle сессии на WAF-403. False → старое
+    поведение (flats+extras в одной сессии, poison-prone).
+
     Resume-aware: если `resume_from_run_id` задан, читаем objects_snapshot и
     progress_obj_index из исходного run и продолжаем с этого индекса. Новый
     run_id создаётся (с resumed_from_run_id ссылкой), исходный остаётся
@@ -1802,6 +2059,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,9 +2142,16 @@ 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 для параллельной записи.
+            #
+            # #1945 ИЗОЛЯЦИЯ: при extras_isolated=True эта (warm) сессия делает
+            # ТОЛЬКО flats — extras на ней НЕ дёргаются (их 403-яд не травит cookies
+            # flats-сессии → flats_count восстанавливается). extras уходят отдельным
+            # best-effort проходом ниже (после закрытия flats-сессии) с recycle на
+            # WAF-403. При extras_isolated=False — старое combined-поведение.
             pdir = Path(photos_dir) if photos_dir else PHOTOS_DIR_DEFAULT
             total = len(all_objects)
             for i in range(start_index, total):
@@ -1898,17 +2166,12 @@ async def run_region_sweep(
                 if fetch_flats:
                     coros.append(_fetch_flats_safe(sess, obj_id))
 
-                if extras:
-                    coros.append(_fetch_sale_graph_safe(sess, obj_id, "apartments"))
-                    coros.append(_fetch_sale_graph_safe(sess, obj_id, "parking"))
-                    coros.append(_fetch_sales_agg_safe(sess, obj_id))
-                    coros.append(_fetch_infrastructure_safe(sess, obj_id))
-                    coros.append(_fetch_photos_safe(sess, obj_id))
-                    coros.append(_fetch_doc_rpd_safe(sess, obj_id))
-                    coros.append(_fetch_doc_developer_report_safe(sess, obj_id))
-                    coros.append(_fetch_doc_project_documentation_safe(sess, obj_id))
-                    coros.append(_fetch_doc_documentation_other_safe(sess, obj_id))
-                    coros.append(_fetch_doc_permits_safe(sess, obj_id))
+                # extras в combined-сессии ТОЛЬКО когда изоляция выключена —
+                # иначе они отравят flats (#1945). В isolated-режиме extras
+                # обрабатываются отдельным проходом ниже.
+                combined_extras = extras and not extras_isolated
+                if combined_extras:
+                    coros.extend(_extras_coros(sess, obj_id))
                     # TODO: obj_checks endpoint not found at /api/object/{id}/checks (404).
                     # 6 чек-боксов "Проверено на наш.дом.рф" вероятно inline в kn/object payload.
                     # Re-enable после investigation структуры объекта (separate PR).
@@ -1922,64 +2185,28 @@ async def run_region_sweep(
 
                 # Sequential upsert — DB session не thread-safe
                 all_docs: list[dict[str, Any]] = []
-                _doc_kinds = frozenset(
-                    (
-                        "doc_rpd",
-                        "doc_developer_report",
-                        "doc_project_documentation",
-                        "doc_documentation_other",
-                        "doc_permits",
-                    )
-                )
                 for kind_tag, full_url, result in results:
-                    if isinstance(result, Exception):
-                        _classify_and_log(db, run_id, obj_id, kind_tag, full_url, result)
-                        continue
-
                     if kind_tag == "flats":
-                        flats_list: list[dict[str, Any]] = result  # type: ignore[assignment]
-                        if flats_list:
-                            total_flats += upsert_flats(db, flats_list, snapshot_date, region_code)
-
-                    elif kind_tag in ("sale_graph_apartments", "sale_graph_parking"):
-                        sg_type = kind_tag.replace("sale_graph_", "")
-                        rows_sg, _ = result  # type: ignore[misc]
-                        extras_counts["sale_graph_rows"] += upsert_sale_graph(
-                            db, obj_id, sg_type, rows_sg, snapshot_date
+                        total_flats += _process_flats_result(
+                            db, run_id, obj_id, full_url, result, snapshot_date, region_code
+                        )
+                    else:
+                        await _process_extras_result(
+                            sess,
+                            db,
+                            run_id,
+                            obj_id,
+                            kind_tag,
+                            full_url,
+                            result,
+                            snapshot_date,
+                            extras_counts,
+                            all_docs,
+                            download_photos_binary=download_photos_binary,
+                            pdir=pdir,
                         )
 
-                    elif kind_tag == "sales_agg":
-                        agg_data, _ = result  # type: ignore[misc]
-                        extras_counts["sales_agg_rows"] += upsert_sales_agg(
-                            db, obj_id, agg_data, snapshot_date
-                        )
-
-                    elif kind_tag == "infrastructure":
-                        pois_data, _ = result  # type: ignore[misc]
-                        extras_counts["infra_rows"] += upsert_infrastructure(
-                            db, obj_id, pois_data, snapshot_date
-                        )
-
-                    elif kind_tag == "photos":
-                        photos_data, _ = result  # type: ignore[misc]
-                        local_paths: dict[str, str] = {}
-                        thumb_paths: dict[str, str] = {}
-                        if download_photos_binary and photos_data:
-                            local_paths, thumb_paths = await download_photos(
-                                sess, obj_id, photos_data, pdir
-                            )
-                            extras_counts["photos_downloaded"] += len(local_paths)
-                        extras_counts["photos_rows"] += upsert_photos(
-                            db, obj_id, photos_data, local_paths, thumb_paths
-                        )
-
-                    elif kind_tag in _doc_kinds:
-                        # Каждый из 5 doc-endpoint'ов отдаёт свой список документов.
-                        # Накапливаем в all_docs — единый upsert после цикла.
-                        doc_items, _ = result  # type: ignore[misc]
-                        all_docs.extend(extract_documents(doc_items or []))
-
-                # Единый upsert всех документов объекта после обработки 5 endpoint'ов.
+                # Единый upsert всех документов объекта после обработки doc-endpoint'ов.
                 if all_docs:
                     ins, _skip = upsert_documents(db, obj_id, all_docs)
                     extras_counts["documents_rows"] += ins
@@ -1998,12 +2225,49 @@ async def run_region_sweep(
                         f" downloaded={extras_counts['photos_downloaded']}"
                         f" docs={extras_counts['documents_rows']}"
                         f" checks={extras_counts['checks_rows']}",
-                        stage="extras" if extras else "fetch_flats",
+                        stage="extras" if combined_extras else "fetch_flats",
                     )
 
             _checkpoint(db, run_id, total)
             request_count = sess.request_count
 
+        # ── Phase C (isolated) — extras best-effort пас в ОТДЕЛЬНОЙ сессии ──────
+        # flats-сессия закрыта. Теперь (только если extras=True И extras_isolated)
+        # тянем extras в disposable-сессиях: на каждый WAF-403 recycle всю
+        # BrowserSession (свежие cookies/контекст) перед следующим объектом, чтобы
+        # яд НЕ накапливался. extras сейчас мертвы (403 с 2026-06-03) — пас целиком
+        # best-effort: любые сбои уходят в kn_scrape_failures и НЕ роняют run.
+        if extras and extras_isolated:
+            try:
+                ex_req = await _run_extras_pass(
+                    db=db,
+                    run_id=run_id,
+                    region_code=region_code,
+                    all_objects=all_objects,
+                    start_index=start_index,
+                    snapshot_date=snapshot_date,
+                    extras_counts=extras_counts,
+                    pdir=pdir,
+                    download_photos_binary=download_photos_binary,
+                    load_state=load_state,
+                    headed=headed,
+                    browser_concurrency=browser_concurrency,
+                    request_jitter_min_ms=request_jitter_min_ms,
+                    request_jitter_max_ms=request_jitter_max_ms,
+                    proxy_url=proxy_url,
+                )
+                request_count += ex_req
+            except Exception as e:
+                # best-effort: extras-пас НИКОГДА не должен ронять уже собранные flats.
+                logger.warning("extras isolated pass failed (best-effort): %s", e)
+                log_progress(
+                    db,
+                    run_id,
+                    f"extras-пас (best-effort) упал: {type(e).__name__}: {str(e)[:200]}",
+                    level="warn",
+                    stage="extras",
+                )
+
         # ── Phase D — finalize ─────────────────────────────────────────────
         db.execute(
             text(

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,45 @@ 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:
+        # НЕ эхо-им сырой 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 = {
@@ -94,6 +131,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 +146,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 +177,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 +250,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,12 @@ def scrape_kn_region(
                 fetch_flats=fetch_flats,
                 extras=extras,
                 download_photos_binary=download_photos,
+                # #1945 anti-ban: throttle + optional proxy + flats/extras isolation.
+                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,
+                extras_isolated=settings.scrape_kn_extras_isolated,
             )
         )
 
@@ -289,5 +295,11 @@ 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 + flats/extras isolation.
+                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,
+                extras_isolated=settings.scrape_kn_extras_isolated,
             )
         )

backend/tests/services/scrapers/test_domrf_kn_isolation.py

@@ -0,0 +1,200 @@
+"""#1945 flats/extras isolation tests (scrapers/domrf_kn).
+
+PROVEN mechanism: extras /сервисы/api/object/{id}/* return 403 (volume-independent,
+dead since 2026-06-03) AND poison the session cookies → subsequent FLATS on the same
+session flip to 403. flats_count collapsed 3670→9. Fix: flats never share a session
+with extras; extras run as a separate best-effort pass that RECYCLES the session on
+any WAF-403 so poison never accumulates (and never touches flats).
+
+These tests assert:
+  • _extras_coros builds ONLY extras endpoints (never flats);
+  • _is_waf_poisoned flips on any WafBlockedError;
+  • _run_extras_pass recycles the BrowserSession on a poisoned object;
+  • _run_extras_pass aborts early when extras are wholesale-403 (no ~1500 relaunches).
+"""
+
+from __future__ import annotations
+
+import datetime
+from typing import Any, ClassVar
+
+import pytest
+
+import app.services.scrapers.domrf_kn as kn
+from app.services.scrapers.stealth import WafBlockedError
+
+
+class TestExtrasCorosPurity:
+    """_extras_coros НИКОГДА не включает flats — иначе изоляция бессмысленна."""
+
+    def test_extras_coros_excludes_flats(self) -> None:
+        coros = kn._extras_coros(sess=object(), obj_id=123)  # type: ignore[arg-type]
+        try:
+            names = {c.__qualname__ for c in coros}
+            assert not any("flats" in n for n in names)
+            # 10 extras: 2 sale_graph + sales_agg + infra + photos + 5 docs.
+            assert len(coros) == 10
+        finally:
+            for c in coros:
+                c.close()  # avoid un-awaited coroutine warnings
+
+
+class TestWafPoisonDetection:
+    def test_poisoned_on_any_waf(self) -> None:
+        results = [
+            ("sales_agg", "u", {"ok": 1}),
+            ("infrastructure", "u", WafBlockedError("403 waf")),
+        ]
+        assert kn._is_waf_poisoned(results) is True
+
+    def test_not_poisoned_without_waf(self) -> None:
+        results = [
+            ("sales_agg", "u", {"ok": 1}),
+            ("infrastructure", "u", RuntimeError("http 500")),  # transient, not WAF
+        ]
+        assert kn._is_waf_poisoned(results) is False
+
+    def test_empty_not_poisoned(self) -> None:
+        assert kn._is_waf_poisoned([]) is False
+
+
+# ── _run_extras_pass session-lifecycle tests ─────────────────────────────────
+
+
+class _FakeSession:
+    """BrowserSession stand-in: tracks bootstrap/warm/exit, no real browser."""
+
+    instances: ClassVar[list[_FakeSession]] = []
+
+    def __init__(self, **_kw: Any) -> None:
+        self.request_count = 7
+        self.warmed = 0
+        self.exited = False
+        _FakeSession.instances.append(self)
+
+    async def __aenter__(self) -> _FakeSession:
+        return self
+
+    async def __aexit__(self, *_exc: Any) -> None:
+        self.exited = True
+
+    async def warm_up(self, force: bool = False) -> None:
+        self.warmed += 1
+
+
+def _objs(n: int) -> list[dict[str, Any]]:
+    return [{"objId": 1000 + i} for i in range(n)]
+
+
+@pytest.fixture(autouse=True)
+def _patch(monkeypatch: pytest.MonkeyPatch) -> None:
+    _FakeSession.instances = []
+    monkeypatch.setattr(kn, "BrowserSession", _FakeSession)
+    monkeypatch.setattr(kn, "log_progress", lambda *a, **k: None)
+    monkeypatch.setattr(kn, "_checkpoint", lambda *a, **k: None)
+    monkeypatch.setattr(kn, "upsert_documents", lambda *a, **k: (0, 0))
+
+    # _process_extras_result is real upsert plumbing — no-op it (we test lifecycle).
+    async def _noop_process(*_a: Any, **_k: Any) -> None:
+        return None
+
+    monkeypatch.setattr(kn, "_process_extras_result", _noop_process)
+
+
+def _make_coros_factory(per_obj_results: Any) -> Any:
+    """Return a _extras_coros replacement: per-object it yields coroutines that each
+    resolve to one staged (kind, url, result) tuple, so real asyncio.gather is used."""
+    state = {"i": 0}
+
+    def _coros(_sess: Any, _oid: int) -> list[Any]:
+        results = per_obj_results(state["i"])
+        state["i"] += 1
+
+        async def _wrap(res: tuple[str, str, Any]) -> tuple[str, str, Any]:
+            return res
+
+        return [_wrap(r) for r in results]
+
+    return _coros
+
+
+async def _drive(objs: list[dict[str, Any]]) -> int:
+    extras_counts = dict.fromkeys(
+        (
+            "sale_graph_rows",
+            "sales_agg_rows",
+            "infra_rows",
+            "photos_rows",
+            "photos_downloaded",
+            "documents_rows",
+            "checks_rows",
+        ),
+        0,
+    )
+    return await kn._run_extras_pass(
+        db=object(),  # type: ignore[arg-type]
+        run_id=1,
+        region_code=66,
+        all_objects=objs,
+        start_index=0,
+        snapshot_date=datetime.date(2026, 6, 28),
+        extras_counts=extras_counts,
+        pdir=kn.PHOTOS_DIR_DEFAULT,
+        download_photos_binary=False,
+        load_state=None,
+        headed=False,
+        browser_concurrency=2,
+        request_jitter_min_ms=1200,
+        request_jitter_max_ms=3000,
+        proxy_url=None,
+    )
+
+
+class TestRunExtrasPassLifecycle:
+    @pytest.mark.asyncio
+    async def test_clean_results_use_single_session(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Every object returns clean JSON → no poison → one session for the whole pass.
+        def clean(_i: int) -> list[tuple[str, str, Any]]:
+            return [("sales_agg", "u", ({"x": 1}, "u")), ("infrastructure", "u", ([], "u"))]
+
+        monkeypatch.setattr(kn, "_extras_coros", _make_coros_factory(clean))
+        req = await _drive(_objs(4))
+        # Exactly 1 BrowserSession constructed for a fully-clean pass.
+        assert len(_FakeSession.instances) == 1
+        assert _FakeSession.instances[0].exited is True
+        assert req == 7  # single session.request_count
+
+    @pytest.mark.asyncio
+    async def test_recycle_on_poison(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Object 0 poisoned (1 WAF among results) but NOT all-WAF (so no early abort);
+        # objects 1+ clean → pass recycles once after object 0, then continues clean.
+        def mixed(i: int) -> list[tuple[str, str, Any]]:
+            if i == 0:
+                return [
+                    ("sales_agg", "u", ({"x": 1}, "u")),  # one ok → not all-WAF
+                    ("infrastructure", "u", WafBlockedError("403")),  # poison
+                ]
+            return [("sales_agg", "u", ({"x": 1}, "u")), ("infrastructure", "u", ([], "u"))]
+
+        monkeypatch.setattr(kn, "_extras_coros", _make_coros_factory(mixed))
+        await _drive(_objs(3))
+        # 1 initial + 1 recycle after the poisoned object 0 = 2 sessions.
+        assert len(_FakeSession.instances) == 2
+        assert all(s.exited for s in _FakeSession.instances)
+
+    @pytest.mark.asyncio
+    async def test_early_abort_on_wholesale_waf(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # Every object fully WAF-403 (extras dead) → abort after the threshold,
+        # NOT grinding through all 1000 objects with a relaunch each time.
+        def all_waf(_i: int) -> list[tuple[str, str, Any]]:
+            return [
+                ("sales_agg", "u", WafBlockedError("403")),
+                ("infrastructure", "u", WafBlockedError("403")),
+            ]
+
+        monkeypatch.setattr(kn, "_extras_coros", _make_coros_factory(all_waf))
+        await _drive(_objs(1000))
+        # Abort kicks in at _EXTRAS_ABORT_AFTER_CONSEC_WAF consecutive all-WAF objects.
+        # Sessions: each poisoned object recycles UNLESS it's the abort object.
+        # Bounded well under 1000 — proves we don't relaunch per object forever.
+        assert len(_FakeSession.instances) <= kn._EXTRAS_ABORT_AFTER_CONSEC_WAF + 1

backend/tests/services/scrapers/test_stealth_throttle_proxy.py

@@ -0,0 +1,125 @@
+"""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")
+
+    def test_error_does_not_leak_credentials(self) -> None:
+        # ValueError всплывает в kn_scrape_runs.error / log_progress / Sentry —
+        # НЕ должен содержать пароль (#1945 sec-review).
+        bad = "http://secretuser:secretpass@:8080"  # есть creds, но нет host
+        with pytest.raises(ValueError) as exc:
+            parse_proxy_url(bad)
+        msg = str(exc.value)
+        assert "secretpass" not in msg
+        assert "secretuser" not in msg
+        assert bad not in msg
+
+
+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