diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 0555a7b6..831623a7 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -53,6 +53,9 @@ Sources: enable manually after deploy verified; window 02:00-05:00 UTC) - cian_city_sweep → run_cian_city_sweep (newbuilding Phase 4, #973; shipped DORMANT — proxy dead, enable manually after proxy restored) + - domclick_city_sweep → run_domclick_city_sweep (scrape_pipeline.py; citywide SERP via + camoufox BrowserFetcher; shipped DORMANT enabled=false, enable + manually after prod-smoke) - geocode_missing_listings → run_geocode_missing_listings (tasks/geocode_missing.py; nightly backfill listings.lat/lon для всех источников с NULL coords; address-dedup batch loop с @@ -83,6 +86,7 @@ from app.services import scrape_runs as runs_mod from app.services.scrape_pipeline import ( run_avito_city_sweep, run_cian_city_sweep, + run_domclick_city_sweep, run_yandex_city_sweep, ) @@ -429,6 +433,43 @@ async def trigger_cian_city_sweep_run(db: Session, schedule_row: dict[str, Any]) return run_id +async def trigger_domclick_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать scrape_runs + launch run_domclick_city_sweep в asyncio.create_task. + + Зеркало trigger_yandex_city_sweep_run, но citywide (city_id, без anchor/geo): + DomClick SERP не поддерживает geo-radius. Shipped DORMANT — seed enabled=false + (122), включается оператором вручную после prod-smoke. + + Returns run_id (или None если skip — есть running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + await run_domclick_city_sweep( + run_db, + run_id=run_id, + city_id=int(params.get("city_id", 4)), + rooms=params.get("rooms"), + pages=int(params.get("pages_per_anchor", 5)), + request_delay_sec=float(params.get("request_delay_sec", 6.0)), + ) + except Exception: + logger.exception("scheduler: run_domclick_city_sweep crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered domclick_city_sweep run_id=%d", run_id) + return run_id + + async def _execute_cian_backfill( db: Session, *, @@ -1315,6 +1356,8 @@ async def scheduler_loop() -> None: await trigger_cian_backfill_run(db, sch) elif source == "cian_city_sweep": await trigger_cian_city_sweep_run(db, sch) + elif source == "domclick_city_sweep": + await trigger_domclick_city_sweep_run(db, sch) elif source == "rosreestr_dkp_import": await trigger_rosreestr_dkp_run(db, sch) elif source == "listing_source_snapshot": diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 793a3101..651b0c52 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -2180,6 +2180,150 @@ async def run_yandex_full_load( raise +# ── DomClick citywide sweep (без anchor'ов) ──────────────────────────────── + + +@dataclass +class DomClickCitySweepCounters: + """Aggregate counters для DomClick citywide sweep run. + + DomClick SERP — citywide (city_id), не geo-bbox: anchor-loop отсутствует. + Один проход fetch_city перебирает rooms × pages внутри scraper'а. + Поля симметричны sibling-counters, но без detail_*/houses_* (DomClick SERP-only) + и без anchors_* (нет anchor-фазы). pages_fetched = rooms × pages worst-case. + """ + + lots_fetched: int = 0 + lots_inserted: int = 0 + lots_updated: int = 0 + pages_fetched: 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)} + + +# Дефолтные параметры sweep'а (EKB city_id=4, все комнатности). +DOMCLICK_DEFAULT_CITY_ID: int = 4 +DOMCLICK_DEFAULT_ROOMS: list[int] = [0, 1, 2, 3, 4] +# Оценка времени одного fetch'а (network + camoufox render + parse) для watchdog. +_DOMCLICK_PER_FETCH_S: float = 12.0 +# Буфер сверху расчётного бюджета (cold browser start, save-фаза). +_DOMCLICK_SWEEP_BUDGET_S: float = 300.0 + + +async def run_domclick_city_sweep( + db: Session, + *, + run_id: int, + city_id: int = DOMCLICK_DEFAULT_CITY_ID, + rooms: list[int] | None = None, + pages: int = 5, + request_delay_sec: float | None = None, +) -> DomClickCitySweepCounters: + """DomClick citywide sweep: SERP по city_id × rooms × pages → save → house-match. + + Структурно зеркалит run_cian_city_sweep / run_yandex_city_sweep, но CITYWIDE: + DomClick SERP не поддерживает geo-radius (fetch_around → NotImplementedError), + поэтому anchor-loop отсутствует. DomClickScraper.fetch_city уже перебирает + rooms × pages через внутренний BrowserFetcher (camoufox headless) и + дедуплицирует по source_id. + + Фазы (единственный citywide проход): + 1. SERP: DomClickScraper().fetch_city(city_id, rooms, pages). + 2. SAVE: save_listings(db, lots, run_id=run_id) — BaseScraper save path, + триггерит существующий house-match hook (адрес-fingerprint tier, т.к. + DomClick SERP не отдаёт lat/lon). + + Watchdog: весь fetch_city оборачивается в asyncio.wait_for. Таймаут считается + из rooms × pages × per-fetch + buffer (минимум ANCHOR_TIMEOUT_SEC). + + Cooperative cancel: is_cancelled(db, run_id) проверяется перед SERP-фазой. + mark_done вызывается ВСЕГДА (кроме cancel / fatal). lat=lon=None у всех lots → + house-match использует только address_fingerprint tier. + + Возвращает DomClickCitySweepCounters. + """ + from app.services.scrapers.domclick import DomClickScraper + + _rooms = rooms if rooms is not None else list(DOMCLICK_DEFAULT_ROOMS) + _resolved_delay = request_delay_sec if request_delay_sec is not None else 8.0 + counters = DomClickCitySweepCounters() + + # Watchdog-таймаут: число fetch'ей = len(rooms) × pages (worst-case, без + # early-break на пустой странице). Каждый fetch ≈ delay + render/parse overhead. + _num_fetches = max(1, len(_rooms)) * max(1, pages) + _sweep_timeout = max( + ANCHOR_TIMEOUT_SEC, + int(_num_fetches * (_resolved_delay + _DOMCLICK_PER_FETCH_S) + _DOMCLICK_SWEEP_BUDGET_S), + ) + + try: + # ── Cooperative cancel перед SERP-фазой ────────────────────────────── + if scrape_runs.is_cancelled(db, run_id): + logger.info("domclick-sweep run_id=%d: cancelled before SERP", run_id) + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + return counters + + logger.info( + "domclick-sweep run_id=%d: citywide SERP city_id=%d rooms=%s pages=%d " + "(watchdog %ds)", + run_id, + city_id, + _rooms, + pages, + _sweep_timeout, + ) + + lots: list[ScrapedLot] = [] + + async def _domclick_phase() -> None: + """Единственная citywide-фаза: fetch_city + save.""" + nonlocal lots + async with DomClickScraper() as 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, 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 + + # pages_fetched: worst-case число запрошенных страниц (rooms × pages). + counters.pages_fetched = _num_fetches + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_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", + run_id, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.pages_fetched, + counters.errors_count, + ) + return counters + + except Exception as exc: + logger.exception("domclick-sweep run_id=%d: fatal error", run_id) + scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + # ── Public re-exports ─────────────────────────────────────────── __all__ = [ "ANCHOR_TIMEOUT_SEC", @@ -2188,6 +2332,7 @@ __all__ = [ "CianCitySweepCounters", "CianFullLoadCounters", "CitySweepCounters", + "DomClickCitySweepCounters", "PipelineCounters", "PipelineResult", "YandexCitySweepCounters", @@ -2196,6 +2341,7 @@ __all__ = [ "run_avito_pipeline", "run_cian_city_sweep", "run_cian_full_load", + "run_domclick_city_sweep", "run_yandex_city_sweep", "run_yandex_full_load", ] diff --git a/tradein-mvp/backend/data/sql/122_enable_domclick_city_sweep.sql b/tradein-mvp/backend/data/sql/122_enable_domclick_city_sweep.sql new file mode 100644 index 00000000..b4844a01 --- /dev/null +++ b/tradein-mvp/backend/data/sql/122_enable_domclick_city_sweep.sql @@ -0,0 +1,56 @@ +-- 122_enable_domclick_city_sweep.sql +-- Seed row для DomClick периодического citywide SERP-sweep. +-- +-- !!! DORMANT BY DESIGN !!! enabled = false. +-- DomClickScraper.fetch_city бьёт DomClick SERP с prod-IP через camoufox headless +-- BrowserFetcher (DataDome пропускает headless Playwright; curl_cffi → 401). +-- Capability полностью wired (run_domclick_city_sweep + trigger_domclick_city_sweep_run +-- + dispatch в scheduler.py), но schedule засеян ВЫКЛЮЧЕННЫМ — live-поведение +-- намеренно НЕ verified до ручного prod-smoke. ВКЛЮЧАТЬ ВРУЧНУЮ оператором: +-- UPDATE scrape_schedules SET enabled = true WHERE source = 'domclick_city_sweep'; +-- Зеркалит cian_city_sweep (107) и yandex_city_sweep (078, тоже enabled=false, dormant). +-- +-- DomClick SERP — CITYWIDE (city_id), не geo-radius: anchor-loop отсутствует. +-- fetch_city перебирает rooms × pages внутри scraper'а и дедуплицирует по source_id. +-- SERP координаты не отдаёт (lat=lon=None) → house-match только по address-fingerprint. +-- SERP-only: нет detail/houses-фаз (в отличие от cian sweep), поэтому default_params +-- несут только city_id / rooms / pages_per_anchor / request_delay_sec. +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)). +-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно. +-- +-- domclick_city_sweep — окно 03:00-06:00 UTC (06:00-09:00 МСК), после avito/cian/yandex +-- sweeps (02:00-05:00 UTC) чтобы не конкурировать за egress в одно окно. +-- default_params (консервативно, под DataDome): +-- city_id — числовой ID города DomClick (4 = Екатеринбург). +-- rooms — список комнатностей (0=студия, 1, 2, 3, 4). +-- pages_per_anchor — максимум страниц SERP на один room-проход (break on empty). +-- request_delay_sec — пауза между страницами SERP (anti-DataDome). + +BEGIN; + +-- next_run_at = следующее наступление night-window (tomorrow + window_start_hour). +-- При enabled=false get_due_schedules() всё равно не подхватит строку, но заполняем +-- по образцу 107: когда оператор включит schedule, next_run_at уже валиден и +-- запуск не выстрелит мгновенно (вне окна). AT TIME ZONE 'UTC' фиксирует wall-clock +-- час в UTC независимо от session TimeZone GUC. +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'domclick_city_sweep', + false, -- DORMANT — enable manually after prod-smoke + 3, + 6, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 3)) AT TIME ZONE 'UTC', + '{"city_id": 4, "rooms": [0, 1, 2, 3, 4], "pages_per_anchor": 5, "request_delay_sec": 6}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_domclick_sweep.py b/tradein-mvp/backend/tests/test_domclick_sweep.py new file mode 100644 index 00000000..e6f9200c --- /dev/null +++ b/tradein-mvp/backend/tests/test_domclick_sweep.py @@ -0,0 +1,291 @@ +"""Offline unit tests for run_domclick_city_sweep + DomClick parser. + +Pure & fast: NO live network, NO DB. Mirrors test_cian_city_sweep orchestration +semantics, but DomClick is CITYWIDE (no anchor-loop) и SERP-only (no detail/houses). + +Coverage: + - fetch_city вызывается, save_listings получает накопленные lots с source='domklik' + - DomClickCitySweepCounters корректно заполняется (lots_fetched/inserted/updated) + - Cooperative cancel прерывает sweep перед SERP + - fetch_city exception не валит sweep (graceful, errors_count++, mark_done) + - _parse_html: минимальная DomClick карточка → ScrapedLot с rooms/area/floor/price +""" + +from __future__ import annotations + +import os + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from app.services import scrape_pipeline +from app.services.scrapers.base import ScrapedLot +from app.services.scrapers.domclick import DomClickScraper + + +def _fake_lot(offer_id: str) -> ScrapedLot: + return ScrapedLot( + source="domklik", + source_url=f"https://domclick.ru/card/sale__flat__{offer_id}", + source_id=offer_id, + address="Екатеринбург, улица Тестовая, 1", + price_rub=5_000_000, + area_m2=50.0, + rooms=2, + lat=None, + lon=None, + ) + + +class _FakeRuns: + """Stub для scrape_runs: отслеживает вызовы, поддерживает cooperative cancel.""" + + def __init__(self, *, cancel: bool = False) -> None: + self.cancel = cancel + self.heartbeats: list[dict[str, int]] = [] + self.done: dict[str, int] | None = None + self.failed: tuple[str, dict[str, int]] | None = None + + def is_cancelled(self, _db: Any, _run_id: int) -> bool: + return self.cancel + + def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None: + self.heartbeats.append(dict(counters)) + + def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None: + self.done = dict(counters) + + def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None: + self.failed = (error, dict(counters)) + + +@pytest.fixture(autouse=True) +def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None: + async def _instant(_secs: float) -> None: + return None + + monkeypatch.setattr(scrape_pipeline.asyncio, "sleep", _instant) + + +@pytest.fixture(autouse=True) +def _stub_domclick_lifecycle(monkeypatch: pytest.MonkeyPatch) -> None: + """Нейтрализуем DomClickScraper.__init__/__aenter__/__aexit__ — нет DB/browser.""" + + def _init(self: DomClickScraper) -> None: + self.request_delay_sec = 0.0 + self.parse_failures = 0 + + async def _aenter(self: DomClickScraper) -> DomClickScraper: + return self + + async def _aexit(self: DomClickScraper, *_args: Any) -> None: + return None + + monkeypatch.setattr(DomClickScraper, "__init__", _init) + monkeypatch.setattr(DomClickScraper, "__aenter__", _aenter) + monkeypatch.setattr(DomClickScraper, "__aexit__", _aexit) + + +def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None: + monkeypatch.setattr(scrape_pipeline.scrape_runs, "is_cancelled", fake.is_cancelled) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "update_heartbeat", fake.update_heartbeat) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_done", fake.mark_done) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_failed", fake.mark_failed) + + +# ── DomClickCitySweepCounters ─────────────────────────────────────────────── + + +def test_domclick_sweep_counters_defaults() -> None: + from app.services.scrape_pipeline import DomClickCitySweepCounters + + c = DomClickCitySweepCounters() + assert c.lots_fetched == 0 + assert c.lots_inserted == 0 + assert c.lots_updated == 0 + assert c.pages_fetched == 0 + assert c.errors_count == 0 + + +def test_domclick_sweep_counters_to_dict_all_keys() -> None: + from dataclasses import fields + + from app.services.scrape_pipeline import DomClickCitySweepCounters + + c = DomClickCitySweepCounters() + d = c.to_dict() + expected = {f.name for f in fields(c)} + assert set(d.keys()) == expected + + +# ── Full sweep — фазы и агрегация ────────────────────────────────────────── + + +async def test_sweep_fetches_and_saves_with_source(monkeypatch: pytest.MonkeyPatch) -> None: + """fetch_city → save_listings получает 2 lots с source='domklik', counters заполнены.""" + fetch_calls: list[tuple[int, int]] = [] + save_calls: list[list[ScrapedLot]] = [] + + async def fake_fetch_city( + self: DomClickScraper, + city_id: int, + rooms: list[int] | None = None, + pages: int = 20, + ) -> list[ScrapedLot]: + fetch_calls.append((city_id, pages)) + return [_fake_lot("100"), _fake_lot("101")] + + def fake_save(_db: Any, lots: list, *, run_id: int | None = None) -> tuple[int, int]: + save_calls.append(list(lots)) + return len(lots), 0 + + monkeypatch.setattr(DomClickScraper, "fetch_city", fake_fetch_city) + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_domclick_city_sweep( + db=MagicMock(), + run_id=1, + city_id=4, + rooms=[1, 2], + pages=3, + request_delay_sec=0.0, + ) + + assert len(fetch_calls) == 1 + assert fetch_calls[0] == (4, 3) + assert len(save_calls) == 1 + assert len(save_calls[0]) == 2 + assert all(lot.source == "domklik" for lot in save_calls[0]) + assert counters.lots_fetched == 2 + assert counters.lots_inserted == 2 + assert counters.lots_updated == 0 + assert counters.errors_count == 0 + assert fake.done is not None + assert fake.done["lots_fetched"] == 2 + assert fake.failed is None + + +async def test_sweep_empty_serp_still_mark_done(monkeypatch: pytest.MonkeyPatch) -> None: + """Пустой SERP → save не вызывается, но mark_done всё равно вызывается.""" + save_called = False + + async def fake_fetch_empty( + self: DomClickScraper, + city_id: int, + rooms: list[int] | None = None, + pages: int = 20, + ) -> list[ScrapedLot]: + return [] + + def fake_save(_db: Any, lots: list, *, run_id: int | None = None) -> tuple[int, int]: + nonlocal save_called + save_called = True + return 0, 0 + + monkeypatch.setattr(DomClickScraper, "fetch_city", fake_fetch_empty) + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_domclick_city_sweep( + db=MagicMock(), run_id=2, request_delay_sec=0.0 + ) + + assert save_called is False + assert counters.lots_fetched == 0 + assert fake.done is not None + + +async def test_cancel_before_serp_skips_fetch(monkeypatch: pytest.MonkeyPatch) -> None: + """Cooperative cancel перед SERP → fetch_city НЕ вызывается, mark_done НЕ вызывается.""" + fetch_called = False + + async def fake_fetch( + self: DomClickScraper, + city_id: int, + rooms: list[int] | None = None, + pages: int = 20, + ) -> list[ScrapedLot]: + nonlocal fetch_called + fetch_called = True + return [_fake_lot("1")] + + monkeypatch.setattr(DomClickScraper, "fetch_city", fake_fetch) + monkeypatch.setattr(scrape_pipeline, "save_listings", lambda *a, **k: (0, 0)) + + fake = _FakeRuns(cancel=True) + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_domclick_city_sweep( + db=MagicMock(), run_id=3, request_delay_sec=0.0 + ) + + assert fetch_called is False + assert counters.lots_fetched == 0 + assert fake.done is None + + +async def test_fetch_city_exception_graceful(monkeypatch: pytest.MonkeyPatch) -> None: + """Исключение в fetch_city → errors_count++, sweep завершается mark_done (не fatal).""" + + async def fake_fetch_fail( + self: DomClickScraper, + city_id: int, + rooms: list[int] | None = None, + pages: int = 20, + ) -> list[ScrapedLot]: + raise RuntimeError("camoufox crashed") + + monkeypatch.setattr(DomClickScraper, "fetch_city", fake_fetch_fail) + monkeypatch.setattr(scrape_pipeline, "save_listings", lambda *a, **k: (0, 0)) + + fake = _FakeRuns() + _install_runs(monkeypatch, fake) + + counters = await scrape_pipeline.run_domclick_city_sweep( + db=MagicMock(), run_id=4, request_delay_sec=0.0 + ) + + assert counters.errors_count == 1 + assert counters.lots_fetched == 0 + assert fake.done is not None, "graceful: mark_done вызывается несмотря на ошибку SERP" + assert fake.failed is None + + +# ── _parse_html parser unit test ──────────────────────────────────────────── + + +def test_parse_html_minimal_card() -> None: + """Минимальная DomClick карточка → ScrapedLot с rooms/area/floor/price/source_id.""" + # __init__ застаблен фикстурой _stub_domclick_lifecycle → нет DB/browser + scraper = DomClickScraper() + html = """ + + +
2-комн. квартира
+ 52,3 м² + 5 / 16 эт. + Екатеринбург, улица Ленина, 10 + 6 500 000 ₽ + + + """ + lots = scraper._parse_html(html) + assert len(lots) == 1 + lot = lots[0] + assert lot.source == "domklik" + assert lot.source_id == "2075671636" + assert lot.rooms == 2 + assert lot.area_m2 == pytest.approx(52.3) + assert lot.floor == 5 + assert lot.total_floors == 16 + assert lot.price_rub == 6_500_000 + assert lot.lat is None and lot.lon is None