lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein): fill houses.year_built/material_walls from ДОМ.РФ КР1.1/1.2 + zhkh_year, propagate to listings (#2013) #2476

Merged
bot-backend merged 1 commit from feat/2013-domrf-year-backfill into main 2026-07-12 13:12:08 +00:00
Conversation 0 Commits 1 Files changed 4 +1059
4 changed files with 1059 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

479
tradein-mvp/backend/app/services/domrf_kapremont_loader.py Normal file
View file

@ -0,0 +1,479 @@
"""ДОМ.РФ капремонт open data loader: houses.year_built/material_walls (issue #2013).
CONTEXT: houses.year_built заполнен только на 38%, houses.material_walls — на 0%.
Существующий ГИС-ЖКХ loader (zhkh_flats_loader.py, мигр. 146/149) заполняет
houses.zhkh_year (70%) и houses.zhkh_floors (70%), НО никогда не копирует их в
houses.year_built/total_floors — это основной пробел. material_walls вообще ни разу
не заполнялся никаким источником. Estimator (app/services/estimator.py) фильтрует
когорту по `year_built BETWEEN ...` — реальные годы напрямую двигают точность оценки.
ИСТОЧНИК: ДОМ.РФ капремонт open data (free, no auth), region 66 (Свердловская обл.):
КР1.1 house registry (export/190) — zip → CSV, delimiter ';', UTF-8 BOM. Колонки:
mkd_code, houseguid (ФИАС GUID дома), address, commission_year (год ввода, int),
total_sq (decimal, запятая — «909,80»), number_floors_max (int), + служебные.
КР1.2 constructive elements (export/275) — zip → CSV, тот же delimiter/encoding.
LONG FORMAT: одна строка на конструктивный элемент на mkd_code; wall_material
заполнен ТОЛЬКО на строке construction_element_type='фасад'. Чтобы получить
материал стен на дом — группируем по mkd_code, берём первую строку с непустым
wall_material.
TLS: домен домрф.рф отдаёт RU-сертификат, который httpx с default trust store не
верифицирует ("unable to get local issuer certificate") — та же ситуация, что и у
sber_index.py. Открытые неавторизованные данные, поэтому используем verify=False,
как остальные RU-gov loader'ы в этом репо.
Матч staging→houses: domrf_kapremont.houseguid = COALESCE(houses.gar_house_guid,
houses.house_fias_id, houses.zhkh_house_guid) — простой приоритетный COALESCE-джойн
(без multi-attempt fallback-если-нет-матча — see PR discussion, "keep it simple/safe").
psycopg v3: SQL через `text(...)` использует `CAST(:x AS type)`, НИКОГДА `:x::type`.
"""
from __future__ import annotations
import csv
import io
import logging
import tempfile
import zipfile
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
import httpx
from sqlalchemy import text
from sqlalchemy.orm import Session
logger = logging.getLogger(__name__)
# ─────────────────────────────────────────────────────────────────────────────
# Константы
# ─────────────────────────────────────────────────────────────────────────────
# export/190 (КР1.1) и export/275 (КР1.2) без доп. параметров отдают Свердловскую
# обл. (region 66) — подтверждено инспекцией скачанных файлов. Для других регионов
# понадобится region-gid параметр (не реализовано — вне скоупа #2013).
KR11_URL = "https://xn--80adsazqn.xn--p1aee.xn--p1ai/opendata/export/190"
KR12_URL = "https://xn--80adsazqn.xn--p1aee.xn--p1ai/opendata/export/275"
DOWNLOAD_TIMEOUT_SEC = 180
# UPSERT в staging чанками (в одном SAVEPOINT), как zhkh_flats_loader.SAVEPOINT_CHUNK —
# сбойный чанк откатывается изолированно, остальные доезжают.
UPSERT_CHUNK_SIZE = 500
# ─────────────────────────────────────────────────────────────────────────────
# Чистые хелперы парсинга (юнит-тестируются без сети/БД)
# ─────────────────────────────────────────────────────────────────────────────
def parse_decimal_comma(raw: str | None) -> float | None:
"""«909,80» / «909.80» / «» / None → float|None. Запятая — decimal separator ДОМ.РФ CSV."""
if raw is None:
return None
s = raw.strip()
if not s:
return None
try:
return float(s.replace(",", "."))
except ValueError:
return None
def parse_int_field(raw: str | None) -> int | None:
"""Устойчивый str → int|None (commission_year / number_floors_max).
Пусто/None/нечисло → None. Терпит decimal-строки («5,0») через float-фоллбек —
ДОМ.РФ CSV в принципе может так отдать целочисленные поля.
"""
if raw is None:
return None
s = raw.strip()
if not s:
return None
try:
return int(s)
except ValueError:
pass
try:
return int(float(s.replace(",", ".")))
except ValueError:
return None
@dataclass(slots=True)
class Kr11Row:
"""Одна строка КР1.1 house registry (house-per-row)."""
mkd_code: str
houseguid: str | None
address: str | None
commission_year: int | None
number_floors_max: int | None
total_sq: float | None
@dataclass(slots=True)
class DomrfHouse:
"""Объединённая КР1.1 + КР1.2(wall_material) строка — готова к UPSERT в staging."""
mkd_code: str
houseguid: str | None
address: str | None
commission_year: int | None
number_floors_max: int | None
total_sq: float | None
wall_material: str | None
def _open_csv(path: str | Path) -> Iterator[dict[str, str]]:
"""csv.DictReader по ДОМ.РФ CSV: delimiter ';', UTF-8 BOM (utf-8-sig)."""
with open(path, encoding="utf-8-sig", newline="") as f:
reader = csv.DictReader(f, delimiter=";")
yield from reader
def parse_kr11_csv(path: str | Path) -> dict[str, Kr11Row]:
"""КР1.1 → dict по mkd_code. Строки без mkd_code пропускаются.
Дубликат mkd_code (не ожидается — mkd_code PK в реестре ДОМ.РФ) — последняя
строка побеждает (совпадает с семантикой ON CONFLICT DO UPDATE ниже).
"""
rows: dict[str, Kr11Row] = {}
for raw in _open_csv(path):
mkd_code = (raw.get("mkd_code") or "").strip()
if not mkd_code:
continue
rows[mkd_code] = Kr11Row(
mkd_code=mkd_code,
houseguid=(raw.get("houseguid") or "").strip() or None,
address=(raw.get("address") or "").strip() or None,
commission_year=parse_int_field(raw.get("commission_year")),
number_floors_max=parse_int_field(raw.get("number_floors_max")),
total_sq=parse_decimal_comma(raw.get("total_sq")),
)
return rows
def parse_kr12_wall_materials(path: str | Path) -> dict[str, str]:
"""КР1.2 (long-format) → dict mkd_code → wall_material.
wall_material непустой ТОЛЬКО на строке construction_element_type='фасад' —
группировка не по этому полю, а просто "первая непустая wall_material на
mkd_code" (устойчиво даже если разметка типа элемента когда-то изменится).
"""
materials: dict[str, str] = {}
for raw in _open_csv(path):
mkd_code = (raw.get("mkd_code") or "").strip()
wall = (raw.get("wall_material") or "").strip()
if mkd_code and wall and mkd_code not in materials:
materials[mkd_code] = wall
return materials
def build_domrf_houses(
kr11: dict[str, Kr11Row], wall_materials: dict[str, str]
) -> list[DomrfHouse]:
"""КР1.1 rows + КР1.2 wall_material lookup → список DomrfHouse (для UPSERT)."""
return [
DomrfHouse(
mkd_code=row.mkd_code,
houseguid=row.houseguid,
address=row.address,
commission_year=row.commission_year,
number_floors_max=row.number_floors_max,
total_sq=row.total_sq,
wall_material=wall_materials.get(row.mkd_code),
)
for row in kr11.values()
]
# ─────────────────────────────────────────────────────────────────────────────
# HTTP: скачивание zip + извлечение CSV
# ─────────────────────────────────────────────────────────────────────────────
def _download_zip(url: str, *, client: httpx.Client) -> bytes:
resp = client.get(url, timeout=DOWNLOAD_TIMEOUT_SEC)
resp.raise_for_status()
return resp.content
def _extract_csv_from_zip(data: bytes, dest_dir: Path) -> Path:
"""Распаковывает первый *.csv из zip-байтов в dest_dir, возвращает путь."""
with zipfile.ZipFile(io.BytesIO(data)) as zf:
csv_names = [n for n in zf.namelist() if n.lower().endswith(".csv")]
if not csv_names:
raise ValueError(f"zip не содержит .csv записей: {zf.namelist()!r}")
extracted = zf.extract(csv_names[0], dest_dir)
return Path(extracted)
def fetch_domrf_csvs(dest_dir: Path, *, client: httpx.Client) -> tuple[Path, Path]:
"""Скачивает КР1.1 + КР1.2 zip'ы, распаковывает в dest_dir. Возвращает (kr11, kr12)."""
kr11_path = _extract_csv_from_zip(_download_zip(KR11_URL, client=client), dest_dir)
kr12_path = _extract_csv_from_zip(_download_zip(KR12_URL, client=client), dest_dir)
return kr11_path, kr12_path
# ─────────────────────────────────────────────────────────────────────────────
# UPSERT staging (domrf_kapremont, мигр. 176)
# ─────────────────────────────────────────────────────────────────────────────
_UPSERT_SQL = text(
"""
INSERT INTO domrf_kapremont (
mkd_code, houseguid, address, commission_year,
number_floors_max, total_sq, wall_material, loaded_at
)
VALUES (
CAST(:mkd_code AS text), CAST(:houseguid AS text), CAST(:address AS text),
CAST(:commission_year AS smallint), CAST(:number_floors_max AS int),
CAST(:total_sq AS numeric), CAST(:wall_material AS text), now()
)
ON CONFLICT (mkd_code) DO UPDATE SET
houseguid = EXCLUDED.houseguid,
address = EXCLUDED.address,
commission_year = EXCLUDED.commission_year,
number_floors_max = EXCLUDED.number_floors_max,
total_sq = EXCLUDED.total_sq,
wall_material = EXCLUDED.wall_material,
loaded_at = now()
WHERE domrf_kapremont.houseguid IS DISTINCT FROM EXCLUDED.houseguid
OR domrf_kapremont.address IS DISTINCT FROM EXCLUDED.address
OR domrf_kapremont.commission_year IS DISTINCT FROM EXCLUDED.commission_year
OR domrf_kapremont.number_floors_max IS DISTINCT FROM EXCLUDED.number_floors_max
OR domrf_kapremont.total_sq IS DISTINCT FROM EXCLUDED.total_sq
OR domrf_kapremont.wall_material IS DISTINCT FROM EXCLUDED.wall_material
"""
)
def _chunk_houses(items: list[DomrfHouse], size: int) -> Iterator[list[DomrfHouse]]:
for i in range(0, len(items), size):
yield items[i : i + size]
def upsert_domrf_kapremont(
db: Session, houses: list[DomrfHouse], *, chunk_size: int = UPSERT_CHUNK_SIZE
) -> int:
"""UPSERT списка DomrfHouse в domrf_kapremont, чанками по SAVEPOINT. Не коммитит (caller).
Идемпотентно (IS DISTINCT FROM gate в _UPSERT_SQL — повторный прогон с теми же
данными не трогает уже актуальные строки). Сбойный чанк откатывается изолированно
(та же схема, что zhkh_flats_loader.match_houses_to_zhkh).
"""
upserted = 0
for chunk in _chunk_houses(houses, chunk_size):
try:
with db.begin_nested():
for h in chunk:
res = db.execute(
_UPSERT_SQL,
{
"mkd_code": h.mkd_code,
"houseguid": h.houseguid,
"address": h.address,
"commission_year": h.commission_year,
"number_floors_max": h.number_floors_max,
"total_sq": h.total_sq,
"wall_material": h.wall_material,
},
)
upserted += res.rowcount
except Exception:
logger.warning(
"domrf_kapremont upsert: чанк из %d строк сбойнул (откат savepoint)",
len(chunk),
exc_info=True,
)
return upserted
def load_domrf_kapremont(
db: Session,
*,
kr11_path: str | Path | None = None,
kr12_path: str | Path | None = None,
work_dir: str | Path | None = None,
chunk_size: int = UPSERT_CHUNK_SIZE,
dry_run: bool = False,
) -> dict[str, int]:
"""Скачивает (если пути не заданы) КР1.1+КР1.2, парсит, UPSERT в domrf_kapremont.
kr11_path/kr12_path — локальные CSV (пропустить скачивание; используется тестами
и ops-дебагом). work_dir — куда распаковывать скачанные zip (по умолчанию —
временный каталог, удаляется после загрузки). dry_run — парс происходит, но НИ
ОДНОЙ записи в БД не делается. Не коммитит (caller).
"""
tmp_ctx: tempfile.TemporaryDirectory[str] | None = None
if kr11_path is None or kr12_path is None:
if work_dir is not None:
dest = Path(work_dir)
dest.mkdir(parents=True, exist_ok=True)
else:
tmp_ctx = tempfile.TemporaryDirectory()
dest = Path(tmp_ctx.name)
# verify=False: домрф.рф отдаёт RU-сертификат, не верифицируемый дефолтным
# trust store'ом (та же ситуация, что sber_index.py). Публичные open data,
# без auth/PII — приемлемо skip'нуть TLS-верификацию.
with httpx.Client(timeout=DOWNLOAD_TIMEOUT_SEC, verify=False) as client:
if kr11_path is None:
kr11_path = _extract_csv_from_zip(_download_zip(KR11_URL, client=client), dest)
if kr12_path is None:
kr12_path = _extract_csv_from_zip(_download_zip(KR12_URL, client=client), dest)
try:
kr11 = parse_kr11_csv(kr11_path)
wall_materials = parse_kr12_wall_materials(kr12_path)
houses = build_domrf_houses(kr11, wall_materials)
upserted = 0 if dry_run else upsert_domrf_kapremont(db, houses, chunk_size=chunk_size)
finally:
if tmp_ctx is not None:
tmp_ctx.cleanup()
result = {
"kr11_rows": len(kr11),
"kr12_wall_rows": len(wall_materials),
"houses_built": len(houses),
"upserted": upserted,
}
logger.info("domrf_kapremont load DONE (dry_run=%s): %s", dry_run, result)
return result
# ─────────────────────────────────────────────────────────────────────────────
# Backfill houses.year_built/material_walls/total_floors
# ─────────────────────────────────────────────────────────────────────────────
# Шаг 1: дома с domrf-матчем — year_built/material_walls/total_floors из domrf,
# С ЗАФОЛЖЕННЫМ zhkh_year/zhkh_floors фоллбеком внутри ТОЙ ЖЕ строки COALESCE
# (COALESCE(year_built, domrf.commission_year, zhkh_year) — ровно как в issue).
# Только заполняет NULL — никогда не перезаписывает существующий non-null year_built.
_BACKFILL_HOUSES_FROM_DOMRF_SQL = text(
"""
UPDATE houses h
SET year_built = COALESCE(h.year_built, d.commission_year, h.zhkh_year),
material_walls = COALESCE(h.material_walls, d.wall_material),
total_floors = COALESCE(h.total_floors, d.number_floors_max, h.zhkh_floors)
FROM domrf_kapremont d
WHERE d.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid)
AND (
(h.year_built IS NULL AND COALESCE(d.commission_year, h.zhkh_year) IS NOT NULL)
OR (h.material_walls IS NULL AND d.wall_material IS NOT NULL)
OR (h.total_floors IS NULL AND COALESCE(d.number_floors_max, h.zhkh_floors) IS NOT NULL)
)
"""
)
_BACKFILL_HOUSES_FROM_DOMRF_COUNT_SQL = text(
"""
SELECT count(*)
FROM houses h
JOIN domrf_kapremont d
ON d.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid)
WHERE (h.year_built IS NULL AND COALESCE(d.commission_year, h.zhkh_year) IS NOT NULL)
OR (h.material_walls IS NULL AND d.wall_material IS NOT NULL)
OR (h.total_floors IS NULL AND COALESCE(d.number_floors_max, h.zhkh_floors) IS NOT NULL)
"""
)
# Шаг 2: ЛЮБОЙ дом (включая без domrf-матча вообще — шаг 1 INNER JOIN их не трогает)
# добивается zhkh_year/zhkh_floors, если ещё NULL. Домам, уже тронутым шагом 1, шаг 2
# ничего не меняет: их COALESCE в шаге 1 уже включал zhkh_* как fallback, поэтому
# гейт `IS NULL AND zhkh_* IS NOT NULL` здесь для них ложен — шаги взаимоисключающи
# (не двойной счёт при суммировании houses_updated).
_BACKFILL_HOUSES_ZHKH_FALLBACK_SQL = text(
"""
UPDATE houses
SET year_built = COALESCE(year_built, zhkh_year),
total_floors = COALESCE(total_floors, zhkh_floors)
WHERE (year_built IS NULL AND zhkh_year IS NOT NULL)
OR (total_floors IS NULL AND zhkh_floors IS NOT NULL)
"""
)
_BACKFILL_HOUSES_ZHKH_FALLBACK_COUNT_SQL = text(
"""
SELECT count(*)
FROM houses
WHERE (year_built IS NULL AND zhkh_year IS NOT NULL)
OR (total_floors IS NULL AND zhkh_floors IS NOT NULL)
"""
)
def backfill_houses_from_domrf(db: Session, *, dry_run: bool = False) -> dict[str, int]:
"""COALESCE(year_built, domrf.commission_year, zhkh_year) + material_walls + total_floors.
Двухшаговый UPDATE (мирроринг zhkh guid-match → cadastre-fallback паттерна из
zhkh_flats_loader.py):
1) дома с domrf-матчем (COALESCE(gar_house_guid, house_fias_id, zhkh_house_guid) =
domrf_kapremont.houseguid) — заполняются из domrf, с zhkh_year/zhkh_floors
фоллбеком внутри той же COALESCE.
2) ЛЮБОЙ дом (в т.ч. без domrf-матча) — добивает year_built/total_floors из
zhkh_year/zhkh_floors, если шаг 1 их не тронул.
Только COALESCE (заполняет NULL, никогда не перезаписывает existing non-null).
Не коммитит (caller). dry_run — ноль записей, только SELECT count(*) по тем же
предикатам.
"""
if dry_run:
domrf_matched = db.execute(_BACKFILL_HOUSES_FROM_DOMRF_COUNT_SQL).scalar_one()
zhkh_fallback = db.execute(_BACKFILL_HOUSES_ZHKH_FALLBACK_COUNT_SQL).scalar_one()
result = {
"domrf_matched": domrf_matched,
"zhkh_fallback": zhkh_fallback,
"houses_updated": 0,
}
logger.info("backfill_houses_from_domrf DRY-RUN: %s", result)
return result
domrf_updated = db.execute(_BACKFILL_HOUSES_FROM_DOMRF_SQL).rowcount
zhkh_updated = db.execute(_BACKFILL_HOUSES_ZHKH_FALLBACK_SQL).rowcount
result = {
"domrf_matched": domrf_updated,
"zhkh_fallback": zhkh_updated,
"houses_updated": domrf_updated + zhkh_updated,
}
logger.info("backfill_houses_from_domrf DONE: %s", result)
return result
# ─────────────────────────────────────────────────────────────────────────────
# Propagate houses.year_built → listings.year_built
# ─────────────────────────────────────────────────────────────────────────────
# Только NULL listings.year_built, только когда houses.year_built уже известен —
# никогда не перезаписывает существующий listings.year_built (source-provided данные
# приоритетнее houses-агрегата).
_PROPAGATE_LISTINGS_YEAR_SQL = text(
"""
UPDATE listings l
SET year_built = h.year_built
FROM houses h
WHERE l.house_id_fk = h.id
AND l.year_built IS NULL
AND h.year_built IS NOT NULL
"""
)
_PROPAGATE_LISTINGS_YEAR_COUNT_SQL = text(
"""
SELECT count(*)
FROM listings l
JOIN houses h ON l.house_id_fk = h.id
WHERE l.year_built IS NULL
AND h.year_built IS NOT NULL
"""
)
def propagate_listings_year_from_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]:
"""UPDATE listings.year_built = houses.year_built где listing.year_built ещё NULL.
Не коммитит (caller). dry_run — ноль записей, только SELECT count(*) по тому же
предикату (ключ результата — would_update вместо listings_updated).
"""
if dry_run:
would_update = db.execute(_PROPAGATE_LISTINGS_YEAR_COUNT_SQL).scalar_one()
logger.info("propagate_listings_year_from_houses DRY-RUN: would_update=%d", would_update)
return {"listings_updated": 0, "would_update": would_update}
updated = db.execute(_PROPAGATE_LISTINGS_YEAR_SQL).rowcount
logger.info("propagate_listings_year_from_houses DONE: listings_updated=%d", updated)
return {"listings_updated": updated}

