feat(tradein): ГАР-loader знаменателя — apartments-per-house → gar_house_flats + match к houses

tradein-mvp/backend/app/services/gar_flats_loader.py

@@ -0,0 +1,651 @@
+"""ГАР/ФИАС loader: «квартир на дом» (знаменатель «доли квартир дома в продаже»).
+
+Парсит региональный ГАР-дамп (новый формат, по-региональная папка, напр. `66/`) и
+наполняет таблицу `gar_house_flats` числом помещений-квартир под каждым домом, затем
+матчит к `houses` по НОРМАЛИЗОВАННОМУ адресу (`tradein_normalize_short_addr`) и
+проставляет `houses.gar_flat_count` (мигр. 143).
+
+Формат ГАР (данные — в АТРИБУТАХ элементов, не в тексте):
+  - AS_ADDR_OBJ_*.XML    — адресные объекты (регион/город/улица).
+        OBJECTID, OBJECTGUID, NAME, TYPENAME, LEVEL, ISACTUAL, ISACTIVE.
+  - AS_HOUSES_*.XML      — дома.
+        OBJECTID, OBJECTGUID, HOUSENUM, ADDNUM1/2, HOUSETYPE, ADDTYPE1/2, ISACTUAL, ISACTIVE.
+  - AS_APARTMENTS_*.XML  — помещения/квартиры.
+        OBJECTID, NUMBER, APARTTYPE, ISACTUAL, ISACTIVE.
+  - AS_MUN_HIERARCHY_*.XML — иерархия (приоритетна; ADM — fallback).
+        OBJECTID, PARENTOBJID, ISACTIVE — объект OBJECTID лежит под PARENTOBJID.
+
+Подсчёт квартир: число OBJECTID помещений, чей PARENTOBJID в иерархии == OBJECTID дома
+(помещения — прямые дети дома). Адрес дома: parent дома в иерархии = улица/адр-объект;
+поднимаемся по parent'ам до города/региона, собирая «<TYPENAME NAME>» сегменты, + номер дома.
+
+Память: ВСЕ файлы (особенно AS_APARTMENTS/AS_MUN_HIERARCHY для региона 66) — многогигабайтные.
+Парсим стримингом через `lxml.etree.iterparse` с очисткой обработанных элементов + их
+предыдущих сиблингов (никогда не грузим дерево целиком). Единственная крупная структура в
+памяти — set OBJECTID активных помещений (≈десятки МБ на регион).
+
+Соответствие нормализации: `norm_address` считается тем же `tradein_normalize_short_addr`,
+что применяется к адресам `houses` при матче (мигр. 063), но — для устойчивости к
+не-ЕКБ-населённым пунктам — поверх «улица + номер дома» строки (эквивалентно
+fn(full_address) для ЕКБ, где fn срезает регион/город). full_address хранится отдельно как
+человекочитаемая provenance-цепочка.
+
+psycopg v3: SQL через `text(...)` использует `CAST(:x AS type)`, НИКОГДА `:x::type`.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import time
+from collections import defaultdict
+from collections.abc import Iterator
+from dataclasses import dataclass, field
+
+from lxml import etree
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Глобы файлов ГАР (матчим case-insensitive: .XML/.xml).
+ADDR_OBJ_PREFIX = "AS_ADDR_OBJ_"
+HOUSES_PREFIX = "AS_HOUSES_"
+APARTMENTS_PREFIX = "AS_APARTMENTS_"
+MUN_HIER_PREFIX = "AS_MUN_HIERARCHY_"
+ADM_HIER_PREFIX = "AS_ADM_HIERARCHY_"  # fallback, если MUN отсутствует
+
+DEFAULT_BATCH_SIZE = 2000
+
+# ADDTYPE дома (AS_ADDHOUSE_TYPES): 1=Корпус, 2=Строение, 3=Сооружение. Рендер — аббревиатура.
+_ADDTYPE_ABBR: dict[str, str] = {"1": "к", "2": "стр", "3": "соор"}
+
+# LEVEL/типы региона-уровня: рендерим «<NAME> <TYPENAME>» (Свердловская обл), остальное —
+# «<TYPENAME> <NAME>» (ул Шаумяна / г Екатеринбург). Косметика full_address (для людей);
+# на матч не влияет (norm_address считается из «улица + номер»).
+_REGION_LIKE_TYPES = {"обл", "область", "край", "респ", "республика", "ао", "аобл"}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# In-memory модели
+# ─────────────────────────────────────────────────────────────────────────────
+@dataclass(slots=True)
+class AddrObj:
+    object_id: int
+    guid: str
+    name: str
+    typename: str
+    level: str
+
+
+@dataclass(slots=True)
+class GarHouse:
+    object_id: int
+    guid: str
+    housenum: str
+    addnum1: str
+    addnum2: str
+    addtype1: str
+    addtype2: str
+
+
+@dataclass(slots=True)
+class GarHouseRow:
+    """Строка-приёмник для gar_house_flats (norm_address считает SQL-fn в INSERT)."""
+
+    house_guid: str
+    object_id: int
+    region_code: str
+    flat_count: int
+    full_address: str
+    street_name: str | None
+    house_num: str
+    norm_src: str  # вход для tradein_normalize_short_addr (улица + номер)
+    gar_version: str
+
+    def to_params(self) -> dict[str, object]:
+        return {
+            "house_guid": self.house_guid,
+            "object_id": self.object_id,
+            "region_code": self.region_code,
+            "flat_count": self.flat_count,
+            "full_address": self.full_address,
+            "street_name": self.street_name,
+            "house_num": self.house_num,
+            "norm_src": self.norm_src,
+            "gar_version": self.gar_version,
+        }
+
+
+@dataclass(slots=True)
+class GarParseStats:
+    addr_objects: int = 0
+    houses: int = 0
+    apartments: int = 0  # суммарно активных помещений (= sum(flat_count))
+    rows: int = 0  # построенных строк gar_house_flats
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Поиск файлов
+# ─────────────────────────────────────────────────────────────────────────────
+def _find_files(dir_path: str, prefix: str) -> list[str]:
+    """Файлы dir_path, чьё имя начинается с prefix и кончается .xml (case-insensitive)."""
+    out: list[str] = []
+    for name in sorted(os.listdir(dir_path)):
+        upper = name.upper()
+        if upper.startswith(prefix.upper()) and upper.endswith(".XML"):
+            out.append(os.path.join(dir_path, name))
+    return out
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Стриминговый парсер (lxml iterparse + очистка)
+# ─────────────────────────────────────────────────────────────────────────────
+def _local_name(tag: object) -> str:
+    """Локальное имя тега без namespace; '' для не-элементов (комментарии/PI)."""
+    if not isinstance(tag, str):
+        return ""
+    return tag.rsplit("}", 1)[-1]
+
+
+def _stream_rows(path: str, marker_attr: str) -> Iterator[etree._Element]:
+    """Yield row-элементов файла, у которых есть атрибут marker_attr, освобождая память.
+
+    После выдачи элемента (caller обязан прочитать атрибуты СРАЗУ в теле цикла) элемент
+    очищается, а его уже обработанные предыдущие сиблинги удаляются из дерева — так
+    память не растёт на многогигабайтных файлах. `huge_tree=True` снимает security-лимит
+    libxml2 на размер; `recover=True` терпит мелкие дефекты XML.
+    """
+    context = etree.iterparse(
+        path, events=("end",), recover=True, huge_tree=True, resolve_entities=False
+    )
+    for _event, elem in context:
+        is_elem = isinstance(elem.tag, str)
+        if is_elem and marker_attr in elem.attrib:
+            yield elem
+        # Чистим текущий элемент и срезаем предшествующие сиблинги (bounded memory).
+        elem.clear()
+        parent = elem.getparent()
+        if parent is not None:
+            while elem.getprevious() is not None:
+                del parent[0]
+    del context
+
+
+def _is_active(elem: etree._Element, *, need_actual: bool) -> bool:
+    """ISACTIVE='1' (+ опц. ISACTUAL='1'). Отсутствующий атрибут считаем НЕ активным."""
+    if elem.get("ISACTIVE") != "1":
+        return False
+    if need_actual and elem.get("ISACTUAL") != "1":
+        return False
+    return True
+
+
+def _to_int(value: str | None) -> int | None:
+    if value is None:
+        return None
+    value = value.strip()
+    if not value:
+        return None
+    try:
+        return int(value)
+    except ValueError:
+        return None
+
+
+def parse_addr_objects(paths: list[str]) -> dict[int, AddrObj]:
+    """OBJECTID → AddrObj по активным (ISACTUAL=1 AND ISACTIVE=1) адресным объектам."""
+    out: dict[int, AddrObj] = {}
+    for path in paths:
+        for elem in _stream_rows(path, "OBJECTGUID"):
+            if not _is_active(elem, need_actual=True):
+                continue
+            oid = _to_int(elem.get("OBJECTID"))
+            guid = elem.get("OBJECTGUID")
+            if oid is None or not guid:
+                continue
+            out[oid] = AddrObj(
+                object_id=oid,
+                guid=guid,
+                name=(elem.get("NAME") or "").strip(),
+                typename=(elem.get("TYPENAME") or "").strip(),
+                level=(elem.get("LEVEL") or "").strip(),
+            )
+    return out
+
+
+def parse_houses(paths: list[str]) -> dict[int, GarHouse]:
+    """OBJECTID → GarHouse по активным (ISACTUAL=1 AND ISACTIVE=1) домам."""
+    out: dict[int, GarHouse] = {}
+    for path in paths:
+        for elem in _stream_rows(path, "HOUSENUM"):
+            if not _is_active(elem, need_actual=True):
+                continue
+            oid = _to_int(elem.get("OBJECTID"))
+            guid = elem.get("OBJECTGUID")
+            if oid is None or not guid:
+                continue
+            out[oid] = GarHouse(
+                object_id=oid,
+                guid=guid,
+                housenum=(elem.get("HOUSENUM") or "").strip(),
+                addnum1=(elem.get("ADDNUM1") or "").strip(),
+                addnum2=(elem.get("ADDNUM2") or "").strip(),
+                addtype1=(elem.get("ADDTYPE1") or "").strip(),
+                addtype2=(elem.get("ADDTYPE2") or "").strip(),
+            )
+    return out
+
+
+def parse_apartment_ids(paths: list[str]) -> set[int]:
+    """Set OBJECTID активных (ISACTUAL=1 AND ISACTIVE=1) помещений/квартир.
+
+    Тип помещения (APARTTYPE) не фильтруем — знаменатель = все активные помещения дома
+    (квартиры + нежилые); это сознательное приближение «всего квартир в доме».
+    """
+    out: set[int] = set()
+    for path in paths:
+        for elem in _stream_rows(path, "OBJECTID"):
+            if not _is_active(elem, need_actual=True):
+                continue
+            oid = _to_int(elem.get("OBJECTID"))
+            if oid is not None:
+                out.add(oid)
+    return out
+
+
+def parse_hierarchy(
+    paths: list[str],
+    apartment_ids: set[int],
+    house_ids: set[int],
+    addr_ids: set[int],
+) -> tuple[dict[int, int], dict[int, int]]:
+    """Единый стриминговый проход по иерархии (только активные ISACTIVE=1).
+
+    Возвращает (apt_count, child_to_parent):
+      - apt_count[parent_house_objectid] = число помещений-детей этого дома. Считается
+        проверкой членства OBJECTID в apartment_ids; каждый помещение учитывается ОДИН раз
+        (discard из set после учёта — защита от дубль-строк иерархии + bounded memory).
+      - child_to_parent[oid] = PARENTOBJID, но ТОЛЬКО для домов и адр-объектов (нужно для
+        сборки адреса) — иерархия региона целиком в память не грузится.
+    """
+    apt_count: dict[int, int] = defaultdict(int)
+    child_to_parent: dict[int, int] = {}
+    for path in paths:
+        for elem in _stream_rows(path, "PARENTOBJID"):
+            if elem.get("ISACTIVE") != "1":
+                continue
+            oid = _to_int(elem.get("OBJECTID"))
+            pid = _to_int(elem.get("PARENTOBJID"))
+            if oid is None or pid is None:
+                continue
+            if oid in apartment_ids:
+                apt_count[pid] += 1
+                apartment_ids.discard(oid)
+            if oid in house_ids or oid in addr_ids:
+                child_to_parent[oid] = pid
+    return dict(apt_count), child_to_parent
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Сборка адреса
+# ─────────────────────────────────────────────────────────────────────────────
+def _render_segment(ao: AddrObj) -> str:
+    """Косметический рендер адр-сегмента для full_address."""
+    name = ao.name.strip()
+    typ = ao.typename.strip()
+    if not typ:
+        return name
+    if typ.lower() in _REGION_LIKE_TYPES:
+        return f"{name} {typ}".strip()
+    return f"{typ} {name}".strip()
+
+
+def render_house_num(house: GarHouse) -> str:
+    """Номер дома: HOUSENUM (+ корпус/строение из ADDTYPE/ADDNUM, если есть)."""
+    parts: list[str] = []
+    if house.housenum:
+        parts.append(house.housenum)
+    if house.addnum1:
+        abbr = _ADDTYPE_ABBR.get(house.addtype1, "")
+        parts.append(f"{abbr}{house.addnum1}".strip())
+    if house.addnum2:
+        abbr = _ADDTYPE_ABBR.get(house.addtype2, "")
+        parts.append(f"{abbr}{house.addnum2}".strip())
+    return " ".join(p for p in parts if p)
+
+
+def _walk_segments(
+    start_parent: int | None,
+    addr_objs: dict[int, AddrObj],
+    child_to_parent: dict[int, int],
+) -> list[AddrObj]:
+    """Сегменты адр-объектов снизу вверх (улица → … → регион), начиная с start_parent."""
+    segments: list[AddrObj] = []
+    visited: set[int] = set()
+    cur = start_parent
+    while cur is not None and cur in addr_objs and cur not in visited:
+        visited.add(cur)
+        segments.append(addr_objs[cur])
+        cur = child_to_parent.get(cur)
+    return segments
+
+
+def assemble_house_address(
+    house: GarHouse,
+    addr_objs: dict[int, AddrObj],
+    child_to_parent: dict[int, int],
+) -> tuple[str, str | None, str]:
+    """(full_address, street_name, house_num) для дома, поднимаясь по иерархии parent'ов.
+
+    street_name = непосредственный адр-родитель дома (обычно улица), «<TYPENAME> <NAME>».
+    full_address = регион…город…улица (косметически) + номер дома.
+    """
+    house_num = render_house_num(house)
+    parent_id = child_to_parent.get(house.object_id)
+    segments = _walk_segments(parent_id, addr_objs, child_to_parent)
+
+    street_name = _render_segment(segments[0]) if segments else None
+
+    # full_address: разворачиваем (регион → улица) + номер.
+    chain = [_render_segment(s) for s in reversed(segments)]
+    if house_num:
+        chain.append(house_num)
+    full_address = ", ".join(c for c in chain if c)
+    return full_address, street_name, house_num
+
+
+def _norm_src(street_name: str | None, house_num: str, full_address: str) -> str:
+    """Вход для tradein_normalize_short_addr: «улица, номер» (устойчиво к не-ЕКБ нас.пунктам).
+
+    Падаем на full_address, если улица не определилась (редкий дом без адр-родителя).
+    """
+    if street_name and house_num:
+        return f"{street_name}, {house_num}"
+    if street_name:
+        return street_name
+    return full_address
+
+
+def build_gar_rows(
+    addr_objs: dict[int, AddrObj],
+    houses: dict[int, GarHouse],
+    apt_count: dict[int, int],
+    child_to_parent: dict[int, int],
+    region_code: str,
+    gar_version: str,
+) -> list[GarHouseRow]:
+    """Строки gar_house_flats: на каждый дом — адрес + flat_count (0, если нет помещений)."""
+    rows: list[GarHouseRow] = []
+    for oid, house in houses.items():
+        full_address, street_name, house_num = assemble_house_address(
+            house, addr_objs, child_to_parent
+        )
+        rows.append(
+            GarHouseRow(
+                house_guid=house.guid,
+                object_id=oid,
+                region_code=region_code,
+                flat_count=apt_count.get(oid, 0),
+                full_address=full_address,
+                street_name=street_name,
+                house_num=house_num,
+                norm_src=_norm_src(street_name, house_num, full_address),
+                gar_version=gar_version,
+            )
+        )
+    return rows
+
+
+def parse_gar_region(
+    dir_path: str, region_code: str, gar_version: str
+) -> tuple[list[GarHouseRow], GarParseStats]:
+    """Парсит распакованный ГАР-каталог региона → (строки gar_house_flats, статистика).
+
+    Чистая (без БД) часть — тестируется на синтетических фикстурах.
+    """
+    if not os.path.isdir(dir_path):
+        raise NotADirectoryError(f"GAR dir not found: {dir_path}")
+
+    addr_paths = _find_files(dir_path, ADDR_OBJ_PREFIX)
+    house_paths = _find_files(dir_path, HOUSES_PREFIX)
+    apt_paths = _find_files(dir_path, APARTMENTS_PREFIX)
+    hier_paths = _find_files(dir_path, MUN_HIER_PREFIX)
+    if not hier_paths:
+        hier_paths = _find_files(dir_path, ADM_HIER_PREFIX)
+        if hier_paths:
+            logger.warning("GAR %s: AS_MUN_HIERARCHY отсутствует, использую ADM fallback", dir_path)
+
+    if not house_paths:
+        raise FileNotFoundError(f"AS_HOUSES_*.XML не найдены в {dir_path}")
+    if not hier_paths:
+        raise FileNotFoundError(f"AS_MUN_HIERARCHY/ADM_HIERARCHY_*.XML не найдены в {dir_path}")
+
+    logger.info(
+        "GAR parse %s: addr=%d houses=%d apartments=%d hierarchy=%d файлов",
+        dir_path,
+        len(addr_paths),
+        len(house_paths),
+        len(apt_paths),
+        len(hier_paths),
+    )
+
+    addr_objs = parse_addr_objects(addr_paths)
+    houses = parse_houses(house_paths)
+    apartment_ids = parse_apartment_ids(apt_paths)
+    apartments_total = len(apartment_ids)
+    logger.info(
+        "GAR parse %s: addr_obj=%d houses=%d active_apartments=%d",
+        dir_path,
+        len(addr_objs),
+        len(houses),
+        apartments_total,
+    )
+
+    apt_count, child_to_parent = parse_hierarchy(
+        hier_paths, apartment_ids, set(houses.keys()), set(addr_objs.keys())
+    )
+
+    rows = build_gar_rows(addr_objs, houses, apt_count, child_to_parent, region_code, gar_version)
+    counted = sum(r.flat_count for r in rows)
+    stats = GarParseStats(
+        addr_objects=len(addr_objs),
+        houses=len(houses),
+        apartments=counted,
+        rows=len(rows),
+    )
+    logger.info(
+        "GAR parse %s: построено строк=%d, помещений учтено под домами=%d (из %d активных)",
+        dir_path,
+        stats.rows,
+        counted,
+        apartments_total,
+    )
+    return rows, stats
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Upsert в gar_house_flats
+# ─────────────────────────────────────────────────────────────────────────────
+# norm_address считает SQL-fn прямо в INSERT (эквивалентно «UPDATE после insert»), вход —
+# «улица, номер» (:norm_src). CAST(:norm_src AS text) — psycopg v3 discipline (не :x::text).
+_INSERT_SQL = text(
+    """
+    INSERT INTO gar_house_flats (
+        house_guid, object_id, region_code, flat_count,
+        full_address, norm_address, street_name, house_num, gar_version
+    )
+    VALUES (
+        :house_guid, :object_id, :region_code, :flat_count,
+        :full_address,
+        tradein_normalize_short_addr(CAST(:norm_src AS text)),
+        :street_name, :house_num, :gar_version
+    )
+    ON CONFLICT (house_guid) DO UPDATE SET
+        object_id    = EXCLUDED.object_id,
+        region_code  = EXCLUDED.region_code,
+        flat_count   = EXCLUDED.flat_count,
+        full_address = EXCLUDED.full_address,
+        norm_address = EXCLUDED.norm_address,
+        street_name  = EXCLUDED.street_name,
+        house_num    = EXCLUDED.house_num,
+        loaded_at    = now(),
+        gar_version  = EXCLUDED.gar_version
+    """
+)
+
+
+def _chunks(items: list[GarHouseRow], size: int) -> Iterator[list[GarHouseRow]]:
+    for i in range(0, len(items), size):
+        yield items[i : i + size]
+
+
+def upsert_gar_houses(
+    db: Session, rows: list[GarHouseRow], *, batch_size: int = DEFAULT_BATCH_SIZE
+) -> int:
+    """Батчевый UPSERT строк в gar_house_flats. НЕ коммитит (коммитит caller).
+
+    SAVEPOINT (`begin_nested`) на КАЖДЫЙ батч: сбойный батч откатывается изолированно, не
+    срывая всю транзакцию и не врёт счётчику. ON CONFLICT (house_guid) → идемпотентно.
+    """
+    upserted = 0
+    failed_batches = 0
+    for batch in _chunks(rows, batch_size):
+        params = [r.to_params() for r in batch]
+        try:
+            with db.begin_nested():
+                db.execute(_INSERT_SQL, params)
+            upserted += len(batch)
+        except Exception:
+            failed_batches += 1
+            logger.warning(
+                "gar upsert: батч из %d строк сбойнул (пропущен)", len(batch), exc_info=True
+            )
+    if failed_batches:
+        logger.warning("gar upsert: сбойных батчей=%d", failed_batches)
+    logger.info("gar upsert: upsert'нуто строк=%d", upserted)
+    return upserted
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Матчер ГАР → houses
+# ─────────────────────────────────────────────────────────────────────────────
+# Идемпотентный set-based UPDATE. На каждый norm_address берём ОДНУ ГАР-строку
+# (DISTINCT ON … ORDER BY flat_count DESC, house_guid) — детерминированный tie-break:
+# максимальный flat_count, при равенстве — лексикографически меньший house_guid.
+# Houses-сторона: tradein_normalize_short_addr(COALESCE(short_address, full_address, address))
+# — тот же нормализатор, что и в gar_house_flats.norm_address. Предикат
+# `gar_flat_count IS DISTINCT FROM` → повторный прогон не трогает уже совпавшие строки.
+_MATCH_SQL = text(
+    """
+    WITH gar_pick AS (
+        SELECT DISTINCT ON (norm_address)
+            norm_address, house_guid, flat_count
+        FROM gar_house_flats
+        WHERE flat_count > 0
+          AND norm_address IS NOT NULL
+          AND norm_address <> ''
+          AND (
+              CAST(:region AS text) IS NULL
+              OR region_code = CAST(:region AS text)
+          )
+        ORDER BY norm_address, flat_count DESC, house_guid
+    )
+    UPDATE houses h
+    SET gar_house_guid   = gp.house_guid,
+        gar_flat_count   = gp.flat_count,
+        gar_matched_at   = now(),
+        gar_match_method = 'norm_address'
+    FROM gar_pick gp
+    WHERE tradein_normalize_short_addr(
+              COALESCE(h.short_address, h.full_address, h.address)
+          ) = gp.norm_address
+      AND h.gar_flat_count IS DISTINCT FROM gp.flat_count
+    """
+)
+
+
+def match_houses_to_gar(db: Session, *, region_code: str | None = None) -> int:
+    """Матч gar_house_flats → houses по нормализованному адресу. НЕ коммитит (caller коммитит).
+
+    Возвращает число обновлённых домов. Идемпотентно (plain UPDATE, IS DISTINCT FROM gate).
+    """
+    result = db.execute(_MATCH_SQL, {"region": region_code})
+    matched = result.rowcount
+    logger.info("gar match: проставлено gar_flat_count домам=%d (region=%s)", matched, region_code)
+    return matched
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Orchestration
+# ─────────────────────────────────────────────────────────────────────────────
+@dataclass(slots=True)
+class GarLoadResult:
+    addr_objects: int = 0
+    houses: int = 0
+    apartments: int = 0
+    upserted: int = 0
+    duration_sec: float = field(default=0.0)
+
+    def to_counts(self) -> dict[str, int]:
+        return {
+            "addr_objects": self.addr_objects,
+            "houses": self.houses,
+            "apartments": self.apartments,
+            "upserted": self.upserted,
+            "duration_sec": int(self.duration_sec),
+        }
+
+
+def load_gar_region(
+    dir_path: str,
+    region_code: str,
+    gar_version: str,
+    *,
+    db: Session | None = None,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> GarLoadResult:
+    """Парс ГАР-каталога региона → UPSERT в gar_house_flats (коммит). Без матча.
+
+    Возвращает счётчики (houses, apartments, upserted). Если db не передан — открывает
+    собственную SessionLocal. Матч (match_houses_to_gar) — отдельный шаг (см. CLI).
+    """
+    start = time.monotonic()
+    own_session = db is None
+    if db is None:
+        from app.core.db import SessionLocal
+
+        db = SessionLocal()
+
+    try:
+        rows, stats = parse_gar_region(dir_path, region_code, gar_version)
+        upserted = upsert_gar_houses(db, rows, batch_size=batch_size)
+        db.commit()
+        result = GarLoadResult(
+            addr_objects=stats.addr_objects,
+            houses=stats.houses,
+            apartments=stats.apartments,
+            upserted=upserted,
+            duration_sec=time.monotonic() - start,
+        )
+        logger.info(
+            "load_gar_region: region=%s version=%s houses=%d apartments=%d upserted=%d "
+            "duration=%.1fs",
+            region_code,
+            gar_version,
+            result.houses,
+            result.apartments,
+            result.upserted,
+            result.duration_sec,
+        )
+        return result
+    except Exception:
+        logger.exception("load_gar_region: region=%s FAILED", region_code)
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        raise
+    finally:
+        if own_session:
+            db.close()

tradein-mvp/backend/app/tasks/gar_flats_load.py

@@ -0,0 +1,74 @@
+"""CLI: загрузка ГАР-знаменателя «квартир на дом» + матч к houses (мигр. 143).
+
+Запуск (каталог — УЖЕ распакованный ГАР региона, напр. папка `66/` из gar_xml.zip):
+
+    python -m app.tasks.gar_flats_load --dir /data/gar/66 --region 66 --version 2026-06-01
+
+Делает два шага в одной транзакционной сессии:
+  1. load_gar_region — стриминговый парс XML → UPSERT gar_house_flats (коммит).
+  2. match_houses_to_gar — UPDATE houses.gar_flat_count по нормализованному адресу (коммит).
+
+Многогигабайтный ДАМП качается/распаковывается отдельно (ops-шаг, см.
+docs/gar-flats-runbook.md) — этот лоадер потребляет уже распакованные XML локально.
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+from datetime import date
+
+from app.core.db import SessionLocal
+from app.services.gar_flats_loader import load_gar_region, match_houses_to_gar
+
+logger = logging.getLogger(__name__)
+
+
+def run_gar_flats_load(dir_path: str, region_code: str, gar_version: str) -> dict[str, int]:
+    """Парс+UPSERT (load_gar_region) затем матч (match_houses_to_gar). Возвращает счётчики."""
+    db = SessionLocal()
+    try:
+        load = load_gar_region(dir_path, region_code, gar_version, db=db)
+        matched = match_houses_to_gar(db, region_code=region_code)
+        db.commit()
+    finally:
+        db.close()
+
+    counts = load.to_counts()
+    counts["houses_matched"] = matched
+    logger.info(
+        "gar_flats_load DONE: region=%s version=%s houses=%d apartments=%d upserted=%d "
+        "houses_matched=%d duration=%ds",
+        region_code,
+        gar_version,
+        counts["houses"],
+        counts["apartments"],
+        counts["upserted"],
+        matched,
+        counts["duration_sec"],
+    )
+    return counts
+
+
+def main() -> None:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    parser = argparse.ArgumentParser(
+        description="ГАР loader: квартир на дом → gar_house_flats + match к houses"
+    )
+    parser.add_argument("--dir", required=True, help="каталог с распакованными ГАР XML региона")
+    parser.add_argument("--region", default="66", help="код региона (по умолчанию 66)")
+    parser.add_argument(
+        "--version",
+        default=date.today().isoformat(),
+        help="версия ГАР-дампа YYYY-MM-DD (по умолчанию сегодня)",
+    )
+    args = parser.parse_args()
+
+    run_gar_flats_load(args.dir, args.region, args.version)
+
+
+if __name__ == "__main__":
+    main()

tradein-mvp/backend/pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
     "httpx[socks]>=0.27.0",        # для geocoder + scrapers; [socks] тянет socksio для socks5://
     "tenacity>=9.0.0",            # retry с exp backoff
     "selectolax>=0.3.0",          # быстрый HTML парсинг для scrapers
+    "lxml>=5.0.0",                # стриминговый iterparse для многогигабайтных ГАР XML (#143)
     "segno>=1.6.0",               # QR-код для PDF shareable URL
     "curl-cffi>=0.7.0",           # impersonate=chrome120 для Cian/Avito (TLS fingerprint)
     "python-multipart>=0.0.9",    # FastAPI UploadFile — загрузка фото квартиры (#394)

tradein-mvp/backend/tests/test_gar_flats_loader.py

@@ -0,0 +1,367 @@
+"""Тесты ГАР-loader'а знаменателя (app/services/gar_flats_loader.py, мигр. 143).
+
+Структура (mirror'ит test_cadastral_geo_match):
+  - ДЕТЕРМИНИРОВАННОЕ ядро (без БД): синтетические крошечные ГАР-XML фикстуры в tmp_path →
+    парс → assert apartment-count на дом, сборка адреса, фильтрация ISACTUAL/ISACTIVE,
+    дом с 0 квартир → flat_count 0, мульти-сегментная сборка адреса.
+  - СТАТИЧЕСКИЕ asserts по SQL (_INSERT_SQL / _MATCH_SQL): ON CONFLICT, psycopg-v3 CAST,
+    DISTINCT ON tie-break, идемпотентный gate.
+  - ОПЦИОНАЛЬНЫЙ real-Postgres тест upsert+match → houses.gar_flat_count (self-SKIP без БД,
+    создаёт TEMP-таблицы + fn в транзакции и откатывает).
+"""
+
+from __future__ import annotations
+
+import inspect
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+# settings нужен DSN на импорте (как в sibling-тестах); ядро статично/без БД.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.services import gar_flats_loader as gfl
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Фикстуры: крошечный синтетический ГАР-дамп
+# ─────────────────────────────────────────────────────────────────────────────
+# Иерархия:
+#   1 регион «обл Свердловская»
+#     └─ 2 город «г Екатеринбург»
+#          ├─ 3 улица «ул Шаумяна»
+#          │    ├─ 100 дом «10»   → квартиры 1001,1002,1003 (3 активные) + 1004(ISACTIVE=0)
+#          │    └─ 200 дом «5 к1» → квартиры 2001,2002 (2 активные)
+#          └─ 30 улица «ул Ленина»
+#               └─ 300 дом «7»    → 0 квартир
+#   inactive: адр-объект 9 (ISACTIVE=0), дом 900 (ISACTUAL=0), кв 1005 (ISACTUAL=0)
+
+_ADDR_OBJ_XML = """<?xml version="1.0" encoding="utf-8"?>
+<ADDRESSOBJECTS>
+  <OBJECT OBJECTID="1" OBJECTGUID="g-region" NAME="Свердловская" TYPENAME="обл"
+          LEVEL="1" ISACTUAL="1" ISACTIVE="1"/>
+  <OBJECT OBJECTID="2" OBJECTGUID="g-city" NAME="Екатеринбург" TYPENAME="г"
+          LEVEL="5" ISACTUAL="1" ISACTIVE="1"/>
+  <OBJECT OBJECTID="3" OBJECTGUID="g-shaumyana" NAME="Шаумяна" TYPENAME="ул"
+          LEVEL="8" ISACTUAL="1" ISACTIVE="1"/>
+  <OBJECT OBJECTID="30" OBJECTGUID="g-lenina" NAME="Ленина" TYPENAME="ул"
+          LEVEL="8" ISACTUAL="1" ISACTIVE="1"/>
+  <OBJECT OBJECTID="9" OBJECTGUID="g-dead" NAME="Снесённая" TYPENAME="ул"
+          LEVEL="8" ISACTUAL="1" ISACTIVE="0"/>
+</ADDRESSOBJECTS>
+"""
+
+_HOUSES_XML = """<?xml version="1.0" encoding="utf-8"?>
+<HOUSES>
+  <HOUSE OBJECTID="100" OBJECTGUID="h-a" HOUSENUM="10" HOUSETYPE="2"
+         ISACTUAL="1" ISACTIVE="1"/>
+  <HOUSE OBJECTID="200" OBJECTGUID="h-b" HOUSENUM="5" ADDNUM1="1" ADDTYPE1="1"
+         HOUSETYPE="2" ISACTUAL="1" ISACTIVE="1"/>
+  <HOUSE OBJECTID="300" OBJECTGUID="h-c" HOUSENUM="7" HOUSETYPE="2"
+         ISACTUAL="1" ISACTIVE="1"/>
+  <HOUSE OBJECTID="900" OBJECTGUID="h-dead" HOUSENUM="99" HOUSETYPE="2"
+         ISACTUAL="0" ISACTIVE="1"/>
+</HOUSES>
+"""
+
+_APARTMENTS_XML = """<?xml version="1.0" encoding="utf-8"?>
+<APARTMENTS>
+  <APARTMENT OBJECTID="1001" NUMBER="1" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
+  <APARTMENT OBJECTID="1002" NUMBER="2" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
+  <APARTMENT OBJECTID="1003" NUMBER="3" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
+  <APARTMENT OBJECTID="1004" NUMBER="4" APARTTYPE="2" ISACTUAL="1" ISACTIVE="0"/>
+  <APARTMENT OBJECTID="1005" NUMBER="5" APARTTYPE="2" ISACTUAL="0" ISACTIVE="1"/>
+  <APARTMENT OBJECTID="2001" NUMBER="1" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
+  <APARTMENT OBJECTID="2002" NUMBER="2" APARTTYPE="2" ISACTUAL="1" ISACTIVE="1"/>
+</APARTMENTS>
+"""
+
+# Иерархия: пара (OBJECTID лежит под PARENTOBJID).
+_MUN_HIER_XML = """<?xml version="1.0" encoding="utf-8"?>
+<ITEMS>
+  <ITEM OBJECTID="2" PARENTOBJID="1" ISACTIVE="1"/>
+  <ITEM OBJECTID="3" PARENTOBJID="2" ISACTIVE="1"/>
+  <ITEM OBJECTID="30" PARENTOBJID="2" ISACTIVE="1"/>
+  <ITEM OBJECTID="100" PARENTOBJID="3" ISACTIVE="1"/>
+  <ITEM OBJECTID="200" PARENTOBJID="3" ISACTIVE="1"/>
+  <ITEM OBJECTID="300" PARENTOBJID="30" ISACTIVE="1"/>
+  <ITEM OBJECTID="1001" PARENTOBJID="100" ISACTIVE="1"/>
+  <ITEM OBJECTID="1002" PARENTOBJID="100" ISACTIVE="1"/>
+  <ITEM OBJECTID="1003" PARENTOBJID="100" ISACTIVE="1"/>
+  <ITEM OBJECTID="1004" PARENTOBJID="100" ISACTIVE="1"/>
+  <ITEM OBJECTID="2001" PARENTOBJID="200" ISACTIVE="1"/>
+  <ITEM OBJECTID="2002" PARENTOBJID="200" ISACTIVE="1"/>
+  <ITEM OBJECTID="9999" PARENTOBJID="100" ISACTIVE="0"/>
+</ITEMS>
+"""
+
+
+@pytest.fixture
+def gar_dir(tmp_path: Path) -> str:
+    """Пишет синтетический ГАР-каталог, возвращает путь."""
+    (tmp_path / "AS_ADDR_OBJ_20260601_x.XML").write_text(_ADDR_OBJ_XML, encoding="utf-8")
+    (tmp_path / "AS_HOUSES_20260601_x.XML").write_text(_HOUSES_XML, encoding="utf-8")
+    (tmp_path / "AS_APARTMENTS_20260601_x.XML").write_text(_APARTMENTS_XML, encoding="utf-8")
+    (tmp_path / "AS_MUN_HIERARCHY_20260601_x.XML").write_text(_MUN_HIER_XML, encoding="utf-8")
+    return str(tmp_path)
+
+
+def _rows_by_guid(rows: list[gfl.GarHouseRow]) -> dict[str, gfl.GarHouseRow]:
+    return {r.house_guid: r for r in rows}
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Парс: отдельные файлы
+# ─────────────────────────────────────────────────────────────────────────────
+def test_parse_addr_objects_filters_inactive(gar_dir: str) -> None:
+    paths = gfl._find_files(gar_dir, gfl.ADDR_OBJ_PREFIX)
+    addr = gfl.parse_addr_objects(paths)
+    assert set(addr.keys()) == {1, 2, 3, 30}  # снесённая (9, ISACTIVE=0) исключена
+    assert addr[3].name == "Шаумяна" and addr[3].typename == "ул"
+
+
+def test_parse_houses_filters_non_actual(gar_dir: str) -> None:
+    paths = gfl._find_files(gar_dir, gfl.HOUSES_PREFIX)
+    houses = gfl.parse_houses(paths)
+    assert set(houses.keys()) == {100, 200, 300}  # 900 (ISACTUAL=0) исключён
+    assert houses[200].housenum == "5" and houses[200].addnum1 == "1"
+
+
+def test_parse_apartment_ids_excludes_inactive_and_non_actual(gar_dir: str) -> None:
+    paths = gfl._find_files(gar_dir, gfl.APARTMENTS_PREFIX)
+    apts = gfl.parse_apartment_ids(paths)
+    # 1004 (ISACTIVE=0) и 1005 (ISACTUAL=0) исключены.
+    assert apts == {1001, 1002, 1003, 2001, 2002}
+
+
+def test_parse_hierarchy_counts_apartments_per_house(gar_dir: str) -> None:
+    addr = gfl.parse_addr_objects(gfl._find_files(gar_dir, gfl.ADDR_OBJ_PREFIX))
+    houses = gfl.parse_houses(gfl._find_files(gar_dir, gfl.HOUSES_PREFIX))
+    apts = gfl.parse_apartment_ids(gfl._find_files(gar_dir, gfl.APARTMENTS_PREFIX))
+    hier_paths = gfl._find_files(gar_dir, gfl.MUN_HIER_PREFIX)
+    apt_count, child_to_parent = gfl.parse_hierarchy(hier_paths, apts, set(houses), set(addr))
+    # ISACTIVE=0 строка иерархии (9999) не учитывается; 1004 не в apartment_ids → не считается.
+    assert apt_count.get(100) == 3
+    assert apt_count.get(200) == 2
+    assert apt_count.get(300, 0) == 0  # дом без квартир
+    # child_to_parent заполнен ТОЛЬКО для домов и адр-объектов (не для квартир).
+    assert child_to_parent[100] == 3
+    assert child_to_parent[3] == 2
+    assert 1001 not in child_to_parent
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Сборка адреса + строки
+# ─────────────────────────────────────────────────────────────────────────────
+def test_build_rows_apartment_counts_and_zero_house(gar_dir: str) -> None:
+    rows, stats = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
+    by_guid = _rows_by_guid(rows)
+    assert by_guid["h-a"].flat_count == 3
+    assert by_guid["h-b"].flat_count == 2
+    assert by_guid["h-c"].flat_count == 0  # дом с 0 квартир → flat_count 0
+    assert stats.houses == 3
+    assert stats.apartments == 5  # суммарно учтено под домами
+    assert stats.rows == 3
+
+
+def test_multi_segment_address_assembly(gar_dir: str) -> None:
+    rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
+    a = _rows_by_guid(rows)["h-a"]
+    # street_name = непосредственный родитель (улица), «<TYPENAME> <NAME>».
+    assert a.street_name == "ул Шаумяна"
+    assert a.house_num == "10"
+    # full_address — мульти-сегмент регион→город→улица + номер (регион — name-first).
+    assert a.full_address == "Свердловская обл, г Екатеринбург, ул Шаумяна, 10"
+    # norm_src (вход нормализатора) — «улица, номер».
+    assert a.norm_src == "ул Шаумяна, 10"
+
+
+def test_house_num_with_korpus(gar_dir: str) -> None:
+    rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
+    b = _rows_by_guid(rows)["h-b"]
+    # ADDNUM1=1 ADDTYPE1=1 (Корпус) → «5 к1».
+    assert b.house_num == "5 к1"
+    assert b.street_name == "ул Шаумяна"
+
+
+def test_region_and_version_propagated(gar_dir: str) -> None:
+    rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
+    assert all(r.region_code == "66" for r in rows)
+    assert all(r.gar_version == "2026-06-01" for r in rows)
+
+
+def test_render_house_num_unit() -> None:
+    h = gfl.GarHouse(
+        object_id=1,
+        guid="x",
+        housenum="12",
+        addnum1="3",
+        addnum2="",
+        addtype1="2",
+        addtype2="",
+    )
+    assert gfl.render_house_num(h) == "12 стр3"  # ADDTYPE1=2 = Строение
+
+
+def test_adm_hierarchy_fallback(tmp_path: Path) -> None:
+    """Без AS_MUN_HIERARCHY использует AS_ADM_HIERARCHY (fallback)."""
+    (tmp_path / "AS_ADDR_OBJ_x.XML").write_text(_ADDR_OBJ_XML, encoding="utf-8")
+    (tmp_path / "AS_HOUSES_x.XML").write_text(_HOUSES_XML, encoding="utf-8")
+    (tmp_path / "AS_APARTMENTS_x.XML").write_text(_APARTMENTS_XML, encoding="utf-8")
+    (tmp_path / "AS_ADM_HIERARCHY_x.XML").write_text(_MUN_HIER_XML, encoding="utf-8")
+    rows, _ = gfl.parse_gar_region(str(tmp_path), "66", "2026-06-01")
+    assert _rows_by_guid(rows)["h-a"].flat_count == 3
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Статические asserts по SQL (psycopg v3, ON CONFLICT, tie-break)
+# ─────────────────────────────────────────────────────────────────────────────
+_INSERT_SQL = str(gfl._INSERT_SQL.text)
+_MATCH_SQL = str(gfl._MATCH_SQL.text)
+
+
+def test_insert_is_idempotent_on_conflict() -> None:
+    flat = re.sub(r"\s+", " ", _INSERT_SQL)
+    assert "INSERT INTO gar_house_flats" in flat
+    assert "ON CONFLICT (house_guid) DO UPDATE SET" in flat
+    # norm_address считает SQL-fn в INSERT (= «UPDATE после insert»).
+    assert "tradein_normalize_short_addr(CAST(:norm_src AS text))" in flat
+
+
+def test_match_distinct_on_tiebreak_max_flat_count() -> None:
+    flat = re.sub(r"\s+", " ", _MATCH_SQL)
+    # Tie-break: на norm_address — строка с макс flat_count, затем меньший house_guid.
+    assert "DISTINCT ON (norm_address)" in flat
+    assert "ORDER BY norm_address, flat_count DESC, house_guid" in flat
+    # houses-сторона: тот же нормализатор поверх COALESCE(short/full/address).
+    assert (
+        "tradein_normalize_short_addr( COALESCE(h.short_address, h.full_address, h.address) )"
+        in flat
+    )
+    # Идемпотентность повторного прогона.
+    assert "h.gar_flat_count IS DISTINCT FROM gp.flat_count" in flat
+    assert "gar_match_method = 'norm_address'" in flat
+    # Только дома с известным знаменателем матчим.
+    assert "flat_count > 0" in flat
+
+
+def test_no_psycopg_v3_colon_colon_cast() -> None:
+    # Только исполняемый SQL (модульный docstring специально содержит анти-паттерн как памятку).
+    assert not re.search(r":\w+::", _INSERT_SQL)
+    assert not re.search(r":\w+::", _MATCH_SQL)
+
+
+def test_upsert_uses_savepoint_per_batch() -> None:
+    src = inspect.getsource(gfl.upsert_gar_houses)
+    assert "with db.begin_nested():" in src
+    # upsert НЕ коммитит сам (коммитит caller).
+    assert "db.commit()" not in src
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Опциональный real-Postgres тест upsert+match (self-skip без БД)
+# ─────────────────────────────────────────────────────────────────────────────
+def _live_session() -> Any | None:
+    try:
+        from sqlalchemy import create_engine, text
+        from sqlalchemy.orm import sessionmaker
+
+        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
+        if not dsn or "localhost:5432/test" in dsn:
+            return None
+        engine = create_engine(dsn, future=True)
+        conn = engine.connect()
+        conn.execute(text("SELECT 1"))
+        conn.close()
+        return sessionmaker(bind=engine, future=True)()
+    except Exception:
+        return None
+
+
+# Тот же нормализатор, что в мигр. 063 — создаём в транзакции теста, откатываем.
+_NORMALIZER_FN = """
+CREATE OR REPLACE FUNCTION tradein_normalize_short_addr(addr text)
+RETURNS text LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$
+  SELECT trim(both ' ,.' FROM
+    regexp_replace(
+      regexp_replace(
+        regexp_replace(
+          regexp_replace(
+            regexp_replace(
+              regexp_replace(addr, '^\\s*(?:Россия|РФ|Российская\\s+Федерация)\\s*,\\s*', '', 'i'),
+              '^[А-ЯЁа-яё][А-ЯЁа-яё\\s-]+\\s+обл(?:асть|\\.)\\s*,\\s*', '', 'i'),
+            '^г\\.?\\s*[А-ЯЁ][А-ЯЁа-яё-]+\\s*,\\s*', '', 'i'),
+          '\\s*·\\s*.+$', '', 'i'),
+        ',\\s*(?:кв\\.?|корп\\.?|к\\.?|оф\\.?|пом\\.?|подъезд)\\s*\\d+.*$', '', 'i'),
+      '\\s{2,}', ' ', 'g'));
+$$;
+"""
+
+
+@pytest.mark.skipif(_live_session() is None, reason="нет доступной Postgres test-БД")
+def test_upsert_and_match_populates_gar_flat_count(gar_dir: str) -> None:
+    """С реальной Postgres: upsert ГАР-строк + match → houses.gar_flat_count проставлен."""
+    from sqlalchemy import text
+
+    db = _live_session()
+    assert db is not None
+    try:
+        db.execute(text(_NORMALIZER_FN))
+        db.execute(
+            text(
+                "CREATE TEMP TABLE gar_house_flats ("
+                "  house_guid text PRIMARY KEY, object_id bigint, region_code text NOT NULL,"
+                "  flat_count int, full_address text, norm_address text, street_name text,"
+                "  house_num text, loaded_at timestamptz NOT NULL DEFAULT now(), gar_version text"
+                ") ON COMMIT DROP"
+            )
+        )
+        db.execute(
+            text(
+                "CREATE TEMP TABLE houses ("
+                "  id serial PRIMARY KEY, short_address text, full_address text, address text,"
+                "  gar_house_guid text, gar_flat_count int, gar_matched_at timestamptz,"
+                "  gar_match_method text"
+                ") ON COMMIT DROP"
+            )
+        )
+        # Дом «ул Шаумяна, 10» совпадёт с ГАР h-a (flat_count=3); «ул Ленина, 7» — h-c (0) не матч.
+        db.execute(
+            text(
+                "INSERT INTO houses (short_address) VALUES " "('ул Шаумяна, 10'), ('ул Ленина, 7')"
+            )
+        )
+
+        rows, _ = gfl.parse_gar_region(gar_dir, "66", "2026-06-01")
+        upserted = gfl.upsert_gar_houses(db, rows, batch_size=2)
+        assert upserted == 3
+
+        # norm_address посчитан SQL-fn = «ул Шаумяна, 10».
+        norm = db.execute(
+            text("SELECT norm_address FROM gar_house_flats WHERE house_guid='h-a'")
+        ).scalar()
+        assert norm == "ул Шаумяна, 10"
+
+        matched = gfl.match_houses_to_gar(db, region_code="66")
+        assert matched == 1
+        # Идемпотентность: повторный матч ничего не трогает.
+        assert gfl.match_houses_to_gar(db, region_code="66") == 0
+
+        shaumyana = db.execute(
+            text(
+                "SELECT gar_flat_count, gar_match_method FROM houses "
+                "WHERE short_address='ул Шаумяна, 10'"
+            )
+        ).fetchone()
+        assert shaumyana is not None and shaumyana[0] == 3
+        assert shaumyana[1] == "norm_address"
+        # Дом без квартир (flat_count=0) НЕ матчится.
+        lenina = db.execute(
+            text("SELECT gar_flat_count FROM houses WHERE short_address='ул Ленина, 7'")
+        ).scalar()
+        assert lenina is None
+    finally:
+        db.rollback()
+        db.close()

