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

feat(mera): загрузчик реестра МКД АИС ФРТ → обогащение houses #3426

Merged
lekss361 merged 1 commit from feat/mera-frt-mkd-loader into main 2026-09-08 21:41:25 +00:00
Conversation 0 Commits 1 Files changed 6 +1383
6 changed files with 1383 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

688
tradein-mvp/backend/app/services/frt_mkd_loader.py Normal file
View file

@ -0,0 +1,688 @@
"""АИС ППК «ФРТ» (бывш. Реформа ЖКХ) 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

44
tradein-mvp/backend/app/services/product_handlers.py
View file

@ -661,6 +661,49 @@ async def _job_domrf_kapremont_load(
ctx.runs.mark_failed(db, run_id, str(exc)[:1000], {})
# ── frt_mkd_load — АИС ППК ФРТ, реестр МКД region 66 (issue #frt-mkd) ────────
async def _job_frt_mkd_load(
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
"""Скачать реестр МКД АИС ФРТ (node_id) → staging frt_mkd → backfill houses.
Тело переиспользует те же функции, что и CLI (app/tasks/frt_mkd_load.py) — не
дублируем логику. node_id/region_code берутся из scrape_schedules.default_params
(мигр. 291), с фоллбеком на дефолты loader'а.
"""
from app.services.frt_mkd_loader import (
DEFAULT_NODE_ID,
DEFAULT_REGION_CODE,
backfill_houses,
load_frt_mkd,
)
node_id = int(params.get("node_id", DEFAULT_NODE_ID))
region_code = int(params.get("region_code", DEFAULT_REGION_CODE))
def _run() -> dict[str, int]:
load_counts = load_frt_mkd(db, node_id=node_id, region_code=region_code)
db.commit()
houses_counts = backfill_houses(db)
db.commit()
return {
"rows": load_counts["rows"],
"upserted": load_counts["upserted"],
"houses_updated": houses_counts["houses_updated"],
"total_seen": load_counts["rows"],
"new_count": houses_counts["houses_updated"],
}
loop = asyncio.get_event_loop()
try:
counters = await loop.run_in_executor(None, _run)
ctx.runs.mark_done(db, run_id, counters)
except Exception as exc:
logger.exception("scheduler: frt_mkd_load crashed run_id=%d", run_id)
db.rollback()
ctx.runs.mark_failed(db, run_id, str(exc)[:1000], {})
# ── purge_expired_trade_in_data — ЭТАП 4 B2C retention (152-ФЗ) ───────────────
async def _job_purge_expired_trade_in_data(
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
@ -833,6 +876,7 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
"house_imv_backfill": Handler(_job_house_imv_backfill, "house_imv_backfill"),
"house_dedup_merge": Handler(_job_house_dedup_merge, "house_dedup_merge"),
"domrf_kapremont_load": Handler(_job_domrf_kapremont_load, "domrf_kapremont_load"),
"frt_mkd_load": Handler(_job_frt_mkd_load, "frt_mkd_load"),
"purge_expired_trade_in_data": Handler(
_job_purge_expired_trade_in_data, "purge_expired_trade_in_data"
),

127
tradein-mvp/backend/app/tasks/frt_mkd_load.py Normal file
View file

@ -0,0 +1,127 @@
"""CLI: АИС ППК ФРТ (реестр МКД) → 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).
Тянет CSV реестра МКД (export/{node_id}, по умолчанию 110 = region 66) с открытых
данных АИС ППК ФРТ, UPSERT'ит в staging `frt_mkd` (мигр. 290), затем backfill_houses —
COALESCE-добор перечисленных houses.* полей (только NULL, никогда не перезаписывает).
Запуск из контейнера tradein-scraper (у него есть интернет к источнику):
python -m app.tasks.frt_mkd_load # полный прогон (node 110, region 66)
python -m app.tasks.frt_mkd_load --dry-run # без записи, только подсчёт
python -m app.tasks.frt_mkd_load --load-only # только staging, без backfill
python -m app.tasks.frt_mkd_load --backfill-only # staging уже загружена — только backfill
python -m app.tasks.frt_mkd_load --src-path /tmp/frt.csv # локальный CSV, без скачивания
В режиме --dry-run скачивание/парс (если применимо) происходят, но НИ ОДНОЙ записи
в БД не делается.
"""
from __future__ import annotations
import argparse
import logging
from app.core.db import SessionLocal
from app.services.frt_mkd_loader import (
DEFAULT_NODE_ID,
DEFAULT_REGION_CODE,
backfill_houses,
load_frt_mkd,
)
logger = logging.getLogger(__name__)
def build_parser() -> argparse.ArgumentParser:
"""Парсер CLI (вынесен для тестируемости флагов без запуска main)."""
parser = argparse.ArgumentParser(
description=(
"АИС ППК ФРТ loader: реестр МКД → frt_mkd staging → "
"houses.area_land/foundation_type/elevators_total + добор смежных полей"
)
)
parser.add_argument(
"--src-path", default=None, help="локальный CSV реестра МКД (пропустить скачивание)"
)
parser.add_argument(
"--work-dir",
default=None,
help="каталог для скачанного/распакованного CSV (по умолчанию — временный, "
"удаляется после)",
)
parser.add_argument(
"--node-id",
type=int,
default=DEFAULT_NODE_ID,
help=f"id ноды export АИС ФРТ (по умолчанию {DEFAULT_NODE_ID} = реестр МКД region 66)",
)
parser.add_argument(
"--region-code",
type=int,
default=DEFAULT_REGION_CODE,
help=f"код региона выгрузки для тега staging-строк (по умолчанию {DEFAULT_REGION_CODE})",
)
parser.add_argument(
"--dry-run", action="store_true", help="без записи в БД — только скачивание/парс/подсчёт"
)
parser.add_argument(
"--load-only",
action="store_true",
help="только скачать+распарсить+UPSERT staging, без backfill houses",
)
parser.add_argument(
"--backfill-only",
action="store_true",
help="пропустить скачивание/load staging (считаем что уже загружена этим же "
"или предыдущим прогоном) — только backfill houses",
)
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_frt_mkd(
db,
src_path=args.src_path,
work_dir=args.work_dir,
node_id=args.node_id,
region_code=args.region_code,
dry_run=args.dry_run,
)
if not args.dry_run:
db.commit()
logger.info("frt_mkd_load: load stage done: %s", load_counts)
houses_counts: dict[str, int] = {}
if not args.load_only:
houses_counts = backfill_houses(db, dry_run=args.dry_run)
if not args.dry_run:
db.commit()
logger.info(
"frt_mkd_load DONE: dry_run=%s load=%s houses=%s",
args.dry_run,
load_counts,
houses_counts,
)
finally:
db.close()
if __name__ == "__main__":
main()

100
tradein-mvp/backend/data/sql/290_frt_mkd_staging.sql Normal file
View file

@ -0,0 +1,100 @@
-- 290_frt_mkd_staging.sql
-- АИС ППК «ФРТ» (бывш. Реформа ЖКХ) — реестр МКД, обогащение houses (issue #frt-mkd).
--
-- КОНТЕКСТ: houses.area_land / foundation_type отсутствуют вовсе; elevators_count в
-- источнике — ОБЩЕЕ число лифтов, а в houses раздельно passenger/cargo — отдельная
-- колонка elevators_total, НЕ маппинг на passenger_elevators (иначе задвоение/потеря
-- типа). year_built/material_walls/total_floors/entrances/is_emergency/flat_count/
-- heat_supply_type/gas_supply_type/hot_water уже существуют — этот источник их
-- ДОБИРАЕТ (COALESCE, только NULL), не переопределяет.
--
-- ИСТОЧНИК: https://xn--80adsazqn.xn--p1aee.xn--p1ai/opendata/export/110 (node 110 =
-- реестр МКД региона 66), zip → CSV `export-reestrmkd-66-YYYYMMDD.csv`, UTF-8 BOM,
-- разделитель ';', 60 колонок, ~41.8k строк по СО. Проверено живьём 08.09.2026.
--
-- ЧТО НЕ ДЕЛАЕМ (осознанно, см. промпт задачи):
-- * project_type НЕ мапим в houses.series_name — свободный текст, дубль материала
-- стен (топ: пусто, «кирпичный», «нет данных», «панельный», разнобой регистра).
-- * energy_efficiency НЕ добавляем в houses — миграция 284 уже решила НЕ заводить
-- это поле; у ~61% домов значение «Не присвоен».
-- * playground/sportsground — id справочника (498/499/500), НЕ булев флаг; в staging
-- как есть, в houses.has_playground не пишем.
--
-- Идемпотентно: CREATE TABLE/ADD COLUMN IF NOT EXISTS, повторный прогон — no-op.
-- lock_timeout (#2752): ALTER houses берёт ACCESS EXCLUSIVE — SET LOCAL только внутри BEGIN.
BEGIN;
SET LOCAL lock_timeout = '5s';
CREATE TABLE IF NOT EXISTS frt_mkd (
houseguid text PRIMARY KEY, -- ФИАС GUID дома (АИС ФРТ houseguid)
region_code smallint NOT NULL, -- код региона выгрузки (66 = Свердловская обл.)
address text,
built_year smallint,
exploitation_start_year smallint,
project_type text, -- сырьё; НЕ мапится в houses (см. шапку)
house_type text,
is_alarm boolean, -- «Да»/«Нет» из источника
floor_count_max smallint,
floor_count_min smallint,
entrance_count smallint,
elevators_count smallint, -- ОБЩЕЕ число лифтов (не passenger/cargo)
energy_efficiency text, -- сырьё; НЕ мапится в houses (мигр. 284)
quarters_count integer,
living_quarters_count integer,
unliving_quarters_count integer,
area_total numeric(12, 2),
area_residential numeric(12, 2),
area_non_residential numeric(12, 2),
area_common_property numeric(12, 2),
area_land numeric(12, 2),
parking_square numeric(12, 2),
playground integer, -- id справочника благоустройства, НЕ булев флаг
sportsground integer, -- id справочника благоустройства, НЕ булев флаг
other_beautification text,
foundation_type text,
floor_type text,
wall_material text,
basement_area numeric(12, 2),
chute_type text,
chute_count smallint,
heating_type text,
hot_water_type text,
cold_water_type text,
sewerage_type text,
gas_type text,
ventilation_type text,
firefighting_type text,
drainage_type text,
management_organization_id bigint,
method_of_forming_overhaul_fund text,
loaded_at timestamptz NOT NULL DEFAULT now()
);
COMMENT ON TABLE frt_mkd IS
'АИС ППК ФРТ (бывш. Реформа ЖКХ) — реестр МКД, open data export/110, region 66. '
'Ключ houseguid (ФИАС GUID). Заполняется app/services/frt_mkd_loader.py, '
'мигр. 290. Матчится к houses через COALESCE(gar_house_guid, house_fias_id, '
'zhkh_house_guid).';
ALTER TABLE houses ADD COLUMN IF NOT EXISTS area_land numeric(12, 2);
COMMENT ON COLUMN houses.area_land IS
'мигр. 290: площадь земельного участка, м2, источник АИС ППК ФРТ (frt_mkd.area_land).';
ALTER TABLE houses ADD COLUMN IF NOT EXISTS foundation_type text;
COMMENT ON COLUMN houses.foundation_type IS
'мигр. 290: тип фундамента, источник АИС ППК ФРТ (frt_mkd.foundation_type).';
ALTER TABLE houses ADD COLUMN IF NOT EXISTS elevators_total integer;
COMMENT ON COLUMN houses.elevators_total IS
'мигр. 290: ОБЩЕЕ число лифтов (без разбивки passenger/cargo), источник АИС ППК ФРТ '
'(frt_mkd.elevators_count). НЕ путать с passenger_elevators/cargo_elevators — те '
'раздельные и из других источников.';
ALTER TABLE houses ADD COLUMN IF NOT EXISTS frt_matched_at timestamptz;
COMMENT ON COLUMN houses.frt_matched_at IS
'мигр. 290: момент, когда backfill_houses() из frt_mkd_loader.py нашёл матч по '
'houseguid для этого дома (проставляется один раз, даже если полей для дозаполнения '
'уже не осталось).';
COMMIT;

37
tradein-mvp/backend/data/sql/291_scrape_schedules_seed_frt_mkd_load.sql Normal file
View file

@ -0,0 +1,37 @@
-- 291_scrape_schedules_seed_frt_mkd_load.sql
-- Расписание для frt_mkd_load (АИС ППК ФРТ, реестр МКД region 66, issue #frt-mkd).
--
-- Источник обновляется нерегулярно (реестр МКД, не суточный поток) — interval_days=30
-- по аналогии с ДОМ.РФ капремонт (мигр. 176/... каденция месячная). node_id/region_code
-- в default_params — параметры handler'а (app/services/product_handlers.py
-- _job_frt_mkd_load), сам loader их же принимает как аргументы load_frt_mkd().
--
-- ENABLED=false — как у domrf_kapremont_load и domclick_detail_backfill: включение
-- отдельным осознанным шагом после деплоя и дымовой пробы (ZIP 3.2 МБ / CSV 41 МБ —
-- смотрим таймауты и нагрузку на первом ручном прогоне, не в фоне).
--
-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)), 290_frt_mkd_staging.sql.
-- Идемпотентно: ON CONFLICT (source) DO NOTHING.
BEGIN;
INSERT INTO scrape_schedules (
source,
enabled,
window_start_hour,
window_end_hour,
next_run_at,
default_params
)
VALUES
(
'frt_mkd_load',
false,
2,
5,
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC',
'{"node_id": 110, "region_code": 66, "interval_days": 30}'::jsonb
)
ON CONFLICT (source) DO NOTHING;
COMMIT;

387
tradein-mvp/backend/tests/test_frt_mkd_loader.py Normal file
View file

@ -0,0 +1,387 @@
"""Тесты АИС ППК ФРТ loader'а (app/services/frt_mkd_loader.py, мигр. 290/291, #frt-mkd).
Coverage:
- Чистые парс-хелперы: clean_text (мусорные значения-заглушки), parse_decimal_comma,
parse_int_field (+ sanity-гейт границ), parse_bool_da_net.
- parse_frt_mkd_csv — фикстуры собраны из РЕАЛЬНЫХ строк export-reestrmkd-66-20260901.csv
(id 6556123 «д. 41» — плотно заполненная строка; id 6477794 «д. 99» — почти вся строка
из значений-заглушек «Не заполнено»/пусто; id 9090614 — is_alarm='Да', project_type
в нижнем регистре 'нет').
- fetch_frt_mkd_csv — скачивание+распаковка zip, HTTP замокан через httpx.MockTransport.
- Статические asserts по SQL: _UPSERT_SQL (ON CONFLICT, IS DISTINCT FROM, CAST),
_BACKFILL_HOUSES_SQL/_COUNT_SQL (COALESCE-only-NULL-гейт, houseguid COALESCE-матч,
без `:x::type`).
- backfill_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
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import httpx
import pytest
from app.services import frt_mkd_loader as fml
# ─────────────────────────────────────────────────────────────────────────────
# Фикстуры: РЕАЛЬНЫЕ строки export-reestrmkd-66-20260901.csv (region 66)
# ─────────────────────────────────────────────────────────────────────────────
_HEADER = (
"id;region_id;area_id;city_id;street_id;shortname_region;formalname_region;"
"shortname_area;formalname_area;shortname_city;formalname_city;shortname_street;"
"formalname_street;house_number;building;block;letter;address;houseguid;"
"management_organization_id;built_year;exploitation_start_year;project_type;house_type;"
"is_alarm;method_of_forming_overhaul_fund;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;electrical_type;electrical_entries_count;"
"heating_type;hot_water_type;cold_water_type;sewerage_type;"
"sewerage_cesspools_volume;gas_type;ventilation_type;firefighting_type;drainage_type"
)
# д. 41 — плотно заполненная строка, is_alarm='Нет', elevators_count=0.
_ROW_DENSE = (
'6556123;;;;;;;;;;;;;41;;;;"д. 41";d6bad775-f0c8-b0d6-87ab-07c43513529e;8915908;1917;'
'1917;Кирпичный;"Многоквартирный дом";Нет;"На счете регионального оператора";8;6;1;0;'
'"Не присвоен";13;13;0;930,60;930,60;0,00;80,40;1100,00;0,00;500;500;;Ленточный;'
"Деревянные;Кирпич;0,00;Отсутствует;0;Центральное;2;Центральное;"
'"Открытая с отбором сетевой воды на горячее водоснабжение из тепловой сети";'
'Центральное;Центральное;0,00;Отсутствует;"Приточная вентиляция";Отсутствует;'
'"Наружные водостоки"'
)
# д. 99 — почти вся строка из значений-заглушек «Не заполнено» / пусто.
_ROW_NOISE = (
'6477794;;;;;;;;;;;;;99;;;;"д. 99";c42aa407-54e8-0917-ca15-439b5b784903;;;1954;;'
'"Многоквартирный дом";Нет;"Не заполнено";;;;;"Не заполнено";0;;;10156,60;7039,70;;;;;'
'498;498;;"Не заполнено";"Не заполнено";"Не заполнено";;"Не заполнено";;'
'"Не заполнено";;"Не заполнено";"Не заполнено";"Не заполнено";"Не заполнено";;'
'"Не заполнено";"Не заполнено";"Не заполнено";"Не заполнено"'
)
# is_alarm='Да' (аварийный), project_type='нет' (нижний регистр — проверка noise-гейта).
_ROW_ALARM = (
"9090614;92b30014-4d52-4e2e-892d-928142b924bf;;a4ab722e-453a-4aed-bb73-728a05e2e27f;"
"36a4e8b6-5c87-4f67-9ea1-12ac63f32939;обл.;Свердловская;;;г.;Алапаевск;км;123;3;;;;"
'"обл. Свердловская, г. Алапаевск, км. 123, д. 3";'
"37e2d973-9a07-4faa-86f5-95e417f86059;8917294;1911;1911;нет;"
'"Многоквартирный дом";Да;"Не определен";1;1;0;0;"Не присвоен";5;5;0;110,50;110,50;'
"0,00;0,00;2400,00;0,00;500;500;нет;Ленточный;Деревянные;Деревянные;0,00;Отсутствует;"
"0;Центральное;1;Печное;Отсутствует;Отсутствует;Отсутствует;0,00;Отсутствует;"
"Отсутствует;Отсутствует;Отсутствует"
)
FRT_CSV = "\n".join([_HEADER, _ROW_DENSE, _ROW_NOISE, _ROW_ALARM]) + "\n"
@pytest.fixture
def frt_csv_path(tmp_path: Path) -> Path:
p = tmp_path / "frt_mkd.csv"
p.write_text(FRT_CSV, encoding="utf-8-sig")
return p
# ─────────────────────────────────────────────────────────────────────────────
# clean_text / parse_decimal_comma / parse_int_field / parse_bool_da_net
# ─────────────────────────────────────────────────────────────────────────────
def test_clean_text_filters_noise_values() -> None:
assert fml.clean_text("Кирпич") == "Кирпич"
assert fml.clean_text("") is None
assert fml.clean_text(None) is None
assert fml.clean_text("нет данных") is None
assert fml.clean_text("Не заполнено") is None
assert fml.clean_text("отсутствует") is None
assert fml.clean_text("данные отсутствуют") is None
assert fml.clean_text("нет") is None
assert fml.clean_text(" Кирпич ") == "Кирпич"
def test_parse_decimal_comma() -> None:
assert fml.parse_decimal_comma("930,60") == pytest.approx(930.60)
assert fml.parse_decimal_comma("100.5") == pytest.approx(100.5)
assert fml.parse_decimal_comma("") is None
assert fml.parse_decimal_comma(None) is None
assert fml.parse_decimal_comma("Не заполнено") is None
def test_parse_int_field() -> None:
assert fml.parse_int_field("1917") == 1917
assert fml.parse_int_field("8") == 8
assert fml.parse_int_field("") is None
assert fml.parse_int_field(None) is None
assert fml.parse_int_field("Не заполнено") is None
assert fml.parse_int_field("5,0") == 5
def test_parse_int_field_sanity_gate() -> None:
ymax = fml.plausible_year_max()
assert fml.parse_int_field("1200", min_value=1850, max_value=ymax) is None
assert fml.parse_int_field(str(ymax + 10), min_value=1850, max_value=ymax) is None
assert fml.parse_int_field("1917", min_value=1850, max_value=ymax) == 1917
def test_parse_bool_da_net() -> None:
assert fml.parse_bool_da_net("Да") is True
assert fml.parse_bool_da_net("да") is True
assert fml.parse_bool_da_net("Нет") is False
assert fml.parse_bool_da_net("") is None
assert fml.parse_bool_da_net(None) is None
assert fml.parse_bool_da_net("Не заполнено") is None
# ─────────────────────────────────────────────────────────────────────────────
# parse_frt_mkd_csv
# ─────────────────────────────────────────────────────────────────────────────
def test_parse_frt_mkd_csv_dense_row(frt_csv_path: Path) -> None:
rows = fml.parse_frt_mkd_csv(frt_csv_path, region_code=66)
assert set(rows.keys()) == {
"d6bad775-f0c8-b0d6-87ab-07c43513529e",
"c42aa407-54e8-0917-ca15-439b5b784903",
"37e2d973-9a07-4faa-86f5-95e417f86059",
}
dense = rows["d6bad775-f0c8-b0d6-87ab-07c43513529e"]
assert dense.region_code == 66
assert dense.built_year == 1917
assert dense.is_alarm is False
assert dense.floor_count_max == 8
assert dense.entrance_count == 1
assert dense.elevators_count == 0
assert dense.wall_material == "Кирпич"
assert dense.floor_type == "Деревянные"
assert dense.area_land == pytest.approx(1100.00)
assert dense.area_total == pytest.approx(930.60)
assert dense.living_quarters_count == 13
assert dense.heating_type == "Центральное"
assert dense.playground == 500 # id справочника, не булев
assert dense.management_organization_id == 8915908
def test_parse_frt_mkd_csv_noise_row_collapses_to_none(frt_csv_path: Path) -> None:
rows = fml.parse_frt_mkd_csv(frt_csv_path, region_code=66)
noise = rows["c42aa407-54e8-0917-ca15-439b5b784903"]
# Почти вся строка — "Не заполнено"/пусто → None, не должно попасть в staging как текст.
assert noise.wall_material is None
assert noise.foundation_type is None
assert noise.floor_type is None
assert noise.built_year is None # built_year пуст в этой строке
assert noise.exploitation_start_year == 1954
assert noise.area_total == pytest.approx(10156.60)
assert noise.is_alarm is False
def test_parse_frt_mkd_csv_alarm_row_and_lowercase_noise(frt_csv_path: Path) -> None:
rows = fml.parse_frt_mkd_csv(frt_csv_path, region_code=66)
alarm = rows["37e2d973-9a07-4faa-86f5-95e417f86059"]
assert alarm.is_alarm is True
# project_type='нет' (нижний регистр) — та же заглушка, что 'Нет'/'НЕТ' → None.
assert alarm.project_type is None
assert alarm.built_year == 1911
def test_parse_frt_mkd_csv_skips_blank_houseguid(tmp_path: Path) -> None:
csv_text = _HEADER + "\n" + (";" * _HEADER.count(";")) + "\n"
p = tmp_path / "blank.csv"
p.write_text(csv_text, encoding="utf-8-sig")
assert fml.parse_frt_mkd_csv(p, region_code=66) == {}
# ─────────────────────────────────────────────────────────────────────────────
# fetch_frt_mkd_csv — скачивание+распаковка 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_frt_mkd_csv_downloads_and_extracts(tmp_path: Path) -> None:
zip_bytes = _zip_bytes("export-reestrmkd-66-20260901.csv", FRT_CSV)
expected_url = f"{fml.BASE_URL}/opendata/export/110"
def handler(request: httpx.Request) -> httpx.Response:
if str(request.url) == expected_url:
return httpx.Response(200, content=zip_bytes)
raise AssertionError(f"unexpected URL {request.url}")
transport = httpx.MockTransport(handler)
with httpx.Client(transport=transport) as client:
out_path = fml.fetch_frt_mkd_csv(tmp_path, node_id=110, client=client)
assert out_path.exists() and out_path.name.endswith(".csv")
rows = fml.parse_frt_mkd_csv(out_path, region_code=66)
assert len(rows) == 3
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"):
fml._extract_csv_from_zip(buf.getvalue(), tmp_path)
# ─────────────────────────────────────────────────────────────────────────────
# load_frt_mkd — dry_run не пишет (упаковано в тонкий MagicMock-flow)
# ─────────────────────────────────────────────────────────────────────────────
def test_load_frt_mkd_dry_run_parses_but_does_not_upsert(frt_csv_path: Path) -> None:
db = MagicMock()
result = fml.load_frt_mkd(db, src_path=frt_csv_path, region_code=66, dry_run=True)
assert result == {"rows": 3, "upserted": 0}
db.execute.assert_not_called()
# ─────────────────────────────────────────────────────────────────────────────
# Статические asserts по _UPSERT_SQL (psycopg v3, ON CONFLICT, идемпотентность)
# ─────────────────────────────────────────────────────────────────────────────
_UPSERT_SQL = str(fml._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(:houseguid AS text)" in _UPSERT_SQL
assert "CAST(:built_year AS smallint)" in _UPSERT_SQL
assert "CAST(:is_alarm AS boolean)" in _UPSERT_SQL
assert "CAST(:area_land AS numeric)" in _UPSERT_SQL
def test_upsert_sql_idempotent_on_conflict() -> None:
flat = re.sub(r"\s+", " ", _UPSERT_SQL)
assert "INSERT INTO frt_mkd" in flat
assert "ON CONFLICT (houseguid) DO UPDATE SET" in flat
assert "frt_mkd.wall_material IS DISTINCT FROM EXCLUDED.wall_material" in flat
assert "frt_mkd.area_land IS DISTINCT FROM EXCLUDED.area_land" in flat
def test_upsert_uses_savepoint_per_chunk_and_does_not_commit() -> None:
import inspect
src = inspect.getsource(fml.upsert_frt_mkd)
assert "with db.begin_nested():" in src
assert "db.commit()" not in src
def test_load_frt_mkd_does_not_commit() -> None:
import inspect
src = inspect.getsource(fml.load_frt_mkd)
assert "db.commit()" not in src
# ─────────────────────────────────────────────────────────────────────────────
# Статические asserts: _BACKFILL_HOUSES_SQL / _COUNT_SQL
# ─────────────────────────────────────────────────────────────────────────────
_BACKFILL_SQL = re.sub(r"\s+", " ", str(fml._BACKFILL_HOUSES_SQL.text))
_BACKFILL_COUNT_SQL = re.sub(r"\s+", " ", str(fml._BACKFILL_HOUSES_COUNT_SQL.text))
def test_backfill_sql_matches_by_houseguid_coalesce_priority() -> None:
assert (
"s.houseguid = COALESCE(h.gar_house_guid, h.house_fias_id, h.zhkh_house_guid)"
in _BACKFILL_SQL
)
def test_backfill_sql_never_overwrites_existing_non_null() -> None:
# Каждый таргет своим гейтом h.<col> IS NULL — существующее значение не трогается.
assert "WHEN h.year_built IS NULL" in _BACKFILL_SQL
assert "COALESCE(h.material_walls, s.wall_material)" in _BACKFILL_SQL
assert "COALESCE(h.material_floors, s.floor_type)" in _BACKFILL_SQL
assert "WHEN h.total_floors IS NULL" in _BACKFILL_SQL
assert "WHEN h.entrances IS NULL" in _BACKFILL_SQL
assert "WHEN h.elevators_total IS NULL" in _BACKFILL_SQL
assert "COALESCE(h.is_emergency, s.is_alarm)" in _BACKFILL_SQL
assert "WHEN h.flat_count IS NULL" in _BACKFILL_SQL
assert "COALESCE(h.heat_supply_type, s.heating_type)" in _BACKFILL_SQL
assert "COALESCE(h.gas_supply_type, s.gas_type)" in _BACKFILL_SQL
assert "COALESCE(h.hot_water, s.hot_water_type)" in _BACKFILL_SQL
assert "WHEN h.area_land IS NULL" in _BACKFILL_SQL
assert "COALESCE(h.foundation_type, s.foundation_type)" in _BACKFILL_SQL
def test_backfill_sql_frt_matched_at_set_once() -> None:
assert "frt_matched_at = COALESCE(h.frt_matched_at, now())" in _BACKFILL_SQL
assert "h.frt_matched_at IS NULL" in _BACKFILL_SQL
def test_backfill_sql_does_not_touch_elevators_or_series_name() -> None:
# НЕ мапим elevators_count в passenger_elevators, НЕ мапим project_type в series_name.
assert "passenger_elevators" not in _BACKFILL_SQL
assert "series_name" not in _BACKFILL_SQL
assert "energy_efficiency" not in _BACKFILL_SQL
def test_backfill_sql_no_psycopg_v3_colon_colon_cast() -> None:
assert not re.search(r":\w+::", _BACKFILL_SQL)
assert not re.search(r":\w+::", _BACKFILL_COUNT_SQL)
def test_backfill_houses_does_not_commit() -> None:
import inspect
src = inspect.getsource(fml.backfill_houses)
assert "db.commit()" not in src
# ─────────────────────────────────────────────────────────────────────────────
# backfill_houses — dry_run vs real-run (MagicMock db)
# ─────────────────────────────────────────────────────────────────────────────
def test_backfill_houses_real_run_reads_rowcount() -> None:
db = MagicMock()
db.execute.return_value.rowcount = 17
out = fml.backfill_houses(db, dry_run=False)
assert out == {"houses_updated": 17}
sql_text = str(db.execute.call_args[0][0])
assert "UPDATE houses" in sql_text
bind_params = db.execute.call_args[0][1]
assert bind_params["ymin"] == fml.YEAR_BUILT_MIN
assert bind_params["ymax"] == fml.plausible_year_max()
assert bind_params["area_land_max"] == fml.AREA_LAND_MAX_BOUND
def test_backfill_houses_dry_run_only_counts_no_update() -> None:
db = MagicMock()
db.execute.return_value.scalar_one.return_value = 9
out = fml.backfill_houses(db, dry_run=True)
assert out == {"would_update": 9, "houses_updated": 0}
sql_text = str(db.execute.call_args[0][0]).upper()
assert "UPDATE" not in sql_text
assert "SELECT COUNT(*)" in sql_text
def test_duplicate_houseguid_rows_are_merged_field_wise(tmp_path: Path) -> None:
"""Дубли houseguid в источнике взаимодополняющи — слить, а не затереть.
Реальная выгрузка региона 66 (2026-09-01): 41 790 строк, 912 повторяющихся
houseguid = 968 лишних строк, причём у одной строки пары заполнен area_total,
а у парной — нет. «Последняя побеждает» теряла бы эти значения.
"""
header = "houseguid;address;built_year;area_total;wall_material"
csv_text = "\n".join(
[
header,
"6809f339-cf37-4e2c-b689-7ded226a9e51;обл. Свердловская, г. Алапаевск;1960;1698,20;",
"6809f339-cf37-4e2c-b689-7ded226a9e51;обл. Свердловская, г. Алапаевск;;;Кирпич",
]
)
path = tmp_path / "dup.csv"
path.write_text(csv_text, encoding="utf-8-sig")
rows = fml.parse_frt_mkd_csv(path, region_code=66)
assert len(rows) == 1
row = rows["6809f339-cf37-4e2c-b689-7ded226a9e51"]
assert row.built_year == 1960
assert row.area_total is not None
assert row.wall_material == "Кирпич"