diff --git a/backend/app/core/config.py b/backend/app/core/config.py index b0da2a18..d45d0ab9 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -140,6 +140,19 @@ class Settings(BaseSettings): # (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. diff --git a/backend/app/services/scrapers/domrf_kn.py b/backend/app/services/scrapers/domrf_kn.py index e288607d..44fe97cd 100644 --- a/backend/app/services/scrapers/domrf_kn.py +++ b/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__) @@ -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 ───────────────────────────────────────────────────────────── @@ -1664,6 +1905,7 @@ async def run_region_sweep( 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. @@ -1672,6 +1914,12 @@ async def run_region_sweep( прогон (другие скраперы используют модульные дефолты 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 ссылкой), исходный остаётся @@ -1898,6 +2146,12 @@ async def run_region_sweep( # ограничивает запросы. # 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): @@ -1912,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). @@ -1936,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 @@ -2012,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( diff --git a/backend/app/workers/tasks/scrape_kn.py b/backend/app/workers/tasks/scrape_kn.py index 7aec7ffb..d243c047 100644 --- a/backend/app/workers/tasks/scrape_kn.py +++ b/backend/app/workers/tasks/scrape_kn.py @@ -210,11 +210,12 @@ def scrape_kn_region( fetch_flats=fetch_flats, extras=extras, download_photos_binary=download_photos, - # #1945 anti-ban: throttle + optional proxy для KN-sweep. + # #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, ) ) @@ -294,10 +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 для KN-sweep. + # #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, ) ) diff --git a/backend/tests/services/scrapers/test_domrf_kn_isolation.py b/backend/tests/services/scrapers/test_domrf_kn_isolation.py new file mode 100644 index 00000000..eb6b4f5c --- /dev/null +++ b/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