feat(tradein): DaData /clean/address backfill для houses (PR Q3) #606

Merged
Light1YT merged 1 commit from feat/tradein-dadata-house-backfill into main 2026-05-27 14:43:52 +00:00
3 changed files with 1044 additions and 0 deletions
Showing only changes of commit e332052d0e - Show all commits

View file

@ -0,0 +1,62 @@
-- 070_houses_dadata_enrichment.sql
-- PR Q3: DaData backfill полей на таблице `houses`.
--
-- После PR Q1 (#604) сервис `dadata.clean_address()` живёт на проде и обогащает
-- адреса estimator-flow on-demand. Этот скрипт (`scripts/backfill_houses_dadata.py`)
-- проходит по `houses` daily 100/day (DaData demo quota) и заполняет:
-- - lat / lon (3801 rows без координат)
-- - cadastral_number (12437 rows — 100% NULL)
-- - house_fias_id — НОВАЯ колонка, симметрия с trade_in_estimates.house_fias_id (PR Q1)
--
-- Колонки dadata_qc_geo / dadata_qc_house / dadata_enriched_at — tracking:
-- помечают что попытка enrichment была сделана, даже если результат не записан
-- (quality codes 2/3/4 — слишком низкое разрешение, не overwrite coords).
-- Это позволяет:
-- 1) resume-safe backfill: `WHERE dadata_enriched_at IS NULL` пропускает обработанные
-- 2) re-enrich stale: при необходимости фильтр `dadata_enriched_at < NOW() - 30 days`
--
-- Все поля nullable + IF NOT EXISTS → миграция идемпотентна.
--
-- Dependencies: 009_houses.sql + 010_houses_alter.sql + 029_extend_matching_valuation_dynamics.sql
-- Apply after: 069_trade_in_estimates_dadata_fields.sql
BEGIN;
ALTER TABLE houses
ADD COLUMN IF NOT EXISTS house_fias_id text,
ADD COLUMN IF NOT EXISTS dadata_qc_geo integer,
ADD COLUMN IF NOT EXISTS dadata_qc_house integer,
ADD COLUMN IF NOT EXISTS dadata_enriched_at timestamptz;
-- Индекс на ФИАС для будущих JOIN'ов между houses ↔ trade_in_estimates (PR Q1).
CREATE INDEX IF NOT EXISTS houses_fias_id_idx
ON houses (house_fias_id)
WHERE house_fias_id IS NOT NULL;
-- Cadastral_number индекс уже создан в 029_extend_matching_valuation_dynamics.sql
-- (имя houses_cadastral_number_idx) и в 050_search_optimization.sql (houses_kadastr_idx2).
-- IF NOT EXISTS → no-op при повторном создании. Оставляем для idempotent docs.
CREATE INDEX IF NOT EXISTS houses_cadastral_number_idx
ON houses (cadastral_number)
WHERE cadastral_number IS NOT NULL;
-- Partial index для быстрого выбора кандидатов backfill'а:
-- `WHERE dadata_enriched_at IS NULL ORDER BY id`. После полного прохода
-- (12437 row → ~124 дня walltime) индекс схлопнется до пустого set.
CREATE INDEX IF NOT EXISTS houses_dadata_not_enriched_idx
ON houses (id)
WHERE dadata_enriched_at IS NULL;
COMMENT ON COLUMN houses.house_fias_id IS
'DaData UUID ФИАС дома (PR Q3, #604+#605 follow-up). '
'Заполняется backfill-скриптом scripts/backfill_houses_dadata.py.';
COMMENT ON COLUMN houses.dadata_qc_geo IS
'DaData qc_geo: 0=exact, 1=street, 2=settlement, 3=city. '
'Только qc_geo IN (0,1) триггерит overwrite lat/lon в backfill.';
COMMENT ON COLUMN houses.dadata_qc_house IS
'DaData qc_house: 2=в ФИАС, 10=на картах, иначе не найден.';
COMMENT ON COLUMN houses.dadata_enriched_at IS
'Когда последний раз пробовали enrich через DaData (NULL = ни разу). '
'Используется для resume-safe backfill — `WHERE dadata_enriched_at IS NULL`.';
COMMIT;

View file