124
tradein-mvp/backend/app/tasks/domrf_kapremont_load.py Normal file
View file

@ -0,0 +1,124 @@
"""CLI: ДОМ.РФ капремонт open data → houses.year_built/material_walls (issue #2013).
Тянет КР1.1 (house registry) + КР1.2 (constructive elements) с открытых данных
ДОМ.РФ (region 66), UPSERT'ит в staging `domrf_kapremont` (мигр. 176), затем:
1. backfill_houses_from_domrf — COALESCE(houses.year_built, domrf.commission_year,
houses.zhkh_year) + material_walls + total_floors (fold-in уже загруженного
ГИС-ЖКХ года — houses без domrf-матча тоже выигрывают).
2. propagate_listings_year_from_houses — UPDATE listings.year_built = houses.year_built
там, где listing ещё NULL. Именно это открывает estimator'у (year_built BETWEEN
когорта-фильтр) реальные годы на источниках вроде avito (0% year_built, 52.5%
linked к houses).
Запуск из контейнера tradein-backend (у прода есть интернет к домрф.рф):
python -m app.tasks.domrf_kapremont_load # полный прогон
python -m app.tasks.domrf_kapremont_load --dry-run # без записи, только подсчёт
python -m app.tasks.domrf_kapremont_load --load-only # только staging, без backfill
python -m app.tasks.domrf_kapremont_load --backfill-only # staging уже загружена — только
# backfill+propagate
python -m app.tasks.domrf_kapremont_load \\
--kr11-path /tmp/kr11.csv --kr12-path /tmp/kr12.csv # локальные CSV, без скачивания
В режиме --dry-run скачивание/парс (если применимо) происходят, но НИ ОДНОЙ записи
в БД не делается.
"""
from __future__ import annotations
import argparse
import logging
from app.core.db import SessionLocal
from app.services.domrf_kapremont_loader import (
backfill_houses_from_domrf,
load_domrf_kapremont,
propagate_listings_year_from_houses,
)
logger = logging.getLogger(__name__)
def build_parser() -> argparse.ArgumentParser:
"""Парсер CLI (вынесен для тестируемости флагов без запуска main)."""
parser = argparse.ArgumentParser(
description=(
"ДОМ.РФ капремонт loader: КР1.1+КР1.2 → domrf_kapremont staging → "
"houses.year_built/material_walls/total_floors → listings.year_built"
)
)
parser.add_argument(
"--kr11-path", default=None, help="локальный CSV КР1.1 (пропустить скачивание)"
)
parser.add_argument(
"--kr12-path", default=None, help="локальный CSV КР1.2 (пропустить скачивание)"
)
parser.add_argument(
"--work-dir",
default=None,
help="каталог для скачанных/распакованных CSV (по умолчанию — временный, удаляется после)",
)
parser.add_argument(
"--dry-run", action="store_true", help="без записи в БД — только скачивание/парс/подсчёт"
)
parser.add_argument(
"--load-only",
action="store_true",
help="только скачать+распарсить+UPSERT staging, без backfill houses/propagate listings",
)
parser.add_argument(
"--backfill-only",
action="store_true",
help="пропустить скачивание/load staging (считаем что уже загружена этим же "
"или предыдущим прогоном) — только backfill houses + propagate listings",
)
return parser
def main() -> None:
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
parser = build_parser()
args = parser.parse_args()
if args.load_only and args.backfill_only:
parser.error("--load-only и --backfill-only взаимоисключающи")
db = SessionLocal()
try:
load_counts: dict[str, int] = {}
if not args.backfill_only:
load_counts = load_domrf_kapremont(
db,
kr11_path=args.kr11_path,
kr12_path=args.kr12_path,
work_dir=args.work_dir,
dry_run=args.dry_run,
)
if not args.dry_run:
db.commit()
logger.info("domrf_kapremont_load: load stage done: %s", load_counts)
houses_counts: dict[str, int] = {}
listings_counts: dict[str, int] = {}
if not args.load_only:
houses_counts = backfill_houses_from_domrf(db, dry_run=args.dry_run)
listings_counts = propagate_listings_year_from_houses(db, dry_run=args.dry_run)
if not args.dry_run:
db.commit()
logger.info(
"domrf_kapremont_load DONE: dry_run=%s load=%s houses=%s listings=%s",
args.dry_run,
load_counts,
houses_counts,
listings_counts,
)
finally:
db.close()
if __name__ == "__main__":
main()

