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

2026-07-12 13:12:07 +00:00 · 2026-07-12 13:12:07 +00:00 · cc231946c2
commit cc231946c2
parent d22e0e00ae
4 changed files with 1059 additions and 0 deletions
--- a/tradein-mvp/backend/app/services/domrf_kapremont_loader.py
+++ b/tradein-mvp/backend/app/services/domrf_kapremont_loader.py
@ -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}
--- a/tradein-mvp/backend/app/tasks/domrf_kapremont_load.py
+++ b/tradein-mvp/backend/app/tasks/domrf_kapremont_load.py
@ -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()
--- a/tradein-mvp/backend/data/sql/176_domrf_kapremont.sql
+++ b/tradein-mvp/backend/data/sql/176_domrf_kapremont.sql
@ -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;
--- a/tradein-mvp/backend/tests/test_domrf_kapremont_loader.py
+++ b/tradein-mvp/backend/tests/test_domrf_kapremont_loader.py
@ -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