@ -0,0 +1,518 @@
"""Backfill houses table via DaData /clean/address.
PR Q3 follow-up to PR Q1 (#604) — теперь, когда сервис `dadata.clean_address()`
живёт на проде, проходим по таблице `houses` daily 100/day (demo quota) и
дополняем недостающие координаты + кадастровые номера + ФИАС.
DB state on prod (verified 2026-05-27):
- houses: 12437 rows total
- lat IS NULL OR lon IS NULL: 3801 (нужны координаты)
- cadastral_number IS NULL: 12437 (100% все нужно дополнить)
- house_fias_id IS NULL: 12437 (колонка добавлена в migration 070)
Walltime: 12437 / 100 = ~124 дня. Acceptable это batch backfill, не realtime.
Priority order (--priority both, default):
1. NULL lat OR NULL lon (3801 rows) нужны coords для spatial match
2. NULL cadastral_number AND lat IS NOT NULL (~8636 rows) есть coords,
дополняем только cadnum / fias / qc-codes
`--priority coords` только #1, `--priority cadnum` — только #2.
Per-row outcome rules (после `clean_address` ответа):
- result.qc_geo IN (0, 1) AND result.lat IS NOT NULL точное / уличное совпадение,
UPDATE houses SET lat=, lon=, cadastral_number=, house_fias_id=,
dadata_qc_geo=, dadata_qc_house=, dadata_enriched_at=NOW()
- result.qc_geo IN (2, 3, 4) OR result IS None не overwrite координаты,
только UPDATE houses SET dadata_qc_geo=, dadata_qc_house=, dadata_enriched_at=NOW()
(фиксируем факт попытки resume пропустит row при следующем запуске)
Resume-safe: `WHERE dadata_enriched_at IS NULL` фильтрует уже обработанные.
Design choices:
- **Per-row SAVEPOINT** (`db.begin_nested()`) одна DaData / DB ошибка не должна
завалить весь batch (тот же паттерн что и в backfill_house_coords.py).
- **Rate limit**: 0.5s между requests (DaData demo allows ~5 req/sec, мы при limit=100
делаем 50 секунд непрерывной работы далеко от quota burst-detection).
- **Daily budget**: `--limit 100` default matches demo tier (100 req/day).
Usage:
DATABASE_URL=postgresql+psycopg://... \\
DADATA_API_TOKEN=xxx DADATA_API_SECRET=xxx \\
python -m scripts.backfill_houses_dadata --limit 100
Flags:
--limit N max rows to process (default 100)
--batch LABEL log label (default `dadata_YYYY-MM-DD`)
--dry-run log what would happen, no DB writes
--priority X coords | cadnum | both (default both)
"""
from __future__ import annotations
import argparse
import asyncio
import logging
import os
from dataclasses import dataclass, field
from datetime import date
from pathlib import Path
from sqlalchemy import text
from sqlalchemy.orm import Session
# Allow running both as `python -m scripts.backfill_houses_dadata` (preferred,
# матчит pattern из backfill_house_coords.py) и как stand-alone file.
try:
from app.core.db import SessionLocal # type: ignore[import-not-found]
except ImportError: # pragma: no cover — fallback for adhoc invocation
import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from app.core.db import SessionLocal
try:
from app.services.dadata import ( # type: ignore[import-not-found]
DadataAddressResult,
clean_address,
)
except ImportError: # pragma: no cover
import sys
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from app.services.dadata import DadataAddressResult, clean_address
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("backfill_houses_dadata")
# DaData demo tier: 100 requests/day, no burst limit documented but
# best practice — sleep между requests, чтобы не триггерить anti-flood.
# 0.5s = 2 req/sec → 50s walltime для full 100/day batch.
_REQUEST_DELAY_S = 0.5
# qc_geo: 0=exact, 1=street, 2=settlement, 3=city, 4=region, 5=unknown.
# Только 0/1 точны достаточно для overwrite coords дома.
_QC_GEO_ACCEPT = frozenset({0, 1})
# Логировать каждые N rows
_LOG_EVERY = 10
# ---------------------------------------------------------------------------
# Domain types
# ---------------------------------------------------------------------------
@dataclass
class HouseRow:
"""Минимальные поля из houses для enrichment loop'а.
`address` сырой адрес (приоритет houses.address, fallback на full_address).
"""
id: int
address: str
lat: float | None
lon: float | None
cadastral_number: str | None
@dataclass
class Stats:
"""Счётчики для финального summary log'а."""
processed: int = 0
enriched: int = 0 # qc_geo IN (0,1) — coords/cadnum/fias записаны
no_change: int = 0 # qc_geo IN (2,3,4) или canonical_address None — только tracking
failed: int = 0 # exception в clean_address или DB write
skipped: int = 0 # пустой address (нечего geocode'ить)
by_priority: dict[str, int] = field(default_factory=dict)
def bump(self, priority: str, outcome: str) -> None:
"""Внутренний учёт по priority bucket'у."""
bucket = self.by_priority.setdefault(
priority, {"enriched": 0, "no_change": 0, "failed": 0, "skipped": 0}
)
bucket[outcome] = bucket.get(outcome, 0) + 1
# ---------------------------------------------------------------------------
# Source-row query — приоритет: NULL coords → NULL cadnum
# ---------------------------------------------------------------------------
def _select_candidates(
db: Session, *, priority: str, limit: int
) -> list[tuple[HouseRow, str]]:
"""Возвращает список (HouseRow, priority_bucket) для enrichment.
priority:
- 'coords' только rows с NULL coords (max бизнес-value)
- 'cadnum' только rows с NOT NULL coords + NULL cadnum
- 'both' coords first, потом cadnum (FIFO в пределах buckets)
Фильтр `dadata_enriched_at IS NULL` resume-safe (не повторяем уже
обработанные rows). После full enrichment колонка станет NOT NULL у всех
rows которые мы успели тронуть, и партишен-индекс
`houses_dadata_not_enriched_idx` схлопнется.
"""
# Skip rows без какого-либо адреса — geocode'ить нечего.
address_filter = (
"(COALESCE(NULLIF(trim(address), ''), NULLIF(trim(full_address), '')) IS NOT NULL)"
)
out: list[tuple[HouseRow, str]] = []
remaining = limit
def _fetch(where: str, bucket: str, lim: int) -> list[tuple[HouseRow, str]]:
sql = text(
"SELECT id, "
" COALESCE(NULLIF(trim(address), ''), NULLIF(trim(full_address), '')) "
" AS address, "
" lat, lon, cadastral_number "
"FROM houses "
f"WHERE {where} "
" AND dadata_enriched_at IS NULL "
f" AND {address_filter} "
"ORDER BY id "
"LIMIT CAST(:lim AS int)"
)
rows = db.execute(sql, {"lim": lim}).mappings().all()
return [
(
HouseRow(
id=r["id"],
address=r["address"],
lat=r["lat"],
lon=r["lon"],
cadastral_number=r["cadastral_number"],
),
bucket,
)
for r in rows
]
if priority in ("coords", "both"):
out.extend(_fetch("(lat IS NULL OR lon IS NULL)", "coords", remaining))
remaining = limit - len(out)
if priority in ("cadnum", "both") and remaining > 0:
out.extend(
_fetch(
"(lat IS NOT NULL AND lon IS NOT NULL AND cadastral_number IS NULL)",
"cadnum",
remaining,
)
)
return out[:limit]
# ---------------------------------------------------------------------------
# DB writers
# ---------------------------------------------------------------------------
def _update_house_enriched(
db: Session,
*,
house_id: int,
result: DadataAddressResult,
) -> None:
"""UPDATE houses с полным набором полей (qc_geo IN (0,1) → точное матч).
Не overwrite'ит lat/lon если result.lat/lon None (DaData может вернуть
qc_geo=0 для города но без точки edge case). COALESCE'им через
`CASE WHEN :lat IS NOT NULL THEN :lat ELSE lat END` чтобы не ломать
существующие значения.
cadastral_number / house_fias_id overwrite только если DaData вернул
непустое значение (COALESCE с existing).
"""
db.execute(
text(
"UPDATE houses "
" SET lat = CASE WHEN CAST(:lat AS double precision) IS NOT NULL "
" THEN CAST(:lat AS double precision) "
" ELSE lat END, "
" lon = CASE WHEN CAST(:lon AS double precision) IS NOT NULL "
" THEN CAST(:lon AS double precision) "
" ELSE lon END, "
" cadastral_number = COALESCE(CAST(:cadnum AS text), cadastral_number), "
" house_fias_id = COALESCE(CAST(:fias AS text), house_fias_id), "
" dadata_qc_geo = CAST(:qc_geo AS integer), "
" dadata_qc_house = CAST(:qc_house AS integer), "
" dadata_enriched_at = NOW() "
" WHERE id = CAST(:id AS bigint)"
),
{
"id": house_id,
"lat": result.lat,
"lon": result.lon,
"cadnum": result.house_cadnum,
"fias": result.house_fias_id,
"qc_geo": result.qc_geo,
"qc_house": result.qc_house,
},
)
def _update_house_attempt(
db: Session,
*,
house_id: int,
qc_geo: int | None,
qc_house: int | None,
) -> None:
"""UPDATE только tracking-поля (qc_geo вне ACCEPT set, либо None result).
НЕ трогаем lat/lon/cadnum/fias DaData либо не нашёл точно (qc=2,3,4),
либо вообще не распознал адрес. Фиксируем `dadata_enriched_at` чтобы
resume не повторял эту row при следующем запуске.
"""
db.execute(
text(
"UPDATE houses "
" SET dadata_qc_geo = CAST(:qc_geo AS integer), "
" dadata_qc_house = CAST(:qc_house AS integer), "
" dadata_enriched_at = NOW() "
" WHERE id = CAST(:id AS bigint)"
),
{"id": house_id, "qc_geo": qc_geo, "qc_house": qc_house},
)
# ---------------------------------------------------------------------------
# Main loop
# ---------------------------------------------------------------------------
def _is_enriched_result(result: DadataAddressResult | None) -> bool:
"""Решает, можно ли overwrite'ить координаты + cadnum по этому result'у.
Условие: qc_geo IN (0, 1) AND lat / lon заданы.
Возврат False означает «записать только tracking, не overwrite payload».
"""
if result is None:
return False
if result.qc_geo not in _QC_GEO_ACCEPT:
return False
if result.lat is None or result.lon is None:
return False
return True
async def _run_backfill(
db: Session,
candidates: list[tuple[HouseRow, str]],
*,
batch: str,
dry_run: bool,
) -> Stats:
"""Главный цикл — для каждого candidate зовём DaData + UPDATE houses.
Per-row SAVEPOINT защищает batch от падений на одной row. Per-row commit
после успешного nested block resume picks up exactly где упали.
"""
stats = Stats()
for i, (row, priority) in enumerate(candidates, start=1):
result: DadataAddressResult | None = None
outcome = "no_change"
try:
# `clean_address` сам обрабатывает HTTP errors / quota / timeout
# и возвращает None — exception будет только при unexpected error.
result = await clean_address(row.address)
except Exception as exc: # defensive — одна ошибка не должна валить batch
stats.failed += 1
stats.bump(priority, "failed")
logger.warning(
"dadata_call failed for house_id=%s addr=%r: %s",
row.id,
row.address[:60],
exc,
)
# Не пишем в DB при exception — оставляем row для следующей попытки.
continue
if dry_run:
if _is_enriched_result(result):
outcome = "enriched"
logger.info(
"DRY-RUN house_id=%s [%s] WOULD enrich → "
"qc_geo=%s qc_house=%s lat=%s lon=%s cadnum=%s fias=%s",
row.id,
priority,
result.qc_geo if result else None,
result.qc_house if result else None,
result.lat if result else None,
result.lon if result else None,
result.house_cadnum if result else None,
result.house_fias_id if result else None,
)
else:
outcome = "no_change"
logger.info(
"DRY-RUN house_id=%s [%s] WOULD record attempt only → "
"qc_geo=%s qc_house=%s (result_present=%s)",
row.id,
priority,
result.qc_geo if result else None,
result.qc_house if result else None,
result is not None,
)
stats.processed += 1
if outcome == "enriched":
stats.enriched += 1
else:
stats.no_change += 1
stats.bump(priority, outcome)
else:
try:
with db.begin_nested():
if _is_enriched_result(result):
assert result is not None # narrowed by _is_enriched_result
_update_house_enriched(db, house_id=row.id, result=result)
outcome = "enriched"
stats.enriched += 1
else:
_update_house_attempt(
db,
house_id=row.id,
qc_geo=result.qc_geo if result else None,
qc_house=result.qc_house if result else None,
)
outcome = "no_change"
stats.no_change += 1
# Per-row commit чтобы resume picks up exactly where we crashed.
db.commit()
stats.processed += 1
stats.bump(priority, outcome)
except Exception as exc: # defensive — DB write failure isolation
db.rollback()
stats.failed += 1
stats.bump(priority, "failed")
logger.warning(
"db_write failed for house_id=%s: %s", row.id, exc
)
if i % _LOG_EVERY == 0:
logger.info(
"batch=%s progress %d/%d enriched=%d no_change=%d failed=%d",
batch,
i,
len(candidates),
stats.enriched,
stats.no_change,
stats.failed,
)
# Rate-limit: пауза только если ещё есть rows впереди.
if i < len(candidates):
await asyncio.sleep(_REQUEST_DELAY_S)
return stats
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"""argparse setup, выделено отдельно для testability."""
p = argparse.ArgumentParser(
description=(
"PR Q3 — backfill houses.lat/lon/cadastral_number/house_fias_id "
"через DaData /clean/address (demo tier, 100/day)."
),
)
p.add_argument(
"--limit",
type=int,
default=100,
help="Max rows to process (default 100 — matches DaData demo daily quota).",
)
p.add_argument(
"--batch",
default=f"dadata_{date.today().isoformat()}",
help="Log batch label. Не влияет на DB фильтры — только для логов.",
)
p.add_argument(
"--dry-run",
action="store_true",
help="Логирует что бы произошло, но не пишет в DB.",
)
p.add_argument(
"--priority",
choices=("coords", "cadnum", "both"),
default="both",
help=(
"Какой набор кандидатов обрабатывать. coords = только NULL coords, "
"cadnum = NOT NULL coords + NULL cadnum, both (default) = первый "
"приоритет coords."
),
)
return p.parse_args(argv)
async def main(argv: list[str] | None = None) -> int:
"""CLI entry point. Возвращает кол-во processed rows."""
args = _parse_args(argv)
# Сервис graceful'но disable'ится без creds, но script без них бессмысленен —
# лучше fail fast, чтобы cron-runs не съели DB-locks впустую.
token = os.environ.get("DADATA_API_TOKEN", "").strip()
secret = os.environ.get("DADATA_API_SECRET", "").strip()
if not token or not secret:
raise SystemExit(
"DADATA_API_TOKEN + DADATA_API_SECRET required — "
"DaData /clean/address без secret не работает."
)
logger.info(
"starting batch=%s limit=%s priority=%s dry_run=%s",
args.batch,
args.limit,
args.priority,
args.dry_run,
)
db = SessionLocal()
try:
candidates = _select_candidates(db, priority=args.priority, limit=args.limit)
logger.info(
"loaded candidates: %d (priority=%s)", len(candidates), args.priority
)
if not candidates:
logger.info(
"nothing to do — нет rows с dadata_enriched_at IS NULL для priority=%s",
args.priority,
)
return 0
stats = await _run_backfill(
db, candidates, batch=args.batch, dry_run=args.dry_run
)
logger.info(
"done: batch=%s processed=%d enriched=%d no_change=%d "
"failed=%d skipped=%d by_priority=%s",
args.batch,
stats.processed,
stats.enriched,
stats.no_change,
stats.failed,
stats.skipped,
stats.by_priority,
)
return stats.processed
finally:
db.close()
if __name__ == "__main__": # pragma: no cover
asyncio.run(main())

