feat(tradein): бэкфилл координат листингов из ekb_geoportal_buildings (street+house match)

tradein-mvp/backend/app/tasks/backfill_listings_coords_geoportal.py

@@ -0,0 +1,309 @@
+"""Бэкфилл координат листингов из реестра зданий ЕКБ (ekb_geoportal_buildings).
+
+Назначение: проставить lat/lon/geom для листингов, у которых геокодер не отработал
+(lat IS NULL), используя точный street_norm+house_norm матч по таблице ekb_geoportal_buildings.
+
+Нормализация совпадает с ekb_geoportal_ingest:
+  street_norm = lower(trim(street))
+  house_norm  = lower(house) без внутренних пробелов («7 б» → «7б»)
+
+Парсинг адреса — _parse_street_house из app.services.geocoder (готовый парсер),
+работающий с формами «г. Екатеринбург, ул. Малышева, 30, кв. 28».
+
+Запуск:
+  python -m app.tasks.backfill_listings_coords_geoportal
+  python -m app.tasks.backfill_listings_coords_geoportal --limit 5000 --batch-size 200
+
+Идемпотентность: UPDATE применяется только к строкам с lat IS NULL (WHERE id=:id AND lat IS NULL).
+Повторный прогон не затирает уже проставленные координаты.
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import time
+from dataclasses import dataclass, field
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.db import SessionLocal
+from app.services.geocoder import _geoportal_house_match, _parse_street_house
+
+logger = logging.getLogger(__name__)
+
+# Размер батча на SELECT + commit-интервал.
+DEFAULT_BATCH_SIZE = 500
+DEFAULT_LIMIT: int | None = None  # None = без ограничений (весь backlog)
+
+
+@dataclass
+class BackfillCoordsResult:
+    """Счётчики прогона бэкфилла."""
+
+    candidates: int = 0  # листингов-кандидатов (lat IS NULL)
+    matched: int = 0  # сматчено с ekb_geoportal_buildings
+    updated: int = 0  # реально обновлено (UPDATE rowcount)
+    no_address: int = 0  # listing.address IS NULL / не распарсился
+    no_match: int = 0  # адрес распарсился, но в реестре здания нет
+    errors: int = 0  # исключения при обработке отдельной записи
+    duration_sec: float = field(default=0.0)
+
+
+def _update_listing_coords(
+    db: Session,
+    listing_id: int,
+    lat: float,
+    lon: float,
+) -> int:
+    """UPDATE lat/lon/geom для одного листинга (только если lat IS NULL — идемпотентно).
+
+    geom проставляется через ST_SetSRID(ST_MakePoint(...), 4326). Не тригерит
+    отдельный trigger — явный UPDATE.
+
+    Возвращает rowcount (1 = обновлено, 0 = уже было заполнено или не найдено).
+    """
+    result = db.execute(
+        text("""
+            UPDATE listings
+            SET lat  = CAST(:lat AS double precision),
+                lon  = CAST(:lon AS double precision),
+                geom = ST_SetSRID(
+                    ST_MakePoint(
+                        CAST(:lon AS double precision),
+                        CAST(:lat AS double precision)
+                    ),
+                    4326
+                )
+            WHERE id = CAST(:id AS bigint)
+              AND lat IS NULL
+        """),
+        {"lat": lat, "lon": lon, "id": listing_id},
+    )
+    return result.rowcount  # type: ignore[return-value]
+
+
+def backfill_coords_from_geoportal(
+    db: Session,
+    *,
+    limit: int | None = None,
+    sources: list[str] | None = None,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> BackfillCoordsResult:
+    """Бэкфилл координат листингов из ekb_geoportal_buildings.
+
+    Алгоритм:
+    1. SELECT листингов с lat IS NULL + ЕКБ-скоуп (опционально), батчами batch_size.
+    2. Для каждого: распарсить street+house через _parse_street_house (geocoder),
+       нормализовать (как ekb_geoportal_ingest) и сматчить в ekb_geoportal_buildings.
+    3. Матч есть → UPDATE lat/lon/geom (только при lat IS NULL — идемпотентно).
+    4. commit каждые batch_size записей, SAVEPOINT-защита на батч.
+
+    Args:
+        db: SQLAlchemy сессия (tradein БД).
+        limit: ограничение общего числа обрабатываемых листингов (None = всё).
+        sources: список source-значений для фильтрации (None = все источники).
+        batch_size: размер батча SELECT + интервал commit.
+
+    Returns:
+        BackfillCoordsResult со счётчиками.
+    """
+    start = time.monotonic()
+    res = BackfillCoordsResult()
+
+    offset = 0
+    total_limit = limit  # общий лимит листингов, None = без ограничений
+
+    while True:
+        # Сколько брать в этой итерации
+        fetch_n = batch_size
+        if total_limit is not None:
+            remaining = total_limit - res.candidates
+            if remaining <= 0:
+                break
+            fetch_n = min(batch_size, remaining)
+
+        # Собираем WHERE с опциональным source-фильтром
+        source_clause = ""
+        params: dict = {"limit": fetch_n, "offset": offset}
+        if sources:
+            source_clause = " AND source = ANY(CAST(:sources AS text[]))"
+            params["sources"] = sources
+
+        rows = (
+            db.execute(
+                text(f"""
+                    SELECT id, address
+                    FROM listings
+                    WHERE lat IS NULL
+                      AND geom IS NULL
+                      AND address IS NOT NULL
+                      {source_clause}
+                    ORDER BY id
+                    LIMIT CAST(:limit AS integer) OFFSET CAST(:offset AS integer)
+                """),
+                params,
+            )
+            .mappings()
+            .all()
+        )
+
+        if not rows:
+            break
+
+        res.candidates += len(rows)
+        batch_updated = 0
+
+        try:
+            with db.begin_nested():
+                for row in rows:
+                    listing_id: int = row["id"]
+                    address: str | None = row["address"]
+
+                    if not address or not address.strip():
+                        res.no_address += 1
+                        continue
+
+                    # Парсинг адреса — переиспользуем парсер geocoder'а
+                    parsed = _parse_street_house(address)
+                    if parsed is None:
+                        res.no_address += 1
+                        continue
+
+                    street, house = parsed
+
+                    # Матч в ekb_geoportal_buildings (точный по street_norm+house_norm)
+                    try:
+                        hit = _geoportal_house_match(db, street, house)
+                    except Exception:
+                        logger.warning(
+                            "backfill_coords: geoportal match raised for listing_id=%d addr=%r",
+                            listing_id,
+                            address[:80],
+                            exc_info=True,
+                        )
+                        res.errors += 1
+                        continue
+
+                    if hit is None:
+                        res.no_match += 1
+                        continue
+
+                    res.matched += 1
+
+                    # UPDATE (идемпотентно — WHERE lat IS NULL)
+                    try:
+                        rc = _update_listing_coords(db, listing_id, hit.lat, hit.lon)
+                        res.updated += rc
+                        batch_updated += rc
+                    except Exception:
+                        logger.warning(
+                            "backfill_coords: UPDATE failed for listing_id=%d",
+                            listing_id,
+                            exc_info=True,
+                        )
+                        res.errors += 1
+
+        except Exception:
+            logger.warning(
+                "backfill_coords: batch SAVEPOINT failed (offset=%d), batch rolled back",
+                offset,
+                exc_info=True,
+            )
+            res.errors += 1
+
+        db.commit()
+
+        logger.info(
+            "backfill_coords: batch offset=%d rows=%d matched=%d updated=%d "
+            "(total: candidates=%d matched=%d updated=%d no_addr=%d no_match=%d err=%d)",
+            offset,
+            len(rows),
+            batch_updated,
+            batch_updated,
+            res.candidates,
+            res.matched,
+            res.updated,
+            res.no_address,
+            res.no_match,
+            res.errors,
+        )
+
+        # Если вернули меньше чем просили — конец данных
+        if len(rows) < fetch_n:
+            break
+
+        offset += len(rows)
+
+    res.duration_sec = time.monotonic() - start
+
+    logger.info(
+        "backfill_coords: DONE — candidates=%d matched=%d updated=%d "
+        "no_address=%d no_match=%d errors=%d duration=%.1fs",
+        res.candidates,
+        res.matched,
+        res.updated,
+        res.no_address,
+        res.no_match,
+        res.errors,
+        res.duration_sec,
+    )
+
+    return res
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    parser = argparse.ArgumentParser(
+        description="Бэкфилл координат листингов из ekb_geoportal_buildings"
+    )
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=None,
+        help="Максимальное число листингов (default: без ограничений)",
+    )
+    parser.add_argument(
+        "--batch-size",
+        type=int,
+        default=DEFAULT_BATCH_SIZE,
+        help=f"Размер батча (default: {DEFAULT_BATCH_SIZE})",
+    )
+    parser.add_argument(
+        "--sources",
+        nargs="*",
+        default=None,
+        help="Фильтр по source (avito/cian/yandex); без флага — все источники",
+    )
+    args = parser.parse_args()
+
+    db = SessionLocal()
+    try:
+        result = backfill_coords_from_geoportal(
+            db,
+            limit=args.limit,
+            sources=args.sources,
+            batch_size=args.batch_size,
+        )
+    finally:
+        db.close()
+
+    logger.info(
+        "Готово: кандидатов=%d сматчено=%d обновлено=%d "
+        "без_адреса=%d без_матча=%d ошибок=%d время=%.1fs",
+        result.candidates,
+        result.matched,
+        result.updated,
+        result.no_address,
+        result.no_match,
+        result.errors,
+        result.duration_sec,
+    )
+
+
+if __name__ == "__main__":
+    main()

tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py

@@ -0,0 +1,364 @@
+"""Unit tests for backfill_listings_coords_geoportal task.
+
+Покрывает:
+- Листинг с адресом «ул. Малышева, 30» + geoportal-строка → coords проставляются.
+- Листинг без матча в geoportal → не трогается (no_match++).
+- Листинг с lat уже заполненным (не попадает в SELECT WHERE lat IS NULL) → skip.
+- Листинг с нераспарсимым адресом (NULL / мусор) → no_address++.
+- Пустой SELECT → нулевой результат.
+- limit/sources параметры прокидываются в SQL.
+- Идемпотентность: UPDATE с WHERE lat IS NULL не затирает уже заполненное.
+- SAVEPOINT откат батча при исключении в DB.execute.
+
+Все сетевые вызовы и DB замоканы.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+_wp_mock = MagicMock()
+sys.modules.setdefault("weasyprint", _wp_mock)
+
+from app.services.geocoder import GeocodeSuggestion  # noqa: E402
+from app.tasks.backfill_listings_coords_geoportal import (  # noqa: E402
+    BackfillCoordsResult,
+    backfill_coords_from_geoportal,
+)
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+
+_HIT = GeocodeSuggestion(
+    label="Малышева, 30",
+    full_address="Малышева, 30",
+    lat=56.838,
+    lon=60.605,
+    kind="house",
+)
+
+
+def _make_db(select_pages: list[list[dict]], update_rowcount: int = 1) -> MagicMock:
+    """Mock Session. SELECT-вызовы возвращают страницы из очереди по порядку.
+
+    UPDATE и прочие execute()-вызовы возвращают generic-мок с rowcount=update_rowcount.
+    Очередь SELECT определяется по наличию .mappings()-обращений внутри задачи:
+    задача вызывает .mappings().all() только для SELECT-запроса (первый execute
+    в каждой итерации батча). UPDATE-вызовы внутри begin_nested не используют
+    .mappings() — это позволяет различить их.
+    """
+    db = MagicMock()
+
+    nested_ctx = MagicMock()
+    nested_ctx.__enter__ = MagicMock(return_value=None)
+    nested_ctx.__exit__ = MagicMock(return_value=False)
+    db.begin_nested.return_value = nested_ctx
+
+    # SELECT-очередь (только для .mappings().all()-вызовов)
+    select_queue: list[list[dict]] = list(select_pages)
+
+    # generic UPDATE-мок
+    upd = MagicMock()
+    upd.rowcount = update_rowcount
+
+    # Отдельный счётчик: первый execute каждого батча = SELECT (mappings)
+    # Определяем по тексту SQL: SELECT vs UPDATE
+    def _side(*args, **kwargs):
+        sql = str(args[0]) if args else ""
+        if "SELECT" in sql or "select" in sql:
+            if select_queue:
+                page = select_queue.pop(0)
+            else:
+                page = []
+            sel = MagicMock()
+            sel.mappings.return_value.all.return_value = page
+            return sel
+        # UPDATE / прочее
+        return upd
+
+    db.execute.side_effect = _side
+    return db
+
+
+# ── basic match ──────────────────────────────────────────────────────────────
+
+
+def test_match_updates_coords() -> None:
+    """Листинг с адресом Малышева 30 → geoportal матч → lat/lon/geom UPDATE."""
+    rows = [{"id": 1, "address": "г. Екатеринбург, ул. Малышева, 30, кв. 28"}]
+    db = _make_db([rows, []])  # второй SELECT пустой → конец
+
+    with (
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._geoportal_house_match",
+            return_value=_HIT,
+        ) as mock_geo,
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._parse_street_house",
+            return_value=("малышева", "30"),
+        ),
+    ):
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    assert res.candidates == 1
+    assert res.matched == 1
+    assert res.updated == 1
+    assert res.no_address == 0
+    assert res.no_match == 0
+    assert res.errors == 0
+    mock_geo.assert_called_once_with(db, "малышева", "30")
+    # UPDATE должен быть вызван с правильными lat/lon
+    update_calls = [c for c in db.execute.call_args_list if "UPDATE" in str(c.args[0])]
+    assert len(update_calls) == 1
+    params = update_calls[0].args[1]
+    assert params["lat"] == pytest.approx(56.838)
+    assert params["lon"] == pytest.approx(60.605)
+    assert params["id"] == 1
+
+
+def test_no_match_does_not_update() -> None:
+    """Geoportal не нашёл → no_match++, UPDATE не вызывается."""
+    rows = [{"id": 2, "address": "ул. Несуществующая, 999"}]
+    db = _make_db([rows, []])
+
+    with (
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._geoportal_house_match",
+            return_value=None,
+        ),
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._parse_street_house",
+            return_value=("несуществующая", "999"),
+        ),
+    ):
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    assert res.candidates == 1
+    assert res.matched == 0
+    assert res.updated == 0
+    assert res.no_match == 1
+    # UPDATE не вызывался (только SELECT)
+    update_calls = [c for c in db.execute.call_args_list if "UPDATE" in str(c.args[0])]
+    assert len(update_calls) == 0
+
+
+def test_listing_with_coords_excluded_by_where() -> None:
+    """SELECT WHERE lat IS NULL не вернёт листинги с координатами.
+
+    Моделируем это пустым SELECT → нулевой результат.
+    """
+    db = _make_db([[]])  # SELECT сразу пустой
+
+    with patch("app.tasks.backfill_listings_coords_geoportal._geoportal_house_match") as mock_geo:
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    assert res.candidates == 0
+    assert res.matched == 0
+    assert res.updated == 0
+    mock_geo.assert_not_called()
+
+
+def test_null_address_counted_as_no_address() -> None:
+    """Листинг с адресом None (NULL в БД) → no_address++."""
+    rows = [{"id": 3, "address": None}]
+    db = _make_db([rows, []])
+
+    with patch("app.tasks.backfill_listings_coords_geoportal._geoportal_house_match") as mock_geo:
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    assert res.no_address == 1
+    assert res.matched == 0
+    mock_geo.assert_not_called()
+
+
+def test_unparseable_address_counted_as_no_address() -> None:
+    """_parse_street_house вернул None → no_address++, geoportal не вызывается."""
+    rows = [{"id": 4, "address": "полный мусор без дома"}]
+    db = _make_db([rows, []])
+
+    with (
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._parse_street_house",
+            return_value=None,
+        ),
+        patch("app.tasks.backfill_listings_coords_geoportal._geoportal_house_match") as mock_geo,
+    ):
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    assert res.no_address == 1
+    assert res.matched == 0
+    mock_geo.assert_not_called()
+
+
+# ── idempotency ───────────────────────────────────────────────────────────────
+
+
+def test_update_idempotent_where_lat_is_null() -> None:
+    """UPDATE содержит WHERE lat IS NULL — повторный прогон не затрёт существующие coords."""
+    rows = [{"id": 5, "address": "ул. Малышева, 30"}]
+    db = _make_db([rows, []])
+    # rowcount=0 симулирует «уже было заполнено»
+    db.execute.side_effect = None
+    calls_log: list = []
+
+    nested_ctx = MagicMock()
+    nested_ctx.__enter__ = MagicMock(return_value=None)
+    nested_ctx.__exit__ = MagicMock(return_value=False)
+    db.begin_nested.return_value = nested_ctx
+
+    select_page1 = MagicMock()
+    select_page1.mappings.return_value.all.return_value = rows
+    select_page2 = MagicMock()
+    select_page2.mappings.return_value.all.return_value = []
+
+    upd_already_filled = MagicMock()
+    upd_already_filled.rowcount = 0  # lat уже был заполнен
+
+    execute_returns = [select_page1, upd_already_filled, select_page2]
+
+    def _side(*args, **kwargs):
+        calls_log.append(str(args[0]))
+        return execute_returns.pop(0)
+
+    db.execute.side_effect = _side
+
+    with (
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._geoportal_house_match",
+            return_value=_HIT,
+        ),
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._parse_street_house",
+            return_value=("малышева", "30"),
+        ),
+    ):
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    # matched=1 (geoportal нашёл), но updated=0 (WHERE lat IS NULL вернул 0)
+    assert res.matched == 1
+    assert res.updated == 0
+    # Проверяем что UPDATE SQL содержит WHERE lat IS NULL
+    update_sqls = [s for s in calls_log if "UPDATE" in s]
+    assert len(update_sqls) == 1
+    assert "lat IS NULL" in update_sqls[0]
+
+
+# ── limit / sources ───────────────────────────────────────────────────────────
+
+
+def test_limit_stops_processing() -> None:
+    """limit=1 → берём не более 1 листинга."""
+    rows = [{"id": 10, "address": "ул. Малышева, 30"}]
+    db = _make_db([rows])  # только одна страница
+
+    with (
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._geoportal_house_match",
+            return_value=None,
+        ),
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._parse_street_house",
+            return_value=("малышева", "30"),
+        ),
+    ):
+        res = backfill_coords_from_geoportal(db, limit=1, batch_size=500)
+
+    assert res.candidates == 1
+
+
+def test_sources_filter_passed_to_sql() -> None:
+    """sources=['avito'] → SQL содержит source = ANY(...)."""
+    db = MagicMock()
+    sel = MagicMock()
+    sel.mappings.return_value.all.return_value = []
+    db.execute.return_value = sel
+
+    nested_ctx = MagicMock()
+    nested_ctx.__enter__ = MagicMock(return_value=None)
+    nested_ctx.__exit__ = MagicMock(return_value=False)
+    db.begin_nested.return_value = nested_ctx
+
+    backfill_coords_from_geoportal(db, sources=["avito"], batch_size=500)
+
+    first_call = db.execute.call_args_list[0]
+    sql_text = str(first_call.args[0])
+    params = first_call.args[1]
+    assert "ANY" in sql_text
+    assert params["sources"] == ["avito"]
+
+
+# ── error handling ────────────────────────────────────────────────────────────
+
+
+def test_geoportal_match_exception_counted_as_error() -> None:
+    """_geoportal_house_match raises → errors++, обработка продолжается."""
+    rows = [
+        {"id": 20, "address": "ул. Сломанная, 1"},
+        {"id": 21, "address": "ул. Малышева, 30"},
+    ]
+    db = _make_db([rows, []])
+
+    side_effects = [RuntimeError("DB error"), _HIT]
+
+    with (
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._geoportal_house_match",
+            side_effect=side_effects,
+        ),
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._parse_street_house",
+            return_value=("малышева", "30"),
+        ),
+    ):
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    assert res.errors == 1
+    assert res.matched == 1
+    assert res.candidates == 2
+
+
+def test_empty_select_returns_zero_result() -> None:
+    """Нет листингов с lat IS NULL → нулевой результат."""
+    db = _make_db([[]])
+
+    with patch("app.tasks.backfill_listings_coords_geoportal._geoportal_house_match") as mock_geo:
+        res = backfill_coords_from_geoportal(db, batch_size=500)
+
+    assert isinstance(res, BackfillCoordsResult)
+    assert res.candidates == 0
+    assert res.matched == 0
+    assert res.updated == 0
+    mock_geo.assert_not_called()
+
+
+# ── batch pagination ──────────────────────────────────────────────────────────
+
+
+def test_batch_pagination_two_pages() -> None:
+    """batch_size=2: 2+1 листингов → 2 страницы SELECT, 3 листинга кандидата."""
+    page1 = [
+        {"id": 30, "address": "ул. Малышева, 30"},
+        {"id": 31, "address": "ул. Ленина, 10"},
+    ]
+    page2 = [{"id": 32, "address": "ул. Горького, 5"}]
+    db = _make_db([page1, page2, []])
+
+    with (
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._geoportal_house_match",
+            return_value=None,
+        ),
+        patch(
+            "app.tasks.backfill_listings_coords_geoportal._parse_street_house",
+            return_value=("малышева", "30"),
+        ),
+    ):
+        res = backfill_coords_from_geoportal(db, batch_size=2)
+
+    assert res.candidates == 3
+    assert res.no_match == 3