gendesign/tradein-mvp/backend/app/services/zhkh_flats_loader.py

"""ГИС ЖКХ loader: «квартир на дом» (знаменатель «доли квартир дома в продаже»).

Тянет точный счёт квартир (residentialPremiseCount) из публичного API ГИС ЖКХ
(dom.gosuslugi.ru) и проставляет `houses.zhkh_*` (мигр. 146/149). В отличие от ГАР-loader'а
(стриминговый парс многогигабайтного XML + fuzzy canon-match адреса) здесь матч ТРИВИАЛЬНЫЙ:
`houses.gar_house_guid` — это И ЕСТЬ ЖКХ `houseGuid` (проверено), поэтому ищем дом строго по
этому guid, без нормализации/канона адреса.

ГИС ЖКХ — точнее ГАР (ГАР дико недосчитывает квартиры в МКД, см. мигр. 149): резолвит ~90%
вторички и даёт год ввода (operationYear) + этажность (maxFloorCount) как бонус.

API-рецепт (verified, прод):
  POST searchByAddress?pageIndex=1&elementsPerPage=100 — БЕЗ auth/cookies.
  · pageIndex — 1-based: pageIndex=0 молча отдаёт пусто (200). Всегда 1.
  · elementsPerPage cap = 100 (≥200 → 403) → батч ≤100 guid на запрос, одна страница.
  · Тело — ПОЛНЫЙ DTO (все поля, неиспользуемые = None); fiasHouseCodeList — батч guid.
  · Ответ {"total":N,"items":[...]}; per item residentialPremiseCount (строка) = счёт квартир,
    operationYear/buildingYear = год, maxFloorCount (строка-диапазон "10-30") = этажность,
    address.house.houseGuid = входной guid (ключ для обратного маппинга).

psycopg v3: SQL через `text(...)` использует `CAST(:x AS type)`, НИКОГДА `:x::type`.
"""

from __future__ import annotations

import logging
import time
import uuid
from collections.abc import Iterator
from dataclasses import dataclass
from typing import Any

import httpx
from sqlalchemy import text
from sqlalchemy.orm import Session

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────────────────────
# Константы API
# ─────────────────────────────────────────────────────────────────────────────
# pageIndex/elementsPerPage — query-параметры (НЕ в теле). pageIndex 1-based (=0 → пусто),
# elementsPerPage cap=100 (≥200 → 403).
ENDPOINT = (
    "https://dom.gosuslugi.ru/homemanagement/api/rest/services/houses/public/"
    "searchByAddress?pageIndex=1&elementsPerPage=100"
)
# FIAS GUID Свердловской обл. (region 66) — обязателен в теле DTO.
REGION_GUID = "92b30014-4d52-4e2e-892d-928142b924bf"
# Batch = elementsPerPage cap: ≤100 guid/запрос, одна страница (pageIndex=1).
BATCH_SIZE = 100
# Группировка UPDATE в один SAVEPOINT: сбойный ЧАНК (≤200 строк) откатывается изолированно,
# остальные чанки доезжают; повторный прогон идемпотентен (IS DISTINCT FROM gate).
SAVEPOINT_CHUNK = 200
DEFAULT_UA = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
    "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
)


# ─────────────────────────────────────────────────────────────────────────────
# Чистые хелперы парсинга (юнит-тестируются без сети/БД)
# ─────────────────────────────────────────────────────────────────────────────
# NB: maxFloorCount хранится СЫРЫМ ("10-30"/"17,23") в zhkh_floors_raw — числовой парс
# этажности отложен до PR фильтра новостроек (#2), который и решит, как трактовать диапазон.
def parse_int(raw: object) -> int | None:
    """Устойчивый str/int → int|None.

    None/«»/«0»/нечисло → None (0 квартир — не пригодный знаменатель; ≤0 тоже отбрасываем).
    """
    if raw is None or isinstance(raw, bool):
        return None
    if isinstance(raw, int):
        return raw if raw > 0 else None
    s = str(raw).strip()
    if not s:
        return None
    try:
        n = int(s)
    except ValueError:
        return None
    return n if n > 0 else None


@dataclass(slots=True)
class ZhkhHouse:
    """Распарсенная ЖКХ-строка дома (ключ обратного маппинга — house_guid)."""

    house_guid: str
    flat_count: int | None
    year: int | None
    floors_raw: str | None


def _parse_item(item: dict[str, Any]) -> ZhkhHouse | None:
    """Извлечь guid + поля из item ответа ГИС ЖКХ. None, если нет houseGuid.

    year = operationYear (предпочтительно) иначе parse_int(buildingYear);
    flat_count = parse_int(residentialPremiseCount); floors_raw = сырая maxFloorCount.
    """
    address = item.get("address") or {}
    house = address.get("house") or {}
    guid = house.get("houseGuid")
    if not guid:
        return None

    year = parse_int(item.get("operationYear"))
    if year is None:
        year = parse_int(item.get("buildingYear"))

    floors_value = item.get("maxFloorCount")
    floors_raw = None if floors_value is None else str(floors_value)

    return ZhkhHouse(
        house_guid=str(guid),
        flat_count=parse_int(item.get("residentialPremiseCount")),
        year=year,
        floors_raw=floors_raw,
    )


# ─────────────────────────────────────────────────────────────────────────────
# HTTP: заголовки + тело + fetch
# ─────────────────────────────────────────────────────────────────────────────
def _headers() -> dict[str, str]:
    """Заголовки запроса. request-guid/session-guid — свежие random uuid4 (random OK)."""
    return {
        "content-type": "application/json;charset=UTF-8",
        "accept": "application/json; charset=utf-8",
        "state-guid": "/houses",
        "request-guid": str(uuid.uuid4()),
        "session-guid": str(uuid.uuid4()),
        "user-agent": DEFAULT_UA,
        "referer": "https://dom.gosuslugi.ru/",
        "origin": "https://dom.gosuslugi.ru",
    }


def _build_body(batch: list[str]) -> dict[str, Any]:
    """ПОЛНЫЙ DTO searchByAddress (15 полей; неиспользуемые = None). batch → fiasHouseCodeList."""
    return {
        "regionCode": REGION_GUID,
        "fiasHouseCodeList": batch,
        "estStatus": None,
        "strStatus": None,
        "calcCount": True,
        "houseConditionRefList": None,
        "houseTypeRefList": None,
        "houseManagementTypeRefList": None,
        "cadastreNumber": None,
        "oktmo": None,
        "statuses": ["APPROVED"],
        "regionProperty": None,
        "municipalProperty": None,
        "hostelTypeCodes": None,
        "useReadOnlyDataSource": True,
    }


def _chunk_guids(guids: list[str], size: int) -> Iterator[list[str]]:
    for i in range(0, len(guids), size):
        yield guids[i : i + size]


def _fetch_batch(
    batch: list[str], *, client: httpx.Client, max_retries: int
) -> list[dict[str, Any]]:
    """POST одного батча guid (pageIndex=1) → items. Ретрай на 403/5xx/httpx-ошибках.

    Backoff 1,2,4с между попытками. На финальном провале — warning + пустой список (батч
    пропускается, весь прогон не срывается). Прочие 4xx (битый запрос) — не ретраим.
    """
    body = _build_body(batch)
    last_err = ""
    for attempt in range(1, max_retries + 1):
        try:
            resp = client.post(ENDPOINT, json=body, headers=_headers())
            if resp.status_code != 403 and resp.status_code < 500:
                resp.raise_for_status()  # прочие 4xx → невосстановимый HTTPStatusError
                data = resp.json()
                items = data.get("items")
                return items if isinstance(items, list) else []
            last_err = f"HTTP {resp.status_code}"
        except httpx.HTTPStatusError as exc:
            logger.warning(
                "zhkh fetch: батч %d guid — невосстановимый %s (пропущен)", len(batch), exc
            )
            return []
        except (httpx.HTTPError, ValueError) as exc:
            last_err = repr(exc)
        if attempt < max_retries:
            backoff = 2 ** (attempt - 1)
            logger.warning(
                "zhkh fetch: батч %d guid сбой (%s), retry %d/%d через %dс",
                len(batch),
                last_err,
                attempt,
                max_retries,
                backoff,
            )
            time.sleep(backoff)
    logger.warning(
        "zhkh fetch: батч %d guid сбойнул после %d попыток (%s), пропущен",
        len(batch),
        max_retries,
        last_err,
    )
    return []


def fetch_zhkh(
    guids: list[str],
    *,
    client: httpx.Client,
    sleep_sec: float = 0.4,
    max_retries: int = 3,
) -> dict[str, ZhkhHouse]:
    """Бьёт guid на батчи по BATCH_SIZE, POST'ит каждый, парсит items → dict по house_guid.

    Между батчами — пауза sleep_sec (вежливость). Прогресс логируется каждые 10 батчей.
    Сбойный батч пропускается (см. _fetch_batch), а не срывает весь прогон.
    """
    result: dict[str, ZhkhHouse] = {}
    batches = list(_chunk_guids(guids, BATCH_SIZE))
    total = len(batches)
    for idx, batch in enumerate(batches, start=1):
        for item in _fetch_batch(batch, client=client, max_retries=max_retries):
            parsed = _parse_item(item)
            if parsed is not None:
                result[parsed.house_guid] = parsed
        if idx % 10 == 0 or idx == total:
            logger.info("zhkh fetch: батч %d/%d, резолвлено домов=%d", idx, total, len(result))
        if idx < total and sleep_sec > 0:
            time.sleep(sleep_sec)
    return result


