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

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

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,

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;

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

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