diff --git a/tradein-mvp/backend/app/tasks/backfill_listings_coords_geoportal.py b/tradein-mvp/backend/app/tasks/backfill_listings_coords_geoportal.py new file mode 100644 index 00000000..a31a3f2e --- /dev/null +++ b/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() diff --git a/tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py b/tradein-mvp/backend/tests/tasks/test_backfill_listings_coords_geoportal.py new file mode 100644 index 00000000..51afc3a2 --- /dev/null +++ b/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