feat(domclick): wire DomClick citywide sweep into pipeline+scheduler (seeded disabled)
All checks were successful
CI / changes (pull_request) Successful in 6s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped

This commit is contained in:
bot-backend 2026-06-17 23:21:49 +03:00
parent 97fce43484
commit 2ed3ae81bc
4 changed files with 536 additions and 0 deletions

View file

@ -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":

View file

@ -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",
]

View file

@ -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;

View file

@ -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 = """
<html><body>
<a href="/card/sale__flat__2075671636">
<div>2-комн. квартира</div>
<span>52,3 м²</span>
<span>5 / 16 эт.</span>
<span>Екатеринбург, улица Ленина, 10</span>
<span>6 500 000 </span>
</a>
</body></html>
"""
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