From 52716a46c6b1ccba53b83ce665824caf360bc025 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sat, 23 May 2026 20:03:48 +0000 Subject: [PATCH] fix(tradein): geocode backfill + remove Avito exclusion from estimator (#490) --- tradein-mvp/backend/app/api/v1/admin.py | 112 +++++- tradein-mvp/backend/app/services/estimator.py | 10 +- .../backend/app/tasks/geocode_missing.py | 186 ++++++++++ tradein-mvp/backend/tests/tasks/__init__.py | 0 .../tests/tasks/test_geocode_missing.py | 318 ++++++++++++++++++ 5 files changed, 618 insertions(+), 8 deletions(-) create mode 100644 tradein-mvp/backend/app/tasks/geocode_missing.py create mode 100644 tradein-mvp/backend/tests/tasks/__init__.py create mode 100644 tradein-mvp/backend/tests/tasks/test_geocode_missing.py diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index d895abd0..fc6b61cc 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -8,9 +8,9 @@ from __future__ import annotations import json import logging import time -from typing import Annotated, Literal +from typing import Annotated, Any, Literal -from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException +from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query from pydantic import BaseModel, Field from sqlalchemy import text from sqlalchemy.orm import Session @@ -39,6 +39,7 @@ from app.services.scrapers.yandex_detail import YandexDetailScraper from app.services.scrapers.yandex_newbuilding import YandexNewbuildingScraper from app.services.scrapers.yandex_realty import YandexRealtyScraper from app.services.scrapers.yandex_valuation import YandexValuationScraper +from app.tasks.geocode_missing import GeocodeBackfillResult, geocode_missing_listings logger = logging.getLogger(__name__) @@ -316,6 +317,113 @@ async def test_cian_auth( return {"authenticated": True, "userId": user_id, "reason": None} +# ── Geocode backfill: batch address-dedup geocoder (all sources) ───────────── + + +@router.get("/scrape/geocode-missing-listings/status") +async def get_geocode_backfill_status( + db: Annotated[Session, Depends(get_db)], +) -> dict[str, Any]: + """Текущая статистика backfill — сколько ещё pending geocode. + + Включает per-source breakdown и оценку времени при Nominatim 1 req/sec. + """ + stats = db.execute( + text( + """ + SELECT + source, + COUNT(*) AS total, + COUNT(*) FILTER (WHERE lat IS NULL AND address IS NOT NULL) AS pending_geocode, + COUNT(*) FILTER (WHERE lat IS NOT NULL) AS has_coords + FROM listings + WHERE is_active = true + GROUP BY source + ORDER BY pending_geocode DESC + """ + ) + ).mappings().all() + + total_pending = sum(s["pending_geocode"] for s in stats) + unique_addresses_pending = db.execute( + text( + """ + SELECT COUNT(DISTINCT address) FROM listings + WHERE lat IS NULL AND address IS NOT NULL AND length(trim(address)) >= 5 + """ + ) + ).scalar() or 0 + + cache_size = db.execute( + text("SELECT COUNT(*) FROM geocode_cache WHERE expires_at > NOW()") + ).scalar() or 0 + + return { + "total_pending_listings": total_pending, + "unique_addresses_pending": int(unique_addresses_pending), + "estimated_nominatim_minutes": round(int(unique_addresses_pending) / 60.0, 1), + "cache_size_active": int(cache_size), + "per_source": [dict(s) for s in stats], + } + + +@router.post("/scrape/geocode-missing-listings") +async def trigger_geocode_missing_listings( + background_tasks: BackgroundTasks, + db: Annotated[Session, Depends(get_db)], + batch_size: int = Query(default=200, ge=1, le=2000), + dry_run: bool = Query(default=False), + background: bool = Query( + default=False, + description="True → запустить в BackgroundTasks (не ждать результат).", + ), +) -> dict[str, Any]: + """Manual trigger для geocode backfill всех listings с NULL coords (address-dedup). + + batch_size=200 ≈ 3.5 мин при Nominatim 1 req/sec (без cache hits — быстрее с cache). + Безопасно для повторного запуска (ON CONFLICT в geocode_cache, UPDATE WHERE lat IS NULL). + + dry_run=true — показывает что бы сделалось, без UPDATE listings. + background=true — запускает через BackgroundTasks, возвращает немедленно (для больших batch). + """ + from app.core.db import SessionLocal + + if background: + async def _bg_task() -> None: + bg_db = SessionLocal() + try: + await geocode_missing_listings(bg_db, batch_size=batch_size, dry_run=dry_run) + except Exception: + logger.exception("geocode-missing-listings background task crashed") + finally: + bg_db.close() + + background_tasks.add_task(_bg_task) + return { + "status": "queued", + "batch_size": batch_size, + "dry_run": dry_run, + "note": "Running in background. Check logs for progress.", + } + + result: GeocodeBackfillResult = await geocode_missing_listings( + db, batch_size=batch_size, dry_run=dry_run + ) + return { + "status": "ok", + "batch_size": batch_size, + "dry_run": dry_run, + "addresses_total": result.addresses_total, + "addresses_processed": result.addresses_processed, + "addresses_geocoded": result.addresses_geocoded, + "addresses_failed": result.addresses_failed, + "listings_updated": result.listings_updated, + "cache_hits": result.cache_hits, + "cache_misses": result.cache_misses, + "duration_sec": round(result.duration_sec, 2), + } + + # ── Stage 4a: Avito enrichment endpoints (single-shot debug + manual triggers) ── diff --git a/tradein-mvp/backend/app/services/estimator.py b/tradein-mvp/backend/app/services/estimator.py index 3071e6f9..75b340cd 100644 --- a/tradein-mvp/backend/app/services/estimator.py +++ b/tradein-mvp/backend/app/services/estimator.py @@ -805,12 +805,10 @@ def _fetch_analogs( AND is_active = true AND scraped_at > NOW() - (:fresh_days || ' days')::interval AND price_rub > 0 - -- Avito исключён из радиусного поиска: его «координаты» — это - -- якорь cron-скрейпа ± jitter (DOM Avito реальных coords не - -- отдаёт). Фейковые точки кучкуются на 5 якорях и вытесняют - -- реальные cian/yandex/n1 — оценка считалась почти целиком по - -- Avito. Реальные источники дают честный гео-радиус. - AND source <> 'avito' + -- 2026-05-23: Avito coords теперь real (PR #487 убрал jitter после + -- C-5 audit). Listings с NULL coords отфильтруются через ST_DWithin + -- (geom IS NULL → не matches). geocode-missing-listings backfill + -- подтягивает координаты для address-only Avito листингов. ORDER BY ( -- distance_m — это SELECT-алиас. В ORDER BY-ВЫРАЖЕНИИ (не голым -- термом) PostgreSQL трактует имя как входную колонку listings, diff --git a/tradein-mvp/backend/app/tasks/geocode_missing.py b/tradein-mvp/backend/app/tasks/geocode_missing.py new file mode 100644 index 00000000..5a57a060 --- /dev/null +++ b/tradein-mvp/backend/app/tasks/geocode_missing.py @@ -0,0 +1,186 @@ +"""Batch geocoding для listings с NULL lat/lon. + +Запускается: + - Manual через POST /admin/scrape/geocode-missing-listings + - Future: cron job (Celery beat если bootstrap'нут / OS cron) + +Pattern: dedup по address (1 unique address → 1 geocode call → UPDATE all listings). +Rate limit: Nominatim 1 req/sec. Yandex 25K/day если YANDEX_GEOCODER_KEY set. + +Отличие от /admin/geocode-missing (per-ID): + - Этот модуль группирует по address → меньше API calls (dedup). + - Поддерживает all sources включая Avito (после PR #487 убрали jitter). + - Возвращает GeocodeBackfillResult с детальными counters. +""" +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass, field + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.services.geocoder import geocode + +logger = logging.getLogger(__name__) + + +@dataclass +class GeocodeBackfillResult: + addresses_total: int = 0 # unique addresses pending geocode в этом batch + addresses_processed: int = 0 # фактически обработано + addresses_geocoded: int = 0 # успешно получили coords + addresses_failed: int = 0 # geocoder вернул None + listings_updated: int = 0 # total listings затронуто (1 address → N listings) + cache_hits: int = 0 # из geocode_cache (instant) + cache_misses: int = 0 # реальные geocoder calls + duration_sec: float = field(default=0.0) + + +async def geocode_missing_listings( + db: Session, + *, + batch_size: int = 200, + dry_run: bool = False, +) -> GeocodeBackfillResult: + """Geocode listings с NULL coords (любой source). + + Steps: + 1. SELECT DISTINCT address FROM listings WHERE lat IS NULL AND address IS NOT NULL + GROUP BY address ORDER BY COUNT(*) DESC LIMIT batch_size + (приоритет адресам с большим числом listings — больший ROI per geocode call) + + 2. Для каждого address: + - geocode(address, db) — auto-cache (hit или miss) + - Если есть результат: UPDATE listings SET lat, lon WHERE address = :addr AND lat IS NULL + - PostGIS trigger (listings_set_geom_trg) автоматически обновит geom + + 3. Log progress каждые 50 addresses. + + Args: + batch_size: max addresses to process per call (default 200 ≈ 3.5 min Nominatim) + dry_run: только показать что бы сделалось, без UPDATE + + Returns: + GeocodeBackfillResult с counters. + """ + start = time.monotonic() + result = GeocodeBackfillResult() + + # 1. Найти top-N адресов с NULL coords (DESC by occurrence count) + rows = db.execute( + text( + """ + SELECT address, COUNT(*) AS listings_count + FROM listings + WHERE lat IS NULL + AND address IS NOT NULL + AND length(trim(address)) >= 5 + GROUP BY address + ORDER BY listings_count DESC, address ASC + LIMIT :limit + """ + ), + {"limit": batch_size}, + ).mappings().all() + + result.addresses_total = len(rows) + + if not rows: + logger.info("geocode_missing: 0 pending addresses — nothing to do") + result.duration_sec = time.monotonic() - start + return result + + logger.info( + "geocode_missing: starting batch=%d total_pending_addresses=%d (top by listings count)", + batch_size, + result.addresses_total, + ) + + for idx, row in enumerate(rows): + address: str = row["address"] + listings_count: int = row["listings_count"] + result.addresses_processed += 1 + + try: + geo = await geocode(address, db) + except Exception as exc: + logger.warning( + "geocode_missing: geocode raised for '%s': %s", address[:60], exc + ) + result.addresses_failed += 1 + continue + + if geo is None: + result.addresses_failed += 1 + logger.info( + "geocode_missing: NOT FOUND '%s' (used in %d listings)", + address[:60], + listings_count, + ) + continue + + if geo.provider == "cache": + result.cache_hits += 1 + else: + result.cache_misses += 1 + + result.addresses_geocoded += 1 + + if dry_run: + logger.info( + "geocode_missing[dry]: '%s' → (%.5f, %.5f) provider=%s would update %d listings", + address[:60], + geo.lat, + geo.lon, + geo.provider, + listings_count, + ) + continue + + # UPDATE listings — PostGIS trigger (listings_set_geom_trg) обновит geom автоматически + update_result = db.execute( + text( + """ + UPDATE listings + SET lat = :lat, lon = :lon + WHERE address = :addr AND lat IS NULL + """ + ), + {"lat": geo.lat, "lon": geo.lon, "addr": address}, + ) + db.commit() + result.listings_updated += update_result.rowcount + + if (idx + 1) % 50 == 0: + elapsed = time.monotonic() - start + rate = (idx + 1) / elapsed if elapsed > 0 else 0 + logger.info( + "geocode_missing: progress %d/%d " + "(geocoded=%d failed=%d cache_hits=%d listings_updated=%d) rate=%.1f addr/s", + idx + 1, + len(rows), + result.addresses_geocoded, + result.addresses_failed, + result.cache_hits, + result.listings_updated, + rate, + ) + + result.duration_sec = time.monotonic() - start + + logger.info( + "geocode_missing: DONE batch=%d processed=%d geocoded=%d failed=%d " + "cache=(hit=%d miss=%d) listings_updated=%d duration=%.1fs", + batch_size, + result.addresses_processed, + result.addresses_geocoded, + result.addresses_failed, + result.cache_hits, + result.cache_misses, + result.listings_updated, + result.duration_sec, + ) + + return result diff --git a/tradein-mvp/backend/tests/tasks/__init__.py b/tradein-mvp/backend/tests/tasks/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tradein-mvp/backend/tests/tasks/test_geocode_missing.py b/tradein-mvp/backend/tests/tasks/test_geocode_missing.py new file mode 100644 index 00000000..3534f45f --- /dev/null +++ b/tradein-mvp/backend/tests/tasks/test_geocode_missing.py @@ -0,0 +1,318 @@ +"""Tests for geocode_missing_listings batch task.""" +from __future__ import annotations + +import os +import sys +from unittest.mock import AsyncMock, MagicMock, call, patch + +# DATABASE_URL required by config before any app import. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +# WeasyPrint stub — not installed in CI without GTK. +_wp_mock = MagicMock() +sys.modules.setdefault("weasyprint", _wp_mock) + +import pytest + +from app.services.geocoder import GeocodeResult +from app.tasks.geocode_missing import GeocodeBackfillResult, geocode_missing_listings + + +def _make_geocode_result(provider: str = "nominatim") -> GeocodeResult: + return GeocodeResult( + lat=56.838, + lon=60.605, + full_address="Екатеринбург, ул. Малышева, 30", + provider=provider, # type: ignore[arg-type] + confidence="exact", + ) + + +def _mock_db_rows(rows: list[dict]) -> MagicMock: + """Возвращает mock Session где .execute().mappings().all() → rows.""" + db = MagicMock() + mappings_mock = MagicMock() + mappings_mock.all.return_value = rows + db.execute.return_value.mappings.return_value = mappings_mock + # execute().rowcount для UPDATE + db.execute.return_value.rowcount = len(rows) + return db + + +@pytest.mark.asyncio +async def test_geocode_missing_empty_returns_zero_result() -> None: + """Нет адресов с NULL coords → возвращает нулевой результат без ошибок.""" + db = _mock_db_rows([]) + + with patch("app.tasks.geocode_missing.geocode", new_callable=AsyncMock) as mock_geo: + result = await geocode_missing_listings(db, batch_size=50) + + assert isinstance(result, GeocodeBackfillResult) + assert result.addresses_total == 0 + assert result.addresses_processed == 0 + assert result.listings_updated == 0 + mock_geo.assert_not_called() + + +@pytest.mark.asyncio +async def test_geocode_missing_processes_pending_addresses() -> None: + """2 unique addresses → geocode вызывается 2 раза, listings_updated суммируется.""" + rows = [ + {"address": "ул. Малышева, 30", "listings_count": 3}, + {"address": "ул. Ленина, 10", "listings_count": 1}, + ] + + db = MagicMock() + # Первый call .execute().mappings().all() → rows (SELECT) + # Последующие .execute() → UPDATE (rowcount) + select_result = MagicMock() + select_result.mappings.return_value.all.return_value = rows + + update_result_1 = MagicMock() + update_result_1.rowcount = 3 + + update_result_2 = MagicMock() + update_result_2.rowcount = 1 + + db.execute.side_effect = [select_result, update_result_1, update_result_2] + + with patch( + "app.tasks.geocode_missing.geocode", + new_callable=AsyncMock, + return_value=_make_geocode_result("nominatim"), + ) as mock_geo: + result = await geocode_missing_listings(db, batch_size=200) + + assert result.addresses_total == 2 + assert result.addresses_processed == 2 + assert result.addresses_geocoded == 2 + assert result.addresses_failed == 0 + assert result.listings_updated == 4 # 3 + 1 + assert result.cache_misses == 2 + assert result.cache_hits == 0 + assert mock_geo.call_count == 2 + + +@pytest.mark.asyncio +async def test_geocode_missing_dedup_addresses_single_geocode_call() -> None: + """5 listings с тем же адресом → SELECT группирует → 1 geocode call.""" + rows = [{"address": "ул. Тургенева, 5", "listings_count": 5}] + + db = MagicMock() + select_result = MagicMock() + select_result.mappings.return_value.all.return_value = rows + update_result = MagicMock() + update_result.rowcount = 5 + db.execute.side_effect = [select_result, update_result] + + with patch( + "app.tasks.geocode_missing.geocode", + new_callable=AsyncMock, + return_value=_make_geocode_result("nominatim"), + ) as mock_geo: + result = await geocode_missing_listings(db, batch_size=200) + + assert mock_geo.call_count == 1 + assert result.listings_updated == 5 + + +@pytest.mark.asyncio +async def test_geocode_missing_cache_hit_counted() -> None: + """Адрес из geocode_cache → cache_hits++, cache_misses не растёт.""" + rows = [{"address": "ул. Горького, 7", "listings_count": 2}] + + db = MagicMock() + select_result = MagicMock() + select_result.mappings.return_value.all.return_value = rows + update_result = MagicMock() + update_result.rowcount = 2 + db.execute.side_effect = [select_result, update_result] + + with patch( + "app.tasks.geocode_missing.geocode", + new_callable=AsyncMock, + return_value=_make_geocode_result("cache"), + ): + result = await geocode_missing_listings(db, batch_size=200) + + assert result.cache_hits == 1 + assert result.cache_misses == 0 + assert result.addresses_geocoded == 1 + + +@pytest.mark.asyncio +async def test_geocode_missing_skips_failed_addresses() -> None: + """Geocoder возвращает None → failed++, UPDATE не вызывается.""" + 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) + + assert result.addresses_failed == 1 + assert result.addresses_geocoded == 0 + assert result.listings_updated == 0 + # commit не должен был вызываться после failed geocode + db.commit.assert_not_called() + + +@pytest.mark.asyncio +async def test_geocode_missing_dry_run_does_not_update() -> None: + """dry_run=True → addresses_processed > 0 но listings_updated == 0, нет commit.""" + rows = [ + {"address": "ул. Пушкина, 1", "listings_count": 3}, + {"address": "пр. Ленина, 50", "listings_count": 2}, + ] + + 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=_make_geocode_result("nominatim"), + ): + result = await geocode_missing_listings(db, batch_size=200, dry_run=True) + + assert result.addresses_processed == 2 + assert result.addresses_geocoded == 2 + assert result.listings_updated == 0 + db.commit.assert_not_called() + + +@pytest.mark.asyncio +async def test_geocode_missing_batch_size_respected() -> None: + """batch_size=3 → SQL LIMIT=3 (проверяем параметр в вызове execute).""" + 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=3) + + # Первый вызов execute — SELECT с LIMIT + first_call = db.execute.call_args_list[0] + params = first_call[0][1] # positional arg[1] = params dict + assert params["limit"] == 3 + + +@pytest.mark.asyncio +async def test_geocode_missing_geocode_exception_counted_as_failed() -> None: + """Geocoder raises Exception → failed++, обработка продолжается.""" + rows = [ + {"address": "ул. Сломанная, 1", "listings_count": 1}, + {"address": "ул. Рабочая, 5", "listings_count": 2}, + ] + + db = MagicMock() + select_result = MagicMock() + select_result.mappings.return_value.all.return_value = rows + update_result = MagicMock() + update_result.rowcount = 2 + db.execute.side_effect = [select_result, update_result] + + geo_side_effects = [RuntimeError("network error"), _make_geocode_result("nominatim")] + + with patch( + "app.tasks.geocode_missing.geocode", + new_callable=AsyncMock, + side_effect=geo_side_effects, + ): + result = await geocode_missing_listings(db, batch_size=200) + + assert result.addresses_failed == 1 + assert result.addresses_geocoded == 1 + assert result.listings_updated == 2 + + +# ── Integration-style: estimator Avito exclusion removed ───────────────────── + + +def test_estimator_fetch_analogs_includes_avito() -> None: + """estimator._fetch_analogs не содержит AND source <> 'avito'.""" + import inspect + + from app.services import estimator + + source = inspect.getsource(estimator._fetch_analogs) + assert "source <> 'avito'" not in source, ( + "Avito exclusion должен быть удалён из estimator._fetch_analogs" + ) + + +# ── Admin endpoint smoke (schema-only, no real DB) ──────────────────────────── + + +def test_admin_geocode_missing_endpoint_exists() -> None: + """GET /api/v1/admin/scrape/geocode-missing-listings/status → endpoint registered.""" + from fastapi import FastAPI + from fastapi.testclient import TestClient + + from app.api.v1 import admin as admin_module + from app.core.db import get_db + + app = FastAPI() + app.include_router(admin_module.router, prefix="/api/v1/admin") + + mock_db = MagicMock() + # status endpoint calls execute twice — stats + unique + cache_size + stats_result = MagicMock() + stats_result.mappings.return_value.all.return_value = [] + scalar_result = MagicMock() + scalar_result.scalar.return_value = 0 + mock_db.execute.side_effect = [stats_result, scalar_result, scalar_result] + + app.dependency_overrides[get_db] = lambda: mock_db + + client = TestClient(app) + resp = client.get("/api/v1/admin/scrape/geocode-missing-listings/status") + # Если endpoint зарегистрирован корректно — не 404 + assert resp.status_code != 404 + + +def test_admin_geocode_missing_post_dry_run_endpoint_exists() -> None: + """POST /api/v1/admin/scrape/geocode-missing-listings?dry_run=true → endpoint registered.""" + from fastapi import FastAPI + from fastapi.testclient import TestClient + + from app.api.v1 import admin as admin_module + from app.core.db import get_db + + app = FastAPI() + app.include_router(admin_module.router, prefix="/api/v1/admin") + + # Mock db: не нужен реальный — тест только на routing + mock_db = MagicMock() + app.dependency_overrides[get_db] = lambda: mock_db + + client = TestClient(app) + # Без реального geocode — mocked через patch + with patch( + "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, + duration_sec=0.01, + ), + ): + resp = client.post( + "/api/v1/admin/scrape/geocode-missing-listings?dry_run=true&batch_size=10" + ) + + assert resp.status_code != 404 + if resp.status_code == 200: + data = resp.json() + assert "status" in data + assert "addresses_total" in data