diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 038ad9ce..84db6fb3 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -51,6 +51,11 @@ 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) + - 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) """ from __future__ import annotations @@ -553,6 +558,45 @@ async def trigger_yandex_address_backfill_run( return run_id +async def trigger_geocode_missing_listings_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch run_geocode_missing_listings в asyncio.create_task. + + Nightly backfill listings.lat/lon для всех источников с NULL coords (#1: geom-coverage gap). + Адрес-dedup batch loop с wall-clock budget (default 30 мин). + + Root cause: scrapers сохраняют listings с lat=lon=NULL — «геокод подтянет cron», + но cron геокодирует только deals, не listings. Эта функция замыкает петлю: нигде + снаружи нет OS cron для listings — только этот планировщик. + + Mirrors trigger_yandex_address_backfill_run: claim run → create_task → mark_done/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.geocode_missing import run_geocode_missing_listings + + await run_geocode_missing_listings(run_db, run_id=run_id, params=params) + except Exception: + logger.exception("scheduler: run_geocode_missing_listings 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 geocode_missing_listings run_id=%d", run_id) + return run_id + + async def trigger_rosreestr_dkp_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch import_rosreestr_dkp в executor (sync I/O-bound task).""" run_id = _claim_run(db, schedule_row) @@ -1191,6 +1235,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 == "geocode_missing_listings": + await trigger_geocode_missing_listings_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/tasks/geocode_missing.py b/tradein-mvp/backend/app/tasks/geocode_missing.py index d14af37d..2dcfd9f2 100644 --- a/tradein-mvp/backend/app/tasks/geocode_missing.py +++ b/tradein-mvp/backend/app/tasks/geocode_missing.py @@ -2,7 +2,8 @@ Запускается: - Manual через POST /admin/scrape/geocode-missing-listings - - Future: cron job (Celery beat если bootstrap'нут / OS cron) + - Scheduled: nightly via scrape_schedules (source='geocode_missing_listings', migration 110) + — wired into in-app scheduler, window 06:00-09:00 UTC. Pattern: dedup по address (1 unique address → 1 geocode call → UPDATE all listings). Rate limit: Nominatim 1 req/sec. Yandex 25K/day если YANDEX_GEOCODER_API_KEY set. @@ -11,6 +12,8 @@ Rate limit: Nominatim 1 req/sec. Yandex 25K/day если YANDEX_GEOCODER_API_KEY - Этот модуль группирует по address → меньше API calls (dedup). - Поддерживает all sources включая Avito (после PR #487 убрали jitter). - Возвращает GeocodeBackfillResult с детальными counters. + - Loop-safe: SELECT фильтрует geocode_tried_at IS NULL OR tried_at < 7 days; + при geocode failure помечает tried_at=NOW() → адрес не переотбирается в этом же run. """ from __future__ import annotations @@ -22,6 +25,7 @@ from dataclasses import dataclass, field from sqlalchemy import text from sqlalchemy.orm import Session +from app.services import scrape_runs as runs_mod from app.services.estimator import _geocode_is_coarse from app.services.geocoder import geocode @@ -70,7 +74,12 @@ async def geocode_missing_listings( start = time.monotonic() result = GeocodeBackfillResult() - # 1. Найти top-N адресов с NULL coords (DESC by occurrence count) + # 1. Найти top-N адресов с NULL coords (DESC by occurrence count). + # Фильтруем адреса, по которым геокодер уже пробовал и не нашёл — они помечены + # geocode_tried_at. Повторяем попытку только если tried_at старше 7 дней (возможен + # переезд адреса в кэше или смена провайдера), либо tried_at IS NULL (ещё не пробовали). + # Это делает функцию loop-safe: при вызове несколько раз в одном прогоне + # failed-адреса не переотбираются бесконечно. rows = ( db.execute( text( @@ -80,6 +89,8 @@ async def geocode_missing_listings( WHERE lat IS NULL AND address IS NOT NULL AND length(trim(address)) >= 5 + AND (geocode_tried_at IS NULL + OR geocode_tried_at < NOW() - INTERVAL '7 days') GROUP BY address ORDER BY listings_count DESC, address ASC LIMIT :limit @@ -114,6 +125,17 @@ async def geocode_missing_listings( except Exception as exc: logger.warning("geocode_missing: geocode raised for '%s': %s", address[:60], exc) result.addresses_failed += 1 + if not dry_run: + # Пометить tried_at чтобы адрес не переотбирался в следующих batch'ах + # этого же прогона (loop-safe backoff 7 дней). + db.execute( + text( + "UPDATE listings SET geocode_tried_at = NOW()" + " WHERE address = :addr AND lat IS NULL" + ), + {"addr": address}, + ) + db.commit() continue if geo is None: @@ -123,6 +145,16 @@ async def geocode_missing_listings( address[:60], listings_count, ) + if not dry_run: + # Пометить tried_at — geocoder не нашёл адрес, backoff 7 дней. + db.execute( + text( + "UPDATE listings SET geocode_tried_at = NOW()" + " WHERE address = :addr AND lat IS NULL" + ), + {"addr": address}, + ) + db.commit() continue if geo.provider == "cache": @@ -150,12 +182,13 @@ async def geocode_missing_listings( precision: str | None = "city" if _geocode_is_coarse(geo) else None # UPDATE listings — PostGIS trigger (listings_set_geom_trg) обновит geom автоматически. - # geo_precision проставляется одновременно с координатами. + # geo_precision и geocode_tried_at проставляются одновременно с координатами. update_result = db.execute( text( """ UPDATE listings - SET lat = :lat, lon = :lon, geo_precision = :precision + SET lat = :lat, lon = :lon, geo_precision = :precision, + geocode_tried_at = NOW() WHERE address = :addr AND lat IS NULL """ ), @@ -195,3 +228,109 @@ async def geocode_missing_listings( ) return result + + +async def run_geocode_missing_listings( + db: Session, + *, + run_id: int, + params: dict, +) -> GeocodeBackfillResult: + """Run-lifecycle wrapper: batch loop с wall-clock budget. + + Запускается планировщиком (source='geocode_missing_listings') или вручную. + Гоняет geocode_missing_listings() в цикле до тех пор пока: + - res.addresses_total == 0 (нет pending адресов) + - res.addresses_total < batch_size (дренаж — последний batch меньше полного) + - истёк budget_sec + + Params (из default_params jsonb в scrape_schedules): + batch_size: int — адресов за один вызов geocode_missing_listings (default 200). + budget_sec: float — максимальное время прогона в секундах (default 1800 = 30 мин). + + Lifecycle: + update_heartbeat перед loop → аккумуляция counters → mark_done / mark_failed. + """ + batch_size = int(params.get("batch_size", 200)) + budget_sec = float(params.get("budget_sec", 1800)) + + total = GeocodeBackfillResult() + counters: dict[str, int] = { + "checked": 0, + "saved": 0, + "skipped": 0, + } + + try: + runs_mod.update_heartbeat(db, run_id, counters) + + start = time.monotonic() + + while True: + res = await geocode_missing_listings(db, batch_size=batch_size, dry_run=False) + + # Аккумулируем counters + total.addresses_total += res.addresses_total + total.addresses_processed += res.addresses_processed + total.addresses_geocoded += res.addresses_geocoded + total.addresses_failed += res.addresses_failed + total.listings_updated += res.listings_updated + total.cache_hits += res.cache_hits + total.cache_misses += res.cache_misses + + counters = { + "checked": total.addresses_processed, + "saved": total.listings_updated, + "skipped": total.addresses_failed, + } + runs_mod.update_heartbeat(db, run_id, counters) + + elapsed = time.monotonic() - start + if res.addresses_total == 0: + logger.info( + "run_geocode_missing_listings: run_id=%d — нет pending адресов, завершаем", + run_id, + ) + break + if res.addresses_total < batch_size: + logger.info( + "run_geocode_missing_listings: run_id=%d — дренаж " + "(addresses_total=%d < batch_size=%d), завершаем", + run_id, + res.addresses_total, + batch_size, + ) + break + if elapsed > budget_sec: + logger.info( + "run_geocode_missing_listings: run_id=%d — бюджет %.0fs исчерпан " + "(elapsed=%.1fs), завершаем", + run_id, + budget_sec, + elapsed, + ) + break + + total.duration_sec = time.monotonic() - start + runs_mod.mark_done(db, run_id, counters) + logger.info( + "run_geocode_missing_listings: run_id=%d DONE — " + "processed=%d geocoded=%d failed=%d listings_updated=%d duration=%.1fs", + run_id, + total.addresses_processed, + total.addresses_geocoded, + total.addresses_failed, + total.listings_updated, + total.duration_sec, + ) + return total + + except Exception as exc: + total.duration_sec = time.monotonic() - start + logger.exception( + "run_geocode_missing_listings: run_id=%d FAILED after %.1fs", + run_id, + total.duration_sec, + ) + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/data/sql/110_scrape_schedules_seed_geocode_missing_listings.sql b/tradein-mvp/backend/data/sql/110_scrape_schedules_seed_geocode_missing_listings.sql new file mode 100644 index 00000000..a34a17eb --- /dev/null +++ b/tradein-mvp/backend/data/sql/110_scrape_schedules_seed_geocode_missing_listings.sql @@ -0,0 +1,65 @@ +-- 110_scrape_schedules_seed_geocode_missing_listings.sql +-- Seed row для nightly geocode backfill listings.lat/lon (все источники, все города). +-- +-- Root cause (обнаружен 2026-06-16): +-- Scrapers (avito, yandex, n1) сохраняют поисковые карточки с lat=lon=NULL +-- в расчёте на то что «geocode-missing cron подтянет». +-- OS crontab геокодирует ТОЛЬКО rosreestr_deals — для listings такого cron нет. +-- На prod: 6959 avito + 590 yandex + 125 n1 = 7674 listings с null-geom при +-- ненулевом address; geocode_tried_at IS NULL (ни разу не пробовали с 2026-05-23). +-- Dry-run показал 85% geocode success (51/60) — wiring нотночного scheduler'а +-- материально поднимает геопокрытие. +-- +-- Решение: +-- Вместо нового cron — wire существующую geocode_missing_listings в in-app scheduler. +-- Функция address-dedup (1 geocode call на уникальный адрес → UPDATE all listings). +-- Loop-safe: addresses помечаются geocode_tried_at=NOW() при failure → backoff 7 дней, +-- нет бесконечного переотбора. +-- +-- Окно 06:00-09:00 UTC (09:00-12:00 МСК): +-- - После city sweeps (~03-04 UTC), которые depositing новые listings с null-geom. +-- - После deals OS-geocode cron (05:30 / 05:45 UTC) — снижаем Nominatim contention. +-- - До open-hours peaks на Nominatim (дневные задачи других систем). +-- +-- default_params: +-- batch_size — addresses за один вызов geocode_missing_listings (200 ≈ 3.5 мин Nominatim). +-- budget_sec — wall-clock budget одного прогона; 1800 = 30 мин (fits в 3h window). +-- +-- ЗАВИСИМОСТИ: 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 +( + 'geocode_missing_listings', + true, -- ACTIVE: нет внешней зависимости (anti-bot), + -- только Nominatim rate-limited geocoder + 6, + 9, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 6)) AT TIME ZONE 'UTC', + '{"batch_size": 200, "budget_sec": 1800}'::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).'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/tasks/test_geocode_missing.py b/tradein-mvp/backend/tests/tasks/test_geocode_missing.py index 3534f45f..b8ccdc75 100644 --- a/tradein-mvp/backend/tests/tasks/test_geocode_missing.py +++ b/tradein-mvp/backend/tests/tasks/test_geocode_missing.py @@ -1,9 +1,10 @@ """Tests for geocode_missing_listings batch task.""" + from __future__ import annotations import os import sys -from unittest.mock import AsyncMock, MagicMock, call, patch +from unittest.mock import AsyncMock, MagicMock, patch # DATABASE_URL required by config before any app import. os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") @@ -12,10 +13,14 @@ os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost: _wp_mock = MagicMock() sys.modules.setdefault("weasyprint", _wp_mock) -import pytest +import pytest # noqa: E402 -from app.services.geocoder import GeocodeResult -from app.tasks.geocode_missing import GeocodeBackfillResult, geocode_missing_listings +from app.services.geocoder import GeocodeResult # noqa: E402 +from app.tasks.geocode_missing import ( # noqa: E402 + GeocodeBackfillResult, + geocode_missing_listings, + run_geocode_missing_listings, +) def _make_geocode_result(provider: str = "nominatim") -> GeocodeResult: @@ -142,13 +147,16 @@ async def test_geocode_missing_cache_hit_counted() -> None: @pytest.mark.asyncio async def test_geocode_missing_skips_failed_addresses() -> None: - """Geocoder возвращает None → failed++, UPDATE не вызывается.""" + """Geocoder возвращает None → failed++, lat UPDATE не вызывается, tried_at помечается.""" rows = [{"address": "несуществующий адрес", "listings_count": 1}] db = MagicMock() select_result = MagicMock() select_result.mappings.return_value.all.return_value = rows - db.execute.return_value = select_result + tried_at_update_result = MagicMock() + tried_at_update_result.rowcount = 1 + # execute side_effect: [SELECT, UPDATE tried_at] + db.execute.side_effect = [select_result, tried_at_update_result] with patch( "app.tasks.geocode_missing.geocode", @@ -160,8 +168,8 @@ async def test_geocode_missing_skips_failed_addresses() -> None: assert result.addresses_failed == 1 assert result.addresses_geocoded == 0 assert result.listings_updated == 0 - # commit не должен был вызываться после failed geocode - db.commit.assert_not_called() + # commit вызывается для tried_at UPDATE (loop-safe backoff) + db.commit.assert_called_once() @pytest.mark.asyncio @@ -209,7 +217,7 @@ async def test_geocode_missing_batch_size_respected() -> None: @pytest.mark.asyncio async def test_geocode_missing_geocode_exception_counted_as_failed() -> None: - """Geocoder raises Exception → failed++, обработка продолжается.""" + """Geocoder raises Exception → failed++, tried_at помечается, обработка продолжается.""" rows = [ {"address": "ул. Сломанная, 1", "listings_count": 1}, {"address": "ул. Рабочая, 5", "listings_count": 2}, @@ -218,9 +226,12 @@ async def test_geocode_missing_geocode_exception_counted_as_failed() -> None: db = MagicMock() select_result = MagicMock() select_result.mappings.return_value.all.return_value = rows + tried_at_result = MagicMock() + tried_at_result.rowcount = 1 update_result = MagicMock() update_result.rowcount = 2 - db.execute.side_effect = [select_result, update_result] + # SELECT → tried_at UPDATE (exception path) → lat/lon UPDATE (success path) + db.execute.side_effect = [select_result, tried_at_result, update_result] geo_side_effects = [RuntimeError("network error"), _make_geocode_result("nominatim")] @@ -236,6 +247,168 @@ async def test_geocode_missing_geocode_exception_counted_as_failed() -> None: assert result.listings_updated == 2 +# ── New tests: geocode_tried_at / loop-safe / run_geocode_missing_listings ─── + + +@pytest.mark.asyncio +async def test_geocode_missing_failure_marks_tried_at() -> None: + """geo=None → UPDATE geocode_tried_at выполняется (loop-safe backoff).""" + rows = [{"address": "несуществующий адрес", "listings_count": 2}] + + db = MagicMock() + select_result = MagicMock() + select_result.mappings.return_value.all.return_value = rows + tried_at_result = MagicMock() + db.execute.side_effect = [select_result, tried_at_result] + + with patch( + "app.tasks.geocode_missing.geocode", + new_callable=AsyncMock, + return_value=None, + ): + result = await geocode_missing_listings(db, batch_size=200) + + assert result.addresses_failed == 1 + assert result.listings_updated == 0 + # Второй execute — UPDATE tried_at + assert db.execute.call_count == 2 + second_call_sql = str(db.execute.call_args_list[1][0][0]) + assert "geocode_tried_at" in second_call_sql + db.commit.assert_called_once() + + +@pytest.mark.asyncio +async def test_geocode_missing_exception_marks_tried_at() -> None: + """geocode() raises → UPDATE geocode_tried_at выполняется, commit вызывается.""" + rows = [{"address": "ул. Битая, 99", "listings_count": 1}] + + db = MagicMock() + select_result = MagicMock() + select_result.mappings.return_value.all.return_value = rows + tried_at_result = MagicMock() + db.execute.side_effect = [select_result, tried_at_result] + + with patch( + "app.tasks.geocode_missing.geocode", + new_callable=AsyncMock, + side_effect=RuntimeError("timeout"), + ): + result = await geocode_missing_listings(db, batch_size=200) + + assert result.addresses_failed == 1 + assert db.execute.call_count == 2 + second_call_sql = str(db.execute.call_args_list[1][0][0]) + assert "geocode_tried_at" in second_call_sql + db.commit.assert_called_once() + + +@pytest.mark.asyncio +async def test_geocode_missing_dry_run_no_tried_at_update() -> None: + """dry_run=True + geo=None → tried_at НЕ обновляется, commit не вызывается.""" + rows = [{"address": "несуществующий адрес", "listings_count": 1}] + + db = MagicMock() + select_result = MagicMock() + select_result.mappings.return_value.all.return_value = rows + db.execute.return_value = select_result + + with patch( + "app.tasks.geocode_missing.geocode", + new_callable=AsyncMock, + return_value=None, + ): + result = await geocode_missing_listings(db, batch_size=200, dry_run=True) + + assert result.addresses_failed == 1 + # Только SELECT, никаких UPDATE + assert db.execute.call_count == 1 + db.commit.assert_not_called() + + +@pytest.mark.asyncio +async def test_geocode_missing_recent_tried_at_excluded_via_where() -> None: + """SELECT содержит geocode_tried_at IS NULL OR tried_at < NOW() - 7d (loop-safe WHERE).""" + db = MagicMock() + select_result = MagicMock() + select_result.mappings.return_value.all.return_value = [] + db.execute.return_value = select_result + + with patch("app.tasks.geocode_missing.geocode", new_callable=AsyncMock): + await geocode_missing_listings(db, batch_size=10) + + first_call = db.execute.call_args_list[0] + sql_text = str(first_call[0][0]) + assert "geocode_tried_at IS NULL" in sql_text + assert "7 days" in sql_text + + +@pytest.mark.asyncio +async def test_run_geocode_missing_listings_terminates_on_drained() -> None: + """run_geocode_missing_listings завершается когда addresses_total == 0 (ничего pending).""" + db = MagicMock() + drained = GeocodeBackfillResult(addresses_total=0) + + with ( + patch( + "app.tasks.geocode_missing.geocode_missing_listings", + new_callable=AsyncMock, + return_value=drained, + ), + patch("app.tasks.geocode_missing.runs_mod") as mock_runs, + ): + result = await run_geocode_missing_listings(db, run_id=42, params={}) + + mock_runs.update_heartbeat.assert_called() + mock_runs.mark_done.assert_called_once() + mock_runs.mark_failed.assert_not_called() + assert result.addresses_total == 0 + + +@pytest.mark.asyncio +async def test_run_geocode_missing_listings_terminates_on_partial_batch() -> None: + """Последний batch меньше batch_size → завершаем (дренаж).""" + db = MagicMock() + partial = GeocodeBackfillResult( + addresses_total=50, # < batch_size=200 + addresses_processed=50, + listings_updated=30, + ) + + with ( + patch( + "app.tasks.geocode_missing.geocode_missing_listings", + new_callable=AsyncMock, + return_value=partial, + ), + patch("app.tasks.geocode_missing.runs_mod") as mock_runs, + ): + result = await run_geocode_missing_listings(db, run_id=7, params={"batch_size": 200}) + + mock_runs.mark_done.assert_called_once() + assert result.addresses_processed == 50 + assert result.listings_updated == 30 + + +@pytest.mark.asyncio +async def test_run_geocode_missing_listings_mark_failed_on_exception() -> None: + """При исключении внутри loop: mark_failed вызывается, re-raise.""" + db = MagicMock() + + with ( + patch( + "app.tasks.geocode_missing.geocode_missing_listings", + new_callable=AsyncMock, + side_effect=RuntimeError("DB down"), + ), + patch("app.tasks.geocode_missing.runs_mod") as mock_runs, + ): + with pytest.raises(RuntimeError, match="DB down"): + await run_geocode_missing_listings(db, run_id=99, params={}) + + mock_runs.mark_failed.assert_called_once() + mock_runs.mark_done.assert_not_called() + + # ── Integration-style: estimator Avito exclusion removed ───────────────────── @@ -246,9 +419,9 @@ def test_estimator_fetch_analogs_includes_avito() -> None: from app.services import estimator source = inspect.getsource(estimator._fetch_analogs) - assert "source <> 'avito'" not in source, ( - "Avito exclusion должен быть удалён из estimator._fetch_analogs" - ) + assert ( + "source <> 'avito'" not in source + ), "Avito exclusion должен быть удалён из estimator._fetch_analogs" # ── Admin endpoint smoke (schema-only, no real DB) ──────────────────────────── @@ -302,8 +475,13 @@ def test_admin_geocode_missing_post_dry_run_endpoint_exists() -> None: "app.tasks.geocode_missing.geocode_missing_listings", new_callable=AsyncMock, return_value=GeocodeBackfillResult( - addresses_total=0, addresses_processed=0, addresses_geocoded=0, - addresses_failed=0, listings_updated=0, cache_hits=0, cache_misses=0, + addresses_total=0, + addresses_processed=0, + addresses_geocoded=0, + addresses_failed=0, + listings_updated=0, + cache_hits=0, + cache_misses=0, duration_sec=0.01, ), ):