From fd9b27f87cd4ee08bf0ed403107c79d11dcadec8 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Sat, 4 Jul 2026 02:22:41 +0300 Subject: [PATCH] =?UTF-8?q?fix(tradein/scrapers):=20=D0=BF=D0=BE=D0=B4?= =?UTF-8?q?=D0=BA=D0=BB=D1=8E=D1=87=D0=B8=D1=82=D1=8C=20geoportal=20coords?= =?UTF-8?q?=20backfill=20=D0=B2=20scheduler=20(#1967)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Живой прод-замер показал, что исходная постановка issue (avito 47% NULL lat) уже почти закрыта побочно: dedup-миграции 159-167 сократили active avito-set с 30k до 9.2k, nightly geocode_missing_listings крутится с PR #1549 — текущий NULL lat = 4.3% (393/9181), не 47%. Реальная находка: точный street+house геокодер против локального ekb_geoportal_buildings (#1841, backfill_listings_coords_geoportal.py) существовал, но был manual-only script, ни разу не запускавшийся recurring. Подключил его в in-app scheduler по паттерну cadastral_geo_match — run-lifecycle wrapper, dispatch, kit-registry entry (обязательно для scraper_kit strangler-migration completeness), seed-миграция на окно 05:00-06:00 UTC (до geocode_missing_listings 06:00-09:00 — точный house-level матч забирает адрес раньше грубого Nominatim-геокода). Dedup-на-ингесте (вторая половина исходного DoD) — отдельный follow-up, out of scope здесь. --- .../backend/app/services/product_handlers.py | 16 +++ tradein-mvp/backend/app/services/scheduler.py | 61 ++++++++- .../backfill_listings_coords_geoportal.py | 82 ++++++++++++ ...hedules_seed_geoportal_coords_backfill.sql | 71 +++++++++++ ...test_backfill_listings_coords_geoportal.py | 119 ++++++++++++++++++ .../test_scraper_kit_scheduler_parity.py | 1 + 6 files changed, 349 insertions(+), 1 deletion(-) create mode 100644 tradein-mvp/backend/data/sql/171_scrape_schedules_seed_geoportal_coords_backfill.sql diff --git a/tradein-mvp/backend/app/services/product_handlers.py b/tradein-mvp/backend/app/services/product_handlers.py index 46016f96..2aa5dcdd 100644 --- a/tradein-mvp/backend/app/services/product_handlers.py +++ b/tradein-mvp/backend/app/services/product_handlers.py @@ -251,6 +251,19 @@ async def _job_yandex_detail_backfill( await run_yandex_detail_backfill(db, run_id=run_id, params=params) +# ── geoportal_coords_backfill — sync local exact match в executor (#1967) ───── +async def _job_geoportal_coords_backfill( + db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext +) -> None: + from app.tasks.backfill_listings_coords_geoportal import run_geoportal_coords_backfill + + loop = asyncio.get_event_loop() + await loop.run_in_executor( + None, + lambda: run_geoportal_coords_backfill(db, run_id=run_id, params=params), + ) + + # ── cadastral_geo_match — sync FDW-refresh+match в executor ─────────────────── async def _job_cadastral_geo_match( db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext @@ -376,6 +389,9 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]: "yandex_newbuilding_sweep": Handler( _job_yandex_newbuilding_sweep, "yandex_newbuilding_sweep" ), + "geoportal_coords_backfill": Handler( + _job_geoportal_coords_backfill, "geoportal_coords_backfill" + ), "geocode_missing_listings": Handler( _job_geocode_missing_listings, "geocode_missing_listings" ), diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 3d4a96b6..8c7eb6e4 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -78,11 +78,23 @@ Sources: - domclick_city_sweep → run_domclick_city_sweep (scrape_pipeline.py; citywide SERP via camoufox BrowserFetcher; shipped DORMANT enabled=false, enable manually after prod-smoke) + - geoportal_coords_backfill → run_geoportal_coords_backfill + (tasks/backfill_listings_coords_geoportal.py, #1967; nightly + exact street+house match против ekb_geoportal_buildings (местный + реестр зданий ЕКБ) для listings.lat IS NULL. Pure internal DB op — + никакого внешнего HTTP/rate-limit (в отличие от Nominatim ниже). + Ранее был manual-only script (#1841); recurring нужен потому что + иначе новые NULL-coord листинги ждут geocode_missing_listings и + получают более грубый (coarse) city-centroid geocode вместо + точного house-level матча. Window 05:00-06:00 UTC — ДО + geocode_missing_listings (06:00-09:00 UTC), чтобы точный матч + имел приоритет над Nominatim.) - geocode_missing_listings → run_geocode_missing_listings (tasks/geocode_missing.py; nightly backfill listings.lat/lon для всех источников с NULL coords; address-dedup batch loop с wall-clock budget; window 06:00-09:00 UTC — после city sweeps - ~03-04 UTC и deals OS-cron 05:30/05:45 UTC) + ~03-04 UTC, deals OS-cron 05:30/05:45 UTC и geoportal_coords_backfill + 05:00-06:00 UTC) - yandex_detail_backfill → run_yandex_detail_backfill (tasks/yandex_detail_backfill.py, #1553; nightly backfill of yandex detail_enriched_at for ~3952 listings never enriched by @@ -870,6 +882,51 @@ async def trigger_yandex_address_backfill_run( return run_id +async def trigger_geoportal_coords_backfill_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch run_geoportal_coords_backfill в executor (sync DB-only task). + + Exact street+house match против ekb_geoportal_buildings (#1967 — was manual-only #1841). + Pure local DB op — никакого внешнего HTTP/rate-limit, поэтому запускается ПЕРЕД + geocode_missing_listings (Nominatim): точный house-level матч должен получить шанс + первым, иначе более грубый (coarse) geocode Nominatim'а забирает адрес раньше и + listing выпадает из WHERE lat IS NULL до следующего цикла. + + Sync task (SELECT + per-row local lookup + UPDATE, no async HTTP) — run in run_in_executor + by the same pattern as trigger_cadastral_geo_match_run. run_geoportal_coords_backfill owns + the scrape_runs lifecycle (mark_done/mark_failed). + + Returns run_id или None (skip — already running). + """ + 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: + from app.tasks.backfill_listings_coords_geoportal import ( + run_geoportal_coords_backfill, + ) + + loop = asyncio.get_event_loop() + await loop.run_in_executor( + None, + lambda: run_geoportal_coords_backfill(run_db, run_id=run_id, params=params), + ) + except Exception: + logger.exception("scheduler: run_geoportal_coords_backfill crashed run_id=%d", run_id) + finally: + run_db.close() + + _spawn_tracked(_run()) + logger.info("scheduler: triggered geoportal_coords_backfill run_id=%d", run_id) + return run_id + + async def trigger_geocode_missing_listings_run( db: Session, schedule_row: dict[str, Any] ) -> int | None: @@ -1995,6 +2052,8 @@ async def scheduler_loop() -> None: await trigger_newbuilding_enrich_run(db, sch) elif source == "yandex_newbuilding_sweep": await trigger_yandex_newbuilding_sweep_run(db, sch) + elif source == "geoportal_coords_backfill": + await trigger_geoportal_coords_backfill_run(db, sch) elif source == "geocode_missing_listings": await trigger_geocode_missing_listings_run(db, sch) elif source == "avito_detail_backfill": diff --git a/tradein-mvp/backend/app/tasks/backfill_listings_coords_geoportal.py b/tradein-mvp/backend/app/tasks/backfill_listings_coords_geoportal.py index a31a3f2e..b67a234f 100644 --- a/tradein-mvp/backend/app/tasks/backfill_listings_coords_geoportal.py +++ b/tradein-mvp/backend/app/tasks/backfill_listings_coords_geoportal.py @@ -14,6 +14,13 @@ python -m app.tasks.backfill_listings_coords_geoportal python -m app.tasks.backfill_listings_coords_geoportal --limit 5000 --batch-size 200 +Также запускается nightly через in-app scheduler (source='geoportal_coords_backfill', +migration 171) — run_geoportal_coords_backfill(). Local exact match, никакого внешнего +HTTP/rate-limit, поэтому окно ставится ПЕРЕД geocode_missing_listings (Nominatim/Yandex, +coarse city-centroid fallback): точный house-level матч должен получить шанс первым, +иначе Nominatim успевает проставить грубые coords и адрес выпадает из WHERE lat IS NULL +(#1967 — было единичным manual-прогоном #1841, здесь становится recurring). + Идемпотентность: UPDATE применяется только к строкам с lat IS NULL (WHERE id=:id AND lat IS NULL). Повторный прогон не затирает уже проставленные координаты. """ @@ -29,6 +36,7 @@ from sqlalchemy import text from sqlalchemy.orm import Session from app.core.db import SessionLocal +from app.services import scrape_runs as runs_mod from app.services.geocoder import _geoportal_house_match, _parse_street_house logger = logging.getLogger(__name__) @@ -50,6 +58,17 @@ class BackfillCoordsResult: errors: int = 0 # исключения при обработке отдельной записи duration_sec: float = field(default=0.0) + def to_counters(self) -> dict[str, int]: + return { + "candidates": self.candidates, + "matched": self.matched, + "updated": self.updated, + "no_address": self.no_address, + "no_match": self.no_match, + "errors": self.errors, + "duration_sec": int(self.duration_sec), + } + def _update_listing_coords( db: Session, @@ -253,6 +272,69 @@ def backfill_coords_from_geoportal( return res +# ── Run lifecycle wrapper (scheduler entrypoint) ───────────────────────────── +def run_geoportal_coords_backfill( + db: Session, + *, + run_id: int, + params: dict, +) -> BackfillCoordsResult: + """Run-lifecycle wrapper for backfill_coords_from_geoportal (sync, DB-only task). + + Launched by the in-app scheduler (source='geoportal_coords_backfill') via + trigger_geoportal_coords_backfill_run, или вручную. Pure local exact street+house match + против ekb_geoportal_buildings — никакого внешнего HTTP, поэтому весь backlog (или + ограниченный `limit`) отрабатывается за один прогон без wall-clock budget (в отличие + от geocode_missing_listings, который rate-limited Nominatim'ом). + + Params (default_params jsonb): + limit: максимум листингов за прогон (default None = весь backlog). + batch_size: размер батча SELECT+commit (default 500). + sources: фильтр по source (default None = все источники). + + Finalises scrape_runs (mark_done / mark_failed) с counters. + """ + limit = params.get("limit") + batch_size = int(params.get("batch_size", DEFAULT_BATCH_SIZE)) + sources = params.get("sources") + + counters: dict[str, int] = BackfillCoordsResult().to_counters() + + try: + runs_mod.update_heartbeat(db, run_id, counters) + + res = backfill_coords_from_geoportal( + db, + limit=limit, + sources=sources, + batch_size=batch_size, + ) + + counters = res.to_counters() + runs_mod.mark_done(db, run_id, counters) + logger.info( + "run_geoportal_coords_backfill: run_id=%d DONE candidates=%d matched=%d " + "updated=%d no_address=%d no_match=%d errors=%d duration=%.1fs", + run_id, + res.candidates, + res.matched, + res.updated, + res.no_address, + res.no_match, + res.errors, + res.duration_sec, + ) + return res + except Exception as exc: + logger.exception("run_geoportal_coords_backfill: run_id=%d FAILED", run_id) + try: + db.rollback() + except Exception: + pass + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise + + def main() -> None: logging.basicConfig( level=logging.INFO, diff --git a/tradein-mvp/backend/data/sql/171_scrape_schedules_seed_geoportal_coords_backfill.sql b/tradein-mvp/backend/data/sql/171_scrape_schedules_seed_geoportal_coords_backfill.sql new file mode 100644 index 00000000..c4492400 --- /dev/null +++ b/tradein-mvp/backend/data/sql/171_scrape_schedules_seed_geoportal_coords_backfill.sql @@ -0,0 +1,71 @@ +-- 171_scrape_schedules_seed_geoportal_coords_backfill.sql +-- Seed row делающий recurring уже существовавший manual-only geoportal coords backfill. +-- +-- Context (#1967): avito ~47% активных листингов имели NULL lat/lon → невидимы для +-- ST_DWithin радиус-поиска аналогов в estimator'е. Root cause — avito SERP-карточки +-- намеренно сохраняются с lat=lon=NULL (координаты приходят точнее из detail-страницы +-- или из geocode-cron по address). geocode_missing_listings (migration 110) с 2026-06-16 +-- уже закрывает большую часть разрыва через Nominatim/Yandex (rate-limited, coarse +-- city-centroid когда geocoder не доходит до дома). +-- +-- app/tasks/backfill_listings_coords_geoportal.py существует с #1841 (точный street+house +-- матч против местного реестра зданий ЕКБ, ekb_geoportal_buildings) и даёт house-level +-- precision БЕЗ единого внешнего HTTP-запроса (в отличие от Nominatim) — но был ТОЛЬКО +-- manual script (`python -m app.tasks.backfill_listings_coords_geoportal`), ни разу не +-- запускавшийся на recurring основе. Один прошлый ручной прогон (#1841): 17241 +-- кандидатов → 1008 проставлено (не-ЕКБ адреса не матчатся — корректно, EKB-only реестр). +-- +-- Решение: wire в in-app scheduler (source='geoportal_coords_backfill') по паттерну +-- cadastral_geo_match (migration 125) — pure internal DB op, SAFE to enable=true. +-- +-- Window 05:00-06:00 UTC — ДО geocode_missing_listings (06:00-09:00 UTC, migration 110): +-- точный house-level матч должен получить шанс ПЕРВЫМ. Если Nominatim успевает +-- проставить (более грубые) coords раньше — адрес выпадает из WHERE lat IS NULL и +-- более точный geoportal-матч уже не применяется до следующего появления NULL-строки. +-- +-- default_params: +-- limit — None (без ограничений, весь backlog за один прогон; pure local DB — +-- быстро, никакого wall-clock budget как у geocode_missing_listings не нужно). +-- batch_size — 500 строк на SELECT + commit-интервал (см. DEFAULT_BATCH_SIZE в модуле). +-- sources — None (все источники; ekb_geoportal_buildings сам ограничивает scope до ЕКБ). +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)). +-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'geoportal_coords_backfill', + true, -- ACTIVE: pure internal DB op, no external + -- HTTP/anti-bot/rate-limit exposure. + 5, + 6, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 5)) AT TIME ZONE 'UTC', + '{"batch_size": 500}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMENT ON TABLE scrape_schedules IS + 'In-app scheduler config (заменяет cron-script setup). ' + 'Sources: avito_city_sweep, yandex_city_sweep (dormant, #561), ' + 'cian_history_backfill, rosreestr_dkp_import, listing_source_snapshot (#570), ' + 'asking_to_sold_ratio_refresh (#648), refresh_search_matview (#769), ' + 'yandex_address_backfill (#855, EKB pilot), ' + 'sber_index_pull (#887, monthly СберИндекс city-level price index), ' + 'rosreestr_quarter_poll (#888, monthly Rosreestr new-quarter availability check), ' + 'cian_city_sweep (dormant, #973), ' + 'yandex_newbuilding_sweep (dormant, #974), ' + 'geocode_missing_listings (#1: listings geom backfill, all sources), ' + 'geoportal_coords_backfill (#1967: exact EKB house-level coords match, recurring ' + 'wiring of the former manual-only #1841 script — runs BEFORE geocode_missing_listings).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py b/tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py index 51afc3a2..6397e2c5 100644 --- a/tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py +++ b/tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py @@ -9,14 +9,19 @@ - limit/sources параметры прокидываются в SQL. - Идемпотентность: UPDATE с WHERE lat IS NULL не затирает уже заполненное. - SAVEPOINT откат батча при исключении в DB.execute. +- run_geoportal_coords_backfill (#1967): run-lifecycle wrapper mark_done/mark_failed. +- Scheduler wiring (#1967): trigger fn + dispatch branch + migration 171 seed content. Все сетевые вызовы и DB замоканы. """ from __future__ import annotations +import inspect import os import sys +from pathlib import Path +from typing import Any from unittest.mock import MagicMock, patch import pytest @@ -26,12 +31,17 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost: _wp_mock = MagicMock() sys.modules.setdefault("weasyprint", _wp_mock) +from app.services import scheduler # noqa: E402 from app.services.geocoder import GeocodeSuggestion # noqa: E402 +from app.tasks import backfill_listings_coords_geoportal as bcg # noqa: E402 from app.tasks.backfill_listings_coords_geoportal import ( # noqa: E402 BackfillCoordsResult, backfill_coords_from_geoportal, ) +_SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql" +_MIGRATION_171 = _SQL_DIR / "171_scrape_schedules_seed_geoportal_coords_backfill.sql" + # ── helpers ─────────────────────────────────────────────────────────────────── _HIT = GeocodeSuggestion( @@ -362,3 +372,112 @@ def test_batch_pagination_two_pages() -> None: assert res.candidates == 3 assert res.no_match == 3 + + +# ── run_geoportal_coords_backfill (run-lifecycle wrapper, #1967) ────────────── + + +def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) -> None: + """run_geoportal_coords_backfill: delegates to backfill_coords_from_geoportal, mark_done.""" + marked: dict[str, Any] = {} + monkeypatch.setattr(bcg.runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr( + bcg.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(bcg.runs_mod, "mark_failed", lambda *a, **k: None) + + fake_result = BackfillCoordsResult( + candidates=10, matched=3, updated=3, no_address=2, no_match=5, errors=0 + ) + monkeypatch.setattr( + bcg, + "backfill_coords_from_geoportal", + lambda _db, **k: fake_result, + ) + + out = bcg.run_geoportal_coords_backfill( + MagicMock(), + run_id=42, + params={"batch_size": 500}, + ) + + assert out is fake_result + assert marked["run_id"] == 42 + assert marked["counters"]["candidates"] == 10 + assert marked["counters"]["matched"] == 3 + assert marked["counters"]["updated"] == 3 + + +def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> None: + """run_geoportal_coords_backfill: exception → mark_failed + re-raise (no silent failure).""" + failed: dict[str, Any] = {} + monkeypatch.setattr(bcg.runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr(bcg.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr( + bcg.runs_mod, + "mark_failed", + lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err), + ) + + def _boom(_db: Any, **_k: Any) -> BackfillCoordsResult: + raise RuntimeError("db down") + + monkeypatch.setattr(bcg, "backfill_coords_from_geoportal", _boom) + + class _DB: + def rollback(self) -> None: + pass + + with pytest.raises(RuntimeError): + bcg.run_geoportal_coords_backfill(_DB(), run_id=9, params={}) # type: ignore[arg-type] + + assert failed["run_id"] == 9 + assert "db down" in failed["err"] + + +# ── Scheduler wiring (#1967) ─────────────────────────────────────────────────── + + +def test_scheduler_has_trigger_and_dispatch() -> None: + assert hasattr(scheduler, "trigger_geoportal_coords_backfill_run") + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "geoportal_coords_backfill"' in loop_src + assert "trigger_geoportal_coords_backfill_run(db, sch)" in loop_src + + +def test_trigger_claims_run_and_runs_in_executor() -> None: + src = inspect.getsource(scheduler.trigger_geoportal_coords_backfill_run) + assert "_claim_run(db, schedule_row)" in src + # sync DB-only task → run_in_executor (not a bare async call). + assert "run_in_executor" in src + assert "run_geoportal_coords_backfill" in src + + +def test_dispatch_runs_before_geocode_missing_listings_in_source_docstring() -> None: + """Header docs must place geoportal_coords_backfill BEFORE geocode_missing_listings. + + Ordering is the whole point (#1967): exact house-level match should claim NULL-coord + listings before the coarser Nominatim/Yandex backfill does. + """ + header = scheduler.__doc__ or "" + geoportal_idx = header.find("geoportal_coords_backfill") + nominatim_idx = header.find("geocode_missing_listings") + assert geoportal_idx != -1 + assert nominatim_idx != -1 + assert geoportal_idx < nominatim_idx + + +# ── Migration content (#1967) ────────────────────────────────────────────────── + + +def test_migration_171_seeds_schedule_before_geocode_missing_listings_window() -> None: + sql = _MIGRATION_171.read_text(encoding="utf-8") + assert "'geoportal_coords_backfill'" in sql + assert "ON CONFLICT (source) DO NOTHING" in sql + assert "BEGIN;" in sql and "COMMIT;" in sql + # Window 5-6 UTC — before geocode_missing_listings' 06:00-09:00 window (migration 110). + assert "\n 5,\n 6,\n" in sql + # enabled=true — safe by default (pure internal DB op, no external HTTP). + assert "'geoportal_coords_backfill',\n true," in sql diff --git a/tradein-mvp/backend/tests/test_scraper_kit_scheduler_parity.py b/tradein-mvp/backend/tests/test_scraper_kit_scheduler_parity.py index 10ef69f7..29b7e84c 100644 --- a/tradein-mvp/backend/tests/test_scraper_kit_scheduler_parity.py +++ b/tradein-mvp/backend/tests/test_scraper_kit_scheduler_parity.py @@ -71,6 +71,7 @@ SOURCE_TO_OLD_TRIGGER: dict[str, str] = { "rosreestr_quarter_poll": "trigger_rosreestr_quarter_poll_run", "newbuilding_enrich": "trigger_newbuilding_enrich_run", "yandex_newbuilding_sweep": "trigger_yandex_newbuilding_sweep_run", + "geoportal_coords_backfill": "trigger_geoportal_coords_backfill_run", "geocode_missing_listings": "trigger_geocode_missing_listings_run", "avito_detail_backfill": "trigger_avito_detail_backfill_run", "yandex_detail_backfill": "trigger_yandex_detail_backfill_run", -- 2.45.3