fix(scrapers): geocode-missing listings nightly scheduler — listings geom coverage backfill (all sources) #1549
4 changed files with 447 additions and 19 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
@ -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,
|
||||
),
|
||||
):
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue