feat(tradein): обогащение house_metadata через OSM Overpass (#392)

Когда пользователь не указал год постройки / тип дома, estimator
подтягивает их из OpenStreetMap (Overpass API) — building:levels,
start_date, building:material. Улучшает house-match аналогов (#6).

- app/services/house_metadata.py — Overpass-провайдер + кэш в таблице
  house_metadata (TTL 30 дней, lookup по близости координат, 40 м).
- estimator.estimate_quality — обогащает target_year / target_house_type
  перед _fetch_analogs. Best-effort: ошибка OSM → оценка без обогащения.

OSM по ЕКБ: building:levels населён хорошо, год постройки — частично.

Closes #392

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

@@ -29,6 +29,7 @@ from sqlalchemy.orm import Session
 
 from app.schemas.trade_in import AggregatedEstimate, AnalogLot, TradeInEstimateInput
 from app.services.geocoder import GeocodeResult, geocode
+from app.services.house_metadata import get_house_metadata
 
 logger = logging.getLogger(__name__)
 
@@ -83,14 +84,27 @@ async def estimate_quality(
         logger.warning("geocode failed for %s — returning low-confidence estimate", payload.address)
         return _empty_estimate(payload, reason="address_not_geocoded")
 
-    # 2. Three-tier fallback:
+    # 2. #392: обогащаем год / тип дома из картографии (OSM Overpass), если
-    #    a) 800m + ±15% area
+    #    пользователь их не указал — это улучшает house-match аналогов (#6).
+    #    Best-effort: при недоступности OSM target_* остаются None.
+    target_year = payload.year_built
+    target_house_type = payload.house_type
+    if target_year is None or target_house_type is None:
+        house_meta = await get_house_metadata(geo.lat, geo.lon, db)
+        if house_meta is not None:
+            if target_year is None:
+                target_year = house_meta.year_built
+            if target_house_type is None:
+                target_house_type = house_meta.house_type
+
+    # 3. Three-tier fallback:
+    #    a) 1km + ±15% area
     #    b) 2km + ±15% area  (fallback_used = True)
     #    c) 2km + ±25% area  (fallback_used = True, area_widened = True)
     listings, fallback_used = _fetch_analogs(
         db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
         radius_m=DEFAULT_RADIUS_M,
-        year_built=payload.year_built, house_type=payload.house_type,
+        year_built=target_year, house_type=target_house_type,
     )
     area_widened = False
 
@@ -98,7 +112,7 @@ async def estimate_quality(
         listings_wide, _ = _fetch_analogs(
             db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
             radius_m=FALLBACK_RADIUS_M,
-            year_built=payload.year_built, house_type=payload.house_type,
+            year_built=target_year, house_type=target_house_type,
         )
         if len(listings_wide) > len(listings):
             listings = listings_wide
@@ -110,7 +124,7 @@ async def estimate_quality(
         listings_widearea, _ = _fetch_analogs(
             db, lat=geo.lat, lon=geo.lon, rooms=payload.rooms, area=payload.area_m2,
             radius_m=FALLBACK_RADIUS_M, area_tolerance=0.25,
-            year_built=payload.year_built, house_type=payload.house_type,
+            year_built=target_year, house_type=target_house_type,
         )
         if len(listings_widearea) > len(listings):
             listings = listings_widearea
@@ -210,8 +224,8 @@ async def estimate_quality(
             "rooms": payload.rooms,
             "floor": payload.floor,
             "total_floors": payload.total_floors,
-            "year_built": payload.year_built,
+            "year_built": target_year,
-            "house_type": payload.house_type,
+            "house_type": target_house_type,
             "repair_state": payload.repair_state,
             "has_balcony": payload.has_balcony,
             "median_price": median_price,

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

@@ -0,0 +1,213 @@
+"""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 м от геокодированной точки
+
+
+@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()
+
+
+# ── 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:
+        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:  # noqa: BLE001
+        logger.warning(
+            "house_metadata enrichment failed at (%.5f, %.5f)", lat, lon, exc_info=True
+        )
+        try:
+            db.rollback()
+        except Exception:  # noqa: BLE001
+            pass
+        return None