"""АИС ППК «ФРТ» (бывш. Реформа ЖКХ) loader: houses.area_land/foundation_type/ elevators_total + добор year_built/material_walls/total_floors/entrances/is_emergency/ flat_count/heat_supply_type/gas_supply_type/hot_water (issue #frt-mkd). CONTEXT: houses.area_land и houses.foundation_type отсутствовали вовсе; elevators_total — новая колонка (в источнике — ОБЩЕЕ число лифтов, houses раздельно хранит только passenger_elevators/cargo_elevators, поэтому мапить в них нельзя — потеряли бы тип). Остальные поля (year_built, material_walls, ...) уже существуют и заполнены частично другими источниками (ДОМ.РФ капремонт, ГИС-ЖКХ) — этот loader их ДОБИРАЕТ через COALESCE (только NULL), никогда не перезаписывает. ИСТОЧНИК: АИС ППК ФРТ open data, https://xn--80adsazqn.xn--p1aee.xn--p1ai/opendata/export/{node_id} node 110 = реестр МКД региона 66 (Свердловская обл.) — проверено живьём 08.09.2026: HTTP 200, application/octet-stream, zip → один CSV `export-reestrmkd-66-*.csv`, UTF-8 BOM, разделитель ';', 60 колонок, ~41.8k строк по СО. robots.txt источника запрещает /opendata/export/ и требует Crawl-delay 10. Один прогон этого loader'а делает РОВНО ОДИН GET (один node_id) — задержка не нужна. Если когда-нибудь понадобится тянуть несколько node_id за один запуск (напр. node 427 — аварийный фонд РФ, node 1 — реестр УО РФ, оба вне скоупа этого PR) — между запросами обязательна пауза CRAWL_DELAY_SEC (константа ниже), иначе нарушаем Crawl-delay из robots.txt. ЧТО НЕ ДЕЛАЕМ (осознанно): * project_type НЕ мапим в houses.series_name — свободный текст, фактический дубль материала стен (замер: пусто 10635, «кирпичный» 1754, «нет данных» 1496, «панельный» 1012, «Блочный»/«блочный» двумя разными строками). Как серию использовать нельзя. * energy_efficiency НЕ добавляем в houses — миграция 284 уже приняла это решение; реальный класс присвоен лишь ~10% домов (у большинства «Не присвоен»). * elevators_count НЕ мапим в houses.passenger_elevators — источник отдаёт общее число лифтов без разбивки, а houses хранит passenger/cargo раздельно. Пишем в отдельную houses.elevators_total. * playground/sportsground — id справочника (498/499/500), не булев — в staging как есть, houses.has_playground не трогаем. * estimator.py не трогаем — встраивание признаков дома в подбор аналогов вне скоупа. МУСОРНЫЕ ЗНАЧЕНИЯ источника (встречаются как обычные строки текстовых полей): '', 'нет данных', 'Не заполнено', 'отсутствует', 'данные отсутствуют', 'нет'. Единый хелпер `clean_text()` отфильтровывает их при парсе КАЖДОГО текстового поля. TLS: домен xn--80adsazqn.xn--p1aee.xn--p1ai отдаёт RU-сертификат НУЦ Минцифры, не верифицируемый дефолтным trust store'ом httpx — та же ситуация, что sber_index.py и domrf_kapremont_loader.py. Открытые данные без auth/PII — verify=False приемлем. 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, fields from datetime import date from pathlib import Path import httpx from sqlalchemy import text from sqlalchemy.orm import Session logger = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────────────── # Константы # ───────────────────────────────────────────────────────────────────────────── BASE_URL = "https://xn--80adsazqn.xn--p1aee.xn--p1ai" DEFAULT_NODE_ID = 110 # реестр МКД региона 66 — проверено живьём 08.09.2026 DEFAULT_REGION_CODE = 66 DOWNLOAD_TIMEOUT_SEC = 180 UPSERT_CHUNK_SIZE = 500 # robots.txt источника: Crawl-delay 10. Используется только если когда-нибудь понадобится # тянуть несколько node_id за один запуск (сейчас — ровно один GET на прогон). CRAWL_DELAY_SEC = 10 # Мусорные значения текстовых полей источника — не данные, а «пусто» в других обёртках. _NOISE_VALUES = frozenset( {"", "нет данных", "не заполнено", "отсутствует", "данные отсутствуют", "нет"} ) # Правдоподобные границы, используются и на парсе (staging), и повторно в backfill SQL. YEAR_BUILT_MIN = 1850 YEAR_BUILT_FUTURE_SLACK = 2 FLOOR_COUNT_MAX_BOUND = 100 ENTRANCE_COUNT_MAX_BOUND = 50 ELEVATORS_COUNT_MAX_BOUND = 50 FLAT_COUNT_MAX_BOUND = 3000 AREA_LAND_MAX_BOUND = 500_000 # м2, с большим запасом для дворовых территорий МКД def plausible_year_max() -> int: """Верхняя граница правдоподобного года постройки (текущий год + slack).""" return date.today().year + YEAR_BUILT_FUTURE_SLACK # ───────────────────────────────────────────────────────────────────────────── # Чистые хелперы парсинга (юнит-тестируются без сети/БД) # ───────────────────────────────────────────────────────────────────────────── def clean_text(raw: str | None) -> str | None: """Строка источника → строка|None, мусорные значения-заглушки схлопываются в None.""" if raw is None: return None s = raw.strip() if not s or s.lower() in _NOISE_VALUES: return None return s def parse_decimal_comma(raw: str | None) -> float | None: """«930,60» / «930.60» / «» / мусор → float|None. Запятая — decimal separator АИС ФРТ.""" s = clean_text(raw) if s is None: return None try: return float(s.replace(",", ".")) except ValueError: return None def parse_int_field( raw: str | None, *, min_value: int | None = None, max_value: int | None = None ) -> int | None: """Устойчивый str → int|None с опциональным sanity-гейтом [min_value, max_value].""" s = clean_text(raw) if s is None: return None value: int | None = None try: value = int(s) except ValueError: try: value = int(float(s.replace(",", "."))) except ValueError: return None if min_value is not None and value < min_value: return None if max_value is not None and value > max_value: return None return value def parse_bool_da_net(raw: str | None) -> bool | None: """«Да»/«Нет» (регистронезависимо) → bool|None. Пусто/иное → None. НЕ идёт через clean_text() — «нет» само по себе входит в _NOISE_VALUES (заглушка текстовых полей типа «отсутствует»/«нет»), и там это совпадение уместно. Но для is_alarm «Нет» — валидный содержательный ответ («не аварийный»), а не отсутствие данных, поэтому здесь разбираем сырую строку напрямую. """ if raw is None: return None s = raw.strip() if not s: return None low = s.lower() if low == "да": return True if low == "нет": return False return None @dataclass(slots=True) class FrtMkdRow: """Одна строка реестра МКД АИС ФРТ (house-per-row), готова к UPSERT в staging.""" houseguid: str region_code: int address: str | None built_year: int | None exploitation_start_year: int | None project_type: str | None house_type: str | None is_alarm: bool | None floor_count_max: int | None floor_count_min: int | None entrance_count: int | None elevators_count: int | None energy_efficiency: str | None quarters_count: int | None living_quarters_count: int | None unliving_quarters_count: int | None area_total: float | None area_residential: float | None area_non_residential: float | None area_common_property: float | None area_land: float | None parking_square: float | None playground: int | None sportsground: int | None other_beautification: str | None foundation_type: str | None floor_type: str | None wall_material: str | None basement_area: float | None chute_type: str | None chute_count: int | None heating_type: str | None hot_water_type: str | None cold_water_type: str | None sewerage_type: str | None gas_type: str | None ventilation_type: str | None firefighting_type: str | None drainage_type: str | None management_organization_id: int | None method_of_forming_overhaul_fund: str | None def parse_frt_mkd_csv(path: str | Path, *, region_code: int) -> dict[str, FrtMkdRow]: """CSV реестра МКД АИС ФРТ → dict по houseguid. Строки без houseguid пропускаются. Дубликаты houseguid в источнике РЕАЛЬНЫ и взаимодополняющи: в выгрузке региона 66 за 2026-09-01 из 41 790 строк 912 guid'ов повторяются (968 лишних строк), причём у одной строки пары заполнен, например, `area_total`, а у другой — нет. Поэтому «последняя побеждает» терять нельзя: дубликаты СЛИВАЮТСЯ по полям, побеждает первое непустое значение (`_merge_rows`). Иначе бэкфилл недосчитывался бы заполненных полей примерно у 2 % домов области. """ rows: dict[str, FrtMkdRow] = {} with open(path, encoding="utf-8-sig", newline="") as f: reader = csv.DictReader(f, delimiter=";") for raw in reader: houseguid = clean_text(raw.get("houseguid")) if not houseguid: continue parsed = FrtMkdRow( houseguid=houseguid, region_code=region_code, address=clean_text(raw.get("address")), built_year=parse_int_field( raw.get("built_year"), min_value=YEAR_BUILT_MIN, max_value=plausible_year_max() ), exploitation_start_year=parse_int_field( raw.get("exploitation_start_year"), min_value=YEAR_BUILT_MIN, max_value=plausible_year_max(), ), project_type=clean_text(raw.get("project_type")), house_type=clean_text(raw.get("house_type")), is_alarm=parse_bool_da_net(raw.get("is_alarm")), floor_count_max=parse_int_field( raw.get("floor_count_max"), min_value=1, max_value=FLOOR_COUNT_MAX_BOUND ), floor_count_min=parse_int_field( raw.get("floor_count_min"), min_value=1, max_value=FLOOR_COUNT_MAX_BOUND ), entrance_count=parse_int_field( raw.get("entrance_count"), min_value=1, max_value=ENTRANCE_COUNT_MAX_BOUND ), elevators_count=parse_int_field( raw.get("elevators_count"), min_value=0, max_value=ELEVATORS_COUNT_MAX_BOUND ), energy_efficiency=clean_text(raw.get("energy_efficiency")), quarters_count=parse_int_field(raw.get("quarters_count"), min_value=0), living_quarters_count=parse_int_field( raw.get("living_quarters_count"), min_value=0, max_value=FLAT_COUNT_MAX_BOUND ), unliving_quarters_count=parse_int_field( raw.get("unliving_quarters_count"), min_value=0 ), area_total=parse_decimal_comma(raw.get("area_total")), area_residential=parse_decimal_comma(raw.get("area_residential")), area_non_residential=parse_decimal_comma(raw.get("area_non_residential")), area_common_property=parse_decimal_comma(raw.get("area_common_property")), area_land=parse_decimal_comma(raw.get("area_land")), parking_square=parse_decimal_comma(raw.get("parking_square")), # playground/sportsground — id справочника благоустройства, не булев флаг. playground=parse_int_field(raw.get("playground"), min_value=0), sportsground=parse_int_field(raw.get("sportsground"), min_value=0), other_beautification=clean_text(raw.get("other_beautification")), foundation_type=clean_text(raw.get("foundation_type")), floor_type=clean_text(raw.get("floor_type")), wall_material=clean_text(raw.get("wall_material")), basement_area=parse_decimal_comma(raw.get("basement_area")), chute_type=clean_text(raw.get("chute_type")), chute_count=parse_int_field(raw.get("chute_count"), min_value=0), heating_type=clean_text(raw.get("heating_type")), hot_water_type=clean_text(raw.get("hot_water_type")), cold_water_type=clean_text(raw.get("cold_water_type")), sewerage_type=clean_text(raw.get("sewerage_type")), gas_type=clean_text(raw.get("gas_type")), ventilation_type=clean_text(raw.get("ventilation_type")), firefighting_type=clean_text(raw.get("firefighting_type")), drainage_type=clean_text(raw.get("drainage_type")), management_organization_id=parse_int_field( raw.get("management_organization_id"), min_value=0 ), method_of_forming_overhaul_fund=clean_text( raw.get("method_of_forming_overhaul_fund") ), ) prev = rows.get(houseguid) rows[houseguid] = _merge_rows(prev, parsed) if prev is not None else parsed return rows def _merge_rows(prev: FrtMkdRow, new: FrtMkdRow) -> FrtMkdRow: """Слить дубликат houseguid: для каждого поля берём первое непустое значение. Источник отдаёт повторы одного дома разными строками с частично заполненными полями (см. `parse_frt_mkd_csv`). Побеждает уже накопленное значение; новое подставляется только туда, где накопленного нет. """ merged: dict[str, object] = {} for field in fields(FrtMkdRow): current = getattr(prev, field.name) merged[field.name] = current if current is not None else getattr(new, field.name) return FrtMkdRow(**merged) # type: ignore[arg-type] # ───────────────────────────────────────────────────────────────────────────── # 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_frt_mkd_csv(dest_dir: Path, *, node_id: int, client: httpx.Client) -> Path: """Скачивает zip export/{node_id}, распаковывает CSV в dest_dir, возвращает путь.""" url = f"{BASE_URL}/opendata/export/{node_id}" return _extract_csv_from_zip(_download_zip(url, client=client), dest_dir) # ───────────────────────────────────────────────────────────────────────────── # UPSERT staging (frt_mkd, мигр. 290) # ───────────────────────────────────────────────────────────────────────────── _UPSERT_SQL = text( """ INSERT INTO frt_mkd ( houseguid, region_code, address, built_year, exploitation_start_year, project_type, house_type, is_alarm, floor_count_max, floor_count_min, entrance_count, elevators_count, energy_efficiency, quarters_count, living_quarters_count, unliving_quarters_count, area_total, area_residential, area_non_residential, area_common_property, area_land, parking_square, playground, sportsground, other_beautification, foundation_type, floor_type, wall_material, basement_area, chute_type, chute_count, heating_type, hot_water_type, cold_water_type, sewerage_type, gas_type, ventilation_type, firefighting_type, drainage_type, management_organization_id, method_of_forming_overhaul_fund, loaded_at ) VALUES ( CAST(:houseguid AS text), CAST(:region_code AS smallint), CAST(:address AS text), CAST(:built_year AS smallint), CAST(:exploitation_start_year AS smallint), CAST(:project_type AS text), CAST(:house_type AS text), CAST(:is_alarm AS boolean), CAST(:floor_count_max AS smallint), CAST(:floor_count_min AS smallint), CAST(:entrance_count AS smallint), CAST(:elevators_count AS smallint), CAST(:energy_efficiency AS text), CAST(:quarters_count AS int), CAST(:living_quarters_count AS int), CAST(:unliving_quarters_count AS int), CAST(:area_total AS numeric), CAST(:area_residential AS numeric), CAST(:area_non_residential AS numeric), CAST(:area_common_property AS numeric), CAST(:area_land AS numeric), CAST(:parking_square AS numeric), CAST(:playground AS int), CAST(:sportsground AS int), CAST(:other_beautification AS text), CAST(:foundation_type AS text), CAST(:floor_type AS text), CAST(:wall_material AS text), CAST(:basement_area AS numeric), CAST(:chute_type AS text), CAST(:chute_count AS smallint), CAST(:heating_type AS text), CAST(:hot_water_type AS text), CAST(:cold_water_type AS text), CAST(:sewerage_type AS text), CAST(:gas_type AS text), CAST(:ventilation_type AS text), CAST(:firefighting_type AS text), CAST(:drainage_type AS text), CAST(:management_organization_id AS bigint), CAST(:method_of_forming_overhaul_fund AS text), now() ) ON CONFLICT (houseguid) DO UPDATE SET region_code = EXCLUDED.region_code, address = EXCLUDED.address, built_year = EXCLUDED.built_year, exploitation_start_year = EXCLUDED.exploitation_start_year, project_type = EXCLUDED.project_type, house_type = EXCLUDED.house_type, is_alarm = EXCLUDED.is_alarm, floor_count_max = EXCLUDED.floor_count_max, floor_count_min = EXCLUDED.floor_count_min, entrance_count = EXCLUDED.entrance_count, elevators_count = EXCLUDED.elevators_count, energy_efficiency = EXCLUDED.energy_efficiency, quarters_count = EXCLUDED.quarters_count, living_quarters_count = EXCLUDED.living_quarters_count, unliving_quarters_count = EXCLUDED.unliving_quarters_count, area_total = EXCLUDED.area_total, area_residential = EXCLUDED.area_residential, area_non_residential = EXCLUDED.area_non_residential, area_common_property = EXCLUDED.area_common_property, area_land = EXCLUDED.area_land, parking_square = EXCLUDED.parking_square, playground = EXCLUDED.playground, sportsground = EXCLUDED.sportsground, other_beautification = EXCLUDED.other_beautification, foundation_type = EXCLUDED.foundation_type, floor_type = EXCLUDED.floor_type, wall_material = EXCLUDED.wall_material, basement_area = EXCLUDED.basement_area, chute_type = EXCLUDED.chute_type, chute_count = EXCLUDED.chute_count, heating_type = EXCLUDED.heating_type, hot_water_type = EXCLUDED.hot_water_type, cold_water_type = EXCLUDED.cold_water_type, sewerage_type = EXCLUDED.sewerage_type, gas_type = EXCLUDED.gas_type, ventilation_type = EXCLUDED.ventilation_type, firefighting_type = EXCLUDED.firefighting_type, drainage_type = EXCLUDED.drainage_type, management_organization_id = EXCLUDED.management_organization_id, method_of_forming_overhaul_fund = EXCLUDED.method_of_forming_overhaul_fund, loaded_at = now() WHERE frt_mkd.address IS DISTINCT FROM EXCLUDED.address OR frt_mkd.built_year IS DISTINCT FROM EXCLUDED.built_year OR frt_mkd.is_alarm IS DISTINCT FROM EXCLUDED.is_alarm OR frt_mkd.wall_material IS DISTINCT FROM EXCLUDED.wall_material OR frt_mkd.foundation_type IS DISTINCT FROM EXCLUDED.foundation_type OR frt_mkd.area_land IS DISTINCT FROM EXCLUDED.area_land OR frt_mkd.elevators_count IS DISTINCT FROM EXCLUDED.elevators_count """ ) def _chunk_rows(items: list[FrtMkdRow], size: int) -> Iterator[list[FrtMkdRow]]: for i in range(0, len(items), size): yield items[i : i + size] def upsert_frt_mkd( db: Session, rows: list[FrtMkdRow], *, chunk_size: int = UPSERT_CHUNK_SIZE ) -> int: """UPSERT списка FrtMkdRow в frt_mkd, чанками по SAVEPOINT. Не коммитит (caller). Идемпотентно (IS DISTINCT FROM gate на ключевых полях в _UPSERT_SQL). Сбойный чанк откатывается изолированно (SAVEPOINT-паттерн domrf_kapremont_loader/zhkh_flats_loader). """ upserted = 0 for chunk in _chunk_rows(rows, chunk_size): try: with db.begin_nested(): for r in chunk: res = db.execute( _UPSERT_SQL, { "houseguid": r.houseguid, "region_code": r.region_code, "address": r.address, "built_year": r.built_year, "exploitation_start_year": r.exploitation_start_year, "project_type": r.project_type, "house_type": r.house_type, "is_alarm": r.is_alarm, "floor_count_max": r.floor_count_max, "floor_count_min": r.floor_count_min, "entrance_count": r.entrance_count, "elevators_count": r.elevators_count, "energy_efficiency": r.energy_efficiency, "quarters_count": r.quarters_count, "living_quarters_count": r.living_quarters_count, "unliving_quarters_count": r.unliving_quarters_count, "area_total": r.area_total, "area_residential": r.area_residential, "area_non_residential": r.area_non_residential, "area_common_property": r.area_common_property, "area_land": r.area_land, "parking_square": r.parking_square, "playground": r.playground, "sportsground": r.sportsground, "other_beautification": r.other_beautification, "foundation_type": r.foundation_type, "floor_type": r.floor_type, "wall_material": r.wall_material, "basement_area": r.basement_area, "chute_type": r.chute_type, "chute_count": r.chute_count, "heating_type": r.heating_type, "hot_water_type": r.hot_water_type, "cold_water_type": r.cold_water_type, "sewerage_type": r.sewerage_type, "gas_type": r.gas_type, "ventilation_type": r.ventilation_type, "firefighting_type": r.firefighting_type, "drainage_type": r.drainage_type, "management_organization_id": r.management_organization_id, "method_of_forming_overhaul_fund": r.method_of_forming_overhaul_fund, }, ) upserted += res.rowcount except Exception: logger.warning( "frt_mkd upsert: чанк из %d строк сбойнул (откат savepoint)", len(chunk), exc_info=True, ) return upserted def load_frt_mkd( db: Session, *, src_path: str | Path | None = None, work_dir: str | Path | None = None, node_id: int = DEFAULT_NODE_ID, region_code: int = DEFAULT_REGION_CODE, chunk_size: int = UPSERT_CHUNK_SIZE, dry_run: bool = False, ) -> dict[str, int]: """Скачивает (если src_path не задан) реестр МКД node_id, парсит, UPSERT в frt_mkd. src_path — локальный CSV (пропустить скачивание; тесты/ops-дебаг). work_dir — куда распаковывать скачанный zip (по умолчанию — временный каталог, удаляется после). dry_run — парс происходит, но НИ ОДНОЙ записи в БД не делается. Не коммитит (caller). """ tmp_ctx: tempfile.TemporaryDirectory[str] | None = None if src_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 и # domrf_kapremont_loader.py). Открытые данные без auth/PII — приемлемо. with httpx.Client(timeout=DOWNLOAD_TIMEOUT_SEC, verify=False) as client: src_path = fetch_frt_mkd_csv(dest, node_id=node_id, client=client) try: rows_by_guid = parse_frt_mkd_csv(src_path, region_code=region_code) rows = list(rows_by_guid.values()) upserted = 0 if dry_run else upsert_frt_mkd(db, rows, chunk_size=chunk_size) finally: if tmp_ctx is not None: tmp_ctx.cleanup() result = {"rows": len(rows), "upserted": upserted} logger.info("frt_mkd load DONE (dry_run=%s): %s", dry_run, result) return result # ───────────────────────────────────────────────────────────────────────────── # Backfill houses (COALESCE-семантика — только NULL-поля) # ───────────────────────────────────────────────────────────────────────────── # Матч: s.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid) — # тот же приоритетный COALESCE-джойн, что domrf_kapremont_loader/zhkh_flats_loader. # Каждое поле СВОИМ гейтом "h.col IS NULL AND s.col-в-границах" — ни одно поле, уже # заполненное другим источником, не перезаписывается. frt_matched_at проставляется один # раз (COALESCE(h.frt_matched_at, now())) при ЛЮБОМ найденном матче — метка «дом уже # сверялся с этим источником», даже если дозаполнять было уже нечего. _BACKFILL_HOUSES_SQL = text( """ UPDATE houses h SET year_built = CASE WHEN h.year_built IS NULL AND s.built_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int) THEN s.built_year ELSE h.year_built END, material_walls = COALESCE(h.material_walls, s.wall_material), material_floors = COALESCE(h.material_floors, s.floor_type), total_floors = CASE WHEN h.total_floors IS NULL AND s.floor_count_max BETWEEN 1 AND CAST(:floor_max AS int) THEN s.floor_count_max ELSE h.total_floors END, entrances = CASE WHEN h.entrances IS NULL AND s.entrance_count BETWEEN 1 AND CAST(:entrance_max AS int) THEN s.entrance_count ELSE h.entrances END, elevators_total = CASE WHEN h.elevators_total IS NULL AND s.elevators_count BETWEEN 0 AND CAST(:elevator_max AS int) THEN s.elevators_count ELSE h.elevators_total END, is_emergency = COALESCE(h.is_emergency, s.is_alarm), flat_count = CASE WHEN h.flat_count IS NULL AND s.living_quarters_count BETWEEN 0 AND CAST(:flat_max AS int) THEN s.living_quarters_count ELSE h.flat_count END, heat_supply_type = COALESCE(h.heat_supply_type, s.heating_type), gas_supply_type = COALESCE(h.gas_supply_type, s.gas_type), hot_water = COALESCE(h.hot_water, s.hot_water_type), area_land = CASE WHEN h.area_land IS NULL AND s.area_land BETWEEN 0 AND CAST(:area_land_max AS numeric) THEN s.area_land ELSE h.area_land END, foundation_type = COALESCE(h.foundation_type, s.foundation_type), frt_matched_at = COALESCE(h.frt_matched_at, now()) FROM frt_mkd s WHERE s.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid) AND ( (h.year_built IS NULL AND s.built_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)) OR (h.material_walls IS NULL AND s.wall_material IS NOT NULL) OR (h.material_floors IS NULL AND s.floor_type IS NOT NULL) OR (h.total_floors IS NULL AND s.floor_count_max BETWEEN 1 AND CAST(:floor_max AS int)) OR (h.entrances IS NULL AND s.entrance_count BETWEEN 1 AND CAST(:entrance_max AS int)) OR (h.elevators_total IS NULL AND s.elevators_count BETWEEN 0 AND CAST(:elevator_max AS int)) OR (h.is_emergency IS NULL AND s.is_alarm IS NOT NULL) OR (h.flat_count IS NULL AND s.living_quarters_count BETWEEN 0 AND CAST(:flat_max AS int)) OR (h.heat_supply_type IS NULL AND s.heating_type IS NOT NULL) OR (h.gas_supply_type IS NULL AND s.gas_type IS NOT NULL) OR (h.hot_water IS NULL AND s.hot_water_type IS NOT NULL) OR (h.area_land IS NULL AND s.area_land BETWEEN 0 AND CAST(:area_land_max AS numeric)) OR (h.foundation_type IS NULL AND s.foundation_type IS NOT NULL) OR h.frt_matched_at IS NULL ) """ ) _BACKFILL_HOUSES_COUNT_SQL = text( """ SELECT count(*) FROM houses h JOIN frt_mkd s ON s.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid) WHERE ( (h.year_built IS NULL AND s.built_year BETWEEN CAST(:ymin AS int) AND CAST(:ymax AS int)) OR (h.material_walls IS NULL AND s.wall_material IS NOT NULL) OR (h.material_floors IS NULL AND s.floor_type IS NOT NULL) OR (h.total_floors IS NULL AND s.floor_count_max BETWEEN 1 AND CAST(:floor_max AS int)) OR (h.entrances IS NULL AND s.entrance_count BETWEEN 1 AND CAST(:entrance_max AS int)) OR (h.elevators_total IS NULL AND s.elevators_count BETWEEN 0 AND CAST(:elevator_max AS int)) OR (h.is_emergency IS NULL AND s.is_alarm IS NOT NULL) OR (h.flat_count IS NULL AND s.living_quarters_count BETWEEN 0 AND CAST(:flat_max AS int)) OR (h.heat_supply_type IS NULL AND s.heating_type IS NOT NULL) OR (h.gas_supply_type IS NULL AND s.gas_type IS NOT NULL) OR (h.hot_water IS NULL AND s.hot_water_type IS NOT NULL) OR (h.area_land IS NULL AND s.area_land BETWEEN 0 AND CAST(:area_land_max AS numeric)) OR (h.foundation_type IS NULL AND s.foundation_type IS NOT NULL) OR h.frt_matched_at IS NULL ) """ ) def backfill_houses(db: Session, *, dry_run: bool = False) -> dict[str, int]: """COALESCE-добор houses.* из frt_mkd (только NULL-поля). Не коммитит (caller). dry_run — ноль записей, только SELECT count(*) по тому же предикату. """ bounds = { "ymin": YEAR_BUILT_MIN, "ymax": plausible_year_max(), "floor_max": FLOOR_COUNT_MAX_BOUND, "entrance_max": ENTRANCE_COUNT_MAX_BOUND, "elevator_max": ELEVATORS_COUNT_MAX_BOUND, "flat_max": FLAT_COUNT_MAX_BOUND, "area_land_max": AREA_LAND_MAX_BOUND, } if dry_run: would_update = db.execute(_BACKFILL_HOUSES_COUNT_SQL, bounds).scalar_one() result = {"would_update": would_update, "houses_updated": 0} logger.info("backfill_houses (frt_mkd) DRY-RUN: %s", result) return result updated = db.execute(_BACKFILL_HOUSES_SQL, bounds).rowcount result = {"houses_updated": updated} logger.info("backfill_houses (frt_mkd) DONE: %s", result) return result