diff --git a/tradein-mvp/backend/app/services/scraper_adapters.py b/tradein-mvp/backend/app/services/scraper_adapters.py index 55ec419f..e193747f 100644 --- a/tradein-mvp/backend/app/services/scraper_adapters.py +++ b/tradein-mvp/backend/app/services/scraper_adapters.py @@ -28,9 +28,16 @@ from app.services.house_imv_backfill import ( ) from app.services.matching import match_or_create_house as _match_or_create_house from app.services.matching import upsert_listing_source as _upsert_listing_source +from app.services.yandex_address_backfill import ( + _RE_HAS_HOUSE_NUMBER, + _extract_address_from_title, +) from app.services.yandex_address_backfill import ( backfill_yandex_addresses as _backfill_yandex_addresses, ) +from app.services.yandex_price_history import ( + record_yandex_price_history as _record_yandex_price_history, +) if TYPE_CHECKING: from sqlalchemy.orm import Session @@ -261,6 +268,15 @@ class RealEnrichmentJobs: heartbeat=heartbeat, ) + def record_yandex_price_history(self, db: Session, lots: list[Any]) -> int: + return _record_yandex_price_history(db, lots) + + def extract_address_from_title(self, html: str) -> str | None: + return _extract_address_from_title(html) + + def address_has_house_number(self, address: str) -> bool: + return bool(_RE_HAS_HOUSE_NUMBER.search(address)) + __all__ = [ "RealEnrichmentJobs", diff --git a/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py new file mode 100644 index 00000000..aeb6c0f4 --- /dev/null +++ b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py @@ -0,0 +1,481 @@ +"""Golden-parity (F2): kit sweep-оркестраторы ≡ `app.services.scrape_pipeline`. + +Продолжение test_scraper_kit_pipeline_parity.py (F1 покрыл avito city sweep). Здесь — +остальные 7 sweep'ов, скопированных в `scraper_kit.orchestration.pipeline` (#2135 F2): + + City sweep'ы (приоритет — активны в проде): + - run_yandex_city_sweep — combos SERP + save + price-history + - run_cian_city_sweep — SERP + newbuilding_only-фильтр + save + - run_domclick_city_sweep — BFF citywide + честный статус (done / failed) + Плюс: + - run_avito_newbuilding_sweep — citywide novostroyka SERP + save + Full load'ы (smoke-parity — импорт + базовый прогон через on_bucket): + - run_avito_full_load / run_cian_full_load / run_yandex_full_load + +Метод — как в F1: гоняем ОБА orchestrator'а на идентичном мокнутом I/O, сравниваем +итоговый counters.to_dict() + последовательность (method, numeric-counters) вызовов +scrape_runs. Развязка `app.*` (config/matcher/enrichment/runs/shutdown инжектируются) +не должна менять ветвление/счётчики. Без сети, без БД. +""" + +from __future__ import annotations + +import os +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit.orchestration.pipeline import ( + run_avito_full_load as new_avito_full_load, +) +from scraper_kit.orchestration.pipeline import ( + run_avito_newbuilding_sweep as new_nb_sweep, +) +from scraper_kit.orchestration.pipeline import ( + run_cian_city_sweep as new_cian_city_sweep, +) +from scraper_kit.orchestration.pipeline import ( + run_cian_full_load as new_cian_full_load, +) +from scraper_kit.orchestration.pipeline import ( + run_domclick_city_sweep as new_domclick_city_sweep, +) +from scraper_kit.orchestration.pipeline import ( + run_yandex_city_sweep as new_yandex_city_sweep, +) +from scraper_kit.orchestration.pipeline import ( + run_yandex_full_load as new_yandex_full_load, +) + +from app.services.scrape_pipeline import ( + run_avito_full_load as old_avito_full_load, +) +from app.services.scrape_pipeline import ( + run_avito_newbuilding_sweep as old_nb_sweep, +) +from app.services.scrape_pipeline import ( + run_cian_city_sweep as old_cian_city_sweep, +) +from app.services.scrape_pipeline import ( + run_cian_full_load as old_cian_full_load, +) +from app.services.scrape_pipeline import ( + run_domclick_city_sweep as old_domclick_city_sweep, +) +from app.services.scrape_pipeline import ( + run_yandex_city_sweep as old_yandex_city_sweep, +) +from app.services.scrape_pipeline import ( + run_yandex_full_load as old_yandex_full_load, +) + +OLD_PFX = "app.services.scrape_pipeline" +NEW_PFX = "scraper_kit.orchestration.pipeline" + +# ── recording scrape_runs ──────────────────────────────────────────────────── + +_NON_NUMERIC_KEYS = {"enrichment_abort_note", "done_buckets"} + + +class _RunsRecorder: + """Записывает вызовы scrape_runs-финализаторов. is_cancelled всегда False.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + + def is_cancelled(self, db: Any, run_id: int) -> bool: + return False + + def update_heartbeat(self, db: Any, run_id: int, counters: dict[str, Any]) -> None: + self.calls.append(("update_heartbeat", dict(counters))) + + def mark_done(self, db: Any, run_id: int, counters: dict[str, Any]) -> None: + self.calls.append(("mark_done", dict(counters))) + + def mark_banned(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None: + self.calls.append(("mark_banned", dict(counters))) + + def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None: + self.calls.append(("mark_failed", dict(counters))) + + +def _normalize(calls: list[tuple[str, dict[str, Any]]]) -> list[tuple[str, dict[str, int]]]: + """Оставить имя метода + числовые counters (убрать note/done_buckets).""" + out: list[tuple[str, dict[str, int]]] = [] + for method, counters in calls: + numeric = {k: v for k, v in counters.items() if k not in _NON_NUMERIC_KEYS} + out.append((method, numeric)) + return out + + +def _config() -> SimpleNamespace: + """Инжектируемый config (kit) + тот же объект как patched settings (old).""" + return SimpleNamespace( + scraper_fetch_mode="curl_cffi", + browser_http_endpoint="http://browser.test/fetch", + scraper_proxy_url=None, + avito_proxy_rotate_url=None, + avito_proxy_max_rotations=0, + avito_serp_ok_not_banned=True, + avito_proxy_rotate_settle_s=0.0, + proxy_rotate_attempts=1, + proxy_rotate_attempt_timeout_s=1.0, + cian_proxy_rotate_url=None, + cian_proxy_max_rotations=0, + yandex_proxy_rotate_url=None, + yandex_proxy_max_rotations=0, + scraper_skip_seen_today=False, + cian_full_load_per_fetch_timeout_s=0.0, + ) + + +def _ctx_scraper(**attrs: Any) -> MagicMock: + """MagicMock, пригодный и как `Cls()`, и как `async with Cls() as x`. + + __aenter__ возвращает сам объект, __aexit__ — no-op. Доп. атрибуты/методы + задаются через kwargs (async-методы — AsyncMock). + """ + m = MagicMock() + m.__aenter__ = AsyncMock(return_value=m) + m.__aexit__ = AsyncMock(return_value=None) + for k, v in attrs.items(): + setattr(m, k, v) + return m + + +def _async_session_cm() -> MagicMock: + sess = MagicMock() + sess.__aenter__ = AsyncMock(return_value=sess) + sess.__aexit__ = AsyncMock(return_value=None) + sess.close = AsyncMock() + return sess + + +_DriveResult = tuple[dict[str, int], list[tuple[str, dict[str, int]]]] + + +async def _assert_parity(old_res: _DriveResult, new_res: _DriveResult) -> dict[str, int]: + old_counters, old_calls = old_res + new_counters, new_calls = new_res + assert ( + new_counters == old_counters + ), f"counters diverged:\n old={old_counters}\n new={new_counters}" + assert ( + new_calls == old_calls + ), f"scrape_runs calls diverged:\n old={old_calls}\n new={new_calls}" + return new_counters + + +# ── Yandex city sweep ───────────────────────────────────────────────────────── + + +def _yandex_scraper(combos: list[tuple[str, list[Any]]]) -> MagicMock: + """Fake YandexRealtyScraper: fetch_around_multi_room вызывает on_combo по combos.""" + + async def _fetch(*_a: Any, on_combo: Any = None, **_k: Any) -> None: + for label, lots in combos: + on_combo(label, lots) + + return _ctx_scraper(fetch_around_multi_room=_fetch) + + +async def _drive_yandex_city(*, new: bool) -> _DriveResult: + recorder = _RunsRecorder() + db = MagicMock() + combos = [ + ("2к-price0", [MagicMock(source_id="y1"), MagicMock(source_id="y2")]), + ("1к-price0", [MagicMock(source_id="y3")]), + ] + scraper = _yandex_scraper(combos) + save_mock = MagicMock(side_effect=[(2, 0), (1, 0)]) + cfg = _config() + kwargs = dict( + run_id=1, + anchors=None, + pages_per_anchor=1, + request_delay_sec=0.0, + enrich_address=False, + ) + if new: + enrichment = MagicMock() + enrichment.record_yandex_price_history = MagicMock(return_value=5) + with ( + patch(f"{NEW_PFX}.YandexRealtyScraper", return_value=scraper), + patch(f"{NEW_PFX}.save_listings", save_mock), + patch(f"{NEW_PFX}.runs", recorder), + ): + counters = await new_yandex_city_sweep( + db, config=cfg, matcher=MagicMock(), enrichment=enrichment, **kwargs + ) + else: + with ( + patch(f"{OLD_PFX}.YandexRealtyScraper", return_value=scraper), + patch(f"{OLD_PFX}.save_listings", save_mock), + patch(f"{OLD_PFX}.record_yandex_price_history", MagicMock(return_value=5)), + patch(f"{OLD_PFX}.scrape_runs", recorder), + patch(f"{OLD_PFX}.settings", cfg), + patch(f"{OLD_PFX}.shutdown_requested", return_value=False), + ): + counters = await old_yandex_city_sweep(db, **kwargs) + return counters.to_dict(), _normalize(recorder.calls) + + +@pytest.mark.asyncio +async def test_yandex_city_sweep_parity() -> None: + """1 центральный anchor, 2 combos, save + price-history, enrich_address=False.""" + counters = await _assert_parity( + await _drive_yandex_city(new=False), await _drive_yandex_city(new=True) + ) + assert counters["lots_fetched"] == 3 + assert counters["lots_inserted"] == 3 + assert counters["price_history_rows"] == 5 + assert counters["anchors_done"] == 1 + + +# ── Cian city sweep ─────────────────────────────────────────────────────────── + + +def _cian_lot(segment: str) -> MagicMock: + return MagicMock(listing_segment=segment, house_source=None, house_ext_id=None) + + +async def _drive_cian_city(*, new: bool) -> _DriveResult: + recorder = _RunsRecorder() + db = MagicMock() + # 3 novostroyki + 2 secondary → newbuilding_only оставит 3. + lots = [_cian_lot("novostroyki")] * 3 + [_cian_lot("vtorichnaya")] * 2 + scraper = _ctx_scraper(fetch_around_multi_room=AsyncMock(return_value=lots)) + save_mock = MagicMock(side_effect=[(3, 0)]) + cfg = _config() + kwargs = dict( + run_id=1, + anchors=[(56.84, 60.60, "A1")], + radius_m=1000, + pages_per_anchor=1, + request_delay_sec=0.0, + detail_top_n=0, + enrich_houses=False, + newbuilding_only=True, + ) + if new: + with ( + patch(f"{NEW_PFX}.CianScraper", return_value=scraper), + patch(f"{NEW_PFX}.save_listings", save_mock), + patch(f"{NEW_PFX}.runs", recorder), + ): + counters = await new_cian_city_sweep(db, config=cfg, matcher=MagicMock(), **kwargs) + else: + with ( + patch("app.services.scrapers.cian.CianScraper", return_value=scraper), + patch(f"{OLD_PFX}.save_listings", save_mock), + patch(f"{OLD_PFX}.scrape_runs", recorder), + patch(f"{OLD_PFX}.settings", cfg), + patch(f"{OLD_PFX}.shutdown_requested", return_value=False), + ): + counters = await old_cian_city_sweep(db, **kwargs) + return counters.to_dict(), _normalize(recorder.calls) + + +@pytest.mark.asyncio +async def test_cian_city_sweep_parity() -> None: + """1 anchor, newbuilding_only отбрасывает вторичку, save новостроек, mark_done.""" + counters = await _assert_parity( + await _drive_cian_city(new=False), await _drive_cian_city(new=True) + ) + assert counters["lots_fetched"] == 5 + assert counters["lots_dropped_secondary"] == 2 + assert counters["lots_inserted"] == 3 + assert counters["anchors_done"] == 1 + + +# ── DomClick city sweep ─────────────────────────────────────────────────────── + + +async def _drive_domclick(*, new: bool, lots_n: int, blocked: bool) -> _DriveResult: + recorder = _RunsRecorder() + db = MagicMock() + lots = [MagicMock() for _ in range(lots_n)] + scraper = _ctx_scraper( + fetch_city=AsyncMock(return_value=lots), + blocked=blocked, + geo_filtered=0, + fetch_errors=0, + ) + save_mock = MagicMock(side_effect=[(lots_n, 0)] if lots_n else []) + cfg = _config() + kwargs = dict(run_id=1, city_id=4, pages=1, request_delay_sec=0.0) + if new: + with ( + patch(f"{NEW_PFX}.DomClickScraper", return_value=scraper), + patch(f"{NEW_PFX}.save_listings", save_mock), + patch(f"{NEW_PFX}.runs", recorder), + ): + counters = await new_domclick_city_sweep(db, config=cfg, matcher=MagicMock(), **kwargs) + else: + with ( + patch("app.services.scrapers.domclick.DomClickScraper", return_value=scraper), + patch(f"{OLD_PFX}.save_listings", save_mock), + patch(f"{OLD_PFX}.scrape_runs", recorder), + patch(f"{OLD_PFX}.settings", cfg), + patch(f"{OLD_PFX}.shutdown_requested", return_value=False), + ): + counters = await old_domclick_city_sweep(db, **kwargs) + return counters.to_dict(), _normalize(recorder.calls) + + +@pytest.mark.asyncio +async def test_domclick_city_sweep_parity_done() -> None: + """Citywide fetch_city → 4 lots saved → mark_done.""" + counters = await _assert_parity( + await _drive_domclick(new=False, lots_n=4, blocked=False), + await _drive_domclick(new=True, lots_n=4, blocked=False), + ) + assert counters["lots_fetched"] == 4 + assert counters["lots_inserted"] == 4 + + +@pytest.mark.asyncio +async def test_domclick_city_sweep_parity_blocked_failed() -> None: + """QRATOR-блок + 0 lots → mark_failed (честный статус #1968).""" + counters = await _assert_parity( + await _drive_domclick(new=False, lots_n=0, blocked=True), + await _drive_domclick(new=True, lots_n=0, blocked=True), + ) + assert counters["lots_fetched"] == 0 + assert counters["blocked"] == 1 + + +# ── Avito newbuilding sweep ─────────────────────────────────────────────────── + + +async def _drive_nb_sweep(*, new: bool) -> _DriveResult: + recorder = _RunsRecorder() + db = MagicMock() + lots = [MagicMock() for _ in range(6)] + scraper = MagicMock() + scraper._cffi = None + scraper._browser = None + scraper.fetch_newbuildings = AsyncMock(return_value=lots) + save_mock = MagicMock(side_effect=[(5, 1)]) + cfg = _config() + kwargs = dict(run_id=1, pages=2, request_delay_sec=0.0) + if new: + with ( + patch(f"{NEW_PFX}.AvitoScraper", return_value=scraper), + patch(f"{NEW_PFX}.save_listings", save_mock), + patch(f"{NEW_PFX}.runs", recorder), + patch(f"{NEW_PFX}.AsyncSession", return_value=_async_session_cm()), + ): + counters = await new_nb_sweep(db, config=cfg, matcher=MagicMock(), **kwargs) + else: + with ( + patch(f"{OLD_PFX}.AvitoScraper", return_value=scraper), + patch(f"{OLD_PFX}.save_listings", save_mock), + patch(f"{OLD_PFX}.scrape_runs", recorder), + patch(f"{OLD_PFX}.settings", cfg), + patch(f"{OLD_PFX}.shutdown_requested", return_value=False), + patch(f"{OLD_PFX}.AsyncSession", return_value=_async_session_cm()), + ): + counters = await old_nb_sweep(db, **kwargs) + return counters.to_dict(), _normalize(recorder.calls) + + +@pytest.mark.asyncio +async def test_avito_newbuilding_sweep_parity() -> None: + """Citywide novostroyka SERP → save → heartbeat → mark_done.""" + counters = await _assert_parity( + await _drive_nb_sweep(new=False), await _drive_nb_sweep(new=True) + ) + assert counters["lots_fetched"] == 6 + assert counters["lots_inserted"] == 5 + assert counters["lots_updated"] == 1 + + +# ── Full loads (smoke-parity через on_bucket) ───────────────────────────────── + + +def _full_load_scraper(buckets: list[tuple[str, list[Any]]]) -> MagicMock: + """Fake scraper: fetch_all_secondary вызывает on_bucket по каждому бакету.""" + + async def _fetch(*_a: Any, on_bucket: Any = None, on_progress: Any = None, **_k: Any) -> None: + for key, lots in buckets: + on_bucket(key, lots) + + scraper = _ctx_scraper(fetch_all_secondary=_fetch) + scraper._browser = None + return scraper + + +async def _drive_full_load(*, new: bool, source: str) -> _DriveResult: + recorder = _RunsRecorder() + db = MagicMock() + buckets = [ + ("2к:0-5m", [MagicMock(source_id=f"{source}1"), MagicMock(source_id=f"{source}2")]), + ("1к:0-5m", [MagicMock(source_id=f"{source}3")]), + ] + scraper = _full_load_scraper(buckets) + save_mock = MagicMock(side_effect=[(2, 0), (1, 0)]) + cfg = _config() + + old_map = { + "avito": (old_avito_full_load, new_avito_full_load, f"{OLD_PFX}.AvitoScraper"), + "cian": (old_cian_full_load, new_cian_full_load, "app.services.scrapers.cian.CianScraper"), + "yandex": ( + old_yandex_full_load, + new_yandex_full_load, + "app.services.scrapers.yandex_realty.YandexRealtyScraper", + ), + } + old_fn, new_fn, old_scraper_target = old_map[source] + + if new: + new_scraper_target = { + "avito": f"{NEW_PFX}.AvitoScraper", + "cian": f"{NEW_PFX}.CianScraper", + "yandex": f"{NEW_PFX}.YandexRealtyScraper", + }[source] + extra: dict[str, Any] = {} + if source == "yandex": + enrichment = MagicMock() + enrichment.record_yandex_price_history = MagicMock(return_value=0) + extra["enrichment"] = enrichment + with ( + patch(new_scraper_target, return_value=scraper), + patch(f"{NEW_PFX}.save_listings", save_mock), + patch(f"{NEW_PFX}.runs", recorder), + ): + counters = await new_fn(db, run_id=1, config=cfg, matcher=MagicMock(), **extra) + else: + import contextlib + + ctx = [ + patch(old_scraper_target, return_value=scraper), + patch(f"{OLD_PFX}.save_listings", save_mock), + patch(f"{OLD_PFX}.scrape_runs", recorder), + patch(f"{OLD_PFX}.settings", cfg), + patch(f"{OLD_PFX}.shutdown_requested", return_value=False), + ] + if source == "yandex": + ctx.append(patch(f"{OLD_PFX}.record_yandex_price_history", MagicMock(return_value=0))) + with contextlib.ExitStack() as stack: + for cm in ctx: + stack.enter_context(cm) + counters = await old_fn(db, run_id=1) + return counters.to_dict(), _normalize(recorder.calls) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source", ["avito", "cian", "yandex"]) +async def test_full_load_smoke_parity(source: str) -> None: + """Full load: 2 бакета через on_bucket → save + heartbeat → mark_done. Counters ≡.""" + counters = await _assert_parity( + await _drive_full_load(new=False, source=source), + await _drive_full_load(new=True, source=source), + ) + assert counters["unique_fetched"] == 3 + assert counters["saved_inserted"] == 3 + assert counters["saved_updated"] == 0 diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py index a865d384..e2e065d9 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py @@ -208,6 +208,36 @@ class EnrichmentJobs(Protocol): """ ... + # ── Yandex sweep-хелперы (#2135 F2) ────────────────────────────────────── + # Продуктовая логика, дёргаемая yandex city/full sweep-оркестраторами напрямую + # (не через отдельные backfill-джобы). Инжектируются вместо прямых импортов + # app.services.yandex_price_history / yandex_address_backfill. + + def record_yandex_price_history(self, db: Session, lots: list[Any]) -> int: + """Записать price-history из gate price.previous/trend. Проксирует + record_yandex_price_history. Returns число вставленных строк истории. + + `lots` — список kit-`ScrapedLot` (тип оставлен `Any`, чтобы не тянуть + доменный тип в контракт). Читается duck-typing'ом (source_id/price/…). + """ + ... + + def extract_address_from_title(self, html: str) -> str | None: + """Извлечь полный адрес из detail-страницы Yandex. + + Проксирует `_extract_address_from_title` (yandex_address_backfill). + Returns адрес с номером дома или None если не распарсилось. + """ + ... + + def address_has_house_number(self, address: str) -> bool: + """True если адрес содержит номер дома (`,\\s*\\d+`). + + Проксирует regex `_RE_HAS_HOUSE_NUMBER` из yandex_address_backfill — + gate для решения, обновлять ли address на обогащённый. + """ + ... + __all__ = [ "EnrichmentJobs", diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py index c2e5dcbf..49ddda42 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py @@ -12,14 +12,22 @@ Strangler-COPY боевого `app.services.scrape_pipeline` (шаг оркес - провайдеры/база → `scraper_kit.providers.*` / `scraper_kit.base` Скопирована КРИТИЧНАЯ логика оркестрации (#2135): - - `run_avito_pipeline` — одиночный anchor (SEARCH→SAVE→GROUP→HOUSES→DETAIL) - - `run_avito_city_sweep` — полный city-sweep + ban/rotation + partial-intake + IMV - -Остаток (yandex/cian/domclick city-sweep, *_full_load, newbuilding sweep) — -follow-up: та же механическая развязка, но не входит в первый срез #2135. + - `run_avito_pipeline` — одиночный anchor (SEARCH→SAVE→GROUP→HOUSES→DETAIL) + - `run_avito_city_sweep` — city-sweep + ban/rotation + partial-intake + IMV (F1) + - `run_avito_newbuilding_sweep`— citywide novostroyka-обход + save (F2) + - `run_yandex_city_sweep` — combos (room×price) + save + address-enrich (F2) + - `run_cian_city_sweep` — anchor SERP → detail(+price-hist) → newbuilding (F2) + - `run_cian_full_load` — region-wide bisection + resume/checkpoint (F2) + - `run_yandex_full_load` — region-wide bisection + price-history (F2) + - `run_avito_full_load` — citywide bisection (exhaustive/incremental) (F2) + - `run_domclick_city_sweep` — BFF citywide sweep + честный статус (F2) Ban-rotation state-machine сведена в единый хелпер `_try_rotate_within_budget` -(в оригинале инлайн-дублировалась в house/detail/house-detail фазах). +(в оригинале инлайн-дублировалась в house/detail/house-detail/enrich фазах). + +Yandex sweep'ы дёргают продуктовую логику (price-history + address-parse) через +инжектируемый `EnrichmentJobs` (record_yandex_price_history / extract_address_from_title +/ address_has_house_number) вместо прямых импортов app.services.yandex_*. """ from __future__ import annotations @@ -30,7 +38,8 @@ import random from collections.abc import Callable from contextlib import AsyncExitStack from dataclasses import dataclass, field, fields -from typing import TYPE_CHECKING +from datetime import date, timedelta +from typing import TYPE_CHECKING, Any from urllib.parse import urlparse from curl_cffi.requests import AsyncSession @@ -46,6 +55,19 @@ from scraper_kit.providers.avito.houses import ( save_house_catalog_enrichment, ) from scraper_kit.providers.avito.serp import AvitoScraper +from scraper_kit.providers.cian.detail import fetch_detail as cian_fetch_detail +from scraper_kit.providers.cian.detail import save_detail_enrichment as cian_save_detail_enrichment +from scraper_kit.providers.cian.newbuilding import ( + fetch_newbuilding, + save_newbuilding_enrichment, +) +from scraper_kit.providers.cian.serp import CianScraper +from scraper_kit.providers.domclick.serp import ROOM_BUCKETS, DomClickScraper +from scraper_kit.providers.yandex.serp import ( + DEFAULT_PRICE_RANGES, + ROOM_PATH, + YandexRealtyScraper, +) if TYPE_CHECKING: from sqlalchemy.orm import Session @@ -1410,3 +1432,1955 @@ async def run_avito_city_sweep( logger.exception("city-sweep run_id=%d: fatal error", run_id) runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise + + +# ── Avito newbuilding (novostroyka) citywide sweep ──────────────── + + +@dataclass +class NewbuildingSweepCounters: + """Aggregate counters для Avito novostroyka citywide sweep run.""" + + lots_fetched: int = 0 + lots_inserted: int = 0 + lots_updated: int = 0 + errors_count: int = 0 + + def to_dict(self) -> dict[str, int]: + return {f.name: getattr(self, f.name) for f in fields(self)} + + +async def run_avito_newbuilding_sweep( + db: Session, + *, + run_id: int, + config: ScraperConfig, + matcher: HouseMatcher, + shutdown_requested: Callable[[], bool] = lambda: False, + pages: int = 20, + request_delay_sec: float = 7.0, + region_code: int = DEFAULT_REGION_CODE, +) -> NewbuildingSweepCounters: + """Citywide-обход ЕКБ-выборки только новостроек (novostroyka-filter) → save. + + В отличие от run_avito_city_sweep (anchor-based fetch_around + houses/detail/IMV), + это простой citywide paginated-обход через AvitoScraper.fetch_newbuildings: + dedicated novostroyka-SERP отдаёт 100% new-build карточки (listing_segment= + "novostroyki", newbuilding_id/newbuilding_url заполнены парсером). Сохраняем через + стандартный save_listings, ведём SERP-счётчики через runs. + + - Single shared AsyncSession/BrowserFetcher на весь sweep (один TLS fingerprint) + - AvitoBlockedError/AvitoRateLimitedError → mark_banned (status='banned') + - is_cancelled-чек до старта SERP-фазы + + Инжекция (#2135 F2): config/matcher/shutdown_requested приходят снаружи вместо + прямых импортов app.* (см. scraper_kit.contracts). + """ + counters = NewbuildingSweepCounters() + + browser_mode = config.scraper_fetch_mode == "browser" + async with AsyncExitStack() as stack: + session: AsyncSession | None = None + shared_bf: BrowserFetcher | None = None + if browser_mode: + shared_bf = await stack.enter_async_context( + BrowserFetcher(source="avito", endpoint=config.browser_http_endpoint) + ) + else: + session = await stack.enter_async_context( + AsyncSession( + impersonate="chrome120", + timeout=25, + headers=_CHROME_HEADERS, + proxies=_avito_proxies(config), + ) + ) + try: + if runs.is_cancelled(db, run_id): + logger.info("nb-sweep run_id=%d: cancelled before SERP phase", run_id) + runs.mark_done(db, run_id, counters.to_dict()) + return counters + elif shutdown_requested(): + logger.info( + "nb-sweep run_id=%d: SIGTERM-drain — stopping before SERP phase", run_id + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_done(db, run_id, counters.to_dict()) + return counters + + scraper = AvitoScraper(config) + if browser_mode: + scraper._browser = shared_bf + # Shared-browser режим: _cffi=None → curl_cffi-fallback на firewall + # пропускается (by design). Логируем для observability. + logger.debug("avito pipeline-mode: shared browser only, no cffi fallback") + else: + scraper._cffi = session + + try: + lots: list[ScrapedLot] = await scraper.fetch_newbuildings( + pages=pages, + delay_override_sec=request_delay_sec, + ) + except (AvitoBlockedError, AvitoRateLimitedError) as e: + logger.error("nb-sweep run_id=%d: SERP BLOCKED — %s", run_id, e) + runs.mark_banned(db, run_id, str(e), counters.to_dict()) + return counters + + counters.lots_fetched += len(lots) + if lots: + try: + ins, upd = save_listings(db, lots, matcher=matcher, region_code=region_code) + counters.lots_inserted += ins + counters.lots_updated += upd + except Exception as save_exc: + logger.exception( + "nb-sweep run_id=%d: save_listings failed: %s", run_id, save_exc + ) + counters.errors_count += 1 + try: + db.rollback() + except Exception: + pass + + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_done(db, run_id, counters.to_dict()) + logger.info( + "nb-sweep run_id=%d done: lots=%d (ins=%d/upd=%d) errors=%d", + run_id, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.errors_count, + ) + return counters + + except Exception as exc: + logger.exception("nb-sweep run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + +# ── Yandex city sweep ─────────────────────────────────────────── + +# Константы для расчёта watchdog-таймаута combos-режима Yandex sweep. +# В combos-режиме единственный "anchor" делает num_combos × max_pages fetch'ей; +# каждый fetch занимает request_delay_sec + сетевой overhead. ADDRESS_ENRICH_BUDGET_S +# добавляет буфер на address-enrich фазу (per-listing HTTP + sleep × N листингов). +_YANDEX_COMBOS_PER_FETCH_S: float = 12.0 # network + parse margin (секунды на страницу) +_YANDEX_ADDRESS_ENRICH_BUDGET_S: float = 300.0 # бюджет на address-enrich фазу + + +@dataclass +class YandexCitySweepCounters: + """Aggregate counters для Yandex.Недвижимость city sweep run. + + Фазы: SERP (fetch_around_multi_room) → save_listings инкрементально → address-enrich. + address_* — результат T10-обогащения (detail <title> → полный адрес с номером дома). + combos_skipped — сколько (room×price) combo пропущено из-за таймаутов gate-API. + """ + + anchors_total: int = 0 + anchors_done: int = 0 + lots_fetched: int = 0 + lots_inserted: int = 0 + lots_updated: int = 0 + price_history_rows: int = 0 + combos_skipped: int = 0 + address_attempted: int = 0 + address_enriched: int = 0 + address_failed: int = 0 + errors_count: int = 0 + + def to_dict(self) -> dict[str, int]: + return {f.name: getattr(self, f.name) for f in fields(self)} + + +# Defensive abort threshold: у YandexRealtyScraper нет block/rate-exception классов +# (fetch_around глотает ошибки и возвращает []). Чтобы не молотить Yandex впустую при +# системном блоке/бане IP, прерываем sweep после N подряд anchor'ов, упавших с +# исключением. NB: пустой результат (0 lots) — это НЕ ошибка, счётчик не растёт. +YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES = 3 + + +async def run_yandex_city_sweep( + db: Session, + *, + run_id: int, + config: ScraperConfig, + matcher: HouseMatcher, + enrichment: EnrichmentJobs, + shutdown_requested: Callable[[], bool] = lambda: False, + anchors: list[tuple[float, float, str]] | None = None, + radius_m: int = 25000, + pages_per_anchor: int = 2, + request_delay_sec: float | None = None, + enrich_address: bool = True, + rooms_list: list[str] | None = None, + price_ranges: list[tuple[int | None, int | None]] | None = None, + segments: list[str] | None = None, + region_code: int = DEFAULT_REGION_CODE, +) -> YandexCitySweepCounters: + """Yandex.Недвижимость city sweep: rooms × price combos от центра ЕКБ → save → address-enrich. + + Вместо 5 географических anchor'ов итерирует ВСЕ combinations (room × price_range) + из одной центральной точки ЕКБ (56.8400, 60.6050). radius_m=25000 охватывает весь + город; price/room segmentation обходит Yandex SERP ~575-card cap без geo-шрапнели. + + Фазы (единственный центральный якорь): + 1. SERP: YandexRealtyScraper.fetch_around_multi_room(...) — combos mode (on_combo save). + 2. ADDRESS-ENRICH (если enrich_address=True): для yandex-листингов без номера дома + запрашиваем detail-страницу и извлекаем полный адрес из <title>. + + Инжекция (#2135 F2): config/matcher/shutdown_requested + продуктовые хелперы + (record_yandex_price_history / extract_address_from_title / address_has_house_number) + через enrichment (EnrichmentJobs) вместо прямых импортов app.services.yandex_*. + + Возвращает YandexCitySweepCounters. mark_done вызывается ВСЕГДА (кроме cancel/ban/fatal). + """ + # EKB center — единственный anchor для citywide combos sweep. + # anchors-параметр принимается для backward-compat тестов. + if anchors is not None: + _anchors: list[tuple[float, float, str]] = anchors + else: + _anchors = [(56.8400, 60.6050, "Центр (combos)")] + + _rooms_list = rooms_list or list(ROOM_PATH.keys()) + _price_ranges = price_ranges or DEFAULT_PRICE_RANGES + # newFlat-сегменты: оба прохода (vtorichka + novostroyki) по одним combos. + _segments = segments or ["NO", "YES"] + + counters = YandexCitySweepCounters(anchors_total=len(_anchors)) + inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0 + enrich_delay = request_delay_sec if request_delay_sec is not None else 3.0 + _resolved_delay = request_delay_sec if request_delay_sec is not None else 9.0 + consecutive_failures = 0 + yandex_rotations_done = 0 # #1848: бюджет IP-ротаций на весь sweep + + # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None). + _num_combos = len(_rooms_list) * len(_price_ranges) + if anchors is None and _num_combos > 0: + _sweep_timeout = max( + ANCHOR_TIMEOUT_SEC, + len(_segments) + * _num_combos + * pages_per_anchor + * (_resolved_delay + _YANDEX_COMBOS_PER_FETCH_S) + + _YANDEX_ADDRESS_ENRICH_BUDGET_S, + ) + else: + # Explicit anchors (backward-compat тесты или ручной override) — legacy timeout. + _sweep_timeout = ANCHOR_TIMEOUT_SEC + + # Прокси для address-enrich curl_cffi сессии (зеркало yandex_address_backfill). + _proxy_url = config.scraper_proxy_url + _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None + + try: + for idx, (lat, lon, name) in enumerate(_anchors, start=1): + # ── Cooperative cancel ─────────────────────────────────────────── + if runs.is_cancelled(db, run_id): + logger.info( + "yandex-sweep run_id=%d: cancelled at anchor #%d/%d (%s)", + run_id, + idx, + len(_anchors), + name, + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + return counters + elif shutdown_requested(): + # SIGTERM-drain: чистый стоп на границе anchor'а → mark_done(partial). + logger.info( + "yandex-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)", + run_id, + idx, + len(_anchors), + name, + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_done(db, run_id, counters.to_dict()) + return counters + + logger.info( + "yandex-sweep run_id=%d center-combos #%d/%d (%s, %.4f, %.4f) " + "radius_m=%d rooms=%d price_ranges=%d", + run_id, + idx, + len(_anchors), + name, + lat, + lon, + radius_m, + len(_rooms_list), + len(_price_ranges), + ) + + # ── Per-anchor phases под watchdog-таймаутом ──────────────────── + anchor_lots: list[ScrapedLot] = [] + _anchor_timed_out = False + _lat, _lon, _name = lat, lon, name + _combo_lots_accumulator: list[ScrapedLot] = anchor_lots + + async def _yandex_anchor_phases( + _a_lat: float = _lat, + _a_lon: float = _lon, + _a_name: str = _name, + _al: list[ScrapedLot] = _combo_lots_accumulator, + ) -> None: + """Все фазы одного anchor'а (SERP + address-enrich).""" + nonlocal consecutive_failures, yandex_rotations_done + + # ── Phase 1+2: SERP + инкрементальный save per-combo ────── + def _on_combo( + combo_label: str, + new_lots: list[ScrapedLot], + _accumulator: list[ScrapedLot] = _al, + ) -> None: + """Callback: вызывается scraper'ом после каждого combo с новыми лотами.""" + _accumulator.extend(new_lots) + counters.lots_fetched += len(new_lots) + try: + ins, upd = save_listings( + db, new_lots, matcher=matcher, region_code=region_code, run_id=run_id + ) + counters.lots_inserted += ins + counters.lots_updated += upd + except Exception as save_exc: + counters.errors_count += 1 + logger.warning( + "yandex-sweep run_id=%d combo %s: save_listings failed: %s", + run_id, + combo_label, + save_exc, + ) + try: + db.rollback() + except Exception: + pass + runs.update_heartbeat(db, run_id, counters.to_dict()) + logger.debug( + "yandex-sweep run_id=%d combo %s: saved %d lots " + "(ins=%d upd=%d total_fetched=%d)", + run_id, + combo_label, + len(new_lots), + counters.lots_inserted, + counters.lots_updated, + counters.lots_fetched, + ) + + async with YandexRealtyScraper(config) as scraper: + await scraper.fetch_around_multi_room( + _a_lat, + _a_lon, + radius_m, + max_pages=pages_per_anchor, + rooms_list=_rooms_list, + price_ranges=_price_ranges, + segments=_segments, + on_combo=_on_combo, + ) + # _al накоплен on_combo; save уже вызван per-combo. + + # Price-history из gate price.previous/trend (graceful — провал + # истории НЕ должен ронять сейв листингов). + if _al: + try: + counters.price_history_rows += enrichment.record_yandex_price_history( + db, _al + ) + except Exception as ph_exc: + logger.warning( + "yandex-sweep run_id=%d: price-history failed: %s", + run_id, + ph_exc, + ) + try: + db.rollback() + except Exception: + pass + + consecutive_failures = 0 + logger.info( + "yandex-sweep run_id=%d center-combos %s: fetched=%d ins=%d upd=%d", + run_id, + _a_name, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + ) + + # ── Phase 3: address-enrich ──────────────────────────────── + if not (enrich_address and _al): + return + source_ids = [lot.source_id for lot in _al if lot.source_id] + if not source_ids: + return + id_rows = ( + db.execute( + text(""" + SELECT id, source_id, address, source_url + FROM listings + WHERE source = 'yandex' + AND source_id = ANY(CAST(:ids AS text[])) + AND address IS NOT NULL + AND NOT (address ~ ',\\s*\\d+') + AND source_url IS NOT NULL + """), + {"ids": source_ids}, + ) + .mappings() + .all() + ) + items = [dict(r) for r in id_rows] + counters.address_attempted += len(items) + + if items: + # #1848: rotation tracking для Yandex address-enrich (stuck-proxy) + _yandex_enrich_consec_fail = 0 + _yandex_enrich_abort = 3 # abort address-enrich после N подряд + + async with AsyncSession( + impersonate="chrome120", + timeout=30.0, + proxies=_proxies, + headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"}, + ) as enrich_session: + eidx = 0 + while eidx < len(items): + item = items[eidx] + lid: int = item["id"] + old_addr: str = item["address"] + url: str = item["source_url"] + try: + resp = await enrich_session.get(url, allow_redirects=True) + if resp.status_code not in {200}: + logger.warning( + "yandex-sweep run_id=%d address-enrich:" + " HTTP %d listing_id=%d", + run_id, + resp.status_code, + lid, + ) + counters.address_failed += 1 + # 503/429 = прокси заблокирован → считаем как failure + if resp.status_code in {429, 503}: + _yandex_enrich_consec_fail += 1 + else: + _yandex_enrich_consec_fail = 0 + else: + new_addr = enrichment.extract_address_from_title(resp.text) + if ( + new_addr is None + or new_addr == old_addr + or not enrichment.address_has_house_number(new_addr) + ): + pass + else: + try: + db.execute( + text(""" + UPDATE listings + SET address = :addr, + geocode_tried_at = NULL + WHERE id = CAST(:id AS bigint) + """), + {"addr": new_addr, "id": lid}, + ) + db.commit() + counters.address_enriched += 1 + logger.info( + "yandex-sweep run_id=%d: " + "address updated listing_id=%d " + "%r -> %r", + run_id, + lid, + old_addr, + new_addr, + ) + except Exception as upd_exc: + counters.address_failed += 1 + counters.errors_count += 1 + logger.warning( + "yandex-sweep run_id=%d: " + "address DB update failed " + "listing_id=%d: %s", + run_id, + lid, + upd_exc, + ) + try: + db.rollback() + except Exception: + pass + _yandex_enrich_consec_fail = 0 + except Exception as fetch_exc: + counters.address_failed += 1 + _yandex_enrich_consec_fail += 1 + logger.warning( + "yandex-sweep run_id=%d: address-enrich " + "fetch failed listing_id=%d " + "(consecutive=%d): %s", + run_id, + lid, + _yandex_enrich_consec_fail, + fetch_exc, + ) + # Ротация Yandex-прокси при stuck/ban (#1848) + if _yandex_enrich_consec_fail >= _yandex_enrich_abort: + rotated, yandex_rotations_done = await _try_rotate_within_budget( + config, + reason=( + f"yandex enrich consecutive={_yandex_enrich_consec_fail}" + ), + rotations_done=yandex_rotations_done, + source="yandex", + ) + if rotated: + _yandex_enrich_consec_fail = 0 + logger.info( + "yandex-sweep run_id=%d: enrich ROTATE — " + "IP changed, reset consecutive " + "(rotation #%d/%d)", + run_id, + yandex_rotations_done, + config.yandex_proxy_max_rotations, + ) + # retry текущего item после ротации + continue + logger.error( + "yandex-sweep run_id=%d: address-enrich ABORT — " + "%d consecutive failures (proxy likely stuck/banned)", + run_id, + _yandex_enrich_consec_fail, + ) + break + if eidx < len(items) - 1: + await asyncio.sleep(enrich_delay) + eidx += 1 + + logger.info( + "yandex-sweep run_id=%d anchor %s: address enrich=%d/%d failed=%d", + run_id, + _a_name, + counters.address_enriched, + counters.address_attempted, + counters.address_failed, + ) + + try: + await asyncio.wait_for(_yandex_anchor_phases(), timeout=_sweep_timeout) + except TimeoutError: + logger.warning( + "yandex-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " + "timed out after %.0fs — skipping", + run_id, + idx, + len(_anchors), + name, + lat, + lon, + _sweep_timeout, + ) + counters.errors_count += 1 + consecutive_failures += 1 + _anchor_timed_out = True + if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES: + logger.error( + "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " + "%d consecutive failures (last: timeout): " + "IP likely blocked or proxy hung", + run_id, + idx, + len(_anchors), + name, + consecutive_failures, + ) + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_banned( + db, + run_id, + f"yandex sweep aborted: {consecutive_failures} consecutive " + f"anchor failures (last: anchor timeout {_sweep_timeout:.0f}s)", + counters.to_dict(), + ) + return counters + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + if idx < len(_anchors): + await asyncio.sleep(inter_anchor_delay) + continue + except Exception as e: + # YandexRealtyScraper не propagate'ит block/rate exceptions — ловим broad. + logger.exception("yandex-sweep run_id=%d: anchor %s SERP failed", run_id, name) + counters.errors_count += 1 + consecutive_failures += 1 + if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES: + logger.error( + "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " + "%d consecutive failures (IP likely blocked): %s", + run_id, + idx, + len(_anchors), + name, + consecutive_failures, + e, + ) + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_banned( + db, + run_id, + f"yandex sweep aborted: {consecutive_failures} consecutive " + f"anchor failures (last: {e})", + counters.to_dict(), + ) + return counters + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + # Пауза перед следующим anchor'ом даже при ошибке. + if idx < len(_anchors): + await asyncio.sleep(inter_anchor_delay) + continue + + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + + # Доп. пауза между anchor'ами (поверх per-page sleep внутри scraper'а). + if idx < len(_anchors): + await asyncio.sleep(inter_anchor_delay) + + runs.mark_done(db, run_id, counters.to_dict()) + logger.info( + "yandex-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) " + "address=%d/%d failed=%d errors=%d", + run_id, + counters.anchors_done, + counters.anchors_total, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.address_enriched, + counters.address_attempted, + counters.address_failed, + counters.errors_count, + ) + return counters + + except Exception as exc: + logger.exception("yandex-sweep run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + +# ── Cian city sweep (#860) ────────────────────────────────────────────────── + + +@dataclass +class CianCitySweepCounters: + """Aggregate counters для Cian city sweep run. + + Поля симметричны CitySweepCounters, но без imv_* (у Cian нет IMV-фазы). + detail_* — обогащение через cian_detail (вкл. price-history). + houses_* — обогащение newbuilding/ЖК через cian_newbuilding. + """ + + anchors_total: int = 0 + anchors_done: int = 0 + lots_fetched: int = 0 + lots_dropped_secondary: int = 0 # вторичка, отброшенная при newbuilding_only=True + lots_inserted: int = 0 + lots_updated: int = 0 + detail_attempted: int = 0 + detail_enriched: int = 0 + detail_failed: int = 0 + houses_attempted: int = 0 + houses_enriched: int = 0 + houses_failed: int = 0 + errors_count: int = 0 + + def to_dict(self) -> dict[str, int]: + return {f.name: getattr(self, f.name) for f in fields(self)} + + +# Defensive abort threshold: если N подряд anchor'ов упали с исключением +# (не просто вернули пустой результат) — прерываем sweep. +CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES = 3 + + +async def run_cian_city_sweep( + db: Session, + *, + run_id: int, + config: ScraperConfig, + matcher: HouseMatcher, + shutdown_requested: Callable[[], bool] = lambda: False, + anchors: list[tuple[float, float, str]] | None = None, + radius_m: int = 1500, + pages_per_anchor: int = 3, + request_delay_sec: float = 5.0, + detail_top_n: int = 10, + enrich_houses: bool = True, + newbuilding_only: bool = True, + region_code: int = DEFAULT_REGION_CODE, +) -> CianCitySweepCounters: + """Cian newbuilding city sweep: SERP → detail(+price-history) → newbuilding/houses. + + Симметрично run_avito_city_sweep. Использует EKB_ANCHORS (общий список). + + newbuilding_only (default True): в SERP-фазе сохраняются только лоты с + listing_segment == "novostroyki" — вторичку отбрасываем, ею авторитетно владеет + run_cian_full_load (exhaustive региональный сбор). + + Инжекция (#2135 F2): config/matcher/shutdown_requested приходят снаружи вместо + прямых импортов app.* (см. scraper_kit.contracts). + + Cooperative cancel: runs.is_cancelled(db, run_id) перед каждым anchor. + mark_done вызывается ВСЕГДА (finally outer). + """ + _anchors = anchors if anchors is not None else EKB_ANCHORS + counters = CianCitySweepCounters(anchors_total=len(_anchors)) + consecutive_failures = 0 + cian_rotations_done = 0 # #1848: бюджет IP-ротаций на весь sweep + + try: + for idx, (lat, lon, name) in enumerate(_anchors, start=1): + # Cooperative cancel перед каждым anchor + if runs.is_cancelled(db, run_id): + logger.info( + "cian-sweep run_id=%d: cancelled at anchor #%d/%d (%s)", + run_id, + idx, + len(_anchors), + name, + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + return counters + elif shutdown_requested(): + # SIGTERM-drain: чистый стоп на границе anchor'а → mark_done(partial). + logger.info( + "cian-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)", + run_id, + idx, + len(_anchors), + name, + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_done(db, run_id, counters.to_dict()) + return counters + + logger.info( + "cian-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", + run_id, + idx, + len(_anchors), + name, + lat, + lon, + ) + + # ── Per-anchor phases под watchdog-таймаутом ──────────────────── + anchor_lots: list[ScrapedLot] = [] + _c_lat, _c_lon, _c_name = lat, lon, name + + async def _cian_anchor_phases( + _a_lat: float = _c_lat, + _a_lon: float = _c_lon, + _a_name: str = _c_name, + ) -> None: + """Все фазы одного cian anchor'а (SERP + detail + houses).""" + nonlocal anchor_lots, consecutive_failures, cian_rotations_done + + # ── Phase 1+2: SERP + save ───────────────────────────────── + async with CianScraper(config) as scraper: + anchor_lots = await scraper.fetch_around_multi_room( + _a_lat, + _a_lon, + radius_m, + pages=pages_per_anchor, + ) + counters.lots_fetched += len(anchor_lots) + # ── Фильтр вторички (newbuilding_only) ───────────────────── + if newbuilding_only: + _before = len(anchor_lots) + anchor_lots = [ + lot for lot in anchor_lots if lot.listing_segment == "novostroyki" + ] + counters.lots_dropped_secondary += _before - len(anchor_lots) + if anchor_lots: + inserted, updated = save_listings( + db, anchor_lots, matcher=matcher, region_code=region_code, run_id=run_id + ) + counters.lots_inserted += inserted + counters.lots_updated += updated + consecutive_failures = 0 + logger.info( + "cian-sweep run_id=%d anchor %s: SERP fetched=%d nb_kept=%d " + "dropped_secondary=%d ins=%d upd=%d", + run_id, + _a_name, + counters.lots_fetched, + len(anchor_lots), + counters.lots_dropped_secondary, + counters.lots_inserted, + counters.lots_updated, + ) + + # ── Phase 3: detail enrichment (top-N) ──────────────────── + if detail_top_n > 0: + priority_rows = ( + db.execute( + text(""" + SELECT id, source_url + FROM listings + WHERE source = 'cian' + AND source_url IS NOT NULL + AND detail_enriched_at IS NULL + AND scraped_at > NOW() - INTERVAL '2 hours' + ORDER BY scraped_at DESC NULLS LAST + LIMIT :lim + """), + {"lim": detail_top_n}, + ) + .mappings() + .all() + ) + # #1848: Cian bans/stuck-proxy tracking + rotation (зеркало avito) + _cian_detail_consec_failures = 0 + _cian_detail_abort = 3 # abort detail-фазы после N подряд + didx = 0 + while didx < len(priority_rows): + row = priority_rows[didx] + listing_id: int = row["id"] + source_url: str = row["source_url"] + counters.detail_attempted += 1 + try: + cian_enrichment = await cian_fetch_detail(source_url) + if cian_enrichment is not None: + cian_save_detail_enrichment(db, listing_id, cian_enrichment) + counters.detail_enriched += 1 + else: + counters.detail_failed += 1 + logger.warning( + "cian-sweep run_id=%d: detail returned None for " + "listing_id=%d url=%s", + run_id, + listing_id, + source_url, + ) + _cian_detail_consec_failures = 0 + except Exception as exc: + counters.detail_failed += 1 + counters.errors_count += 1 + _cian_detail_consec_failures += 1 + logger.warning( + "cian-sweep run_id=%d: detail failed listing_id=%d " + "(consecutive=%d): %s", + run_id, + listing_id, + _cian_detail_consec_failures, + exc, + ) + if _cian_detail_consec_failures >= _cian_detail_abort: + # Попытка ротации Cian-прокси перед abort'ом (#1848) + rotated, cian_rotations_done = await _try_rotate_within_budget( + config, + reason=( + f"cian detail consecutive={_cian_detail_consec_failures}" + ), + rotations_done=cian_rotations_done, + source="cian", + ) + if rotated: + _cian_detail_consec_failures = 0 + logger.info( + "cian-sweep run_id=%d: detail ROTATE — " + "IP changed, reset consecutive " + "(rotation #%d/%d)", + run_id, + cian_rotations_done, + config.cian_proxy_max_rotations, + ) + # retry текущего detail (не инкрементируем didx) + continue + logger.error( + "cian-sweep run_id=%d: detail ABORT — " + "%d consecutive failures (proxy likely banned/stuck)", + run_id, + _cian_detail_consec_failures, + ) + break + if didx < len(priority_rows) - 1: + jitter = random.uniform(0.8, 1.2) + await asyncio.sleep(request_delay_sec * jitter) + didx += 1 + logger.info( + "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d", + run_id, + _a_name, + counters.detail_enriched, + counters.detail_attempted, + counters.detail_failed, + ) + + # ── Phase 4: newbuilding/houses enrichment ───────────────── + if not (enrich_houses and anchor_lots): + return + cian_nb_ext_ids = { + lot.house_ext_id + for lot in anchor_lots + if lot.house_source == "cian_newbuilding" and lot.house_ext_id + } + if not cian_nb_ext_ids: + return + nb_id_list = list(cian_nb_ext_ids) + try: + house_rows = ( + db.execute( + text(""" + SELECT h.id, h.cian_zhk_url + FROM houses h + JOIN house_sources hs ON hs.house_id = h.id + WHERE hs.ext_source = 'cian_newbuilding' + AND hs.ext_id = ANY(CAST(:ids AS text[])) + AND h.cian_zhk_url IS NOT NULL + """), + {"ids": nb_id_list}, + ) + .mappings() + .all() + ) + except Exception as exc: + logger.exception( + "cian-sweep run_id=%d anchor %s: houses DB query failed — " + "skipping houses phase (%d ids): %s", + run_id, + _a_name, + len(nb_id_list), + exc, + ) + counters.houses_failed += len(nb_id_list) + try: + db.rollback() + except Exception: + pass + return + for hidx, hrow in enumerate(house_rows): + house_id_val: int = hrow["id"] + zhk_url: str = hrow["cian_zhk_url"] + counters.houses_attempted += 1 + try: + nb_enrichment = await fetch_newbuilding(zhk_url) + if nb_enrichment is not None: + save_newbuilding_enrichment(db, house_id_val, nb_enrichment) + counters.houses_enriched += 1 + else: + counters.houses_failed += 1 + logger.warning( + "cian-sweep run_id=%d: newbuilding returned None " + "house_id=%d url=%s", + run_id, + house_id_val, + zhk_url, + ) + except Exception as exc: + counters.houses_failed += 1 + counters.errors_count += 1 + logger.warning( + "cian-sweep run_id=%d: houses failed house_id=%d: %s", + run_id, + house_id_val, + exc, + ) + try: + db.rollback() + except Exception: + pass + if hidx < len(house_rows) - 1: + await asyncio.sleep(request_delay_sec) + logger.info( + "cian-sweep run_id=%d anchor %s: houses=%d/%d failed=%d", + run_id, + _a_name, + counters.houses_enriched, + counters.houses_attempted, + counters.houses_failed, + ) + + try: + await asyncio.wait_for(_cian_anchor_phases(), timeout=ANCHOR_TIMEOUT_SEC) + except TimeoutError: + logger.warning( + "cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " + "timed out after %ds — skipping", + run_id, + idx, + len(_anchors), + name, + lat, + lon, + ANCHOR_TIMEOUT_SEC, + ) + counters.errors_count += 1 + consecutive_failures += 1 + if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES: + logger.error( + "cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " + "%d consecutive failures (last: timeout): " + "IP likely blocked or proxy hung", + run_id, + idx, + len(_anchors), + name, + consecutive_failures, + ) + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_banned( + db, + run_id, + f"cian sweep aborted: {consecutive_failures} consecutive " + f"anchor failures (last: anchor timeout {ANCHOR_TIMEOUT_SEC}s)", + counters.to_dict(), + ) + return counters + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + continue + except Exception as e: + logger.exception("cian-sweep run_id=%d: anchor %s SERP failed", run_id, name) + counters.errors_count += 1 + consecutive_failures += 1 + if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES: + logger.error( + "cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " + "%d consecutive SERP failures (IP likely blocked): %s", + run_id, + idx, + len(_anchors), + name, + consecutive_failures, + e, + ) + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_banned( + db, + run_id, + f"cian sweep aborted: {consecutive_failures} consecutive " + f"anchor SERP failures (last: {e})", + counters.to_dict(), + ) + return counters + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + continue + + counters.anchors_done = idx + runs.update_heartbeat(db, run_id, counters.to_dict()) + + # Пауза между anchor'ами (поверх per-request sleep внутри scraper'а) + if idx < len(_anchors): + await asyncio.sleep(request_delay_sec) + + runs.mark_done(db, run_id, counters.to_dict()) + logger.info( + "cian-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) " + "detail=%d/%d houses=%d/%d errors=%d", + run_id, + counters.anchors_done, + counters.anchors_total, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.detail_enriched, + counters.detail_attempted, + counters.houses_enriched, + counters.houses_attempted, + counters.errors_count, + ) + return counters + + except Exception as exc: + logger.exception("cian-sweep run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + +# ── Cian exhaustive full load (region-wide, без anchor'ов) ───────────────── + + +@dataclass +class CianFullLoadCounters: + """Aggregate counters для run_cian_full_load. + + Без anchor-полей — единственный региональный проход. + detail_* — опциональное обогащение detail-страниц (если enrich_detail=True). + """ + + unique_fetched: int = 0 + saved_inserted: int = 0 + saved_updated: int = 0 + detail_attempted: int = 0 + detail_enriched: int = 0 + detail_failed: int = 0 + errors_count: int = 0 + + def to_dict(self) -> dict[str, int]: + return {f.name: getattr(self, f.name) for f in fields(self)} + + +async def run_cian_full_load( + db: Session, + *, + run_id: int, + config: ScraperConfig, + matcher: HouseMatcher, + shutdown_requested: Callable[[], bool] = lambda: False, + price_cap_per_bucket: int = 1400, + detail_top_n: int = 0, + request_delay_sec: float = 4.0, + enrich_detail: bool = False, + concurrency: int = 5, + resume_run_id: int | None = None, + region_code: int = DEFAULT_REGION_CODE, +) -> CianFullLoadCounters: + """Exhaustive региональный сбор Cian ЕКБ вторички (БЕЗ anchor'ов). + + Один проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное + деление) охватывает весь ЕКБ. Инкрементальный save: on_bucket коммитит каждый + leaf-бакет в БД сразу после сбора. + + Инжекция (#2135 F2): config/matcher/shutdown_requested приходят снаружи вместо + прямых импортов app.* (см. scraper_kit.contracts). + Cooperative cancel: runs.is_cancelled проверяется per-bucket. + """ + counters = CianFullLoadCounters() + + # ── Checkpoint/resume: читаем done_buckets из прошлого run ─────────────── + skip_set: set[str] = set() + if resume_run_id is not None: + _prev_row = db.execute( + text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"), + {"rid": resume_run_id}, + ).fetchone() + if _prev_row is not None and _prev_row.counters: + _prev_counters: dict = ( + _prev_row.counters if isinstance(_prev_row.counters, dict) else {} + ) + skip_set = set(_prev_counters.get("done_buckets", [])) + logger.info( + "cian-full-load run_id=%d: resuming from run %d — %d buckets already done", + run_id, + resume_run_id, + len(skip_set), + ) + + done: set[str] = set(skip_set) # накапливаем завершённые бакеты этого прогона + + def _on_bucket(bucket_key: str, lots: list) -> None: # type: ignore[type-arg] + """Инкрементальный save после каждого leaf-бакета. Дописывает bucket_key в done.""" + nonlocal done + if runs.is_cancelled(db, run_id): + logger.info("cian-full-load run_id=%d: cancel detected in on_bucket", run_id) + raise RuntimeError("cancelled") + elif shutdown_requested(): + # SIGTERM-drain: тот же sentinel-механизм, что и user-cancel, но отдельный + # маркер "shutdown" → handler финализирует mark_done(partial), не mark_failed. + logger.info( + "cian-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", + run_id, + bucket_key, + ) + raise RuntimeError("shutdown") + if not lots: + done.add(bucket_key) + runs.update_heartbeat( + db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} + ) + return + inserted, updated = save_listings( + db, + lots, + matcher=matcher, + region_code=region_code, + run_id=run_id, + skip_seen_today=config.scraper_skip_seen_today, + ) + # save_listings вызывает db.commit() внутри — данные в БД сразу + counters.saved_inserted += inserted + counters.saved_updated += updated + counters.unique_fetched += len(lots) + done.add(bucket_key) + runs.update_heartbeat(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + logger.info( + "cian-full-load run_id=%d: bucket %s saved ins=%d upd=%d total_unique=%d", + run_id, + bucket_key, + inserted, + updated, + counters.unique_fetched, + ) + + def _on_progress(unique_count: int) -> None: + """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов).""" + runs.update_heartbeat(db, run_id, counters.to_dict()) + + # #1949: фоновый heartbeat — обновляет heartbeat_at каждые 60s независимо от on_bucket. + async def _background_heartbeat() -> None: + """Периодический heartbeat каждые 60s, не завязанный на прогресс бакетов.""" + while True: + await asyncio.sleep(60.0) + try: + runs.update_heartbeat(db, run_id, counters.to_dict()) + except Exception: + pass + + _hb_task: asyncio.Task[None] | None = None + try: + _hb_task = asyncio.create_task(_background_heartbeat()) + + async with CianScraper(config) as scraper: + scraper.request_delay_sec = request_delay_sec + + # #1949: per-fetch hard-cancel для browser-fetch'ей внутри bucket-gather. + _fetch_timeout = config.cian_full_load_per_fetch_timeout_s + if _fetch_timeout > 0 and scraper._browser is not None: + _orig_fetch = scraper._browser.fetch + + async def _timed_fetch( + url: str, + _f: Any = _orig_fetch, + _t: float = _fetch_timeout, + _rid: int = run_id, + ) -> str: + try: + return await asyncio.wait_for(_f(url), timeout=_t) + except TimeoutError: + logger.warning( + "cian-full-load run_id=%d: browser-fetch timed out " + "after %.0fs — url=%s", + _rid, + _t, + url, + ) + raise + + scraper._browser.fetch = _timed_fetch # type: ignore[method-assign] + logger.info( + "cian-full-load run_id=%d: per-fetch timeout %.0fs enabled", + run_id, + _fetch_timeout, + ) + + await scraper.fetch_all_secondary( + price_cap_per_bucket=price_cap_per_bucket, + concurrency=concurrency, + secondary_only=True, + on_bucket=_on_bucket, + on_progress=_on_progress, + skip_buckets=skip_set if skip_set else None, + ) + + logger.info( + "cian-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + + # ── Опциональный detail-энричмент ───────────────────────────────────── + if enrich_detail and detail_top_n > 0: + priority_rows = ( + db.execute( + text(""" + SELECT id, source_url + FROM listings + WHERE source = 'cian' + AND source_url IS NOT NULL + AND detail_enriched_at IS NULL + AND scraped_at > NOW() - INTERVAL '4 hours' + ORDER BY scraped_at DESC NULLS LAST + LIMIT :lim + """), + {"lim": detail_top_n}, + ) + .mappings() + .all() + ) + for didx, row in enumerate(priority_rows): + if runs.is_cancelled(db, run_id): + logger.info("cian-full-load run_id=%d: cancelled during detail enrich", run_id) + break + elif shutdown_requested(): + logger.info( + "cian-full-load run_id=%d: SIGTERM-drain — stopping detail at #%d/%d", + run_id, + didx + 1, + len(priority_rows), + ) + break + listing_id: int = row["id"] + source_url: str = row["source_url"] + counters.detail_attempted += 1 + try: + cian_enrichment = await cian_fetch_detail(source_url) + if cian_enrichment is not None: + cian_save_detail_enrichment(db, listing_id, cian_enrichment) + counters.detail_enriched += 1 + else: + counters.detail_failed += 1 + except Exception as exc: + counters.detail_failed += 1 + counters.errors_count += 1 + logger.warning( + "cian-full-load run_id=%d: detail failed listing_id=%d: %s", + run_id, + listing_id, + exc, + ) + if didx < len(priority_rows) - 1: + await asyncio.sleep(request_delay_sec * random.uniform(0.8, 1.2)) + + logger.info( + "cian-full-load run_id=%d: detail done — enriched=%d/%d failed=%d", + run_id, + counters.detail_enriched, + counters.detail_attempted, + counters.detail_failed, + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + logger.info( + "cian-full-load run_id=%d done: unique=%d ins=%d upd=%d detail=%d/%d errors=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + counters.detail_enriched, + counters.detail_attempted, + counters.errors_count, + ) + return counters + + except RuntimeError as exc: + # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, + # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). + if str(exc) == "cancelled": + logger.info( + "cian-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters + if str(exc) == "shutdown": + logger.info( + "cian-full-load run_id=%d: SIGTERM-drain — partial results unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters + logger.exception("cian-full-load run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + except Exception as exc: + logger.exception("cian-full-load run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + finally: + # #1949: отменяем фоновый heartbeat при любом завершении (return / raise / cancel). + if _hb_task is not None: + _hb_task.cancel() + try: + await _hb_task + except asyncio.CancelledError: + pass + + +# ── Yandex exhaustive full load (region-wide, без anchor'ов) ────────────────── + + +@dataclass +class YandexFullLoadCounters: + """Aggregate counters для run_yandex_full_load. + + Без detail_* полей — Yandex full load не делает detail-обогащение. + """ + + unique_fetched: int = 0 + saved_inserted: int = 0 + saved_updated: int = 0 + price_history_rows: int = 0 + errors_count: int = 0 + + def to_dict(self) -> dict[str, int]: + return {f.name: getattr(self, f.name) for f in fields(self)} + + +async def run_yandex_full_load( + db: Session, + *, + run_id: int, + config: ScraperConfig, + matcher: HouseMatcher, + enrichment: EnrichmentJobs, + shutdown_requested: Callable[[], bool] = lambda: False, + price_cap_per_bucket: int = 500, + request_delay_sec: float = 2.0, + concurrency: int = 4, + resume_run_id: int | None = None, + region_code: int = DEFAULT_REGION_CODE, +) -> YandexFullLoadCounters: + """Exhaustive региональный сбор Yandex ЕКБ вторички (БЕЗ anchor'ов). + + Один проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное + деление) охватывает весь ЕКБ. Инкрементальный save: on_bucket коммитит каждый + leaf-бакет в БД сразу после сбора. + + Инжекция (#2135 F2): config/matcher/enrichment/shutdown_requested приходят снаружи + вместо прямых импортов app.* (см. scraper_kit.contracts). record_yandex_price_history + зовётся через enrichment (EnrichmentJobs Protocol). + Cooperative cancel: runs.is_cancelled проверяется per-bucket. + """ + counters = YandexFullLoadCounters() + + skip_set: set[str] = set() + if resume_run_id is not None: + _prev_row = db.execute( + text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"), + {"rid": resume_run_id}, + ).fetchone() + if _prev_row is not None and _prev_row.counters: + _prev_counters: dict = ( + _prev_row.counters if isinstance(_prev_row.counters, dict) else {} + ) + skip_set = set(_prev_counters.get("done_buckets", [])) + logger.info( + "yandex-full-load run_id=%d: resuming from run %d — %d buckets already done", + run_id, + resume_run_id, + len(skip_set), + ) + done: set[str] = set(skip_set) + + def _on_bucket(bucket_key: str, lots: list) -> None: # type: ignore[type-arg] + """Инкрементальный save после каждого leaf-бакета.""" + nonlocal done + if runs.is_cancelled(db, run_id): + logger.info("yandex-full-load run_id=%d: cancel detected in on_bucket", run_id) + raise RuntimeError("cancelled") + elif shutdown_requested(): + logger.info( + "yandex-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", + run_id, + bucket_key, + ) + raise RuntimeError("shutdown") + if not lots: + done.add(bucket_key) + runs.update_heartbeat( + db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} + ) + return + inserted, updated = save_listings( + db, + lots, + matcher=matcher, + region_code=region_code, + run_id=run_id, + skip_seen_today=config.scraper_skip_seen_today, + ) + # save_listings вызывает db.commit() внутри — данные в БД сразу + counters.saved_inserted += inserted + counters.saved_updated += updated + counters.unique_fetched += len(lots) + # Price-history из gate price.previous/trend (graceful — провал истории + # НЕ должен ронять сейв листингов/бакета). + try: + counters.price_history_rows += enrichment.record_yandex_price_history(db, lots) + except Exception as ph_exc: + logger.warning( + "yandex-full-load run_id=%d: price-history failed bucket=%s: %s", + run_id, + bucket_key, + ph_exc, + ) + try: + db.rollback() + except Exception: + pass + done.add(bucket_key) + runs.update_heartbeat(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + logger.info( + "yandex-full-load run_id=%d: bucket=%s saved ins=%d upd=%d total_unique=%d", + run_id, + bucket_key, + inserted, + updated, + counters.unique_fetched, + ) + + def _on_progress(unique_count: int) -> None: + """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов).""" + runs.update_heartbeat(db, run_id, counters.to_dict()) + + try: + async with YandexRealtyScraper(config) as scraper: + scraper.request_delay_sec = request_delay_sec + + await scraper.fetch_all_secondary( + price_cap_per_bucket=price_cap_per_bucket, + concurrency=concurrency, + on_bucket=_on_bucket, + on_progress=_on_progress, + skip_buckets=skip_set if skip_set else None, + ) + + logger.info( + "yandex-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + logger.info( + "yandex-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + counters.errors_count, + ) + return counters + + except RuntimeError as exc: + # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, + # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). + if str(exc) == "cancelled": + logger.info( + "yandex-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters + if str(exc) == "shutdown": + logger.info( + "yandex-full-load run_id=%d: SIGTERM-drain — partial results " + "unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters + logger.exception("yandex-full-load run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + except Exception as exc: + logger.exception("yandex-full-load run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + +# ── Avito exhaustive full load (citywide, без anchor'ов) ────────────────────── + + +@dataclass +class AvitoFullLoadCounters: + """Aggregate counters для run_avito_full_load. + + Зеркало YandexFullLoadCounters — citywide room×price bisection без anchor/detail. + """ + + unique_fetched: int = 0 + saved_inserted: int = 0 + saved_updated: int = 0 + errors_count: int = 0 + + def to_dict(self) -> dict[str, int]: + return {f.name: getattr(self, f.name) for f in fields(self)} + + +async def run_avito_full_load( + db: Session, + *, + run_id: int, + config: ScraperConfig, + matcher: HouseMatcher, + shutdown_requested: Callable[[], bool] = lambda: False, + price_cap_per_bucket: int = 1400, + request_delay_sec: float = 7.0, + concurrency: int = 5, + secondary_only: bool = True, + resume_run_id: int | None = None, + incremental_days: int | None = None, + region_code: int = DEFAULT_REGION_CODE, +) -> AvitoFullLoadCounters: + """Exhaustive ИЛИ инкрементальный региональный сбор Avito ЕКБ вторички (БЕЗ anchor'ов). + + Один проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное + деление ценового диапазона через `pmin`/`pmax`) охватывает весь ЕКБ. + + incremental_days: если задан — shallow инкрементальный обход (since-cutoff). + + Инжекция (#2135 F2): config/matcher/shutdown_requested приходят снаружи вместо + прямых импортов app.* (см. scraper_kit.contracts). + Cooperative cancel: runs.is_cancelled проверяется per-bucket. + AvitoBlockedError/AvitoRateLimitedError → mark_banned (status='banned'). + """ + counters = AvitoFullLoadCounters() + + # ── Инкрементальный режим: since-cutoff для shallow обхода ──────────────── + since: date | None = None + if incremental_days is not None: + since = date.today() - timedelta(days=incremental_days) + logger.info( + "avito-full-load run_id=%d: INCREMENTAL mode — since=%s (last %d days)", + run_id, + since.isoformat(), + incremental_days, + ) + + # ── Checkpoint/resume: читаем done_buckets из прошлого run ─────────────── + skip_set: set[str] = set() + if resume_run_id is not None: + _prev_row = db.execute( + text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"), + {"rid": resume_run_id}, + ).fetchone() + if _prev_row is not None and _prev_row.counters: + _prev_counters: dict = ( + _prev_row.counters if isinstance(_prev_row.counters, dict) else {} + ) + skip_set = set(_prev_counters.get("done_buckets", [])) + logger.info( + "avito-full-load run_id=%d: resuming from run %d — %d buckets already done", + run_id, + resume_run_id, + len(skip_set), + ) + done: set[str] = set(skip_set) + + def _on_bucket(bucket_key: str, lots: list) -> None: # type: ignore[type-arg] + """Инкрементальный save после каждого leaf-бакета.""" + nonlocal done + if runs.is_cancelled(db, run_id): + logger.info("avito-full-load run_id=%d: cancel detected in on_bucket", run_id) + raise RuntimeError("cancelled") + elif shutdown_requested(): + logger.info( + "avito-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", + run_id, + bucket_key, + ) + raise RuntimeError("shutdown") + if not lots: + done.add(bucket_key) + runs.update_heartbeat( + db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} + ) + return + inserted, updated = save_listings( + db, + lots, + matcher=matcher, + region_code=region_code, + run_id=run_id, + skip_seen_today=config.scraper_skip_seen_today, + ) + # save_listings вызывает db.commit() внутри — данные в БД сразу + counters.saved_inserted += inserted + counters.saved_updated += updated + counters.unique_fetched += len(lots) + done.add(bucket_key) + runs.update_heartbeat(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + logger.info( + "avito-full-load run_id=%d: bucket=%s saved ins=%d upd=%d total_unique=%d", + run_id, + bucket_key, + inserted, + updated, + counters.unique_fetched, + ) + + def _on_progress(unique_count: int) -> None: + """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов).""" + runs.update_heartbeat(db, run_id, counters.to_dict()) + + try: + async with AvitoScraper(config) as scraper: + scraper.request_delay_sec = request_delay_sec + + await scraper.fetch_all_secondary( + price_cap_per_bucket=price_cap_per_bucket, + concurrency=concurrency, + secondary_only=secondary_only, + on_bucket=_on_bucket, + on_progress=_on_progress, + skip_buckets=skip_set if skip_set else None, + since=since, + ) + + logger.info( + "avito-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + logger.info( + "avito-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + counters.errors_count, + ) + return counters + + except (AvitoBlockedError, AvitoRateLimitedError) as exc: + # IP-бан/rate-limit: partial results уже сохранены через on_bucket. + logger.error("avito-full-load run_id=%d: blocked/rate-limited — %s", run_id, exc) + counters.errors_count += 1 + runs.mark_banned( + db, + run_id, + f"avito full load aborted: {exc}", + {**counters.to_dict(), "done_buckets": sorted(done)}, + ) + return counters + + except RuntimeError as exc: + # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, + # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). + if str(exc) == "cancelled": + logger.info( + "avito-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters + if str(exc) == "shutdown": + logger.info( + "avito-full-load run_id=%d: SIGTERM-drain — partial results " + "unique=%d ins=%d upd=%d", + run_id, + counters.unique_fetched, + counters.saved_inserted, + counters.saved_updated, + ) + runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) + return counters + logger.exception("avito-full-load run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + except Exception as exc: + logger.exception("avito-full-load run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + +# ── DomClick citywide sweep (без anchor'ов) ──────────────────────────────── + + +@dataclass +class DomClickCitySweepCounters: + """Aggregate counters для DomClick citywide sweep run. + + DomClick BFF — citywide (city_id), не geo-bbox: anchor-loop отсутствует. + Один проход fetch_city перебирает ROOM_BUCKETS × pages внутри scraper'а. + """ + + lots_fetched: int = 0 + lots_inserted: int = 0 + lots_updated: int = 0 + pages_fetched: int = 0 + errors_count: int = 0 + blocked: int = 0 # 1 если QRATOR-блок был во время sweep + geo_filtered: int = 0 # число офферов отфильтрованных geo-guard + + def to_dict(self) -> dict[str, int]: + return {f.name: getattr(self, f.name) for f in fields(self)} + + +# Дефолтные параметры sweep'а (EKB city_id=4). +DOMCLICK_DEFAULT_CITY_ID: int = 4 +DOMCLICK_DEFAULT_ROOMS: list[int] = [0, 1, 2, 3, 4] # vestigial; scraper sweeps all buckets +# Число BFF-бакетов (st/1/2/3/4/5+) — фиксировано; используется для watchdog. +_DOMCLICK_NUM_BUCKETS: int = 6 +# Оценка времени одного fetch'а (network + parse) для watchdog. +_DOMCLICK_PER_FETCH_S: float = 12.0 +# Буфер сверху расчётного бюджета (cold browser start, save-фаза, price-splits). +_DOMCLICK_SWEEP_BUDGET_S: float = 300.0 + + +async def run_domclick_city_sweep( + db: Session, + *, + run_id: int, + config: ScraperConfig, + matcher: HouseMatcher, + shutdown_requested: Callable[[], bool] = lambda: False, + city_id: int = DOMCLICK_DEFAULT_CITY_ID, + rooms: list[int] | None = None, + pages: int = 100, + request_delay_sec: float | None = None, + region_code: int = DEFAULT_REGION_CODE, +) -> DomClickCitySweepCounters: + """DomClick citywide sweep через BFF JSON API. + + Структурно зеркалит run_cian_city_sweep / run_yandex_city_sweep, но CITYWIDE: + DomClick не поддерживает geo-radius (fetch_around → NotImplementedError), + поэтому anchor-loop отсутствует. DomClickScraper.fetch_city перебирает все + ROOM_BUCKETS (st/1/2/3/4/5+) через BrowserFetcher → shared mobile proxy. + + Инжекция (#2135 F2): config/matcher/shutdown_requested приходят снаружи вместо + прямых импортов app.* (см. scraper_kit.contracts). + + ЧЕСТНЫЙ СТАТУС (#1968): если scraper сообщил QRATOR-блок И lots == 0 → + mark_failed. Иначе mark_done. + + Возвращает DomClickCitySweepCounters. + """ + _resolved_delay = request_delay_sec if request_delay_sec is not None else 6.0 + counters = DomClickCitySweepCounters() + + # Watchdog: 6 buckets × pages × per_fetch + budget. + _num_fetches = _DOMCLICK_NUM_BUCKETS * max(1, pages) + _sweep_timeout = max( + ANCHOR_TIMEOUT_SEC, + int(_num_fetches * (_resolved_delay + _DOMCLICK_PER_FETCH_S) + _DOMCLICK_SWEEP_BUDGET_S), + ) + + # Мутируемый контейнер для захвата scraper-ссылки из замыкания. + _scraper_ref: list[DomClickScraper] = [] + + try: + # ── Cooperative cancel перед SERP-фазой ────────────────────────────── + if runs.is_cancelled(db, run_id): + logger.info("domclick-sweep run_id=%d: cancelled before SERP", run_id) + runs.update_heartbeat(db, run_id, counters.to_dict()) + return counters + elif shutdown_requested(): + # SIGTERM-drain до SERP-фазы: финализируем run как mark_done(partial), + # минуя honest-status (это чистый drain, не QRATOR-блок). + logger.info("domclick-sweep run_id=%d: SIGTERM-drain — stopping before SERP", run_id) + runs.update_heartbeat(db, run_id, counters.to_dict()) + runs.mark_done(db, run_id, counters.to_dict()) + return counters + + logger.info( + "domclick-sweep run_id=%d: BFF citywide sweep city_id=%d " + "buckets=%d pages_cap=%d (watchdog %ds)", + run_id, + city_id, + len(ROOM_BUCKETS), + pages, + _sweep_timeout, + ) + + lots: list[ScrapedLot] = [] + + async def _domclick_phase() -> None: + """Единственная citywide-фаза: fetch_city + save.""" + nonlocal lots + async with DomClickScraper(config) as _scraper: + _scraper_ref.append(_scraper) + if request_delay_sec is not None: + _scraper.request_delay_sec = _resolved_delay + lots = await _scraper.fetch_city(city_id=city_id, rooms=rooms, pages=pages) + counters.lots_fetched += len(lots) + if lots: + inserted, updated = save_listings( + db, lots, matcher=matcher, region_code=region_code, run_id=run_id + ) + counters.lots_inserted += inserted + counters.lots_updated += updated + + try: + await asyncio.wait_for(_domclick_phase(), timeout=_sweep_timeout) + except TimeoutError: + logger.warning( + "domclick-sweep run_id=%d: SERP phase timed out after %ds — partial results", + run_id, + _sweep_timeout, + ) + counters.errors_count += 1 + except Exception: + logger.exception("domclick-sweep run_id=%d: SERP phase failed", run_id) + counters.errors_count += 1 + + # Перенести счётчики scraper'а в counters (если scraper был создан). + if _scraper_ref: + _s = _scraper_ref[0] + counters.blocked = 1 if _s.blocked else 0 + counters.geo_filtered = _s.geo_filtered + # Не-block fetch-ошибки скрейпера учитываем в errors_count. + counters.errors_count += _s.fetch_errors + + # pages_fetched: worst-case число страниц (buckets × pages cap). + counters.pages_fetched = _num_fetches + runs.update_heartbeat(db, run_id, counters.to_dict()) + + # ── ЧЕСТНЫЙ СТАТУС (#1968) ──────────────────────────────────────────── + if counters.lots_fetched == 0 and (counters.blocked or counters.errors_count > 0): + logger.error( + "domclick-sweep run_id=%d: 0 listings with blocked=%d errors=%d " + "— marking failed", + run_id, + counters.blocked, + counters.errors_count, + ) + runs.mark_failed( + db, + run_id, + "QRATOR block or fetch errors — 0 listings", + counters.to_dict(), + ) + else: + runs.mark_done(db, run_id, counters.to_dict()) + + logger.info( + "domclick-sweep run_id=%d done: lots=%d (ins=%d/upd=%d) " + "pages=%d errors=%d blocked=%d geo_filtered=%d", + run_id, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.pages_fetched, + counters.errors_count, + counters.blocked, + counters.geo_filtered, + ) + return counters + + except Exception as exc: + logger.exception("domclick-sweep run_id=%d: fatal error", run_id) + runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise