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) ──
 
 

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,

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

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