View file

@ -0,0 +1,464 @@
"""Unit tests for PR Q3 — backfill_houses_dadata.py.
Coverage:
- Happy path: row с NULL coords + qc_geo=0 UPDATE с полным набором полей.
- qc_geo=3 (city-level) tracking only, lat/lon не overwrite'аются.
- DaData returns None tracking only (qc_geo / qc_house = NULL).
- Per-row SAVEPOINT: exception от clean_address не валит batch.
- --dry-run: nothing written в DB.
- --limit N: батч обрезается до N даже если кандидатов больше.
- --priority coords / cadnum / both фильтр source-rows работает.
- _is_enriched_result accept / reject правила.
- main() с отсутствующими DADATA_API_TOKEN / SECRET SystemExit.
No real Postgres. Session MagicMock что записывает UPDATE и роутит SELECT
side-effects (тот же pattern что в test_backfill_house_coords.py /
test_backfill_listing_sources.py).
"""
from __future__ import annotations
import os
from unittest.mock import AsyncMock, MagicMock, patch
# Settings требует DSN на import — set dummy DSN до любого app.* import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
import pytest
from app.services.dadata import DadataAddressResult
from scripts.backfill_houses_dadata import (
HouseRow,
Stats,
_is_enriched_result,
_run_backfill,
_select_candidates,
_update_house_attempt,
_update_house_enriched,
main,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_dadata_result(
*,
qc_geo: int | None = 0,
qc_house: int | None = 2,
lat: float | None = 56.838,
lon: float | None = 60.586,
cadnum: str | None = "66:41:0704045:350",
fias: str | None = "abcdef00-aaaa-bbbb-cccc-1234567890ab",
canonical: str | None = "г Екатеринбург, ул Малышева, д 51",
) -> DadataAddressResult:
"""DadataAddressResult fixture с sensible defaults для tests."""
return DadataAddressResult(
canonical_address=canonical,
house_cadnum=cadnum,
house_fias_id=fias,
lat=lat,
lon=lon,
qc_geo=qc_geo,
qc_house=qc_house,
kladr_id=None,
okato=None,
oktmo=None,
metro=[],
raw={"test": True},
)
def _make_db_mock(
coords_sample: list[dict] | None = None,
cadnum_sample: list[dict] | None = None,
) -> tuple[MagicMock, list[dict]]:
"""MagicMock Session что:
- возвращает coords_sample для SELECT с `lat IS NULL OR lon IS NULL`
- возвращает cadnum_sample для SELECT с `cadastral_number IS NULL`
- записывает UPDATE houses в `updated` список (полные binds dict'ы)
- корректно работает с `db.begin_nested()` контекст-менеджером
"""
coords_sample = coords_sample or []
cadnum_sample = cadnum_sample or []
updated: list[dict] = []
db = MagicMock()
db.begin_nested.return_value.__enter__ = lambda self: self
db.begin_nested.return_value.__exit__ = lambda self, *a: False
def execute_side_effect(sql, params=None):
sql_str = str(sql)
result = MagicMock()
if "FROM houses" in sql_str and "lat IS NULL OR lon IS NULL" in sql_str:
lim = params.get("lim", len(coords_sample)) if params else len(coords_sample)
result.mappings.return_value.all.return_value = coords_sample[:lim]
elif "FROM houses" in sql_str and "cadastral_number IS NULL" in sql_str:
lim = params.get("lim", len(cadnum_sample)) if params else len(cadnum_sample)
result.mappings.return_value.all.return_value = cadnum_sample[:lim]
elif "UPDATE houses" in sql_str:
updated.append(dict(params) if params else {})
return result
db.execute.side_effect = execute_side_effect
db.commit = MagicMock()
db.rollback = MagicMock()
db.close = MagicMock()
return db, updated
# ---------------------------------------------------------------------------
# _is_enriched_result — accept / reject правила
# ---------------------------------------------------------------------------
def test_is_enriched_result_qc_geo_0_lat_lon_present_accepts():
res = _make_dadata_result(qc_geo=0, lat=56.838, lon=60.586)
assert _is_enriched_result(res) is True
def test_is_enriched_result_qc_geo_1_accepts():
"""qc_geo=1 (street-level) — также accept'аем для overwrite."""
res = _make_dadata_result(qc_geo=1, lat=56.838, lon=60.586)
assert _is_enriched_result(res) is True
def test_is_enriched_result_qc_geo_2_rejects():
"""qc_geo=2 (settlement) — слишком грубо для дома."""
res = _make_dadata_result(qc_geo=2)
assert _is_enriched_result(res) is False
def test_is_enriched_result_qc_geo_3_rejects():
"""qc_geo=3 (city) — точно не дом."""
res = _make_dadata_result(qc_geo=3)
assert _is_enriched_result(res) is False
def test_is_enriched_result_none_input_rejects():
assert _is_enriched_result(None) is False
def test_is_enriched_result_qc_geo_ok_but_no_coords_rejects():
"""Defensive: qc_geo=0 но lat/lon отсутствуют → не overwrite."""
res = _make_dadata_result(qc_geo=0, lat=None, lon=None)
assert _is_enriched_result(res) is False
# ---------------------------------------------------------------------------
# _update_house_enriched — bind shape
# ---------------------------------------------------------------------------
def test_update_house_enriched_passes_full_payload():
"""UPDATE houses содержит lat/lon/cadnum/fias/qc-codes и dadata_enriched_at = NOW()."""
db = MagicMock()
res = _make_dadata_result(
qc_geo=0, qc_house=2, lat=56.838, lon=60.586,
cadnum="66:41:0704045:350", fias="fias-uuid-here",
)
_update_house_enriched(db, house_id=42, result=res)
args, _kw = db.execute.call_args
sql_str = str(args[0])
binds = args[1]
assert "UPDATE houses" in sql_str
assert "dadata_enriched_at = NOW()" in sql_str
assert "cadastral_number" in sql_str
assert "house_fias_id" in sql_str
assert binds["id"] == 42
assert binds["lat"] == 56.838
assert binds["lon"] == 60.586
assert binds["cadnum"] == "66:41:0704045:350"
assert binds["fias"] == "fias-uuid-here"
assert binds["qc_geo"] == 0
assert binds["qc_house"] == 2
def test_update_house_attempt_writes_only_tracking_fields():
"""attempt-write не должен затрагивать lat/lon/cadnum/fias columns."""
db = MagicMock()
_update_house_attempt(db, house_id=7, qc_geo=3, qc_house=10)
args, _kw = db.execute.call_args
sql_str = str(args[0])
binds = args[1]
assert "UPDATE houses" in sql_str
assert "dadata_enriched_at = NOW()" in sql_str
# Не должно быть UPDATE на coord-payload поля
assert "lat =" not in sql_str
assert "lon =" not in sql_str
assert "cadastral_number =" not in sql_str
assert "house_fias_id =" not in sql_str
assert binds == {"id": 7, "qc_geo": 3, "qc_house": 10}
# ---------------------------------------------------------------------------
# _run_backfill — happy path qc_geo=0
# ---------------------------------------------------------------------------
async def test_run_backfill_qc_geo_0_writes_enriched_update():
"""Happy path — qc_geo=0 → UPDATE с полным payload, stats.enriched=1."""
row = HouseRow(id=1, address="ул Малышева 51", lat=None, lon=None, cadastral_number=None)
db, updated = _make_db_mock()
fake = _make_dadata_result(qc_geo=0)
with patch(
"scripts.backfill_houses_dadata.clean_address",
new=AsyncMock(return_value=fake),
):
stats = await _run_backfill(
db, [(row, "coords")], batch="b1", dry_run=False
)
assert stats.enriched == 1
assert stats.no_change == 0
assert stats.failed == 0
assert stats.processed == 1
assert len(updated) == 1
assert updated[0]["lat"] == 56.838
assert updated[0]["lon"] == 60.586
assert updated[0]["cadnum"] == "66:41:0704045:350"
assert updated[0]["qc_geo"] == 0
assert db.commit.call_count == 1
assert stats.by_priority["coords"]["enriched"] == 1
# ---------------------------------------------------------------------------
# _run_backfill — qc_geo=3 city-level → tracking-only
# ---------------------------------------------------------------------------
async def test_run_backfill_qc_geo_3_records_attempt_only():
"""qc_geo=3 → НЕ overwrite coords, только tracking UPDATE."""
row = HouseRow(
id=2, address="Екатеринбург (без улицы)", lat=None, lon=None, cadastral_number=None
)
db, updated = _make_db_mock()
fake = _make_dadata_result(qc_geo=3, qc_house=None, lat=None, lon=None, cadnum=None, fias=None)
with patch(
"scripts.backfill_houses_dadata.clean_address",
new=AsyncMock(return_value=fake),
):
stats = await _run_backfill(
db, [(row, "coords")], batch="b2", dry_run=False
)
assert stats.enriched == 0
assert stats.no_change == 1
assert len(updated) == 1
# Tracking-only UPDATE — нет lat/lon/cadnum binds
assert "lat" not in updated[0]
assert "lon" not in updated[0]
assert updated[0]["qc_geo"] == 3
assert stats.by_priority["coords"]["no_change"] == 1
# ---------------------------------------------------------------------------
# _run_backfill — DaData returns None (graceful)
# ---------------------------------------------------------------------------
async def test_run_backfill_dadata_returns_none_records_attempt():
"""clean_address возвращает None (HTTP error / unrecognized) → tracking only с NULL qc."""
row = HouseRow(id=3, address="несуществующая 99", lat=None, lon=None, cadastral_number=None)
db, updated = _make_db_mock()
with patch(
"scripts.backfill_houses_dadata.clean_address",
new=AsyncMock(return_value=None),
):
stats = await _run_backfill(
db, [(row, "coords")], batch="b3", dry_run=False
)
assert stats.no_change == 1
assert stats.enriched == 0
assert len(updated) == 1
assert updated[0]["qc_geo"] is None
assert updated[0]["qc_house"] is None
# ---------------------------------------------------------------------------
# Per-row SAVEPOINT — exception в clean_address не валит batch
# ---------------------------------------------------------------------------
async def test_run_backfill_clean_address_exception_isolated_to_row():
"""Один clean_address raise → этот row failed, остальные processed normally."""
rows = [
(HouseRow(id=10, address="ул Bad", lat=None, lon=None, cadastral_number=None), "coords"),
(HouseRow(id=11, address="ул Good 5", lat=None, lon=None, cadastral_number=None), "coords"),
]
db, updated = _make_db_mock()
good_result = _make_dadata_result(qc_geo=0)
call_count = {"n": 0}
async def fake_clean(addr):
call_count["n"] += 1
if call_count["n"] == 1:
raise RuntimeError("dadata blew up")
return good_result
with patch("scripts.backfill_houses_dadata.clean_address", side_effect=fake_clean):
stats = await _run_backfill(db, rows, batch="b4", dry_run=False)
assert stats.processed == 1 # только 2-й row дошёл до DB-write success
assert stats.failed == 1
assert stats.enriched == 1
# Только один UPDATE — для второго row
assert len(updated) == 1
assert updated[0]["id"] == 11
# ---------------------------------------------------------------------------
# --dry-run — no DB writes
# ---------------------------------------------------------------------------
async def test_run_backfill_dry_run_skips_db_writes():
"""dry_run=True → counters обновляются, но db.execute UPDATE не вызывается."""
row = HouseRow(id=20, address="ул Test 1", lat=None, lon=None, cadastral_number=None)
db, updated = _make_db_mock()
fake = _make_dadata_result(qc_geo=0)
with patch(
"scripts.backfill_houses_dadata.clean_address",
new=AsyncMock(return_value=fake),
):
stats = await _run_backfill(
db, [(row, "coords")], batch="dry", dry_run=True
)
assert stats.processed == 1
assert stats.enriched == 1
assert updated == [] # никаких UPDATE'ов
assert db.commit.call_count == 0 # никаких commit'ов
# ---------------------------------------------------------------------------
# _select_candidates — priority filter работает
# ---------------------------------------------------------------------------
def test_select_candidates_coords_priority_only_fetches_null_coords():
"""priority=coords → запрашивается только `lat IS NULL OR lon IS NULL` SELECT."""
coords_rows = [
{"id": 1, "address": "ул A 1", "lat": None, "lon": None, "cadastral_number": None},
{"id": 2, "address": "ул B 2", "lat": None, "lon": None, "cadastral_number": None},
]
db, _updated = _make_db_mock(coords_sample=coords_rows)
out = _select_candidates(db, priority="coords", limit=10)
assert len(out) == 2
assert all(p == "coords" for _row, p in out)
assert out[0][0].id == 1
assert out[1][0].id == 2
def test_select_candidates_cadnum_priority_only_fetches_null_cadnum():
"""priority=cadnum → только NOT NULL coords + NULL cadastral_number rows."""
cadnum_rows = [
{"id": 50, "address": "ул C 3", "lat": 56.8, "lon": 60.6, "cadastral_number": None},
]
db, _updated = _make_db_mock(cadnum_sample=cadnum_rows)
out = _select_candidates(db, priority="cadnum", limit=10)
assert len(out) == 1
assert out[0][1] == "cadnum"
assert out[0][0].id == 50
def test_select_candidates_both_priority_fetches_coords_first_then_cadnum():
"""priority=both → сначала NULL coords, потом NULL cadnum (в пределах limit)."""
coords_rows = [
{"id": 1, "address": "ул A 1", "lat": None, "lon": None, "cadastral_number": None},
]
cadnum_rows = [
{"id": 100, "address": "ул C 3", "lat": 56.8, "lon": 60.6, "cadastral_number": None},
]
db, _updated = _make_db_mock(coords_sample=coords_rows, cadnum_sample=cadnum_rows)
out = _select_candidates(db, priority="both", limit=10)
assert len(out) == 2
# coords bucket первый
assert out[0][1] == "coords"
assert out[0][0].id == 1
assert out[1][1] == "cadnum"
assert out[1][0].id == 100
# ---------------------------------------------------------------------------
# --limit caps batch size
# ---------------------------------------------------------------------------
async def test_main_respects_limit(monkeypatch):
"""main(--limit 1) при 3 кандидатах → только 1 row processed."""
coords_rows = [
{"id": 1, "address": "ул A 1", "lat": None, "lon": None, "cadastral_number": None},
{"id": 2, "address": "ул B 2", "lat": None, "lon": None, "cadastral_number": None},
{"id": 3, "address": "ул C 3", "lat": None, "lon": None, "cadastral_number": None},
]
db, updated = _make_db_mock(coords_sample=coords_rows)
monkeypatch.setenv("DADATA_API_TOKEN", "TEST_TOKEN")
monkeypatch.setenv("DADATA_API_SECRET", "TEST_SECRET")
fake = _make_dadata_result(qc_geo=0)
with (
patch("scripts.backfill_houses_dadata.SessionLocal", return_value=db),
patch(
"scripts.backfill_houses_dadata.clean_address",
new=AsyncMock(return_value=fake),
),
# Уберём sleep чтобы test не висел 0.5s
patch("scripts.backfill_houses_dadata.asyncio.sleep", new=AsyncMock()),
):
n = await main(["--limit", "1", "--batch", "test_limit"])
assert n == 1
assert len(updated) == 1
# ---------------------------------------------------------------------------
# Missing creds → SystemExit
# ---------------------------------------------------------------------------
async def test_main_requires_dadata_creds(monkeypatch):
"""Без DADATA_API_TOKEN + DADATA_API_SECRET → SystemExit ещё до DB connect."""
monkeypatch.delenv("DADATA_API_TOKEN", raising=False)
monkeypatch.delenv("DADATA_API_SECRET", raising=False)
with pytest.raises(SystemExit):
await main(["--limit", "1"])
async def test_main_requires_secret_not_just_token(monkeypatch):
"""Только TOKEN (без SECRET) → /clean/address не сработает → fail fast."""
monkeypatch.setenv("DADATA_API_TOKEN", "TEST_TOKEN")
monkeypatch.delenv("DADATA_API_SECRET", raising=False)
with pytest.raises(SystemExit):
await main(["--limit", "1"])
# ---------------------------------------------------------------------------
# Stats summary contains by_priority breakdown
# ---------------------------------------------------------------------------
def test_stats_bump_buckets_by_priority():
"""Stats.bump поддерживает несколько priority bucket'ов одновременно."""
s = Stats()
s.bump("coords", "enriched")
s.bump("coords", "enriched")
s.bump("cadnum", "no_change")
assert s.by_priority["coords"]["enriched"] == 2
assert s.by_priority["cadnum"]["no_change"] == 1
# Не существующие outcome'ы должны быть 0
assert s.by_priority["coords"]["failed"] == 0