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

"""House metadata service — обогащение данных дома по координатам (#392).

Когда пользователь не указал год постройки / тип дома, подтягиваем данные
из OpenStreetMap через Overpass API: building:levels, start_date, материал.

Стратегия:
- Cache lookup в `house_metadata` (Postgres) по близости координат — TTL 30 дней
- Cache miss → Overpass API → парсинг тегов → запись в кэш
- Best-effort: ЛЮБАЯ ошибка → None, оценка работает без обогащения (не 500-ит)

Используется в estimator.estimate_quality перед house-match аналогов (#6).
"""

from __future__ import annotations

import json
import logging
import re
from dataclasses import dataclass

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

from app.core.config import settings

logger = logging.getLogger(__name__)

OVERPASS_URL = "https://overpass-api.de/api/interpreter"
_CACHE_RADIUS_M = 40       # дом в пределах 40 м от точки — считаем тем же
_OVERPASS_RADIUS_M = 25    # ищем здание в 25 м от геокодированной точки
_CAD_RADIUS_M = 60         # радиус поиска здания в cad_buildings (#393)


@dataclass(frozen=True, slots=True)
class HouseMetadata:
    lat: float
    lon: float
    year_built: int | None
    total_floors: int | None
    house_type: str | None      # panel / brick / monolith / monolith_brick / other
    total_units: int | None
    source: str                 # 'osm' / 'cache'


# ── Парсинг тегов OSM ────────────────────────────────────────────────────────
_YEAR_KEYS = (
    "start_date", "construction_date", "building:year",
    "year_of_construction", "building:start_date",
)


def _parse_year(tags: dict[str, str]) -> int | None:
    """Год постройки из OSM-тегов. Значение бывает '1975', '1975-06-01',
    '1970s', '1960..1965', 'C20' — берём первое 4-значное число 1800–2100."""
    for key in _YEAR_KEYS:
        raw = tags.get(key)
        if not raw:
            continue
        m = re.search(r"\b(1[89]\d\d|20\d\d)\b", raw)
        if m:
            year = int(m.group(1))
            if 1800 <= year <= 2100:
                return year
    return None


def _parse_int(tags: dict[str, str], *keys: str) -> int | None:
    """Первое валидное целое (0 < n < 1000) из перечисленных тегов."""
    for key in keys:
        raw = (tags.get(key) or "").strip()
        if raw.isdigit():
            value = int(raw)
            if 0 < value < 1000:
                return value
    return None


# OSM building:material → наш house_type. Покрытие в OSM неполное — best-effort.
_MATERIAL_TO_TYPE = {
    "concrete_panels": "panel",
    "panels": "panel",
    "panel": "panel",
    "brick": "brick",
    "concrete": "monolith",
    "reinforced_concrete": "monolith",
}


def _parse_house_type(tags: dict[str, str]) -> str | None:
    material = (tags.get("building:material") or "").lower().strip()
    return _MATERIAL_TO_TYPE.get(material)


# ── Cache (house_metadata, TTL 30 дней) ──────────────────────────────────────
def _cache_get(db: Session, lat: float, lon: float) -> HouseMetadata | None:
    row = db.execute(
        text(
            """
            SELECT lat, lon, year_built, total_floors, house_type, total_units
            FROM house_metadata
            WHERE expires_at > NOW()
              AND ST_DWithin(geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius)
            ORDER BY ST_Distance(geom::geography, ST_MakePoint(:lon, :lat)::geography)
            LIMIT 1
            """
        ),
        {"lat": lat, "lon": lon, "radius": _CACHE_RADIUS_M},
    ).fetchone()
    if row is None:
        return None
    return HouseMetadata(
        lat=row.lat, lon=row.lon,
        year_built=row.year_built, total_floors=row.total_floors,
        house_type=row.house_type, total_units=row.total_units,
        source="cache",
    )


def _cache_put(db: Session, meta: HouseMetadata, raw: dict) -> None:
    db.execute(
        text(
            """
            INSERT INTO house_metadata
                (lat, lon, year_built, total_floors, house_type,
                 total_units, source, raw_payload)
            VALUES (:lat, :lon, :year, :floors, :htype,
                    :units, 'osm', CAST(:raw AS jsonb))
            """
        ),
        {
            "lat": meta.lat, "lon": meta.lon,
            "year": meta.year_built, "floors": meta.total_floors,
            "htype": meta.house_type, "units": meta.total_units,
            "raw": json.dumps(raw, ensure_ascii=False),
        },
    )
    db.commit()


# ── Источник #393: кадастр зданий (gendesign_cad_buildings FDW) ──────────────
def _cad_buildings_get(db: Session, lat: float, lon: float) -> HouseMetadata | None:
    """Ближайшее здание из gendesign_cad_buildings FDW (данные Росреестра) по координатам.

    Приоритетный источник house_metadata — год / этажность полнее, чем OSM
    (~36к домов ЕКБ с годом постройки). Использует bbox+haversine вместо ST_DWithin —
    FDW-таблица не имеет PostGIS-колонки geom (только lat/lon скалярные).
    """
    try:
        row = db.execute(
            text(
                """
                WITH candidates AS (
                  SELECT year_built, floors, lat, lon,
                         111320.0 * sqrt(
                             pow(CAST(:lat AS double precision) - lat, 2) +
                             pow(
                                 cos(radians(CAST(:lat AS double precision)))
                                 * (CAST(:lon AS double precision) - lon), 2
                             )
                         ) AS dist_m
                  FROM gendesign_cad_buildings
                  WHERE (year_built IS NOT NULL OR floors IS NOT NULL)
                    AND lat BETWEEN CAST(:lat AS double precision) - 0.0010
                                AND CAST(:lat AS double precision) + 0.0010
                    AND lon BETWEEN CAST(:lon AS double precision) - 0.0015
                                AND CAST(:lon AS double precision) + 0.0015
                )
                SELECT year_built, floors
                FROM candidates
                WHERE dist_m < CAST(:radius AS double precision)
                ORDER BY dist_m ASC
                LIMIT 1
                """
            ),
            {"lat": lat, "lon": lon, "radius": float(_CAD_RADIUS_M)},
        ).fetchone()
    except Exception:
        logger.warning(
            "gendesign_cad_buildings lookup failed at (%.5f, %.5f)", lat, lon, exc_info=True
        )
        return None

    if row is None:
        return None
    return HouseMetadata(
        lat=lat, lon=lon,
        year_built=row.year_built,
        total_floors=row.floors,
        house_type=None,      # purpose не маппится в panel/brick
        total_units=None,
        source="cadastre",
    )


# ── Provider: OSM Overpass ───────────────────────────────────────────────────
async def _overpass_lookup(lat: float, lon: float) -> tuple[HouseMetadata, dict] | None:
    """Запрос здания у точки через Overpass API. (meta, raw_payload) или None."""
    query = (
        "[out:json][timeout:20];"
        f'way(around:{_OVERPASS_RADIUS_M},{lat},{lon})["building"];'
        "out tags center;"
    )
    headers = {"User-Agent": f"TradeInMVP/0.1 (contact: {settings.contact_email})"}
    async with httpx.AsyncClient(timeout=15.0, headers=headers) as client:
        response = await client.post(OVERPASS_URL, data={"data": query})
        response.raise_for_status()
        data = response.json()

    elements = [e for e in data.get("elements", []) if e.get("tags")]
    if not elements:
        return None

    # Ближайшее здание (по center) к геокодированной точке.
    def _dist2(el: dict) -> float:
        center = el.get("center") or {}
        return (center.get("lat", 0.0) - lat) ** 2 + (center.get("lon", 0.0) - lon) ** 2

    best = min(elements, key=_dist2)
    tags = best.get("tags", {})
    meta = HouseMetadata(
        lat=lat, lon=lon,
        year_built=_parse_year(tags),
        total_floors=_parse_int(tags, "building:levels"),
        house_type=_parse_house_type(tags),
        total_units=_parse_int(tags, "building:flats", "flats"),
        source="osm",
    )
    return meta, {"osm_way_id": best.get("id"), "tags": tags}


# ── Public API ───────────────────────────────────────────────────────────────
async def get_house_metadata(lat: float, lon: float, db: Session) -> HouseMetadata | None:
    """Метаданные дома по координатам: cache (TTL 30д) → Overpass.

    Best-effort: при ЛЮБОЙ ошибке возвращает None и откатывает сессию —
    оценка обязана работать даже если картография недоступна.
    """
    try:
        # #393: кадастр зданий (Росреестр, cad_buildings) — приоритетный источник.
        cad = _cad_buildings_get(db, lat, lon)
        if cad is not None:
            logger.info(
                "house_metadata: cad_buildings (%.5f, %.5f) → year=%s floors=%s",
                lat, lon, cad.year_built, cad.total_floors,
            )
            return cad

        cached = _cache_get(db, lat, lon)
        if cached is not None:
            logger.info("house_metadata cache hit: (%.5f, %.5f)", lat, lon)
            return cached

        result = await _overpass_lookup(lat, lon)
        if result is None:
            logger.info("house_metadata: OSM не нашёл здание у (%.5f, %.5f)", lat, lon)
            return None

        meta, raw = result
        # Пустой результат (ни года, ни этажности) кэшировать смысла нет.
        if meta.year_built is None and meta.total_floors is None:
            return None

        _cache_put(db, meta, raw)
        logger.info(
            "house_metadata OSM: (%.5f, %.5f) → year=%s floors=%s type=%s",
            lat, lon, meta.year_built, meta.total_floors, meta.house_type,
        )
        return meta
    except Exception:
        logger.warning(
            "house_metadata enrichment failed at (%.5f, %.5f)", lat, lon, exc_info=True
        )
        try:
            db.rollback()
        except Exception:
            pass
        return None