# ─────────────────────────────────────────────────────────────────────────────
# UPDATE houses по gar_house_guid (== ЖКХ houseGuid)
# ─────────────────────────────────────────────────────────────────────────────
# Идемпотентный UPDATE: gate `IS DISTINCT FROM` → повторный прогон не трогает уже совпавшие
# строки. CAST(:x AS type) — psycopg v3 discipline (никогда :x::type).
_UPDATE_SQL = text(
    """
    UPDATE houses
    SET zhkh_flat_count = CAST(:fc AS int),
        zhkh_house_guid = CAST(:g AS text),
        zhkh_matched_at = now(),
        zhkh_year = CAST(:yr AS smallint),
        zhkh_floors_raw = CAST(:fr AS text)
    WHERE gar_house_guid = CAST(:g AS text)
      AND (
          zhkh_flat_count IS DISTINCT FROM CAST(:fc AS int)
          OR zhkh_year IS DISTINCT FROM CAST(:yr AS smallint)
          OR zhkh_floors_raw IS DISTINCT FROM CAST(:fr AS text)
      )
    """
)

_SELECT_GUIDS_SQL = text(
    """
    SELECT DISTINCT gar_house_guid
    FROM houses
    WHERE gar_house_guid IS NOT NULL
    """
)

# Тот же запрос с LIMIT для сэмплинга (--limit). Отдельный text() — psycopg v3 не любит
# условную инъекцию LIMIT в строку.
_SELECT_GUIDS_LIMIT_SQL = text(
    """
    SELECT DISTINCT gar_house_guid
    FROM houses
    WHERE gar_house_guid IS NOT NULL
    LIMIT CAST(:limit AS int)
    """
)


def _chunk_houses(items: list[ZhkhHouse], size: int) -> Iterator[list[ZhkhHouse]]:
    for i in range(0, len(items), size):
        yield items[i : i + size]


def match_houses_to_zhkh(
    db: Session,
    *,
    region_code: str = "66",
    limit: int | None = None,
    dry_run: bool = False,
    sleep_sec: float = 0.4,
) -> dict[str, int]:
    """ГИС ЖКХ → houses.zhkh_* по gar_house_guid (== ЖКХ houseGuid). НЕ коммитит (caller).

    region_code — информационно/в лог (все наши guid — region 66; параметр для паритета с
    ГАР-CLI). limit (если задан) ограничивает выборку домов (сэмпл). dry_run — НОЛЬ записей,
    только подсчёт. Возвращает {houses_seen, zhkh_resolved, zhkh_with_flatcount, houses_updated}.
    """
    if limit is not None:
        rows = db.execute(_SELECT_GUIDS_LIMIT_SQL, {"limit": limit}).scalars().all()
    else:
        rows = db.execute(_SELECT_GUIDS_SQL).scalars().all()
    guids = [str(g) for g in rows if g]
    houses_seen = len(guids)
    logger.info(
        "zhkh match: домов с gar_house_guid=%d (region=%s, limit=%s, dry_run=%s)",
        houses_seen,
        region_code,
        limit,
        dry_run,
    )

    with httpx.Client(timeout=40) as client:
        zhkh_by_guid = fetch_zhkh(guids, client=client, sleep_sec=sleep_sec)

    zhkh_resolved = len(zhkh_by_guid)
    with_flatcount = [
        z for z in zhkh_by_guid.values() if z.flat_count is not None and z.flat_count > 0
    ]
    zhkh_with_flatcount = len(with_flatcount)

    houses_updated = 0
    if dry_run:
        logger.info(
            "zhkh match DRY-RUN: резолвлено=%d с flat_count=%d (записей не делаем)",
            zhkh_resolved,
            zhkh_with_flatcount,
        )
    else:
        for chunk in _chunk_houses(with_flatcount, SAVEPOINT_CHUNK):
            chunk_updated = 0
            try:
                with db.begin_nested():
                    for z in chunk:
                        res = db.execute(
                            _UPDATE_SQL,
                            {
                                "fc": z.flat_count,
                                "g": z.house_guid,
                                "yr": z.year,
                                "fr": z.floors_raw,
                            },
                        )
                        chunk_updated += res.rowcount
                houses_updated += chunk_updated
            except Exception:
                logger.warning(
                    "zhkh match: чанк из %d UPDATE сбойнул (откат savepoint)",
                    len(chunk),
                    exc_info=True,
                )

    result = {
        "houses_seen": houses_seen,
        "zhkh_resolved": zhkh_resolved,
        "zhkh_with_flatcount": zhkh_with_flatcount,
        "houses_updated": houses_updated,
    }
    logger.info(
        "zhkh match DONE: seen=%d resolved=%d with_flatcount=%d updated=%d (dry_run=%s)",
        houses_seen,
        zhkh_resolved,
        zhkh_with_flatcount,
        houses_updated,
        dry_run,
    )
    return result