67
tradein-mvp/backend/data/sql/176_domrf_kapremont.sql Normal file
View file

@ -0,0 +1,67 @@
-- 176_domrf_kapremont.sql
--
-- CONTEXT: issue #2013 — houses.year_built заполнен только на 38%, houses.material_walls
-- на 0%. Существующий ГИС-ЖКХ loader (app/tasks/zhkh_flats_load.py, мигр. 146/149)
-- заполняет houses.zhkh_year (70% = 5002/7128) и houses.zhkh_floors (70%), НО НИКОГДА
-- не копирует их в houses.year_built/total_floors — это основной пробел. material_walls
-- вообще ни разу не заполнялся никаким источником (0%). Estimator
-- (app/services/estimator.py) фильтрует когорту по `year_built BETWEEN ...` — реальные
-- годы напрямую двигают точность оценки. Это REAL fix для #2013 (взамен запаркованного
-- когорт-исключения PR #2474).
--
-- Источник: ДОМ.РФ капремонт open data (free, no auth), region 66 (Свердловская обл.):
-- КР1.1 house registry (export/190) — zip → CSV. mkd_code, houseguid (ФИАС GUID дома),
-- address, commission_year (год ввода, int), total_sq, number_floors_max.
-- КР1.2 constructive elements (export/275) — zip → CSV, LONG FORMAT (строка на
-- конструктивный элемент на mkd_code); wall_material заполнен только на строке
-- элемента «фасад».
-- listings→houses линковка через listings.house_id_fk (avito 52.5% linked, year_built
-- 0% на avito → backfill поднимет до ~36-52% через houses.year_built propagation).
--
-- WHAT:
-- Таблица domrf_kapremont — staging-приёмник КР1.1+КР1.2 (region 66), ключ mkd_code.
-- Заполняется отдельным loader'ом (app/services/domrf_kapremont_loader.py,
-- app/tasks/domrf_kapremont_load.py). Backfill houses.year_built/material_walls/
-- total_floors и propagation listings.year_built — отдельный код-путь той же таски,
-- НЕ в этой миграции (только схема здесь, никакого DML над houses/listings).
--
-- DEPENDENCIES (existing prod objects): houses (gar_house_guid, house_fias_id,
-- zhkh_house_guid, zhkh_year, zhkh_floors, year_built, material_walls, total_floors —
-- все уже существуют: мигр. 009/010/070/146/149/152), listings (house_id_fk, year_built —
-- мигр. 002/130).
--
-- SAFETY / IDEMPOTENCY: только CREATE TABLE/INDEX IF NOT EXISTS + COMMENT. Повторный
-- прогон — no-op. Деплой-раннер гонит файл через psql -v ON_ERROR_STOP=on БЕЗ
-- --single-transaction → транзакцию открывает САМ файл (BEGIN/COMMIT ниже), как 143/149.
--
-- NB по нумерации: последний занятый = 175 (ls data/sql | grep '^17[0-9]'); следующий
-- свободный sequential = 176.
--
-- Deploy order: после 175_scrape_schedules_seed_domclick_detail_backfill.sql.
BEGIN;
CREATE TABLE IF NOT EXISTS domrf_kapremont (
mkd_code text PRIMARY KEY, -- ДОМ.РФ код МКД (КР1.1/КР1.2 join-ключ)
houseguid text, -- ФИАС GUID дома (КР1.1 houseguid)
address text,
commission_year smallint, -- год ввода в эксплуатацию (КР1.1 commission_year)
number_floors_max int, -- этажность (КР1.1 number_floors_max)
total_sq numeric(10, 2), -- общая площадь дома, м² (КР1.1 total_sq)
wall_material text, -- материал стен (КР1.2, строка «фасад»; wall_material)
loaded_at timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX IF NOT EXISTS domrf_kapremont_houseguid_idx
ON domrf_kapremont (houseguid)
WHERE houseguid IS NOT NULL;
COMMENT ON TABLE domrf_kapremont IS
'ДОМ.РФ капремонт open data (КР1.1 house registry export/190 + КР1.2 constructive '
'elements export/275), region 66 (Свердловская обл.). Ключ mkd_code. houseguid — ФИАС '
'GUID дома, матчится к houses через COALESCE(gar_house_guid, house_fias_id, '
'zhkh_house_guid) (app/services/domrf_kapremont_loader.py). wall_material берётся из '
'КР1.2 со строки конструктивного элемента «фасад» (единственная строка на mkd_code, '
'где это поле непустое). Заполняется app/tasks/domrf_kapremont_load.py (issue #2013).';
COMMIT;

389
tradein-mvp/backend/tests/test_domrf_kapremont_loader.py Normal file
View file

@ -0,0 +1,389 @@
"""Тесты ДОМ.РФ капремонт loader'а (app/services/domrf_kapremont_loader.py, мигр. 176, #2013).
Coverage:
- Чистые парс-хелперы: parse_decimal_comma / parse_int_field.
- parse_kr11_csv / parse_kr12_wall_materials — фикстуры собраны из РЕАЛЬНЫХ строк
региона 66 (export-kr1_1-66-20260701.csv / export-kr1_2-66-20260701.csv, дом
2170017185 «Ломоносова 3» и 2170025270 «Ломоносова 4»): КР1.2 long-format,
wall_material непустой только на строке construction_element_type='фасад'.
- build_domrf_houses — join КР1.1+КР1.2 по mkd_code.
- fetch_domrf_csvs — скачивание+распаковка zip, HTTP замокан через httpx.MockTransport
(никакой живой сети).
- Статические asserts по SQL: _UPSERT_SQL (ON CONFLICT, IS DISTINCT FROM, CAST),
backfill houses (COALESCE-порядок year_built/domrf/zhkh_year, only-fill-NULL),
propagate listings (только NULL listings.year_built).
- backfill_houses_from_domrf / propagate_listings_year_from_houses — dry_run вызывает
COUNT-SQL и не пишет, real-run суммирует rowcount обоих шагов (MagicMock db).
"""
from __future__ import annotations
import io
import os
import re
import zipfile
from pathlib import Path
from unittest.mock import MagicMock
# settings/SessionLocal не импортируем, но пакет app.* на импорте может тронуть DSN — как в
# sibling-тестах (test_zhkh_flats_loader) задаём заглушку (ядро статично/без реальной БД).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import httpx
import pytest
from app.services import domrf_kapremont_loader as dkl
# ─────────────────────────────────────────────────────────────────────────────
# Фикстуры: РЕАЛЬНЫЕ строки региона 66 (export-kr1_1/kr1_2-66-20260701.csv)
# ─────────────────────────────────────────────────────────────────────────────
# КР1.1 house registry — 2 дома (Ломоносова 3 / Ломоносова 4, п. Арамиль).
_KR11_HEADER = (
"subject_rf;mun_obr_oktmo;mun_obr;mkd_code;houseguid;address;commission_year;"
"architectural_monument_category;total_sq;total_rooms_amount;living_rooms_amount;"
"total_rooms_sq;living_rooms_sq;total_ppl;number_floors_max;money_collecting_way;"
"money_ppl_collected;money_ppl_collected_debts;overhaul_funds_spent_all;"
"overhaul_funds_spent_subsidy;overhaul_fund_spent_other;overhaul_funds_balance;"
"update_date_of_information;money_ppl_collected_date;owners_payment;energy_efficiency;"
"alarm_document_date;exclude_date_from_program;inclusion_date_to_program;comment;"
"last_update;house_id"
)
_KR11_ROW_3 = (
'"Свердловская область";65729000;"Арамильский городской округ Свердловской области";'
"2170017185;04b98bcb-3232-4b26-b0fa-3c4a47adf394;"
'"Арамильский городской округ Свердловской области, пос. Арамиль, ул. Ломоносова, д. 3";'
'1964;;399,90;;;366,60;333,30;20;2;"Счет регионального оператора";486,84407;72,59638;'
"1580,29712;0,00000;0,00000;-1093,45305;2026-05-31;01.10.2014;18,81;"
'"не присвоен";;;22.04.2014;;15.05.2026;8857551'
)
_KR11_ROW_4 = (
'"Свердловская область";65729000;"Арамильский городской округ Свердловской области";'
"2170025270;15064d04-3aec-4266-985b-da26e9a5f392;"
'"Арамильский городской округ Свердловской области, пос. Арамиль, ул. Ломоносова, д. 4";'
'1984;;906,40;;;;;;2;"Счет регионального оператора";1046,79431;204,75296;0,00000;'
'0,00000;0,00000;1046,79431;2026-05-31;01.10.2014;18,81;"не присвоен";;;22.04.2014;;'
"15.05.2026;8857552"
)
KR11_CSV = "\n".join([_KR11_HEADER, _KR11_ROW_3, _KR11_ROW_4]) + "\n"
# КР1.2 constructive elements — long-format: обычная строка (wall_material пуст) +
# строка "фасад" (wall_material='кирпич') на каждый из тех же 2 домов.
_KR12_HEADER = (
"subject_rf;mun_obr_oktmo;mun_obr;mkd_code;address;construction_element_code;"
"construction_element_type;system_type;lift_type;stops_count;commissioning_date;"
"decommissioning_date;roof_type;roofing_area;basement_area;facade_type;facade_area;"
"foundation_type;wall_material;comment;last_update"
)
_KR12_ROW_3_OTHER = (
'"Свердловская область";65729000;"Арамильский городской округ Свердловской области";'
"2170017185;"
'"Арамильский городской округ Свердловской области, пос. Арамиль, ул. Ломоносова, д. 3";'
'580000;"система дымоудаления и (или) противопожарной автоматики";;;;;;;;;;;;;;15.05.2026'
)
_KR12_ROW_3_FACADE = (
'"Свердловская область";65729000;"Арамильский городской округ Свердловской области";'
"2170017185;"
'"Арамильский городской округ Свердловской области, пос. Арамиль, ул. Ломоносова, д. 3";'
'60000;фасад;;;;;;;;;"облицованный камнем";480,00;;кирпич;;15.05.2026'
)
_KR12_ROW_4_OTHER = (
'"Свердловская область";65729000;"Арамильский городской округ Свердловской области";'
"2170025270;"
'"Арамильский городской округ Свердловской области, пос. Арамиль, ул. Ломоносова, д. 4";'
'100000;"инженерная система водоснабжения (ХВС)";центральная;;;;;;;;;;;;;15.05.2026'
)
_KR12_ROW_4_FACADE = (
'"Свердловская область";65729000;"Арамильский городской округ Свердловской области";'
"2170025270;"
'"Арамильский городской округ Свердловской области, пос. Арамиль, ул. Ломоносова, д. 4";'
"60000;фасад;;;;;;;;;;;;кирпич;;15.05.2026"
)
KR12_CSV = (
"\n".join(
[_KR12_HEADER, _KR12_ROW_3_OTHER, _KR12_ROW_3_FACADE, _KR12_ROW_4_OTHER, _KR12_ROW_4_FACADE]
)
+ "\n"
)
@pytest.fixture
def kr11_path(tmp_path: Path) -> Path:
p = tmp_path / "kr11.csv"
p.write_text(KR11_CSV, encoding="utf-8-sig")
return p
@pytest.fixture
def kr12_path(tmp_path: Path) -> Path:
p = tmp_path / "kr12.csv"
p.write_text(KR12_CSV, encoding="utf-8-sig")
return p
# ─────────────────────────────────────────────────────────────────────────────
# parse_decimal_comma / parse_int_field
# ─────────────────────────────────────────────────────────────────────────────
def test_parse_decimal_comma() -> None:
assert dkl.parse_decimal_comma("909,80") == pytest.approx(909.80)
assert dkl.parse_decimal_comma("399,90") == pytest.approx(399.90)
assert dkl.parse_decimal_comma("100.5") == pytest.approx(100.5)
assert dkl.parse_decimal_comma("") is None
assert dkl.parse_decimal_comma(None) is None
assert dkl.parse_decimal_comma("не число") is None
def test_parse_int_field() -> None:
assert dkl.parse_int_field("1964") == 1964
assert dkl.parse_int_field("2") == 2
assert dkl.parse_int_field("") is None
assert dkl.parse_int_field(None) is None
assert dkl.parse_int_field("x") is None
assert dkl.parse_int_field("5,0") == 5 # decimal-фоллбек
# ─────────────────────────────────────────────────────────────────────────────
# parse_kr11_csv — house registry
# ─────────────────────────────────────────────────────────────────────────────
def test_parse_kr11_csv_real_rows(kr11_path: Path) -> None:
rows = dkl.parse_kr11_csv(kr11_path)
assert set(rows.keys()) == {"2170017185", "2170025270"}
house3 = rows["2170017185"]
assert house3.houseguid == "04b98bcb-3232-4b26-b0fa-3c4a47adf394"
assert house3.commission_year == 1964
assert house3.number_floors_max == 2
assert house3.total_sq == pytest.approx(399.90)
assert "Ломоносова, д. 3" in (house3.address or "")
house4 = rows["2170025270"]
assert house4.houseguid == "15064d04-3aec-4266-985b-da26e9a5f392"
assert house4.commission_year == 1984
assert house4.total_sq == pytest.approx(906.40)
def test_parse_kr11_csv_skips_blank_mkd_code(tmp_path: Path) -> None:
csv_text = _KR11_HEADER + "\n" + (";" * (_KR11_HEADER.count(";"))) + "\n"
p = tmp_path / "blank.csv"
p.write_text(csv_text, encoding="utf-8-sig")
assert dkl.parse_kr11_csv(p) == {}
# ─────────────────────────────────────────────────────────────────────────────
# parse_kr12_wall_materials — long-format, wall_material только на строке 'фасад'
# ─────────────────────────────────────────────────────────────────────────────
def test_parse_kr12_wall_materials_extracts_facade_row_only(kr12_path: Path) -> None:
materials = dkl.parse_kr12_wall_materials(kr12_path)
# Обе non-facade строки (дымоудаление / ХВС) имели wall_material='' — исключены.
assert materials == {"2170017185": "кирпич", "2170025270": "кирпич"}
def test_parse_kr12_wall_materials_empty_when_no_facade_row(tmp_path: Path) -> None:
csv_text = "\n".join([_KR12_HEADER, _KR12_ROW_3_OTHER]) + "\n"
p = tmp_path / "no_facade.csv"
p.write_text(csv_text, encoding="utf-8-sig")
assert dkl.parse_kr12_wall_materials(p) == {}
# ─────────────────────────────────────────────────────────────────────────────
# build_domrf_houses — join КР1.1 + КР1.2(wall_material)
# ─────────────────────────────────────────────────────────────────────────────
def test_build_domrf_houses_joins_wall_material(kr11_path: Path, kr12_path: Path) -> None:
kr11 = dkl.parse_kr11_csv(kr11_path)
walls = dkl.parse_kr12_wall_materials(kr12_path)
houses = dkl.build_domrf_houses(kr11, walls)
by_code = {h.mkd_code: h for h in houses}
assert len(houses) == 2
assert by_code["2170017185"].wall_material == "кирпич"
assert by_code["2170017185"].commission_year == 1964
assert by_code["2170025270"].wall_material == "кирпич"
# Дом без строки 'фасад' в КР1.2 → wall_material остаётся None (не KeyError/пустая строка).
def test_wall_material_none_when_house_missing_from_kr12(tmp_path: Path, kr11_path: Path) -> None:
kr11 = dkl.parse_kr11_csv(kr11_path)
houses = dkl.build_domrf_houses(kr11, {}) # пустой КР1.2 lookup
assert all(h.wall_material is None for h in houses)
# ─────────────────────────────────────────────────────────────────────────────
# fetch_domrf_csvs — скачивание+распаковка zip, HTTP замокан (MockTransport)
# ─────────────────────────────────────────────────────────────────────────────
def _zip_bytes(inner_name: str, content: str) -> bytes:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr(inner_name, content.encode("utf-8-sig"))
return buf.getvalue()
def test_fetch_domrf_csvs_downloads_and_extracts(tmp_path: Path) -> None:
kr11_zip = _zip_bytes("export-kr1_1-66-20260701.csv", KR11_CSV)
kr12_zip = _zip_bytes("export-kr1_2-66-20260701.csv", KR12_CSV)
def handler(request: httpx.Request) -> httpx.Response:
url = str(request.url)
if url == dkl.KR11_URL:
return httpx.Response(200, content=kr11_zip)
if url == dkl.KR12_URL:
return httpx.Response(200, content=kr12_zip)
raise AssertionError(f"unexpected URL {url}")
transport = httpx.MockTransport(handler)
with httpx.Client(transport=transport) as client:
kr11_out, kr12_out = dkl.fetch_domrf_csvs(tmp_path, client=client)
assert kr11_out.exists() and kr11_out.name.endswith(".csv")
assert kr12_out.exists() and kr12_out.name.endswith(".csv")
# Круглый путь: распакованные файлы парсятся так же, как локальные фикстуры.
rows = dkl.parse_kr11_csv(kr11_out)
assert set(rows.keys()) == {"2170017185", "2170025270"}
materials = dkl.parse_kr12_wall_materials(kr12_out)
assert materials == {"2170017185": "кирпич", "2170025270": "кирпич"}
def test_extract_csv_from_zip_raises_on_no_csv_entry(tmp_path: Path) -> None:
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
zf.writestr("readme.txt", "not a csv")
with pytest.raises(ValueError, match="csv"):
dkl._extract_csv_from_zip(buf.getvalue(), tmp_path)
# ─────────────────────────────────────────────────────────────────────────────
# Статические asserts по _UPSERT_SQL (psycopg v3, ON CONFLICT, идемпотентность)
# ─────────────────────────────────────────────────────────────────────────────
_UPSERT_SQL = str(dkl._UPSERT_SQL.text)
def test_upsert_sql_uses_psycopg_v3_cast_not_double_colon() -> None:
assert not re.search(r":\w+::", _UPSERT_SQL)
assert "CAST(:mkd_code AS text)" in _UPSERT_SQL
assert "CAST(:commission_year AS smallint)" in _UPSERT_SQL
assert "CAST(:total_sq AS numeric)" in _UPSERT_SQL
def test_upsert_sql_idempotent_on_conflict() -> None:
flat = re.sub(r"\s+", " ", _UPSERT_SQL)
assert "INSERT INTO domrf_kapremont" in flat
assert "ON CONFLICT (mkd_code) DO UPDATE SET" in flat
assert "domrf_kapremont.houseguid IS DISTINCT FROM EXCLUDED.houseguid" in flat
assert "domrf_kapremont.wall_material IS DISTINCT FROM EXCLUDED.wall_material" in flat
def test_upsert_uses_savepoint_per_chunk() -> None:
import inspect
src = inspect.getsource(dkl.upsert_domrf_kapremont)
assert "with db.begin_nested():" in src
assert "db.commit()" not in src # не коммитит сам (коммитит caller)
# ─────────────────────────────────────────────────────────────────────────────
# Статические asserts: backfill houses — COALESCE(year_built, domrf, zhkh_year)
# ─────────────────────────────────────────────────────────────────────────────
_BACKFILL_DOMRF_SQL = re.sub(r"\s+", " ", str(dkl._BACKFILL_HOUSES_FROM_DOMRF_SQL.text))
_BACKFILL_ZHKH_SQL = re.sub(r"\s+", " ", str(dkl._BACKFILL_HOUSES_ZHKH_FALLBACK_SQL.text))
def test_backfill_domrf_sql_coalesce_order_matches_spec() -> None:
# COALESCE(year_built, domrf.commission_year, zhkh_year) — ровно в этом порядке (#2013).
assert "COALESCE(h.year_built, d.commission_year, h.zhkh_year)" in _BACKFILL_DOMRF_SQL
assert "COALESCE(h.material_walls, d.wall_material)" in _BACKFILL_DOMRF_SQL
assert "COALESCE(h.total_floors, d.number_floors_max, h.zhkh_floors)" in _BACKFILL_DOMRF_SQL
def test_backfill_domrf_sql_matches_by_guid_coalesce_priority() -> None:
# gar_house_guid первый (78% coverage), затем house_fias_id, затем zhkh_house_guid.
assert (
"d.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid)"
in _BACKFILL_DOMRF_SQL
)
def test_backfill_domrf_sql_only_fills_null_never_overwrites() -> None:
# Гейт: каждый SET-таргет требует h.<col> IS NULL — COALESCE никогда не перезапишет
# существующее non-null значение (SQL COALESCE тоже это гарантирует, но гейт защищает
# от no-op UPDATE на уже заполненных строках).
assert "h.year_built IS NULL" in _BACKFILL_DOMRF_SQL
assert "h.material_walls IS NULL" in _BACKFILL_DOMRF_SQL
assert "h.total_floors IS NULL" in _BACKFILL_DOMRF_SQL
def test_backfill_zhkh_fallback_sql_folds_zhkh_for_unmatched_houses() -> None:
# Шаг 2 — без JOIN на domrf_kapremont вообще: применяется к ЛЮБОМУ дому, включая
# те, что не заматчились в шаге 1 (issue: "houses without a ДОМ.РФ match still benefit").
assert "domrf_kapremont" not in _BACKFILL_ZHKH_SQL
assert "COALESCE(year_built, zhkh_year)" in _BACKFILL_ZHKH_SQL
assert "COALESCE(total_floors, zhkh_floors)" in _BACKFILL_ZHKH_SQL
assert "year_built IS NULL AND zhkh_year IS NOT NULL" in _BACKFILL_ZHKH_SQL
def test_backfill_sql_no_psycopg_v3_colon_colon_cast() -> None:
assert not re.search(r":\w+::", _BACKFILL_DOMRF_SQL)
assert not re.search(r":\w+::", _BACKFILL_ZHKH_SQL)
# ─────────────────────────────────────────────────────────────────────────────
# backfill_houses_from_domrf — dry_run vs real-run (MagicMock db)
# ─────────────────────────────────────────────────────────────────────────────
def test_backfill_houses_real_run_sums_both_steps_rowcount() -> None:
db = MagicMock()
results = [MagicMock(rowcount=3), MagicMock(rowcount=2)]
db.execute.side_effect = results
out = dkl.backfill_houses_from_domrf(db, dry_run=False)
assert out == {"domrf_matched": 3, "zhkh_fallback": 2, "houses_updated": 5}
assert db.execute.call_count == 2
executed_sql = [str(call.args[0]) for call in db.execute.call_args_list]
assert "UPDATE houses" in executed_sql[0]
assert "UPDATE houses" in executed_sql[1]
def test_backfill_houses_dry_run_only_counts_no_update() -> None:
db = MagicMock()
db.execute.return_value.scalar_one.side_effect = [7, 4]
out = dkl.backfill_houses_from_domrf(db, dry_run=True)
assert out == {"domrf_matched": 7, "zhkh_fallback": 4, "houses_updated": 0}
for call in db.execute.call_args_list:
sql_text = str(call.args[0]).upper()
assert "UPDATE" not in sql_text
assert "SELECT COUNT(*)" in sql_text
# ─────────────────────────────────────────────────────────────────────────────
# Статические asserts + behavior: propagate_listings_year_from_houses
# ─────────────────────────────────────────────────────────────────────────────
_PROPAGATE_SQL = re.sub(r"\s+", " ", str(dkl._PROPAGATE_LISTINGS_YEAR_SQL.text))
def test_propagate_sql_only_fills_null_listings_year() -> None:
assert "SET year_built = h.year_built" in _PROPAGATE_SQL
assert "l.year_built IS NULL" in _PROPAGATE_SQL
assert "h.year_built IS NOT NULL" in _PROPAGATE_SQL
assert "l.house_id_fk = h.id" in _PROPAGATE_SQL
def test_propagate_listings_year_real_run_returns_rowcount() -> None:
db = MagicMock()
db.execute.return_value.rowcount = 42
out = dkl.propagate_listings_year_from_houses(db, dry_run=False)
assert out == {"listings_updated": 42}
sql_text = str(db.execute.call_args[0][0])
assert "UPDATE listings" in sql_text
def test_propagate_listings_year_dry_run_uses_count_sql() -> None:
db = MagicMock()
db.execute.return_value.scalar_one.return_value = 99
out = dkl.propagate_listings_year_from_houses(db, dry_run=True)
assert out == {"listings_updated": 0, "would_update": 99}
sql_text = str(db.execute.call_args[0][0]).upper()
assert "UPDATE" not in sql_text