fix(tradein/scrapers): подключить geoportal coords backfill в scheduler (#1967) #2346

Merged
lekss361 merged 1 commit from fix/tradein-avito-geo-backfill into main 2026-07-03 23:25:48 +00:00
6 changed files with 349 additions and 1 deletions

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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