Light1YT
e332052d0e
Migration 070 — добавляет houses.house_fias_id + dadata_qc_geo / qc_house / enriched_at для tracking. Скрипт scripts/backfill_houses_dadata.py проходит по ~3.8k rows с NULL coords + ~8.6k с NULL cadnum daily 100/day (DaData demo quota → ~124 дня walltime для полного прохода). Per-row SAVEPOINT, resume-safe (фильтр dadata_enriched_at IS NULL), qc_geo IN (0,1) → overwrite coord-payload, иначе только tracking. --priority coords|cadnum|both, --dry-run, --limit N (default 100). 20 тестов покрывают happy path qc_geo=0, qc_geo=3 city-level skip, DaData None response, per-row exception isolation, --dry-run, --limit, --priority filter, и missing-creds fail-fast.
518 lines
20 KiB
Python
518 lines
20 KiB
Python
"""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())
|