tradein-mvp/docs/gar-flats-runbook.md

@@ -0,0 +1,112 @@
+# ГАР/ФИАС loader знаменателя «квартир на дом» (runbook)
+
+Наполняет таблицу `gar_house_flats` числом помещений-квартир под каждым домом из
+официального дампа ГАР/ФИАС и матчит к `houses` по нормализованному адресу, проставляя
+`houses.gar_flat_count`. Это «знаменатель» для страницы **«доля квартир дома в продаже»**
+(`active_vtorichka_listings / total_apartments_in_building`, view `v_building_sale_share`,
+мигр. 143).
+
+Код:
+- `tradein-mvp/backend/app/services/gar_flats_loader.py` — стриминговый парсер + upsert + матчер.
+- `tradein-mvp/backend/app/tasks/gar_flats_load.py` — CLI (`python -m app.tasks.gar_flats_load`).
+
+## 1. Получить ГАР-дамп региона 66 (ops-шаг, вручную)
+
+Лоадер потребляет **уже распакованные** XML локально — многогигабайтную загрузку/распаковку
+делаем отдельно (это не часть лоадера).
+
+1. Источник: ГАР (Государственный адресный реестр), портал ФНС — https://fias.nalog.ru/
+   (раздел «Скачать»). Берём «ГАР. Версия со всеми объектами» — формат XML, `gar_xml.zip`.
+   - Полный РФ-архив `gar_xml.zip` — десятки ГБ. Нам нужен **только регион 66** (Свердловская обл.).
+   - Внутри архива объекты разложены по папкам с кодом региона: распаковываем **только папку `66/`**.
+     Пример (извлечь одну папку без распаковки всего архива):
+     ```bash
+     unzip gar_xml.zip '66/*' -d /data/gar
+     # → /data/gar/66/AS_ADDR_OBJ_*.XML, AS_HOUSES_*.XML, AS_APARTMENTS_*.XML,
+     #   AS_MUN_HIERARCHY_*.XML, ...
+     ```
+   - Размер распакованной папки `66/`: ориентир — несколько ГБ (крупнейшие файлы —
+     `AS_APARTMENTS_*` и `AS_MUN_HIERARCHY_*`, по сотни МБ — единицы ГБ). Поэтому парсер
+     стримит (`lxml.etree.iterparse` + очистка элементов), а не грузит дерево целиком.
+
+   Альтернатива: ГАР обновляется примерно еженедельно дельтами; нам достаточно полного
+   среза. Знаменатель «всего квартир в доме» почти статичен → **ре-прогон раз в квартал** ок.
+
+Нужные файлы в папке региона (остальные лоадер игнорирует):
+- `AS_ADDR_OBJ_*.XML` — адресные объекты (регион/город/улица).
+- `AS_HOUSES_*.XML` — дома.
+- `AS_APARTMENTS_*.XML` — помещения/квартиры.
+- `AS_MUN_HIERARCHY_*.XML` — иерархия (приоритетна; при отсутствии используется `AS_ADM_HIERARCHY_*`).
+
+## 2. Запуск лоадера
+
+Где: в окружении с доступом к БД `tradein` (контейнер `tradein-backend` или локальный venv с
+`DATABASE_URL` на tradein). Каталог `--dir` — путь к распакованной папке региона.
+
+```bash
+# в контейнере tradein-backend (DATABASE_URL уже выставлен)
+docker exec -it tradein-backend \
+  python -m app.tasks.gar_flats_load --dir /data/gar/66 --region 66 --version 2026-06-01
+```
+
+Локально:
+```bash
+cd tradein-mvp/backend
+DATABASE_URL=postgresql+psycopg://<user>:<pw>@localhost:<port>/tradein \
+  uv run python -m app.tasks.gar_flats_load --dir /data/gar/66 --region 66 --version 2026-06-01
+```
+
+Аргументы:
+- `--dir` (обязательный) — каталог с распакованными ГАР XML региона.
+- `--region` (умолч. `66`) — код региона; пишется в `gar_house_flats.region_code`.
+- `--version` (умолч. сегодня, `YYYY-MM-DD`) — метка версии дампа (`gar_house_flats.gar_version`).
+
+Что делает (две фазы, обе коммитятся):
+1. **load** — стримит XML → считает помещения на дом (по иерархии) → собирает адрес →
+   UPSERT в `gar_house_flats` (`ON CONFLICT (house_guid)`, идемпотентно). `norm_address`
+   считает SQL-fn `tradein_normalize_short_addr` поверх «улица, номер».
+2. **match** — `UPDATE houses SET gar_flat_count = …` по нормализованному адресу
+   (`gar_match_method='norm_address'`), идемпотентно (`IS DISTINCT FROM` gate).
+
+В логах — сводка: `houses`, `apartments` (учтено под домами), `upserted`, `houses_matched`.
+
+## 3. Проверка результата
+
+```sql
+-- сколько домов получили ГАР-знаменатель
+SELECT count(*) FILTER (WHERE gar_flat_count IS NOT NULL) AS matched,
+       count(*) AS total
+FROM houses;
+
+-- топ домов по доле квартир в продаже (эффективный знаменатель из ГАР)
+SELECT short_address, active_secondary, flat_count_effective, sale_share_pct, gar_match_method
+FROM v_building_sale_share
+WHERE gar_flat_count IS NOT NULL
+ORDER BY sale_share_pct DESC NULLS LAST
+LIMIT 20;
+```
+
+## 4. Семантика и ограничения
+
+- **flat_count** = число активных (`ISACTUAL=1 AND ISACTIVE=1`) помещений ГАР, чей
+  `PARENTOBJID` в иерархии == `OBJECTID` дома. Тип помещения (`APARTTYPE`) НЕ фильтруется —
+  это «всего помещений в доме» (квартиры + нежилые), сознательное приближение знаменателя.
+- **Адрес/норма.** `norm_address` = `tradein_normalize_short_addr(«<улица>, <номер>»)` —
+  тот же нормализатор, что и для `houses` при матче. Строка строится из «улица + номер»
+  (не из полной цепочки) для устойчивости: для ЕКБ эквивалентно `fn(full_address)` (fn срезает
+  регион/город), но не оставляет префикс для не-«г.» населённых пунктов. `full_address` хранится
+  отдельно как человекочитаемая provenance-цепочка (регион→город→улица→номер).
+- **Tie-break.** Если несколько ГАР-строк дают один `norm_address`, матчер берёт строку с
+  **максимальным `flat_count`** (при равенстве — лексикографически меньший `house_guid`),
+  через `DISTINCT ON (norm_address) ORDER BY norm_address, flat_count DESC, house_guid`.
+- **Дома с 0 квартир** грузятся (`flat_count=0`), но матчер их игнорирует (`flat_count > 0`).
+- **Покрытие матча — точное равенство нормализованного адреса.** Расхождения формата типа
+  улицы (`ул` vs `ул.` vs `улица`) и корпусов (`5 к1` vs `5/1`) не совпадут точно. Это
+  ожидаемо: под `gar_house_flats.norm_address` уже создан GIN-trgm индекс (мигр. 143) для
+  будущего нечёткого (fuzzy) прохода — он вне scope этого лоадера.
+
+## 5. Идемпотентность / ре-прогон
+
+- `gar_house_flats` — `ON CONFLICT (house_guid) DO UPDATE` → повторный прогон обновляет, не дублит.
+- Матчер — plain `UPDATE` с `IS DISTINCT FROM` → повторный прогон ничего не трогает.
+- Ре-прогон с новым `--version` безопасен; рекомендуемая частота — раз в квартал.