"""Trade-In Estimator — endpoints (TI-1 mock + TI-2 PDF). MOCK implementation: returns realistic ЕКБ price bands by rooms/floor/repair. TODO TI-1b: заменить _mock_estimate() на реальный SQL aggregation из objective_lots + rosreestr_deals после OBJ-1/2 merge. """ from __future__ import annotations import json import logging import random from datetime import UTC, datetime, timedelta from typing import Annotated from uuid import UUID, uuid4 from fastapi import APIRouter, Depends, HTTPException, Response from sqlalchemy import text from sqlalchemy.orm import Session from app.core.db import get_db from app.schemas.trade_in import AggregatedEstimate, AnalogLot, TradeInEstimateInput from app.services.exporters.trade_in_pdf import generate_trade_in_pdf logger = logging.getLogger(__name__) router = APIRouter() # ЕКБ-адреса для фейковых аналогов (реальные улицы центра) _EKB_STREETS = [ "ул. Малышева", "ул. Куйбышева", "ул. 8 Марта", "ул. Белинского", "пр. Ленина", "ул. Толмачёва", "ул. Радищева", "ул. Мамина-Сибиряка", "ул. Луначарского", "ул. Первомайская", ] # Базовые ценовые диапазоны по комнатности (ЕКБ, 2026) _PRICE_BANDS: dict[int, dict[str, int | float | str]] = { 0: { # студия ~25 м² "median": 6_500_000, "low": 5_800_000, "high": 7_500_000, "ppm2": 260_000, "ref_area": 25.0, }, 1: { # 1к ~40 м² "median": 9_000_000, "low": 8_000_000, "high": 10_500_000, "ppm2": 225_000, "ref_area": 40.0, }, 2: { # 2к ~60 м² "median": 12_500_000, "low": 11_000_000, "high": 14_000_000, "ppm2": 208_000, "ref_area": 60.0, }, 3: { # 3к ~80 м² "median": 17_000_000, "low": 15_000_000, "high": 19_000_000, "ppm2": 213_000, "ref_area": 80.0, }, } def _floor_factor(floor: int, total_floors: int) -> float: """±5% поправка за этаж: 1й и последний этаж снижают цену.""" if floor == 1: return 0.95 if floor == total_floors: return 0.97 return 1.0 def _repair_factor(repair_state: str | None) -> float: """±10% поправка за состояние отделки.""" factors = { "needs_repair": 0.90, "standard": 1.00, "good": 1.05, "excellent": 1.10, } return factors.get(repair_state or "standard", 1.0) def _confidence(rooms: int) -> str: if 1 <= rooms <= 3: return "high" return "medium" def _gen_analogs( rooms: int, area_m2: float, base_ppm2: int, n: int, *, is_listing: bool, ) -> list[AnalogLot]: """Генерирует список фейковых аналогов (объявления или сделки).""" rng = random.Random(42 + rooms + n) result: list[AnalogLot] = [] today = datetime.now(tz=UTC).date() for i in range(n): street = _EKB_STREETS[i % len(_EKB_STREETS)] building_no = rng.randint(1, 120) apt_no = rng.randint(1, 300) addr = f"г. Екатеринбург, {street}, {building_no}, кв. {apt_no}" area_jitter = area_m2 * rng.uniform(0.85, 1.15) ppm2_jitter = int(base_ppm2 * rng.uniform(0.90, 1.10)) price = int(area_jitter * ppm2_jitter) floor_val = rng.randint(2, 16) total_fl = rng.randint(floor_val, 20) if is_listing: dom = rng.randint(5, 120) listing_dt = today - timedelta(days=dom) else: dom = rng.randint(10, 60) listing_dt = today - timedelta(days=rng.randint(30, 365)) result.append( AnalogLot( address=addr, area_m2=round(area_jitter, 1), rooms=rooms if rooms > 0 else 0, floor=floor_val, total_floors=total_fl, price_rub=price, price_per_m2=ppm2_jitter, listing_date=listing_dt, days_on_market=dom, photo_url=None, ) ) return result def _mock_estimate(payload: TradeInEstimateInput) -> AggregatedEstimate: """Возвращает mock-оценку на основе диапазонов ЕКБ 2026. Логика: - Берём базовый band по rooms (0-3+). - Масштабируем на фактическую площадь относительно референсной. - Применяем поправку за этаж (±5%) и отделку (±10%). - Генерируем 7-10 аналогов (листинги) и 3-5 actual_deals. """ rooms_key = min(payload.rooms, 3) # 4к+ → диапазон 3к band = _PRICE_BANDS[rooms_key] # Масштаб по площади ref_area: float = band["ref_area"] # type: ignore[assignment] area_scale = payload.area_m2 / ref_area ff = _floor_factor(payload.floor, payload.total_floors) rf = _repair_factor(payload.repair_state) combined = area_scale * ff * rf median = int(band["median"] * combined) # type: ignore[operator] low = int(band["low"] * combined) # type: ignore[operator] high = int(band["high"] * combined) # type: ignore[operator] ppm2 = int(band["ppm2"] * ff * rf) # type: ignore[operator] n_analogs = random.randint(7, 10) n_deals = random.randint(3, 5) analogs = _gen_analogs(rooms_key, payload.area_m2, ppm2, n_analogs, is_listing=True) actual_deals = _gen_analogs( rooms_key, payload.area_m2, int(ppm2 * 0.93), n_deals, is_listing=False ) now = datetime.now(tz=UTC) return AggregatedEstimate( estimate_id=uuid4(), median_price_rub=median, range_low_rub=low, range_high_rub=high, median_price_per_m2=ppm2, confidence=_confidence(payload.rooms), n_analogs=n_analogs + n_deals, period_months=24, analogs=analogs, actual_deals=actual_deals, expires_at=now + timedelta(hours=24), ) @router.post("/estimate", response_model=AggregatedEstimate) async def estimate( payload: TradeInEstimateInput, db: Annotated[Session, Depends(get_db)], ) -> AggregatedEstimate: """Реальная оценка через SQL aggregation поверх listings + deals. 1. Geocode address → lat/lon 2. PostGIS ST_DWithin радиус 800м (или 2км fallback) 3. Tukey IQR outlier filter 4. Median + Q1 + Q3 + confidence с explanation """ from app.services.estimator import estimate_quality return await estimate_quality(payload, db) # OLD MOCK PATH — оставлен как fallback для отладки; не вызывается из router. def _estimate_legacy_mock( payload: TradeInEstimateInput, db: Annotated[Session, Depends(get_db)], ) -> AggregatedEstimate: """Старая mock-реализация. НЕ используется в роутере, удалить когда стабилизируем новый estimator.""" result = _mock_estimate(payload) analogs_json = json.dumps( [a.model_dump(mode="json") for a in result.analogs], ensure_ascii=False, ) deals_json = json.dumps( [a.model_dump(mode="json") for a in result.actual_deals], ensure_ascii=False, ) db.execute( text( """ INSERT INTO trade_in_estimates ( id, address, area_m2, rooms, floor, total_floors, year_built, house_type, repair_state, has_balcony, median_price, range_low, range_high, median_price_per_m2, confidence, n_analogs, analogs, actual_deals, expires_at ) VALUES ( CAST(:id AS uuid), :address, :area_m2, :rooms, :floor, :total_floors, :year_built, :house_type, :repair_state, :has_balcony, :median_price, :range_low, :range_high, :median_price_per_m2, :confidence, :n_analogs, CAST(:analogs AS jsonb), CAST(:actual_deals AS jsonb), :expires_at ) """ ), { "id": str(result.estimate_id), "address": payload.address, "area_m2": payload.area_m2, "rooms": payload.rooms, "floor": payload.floor, "total_floors": payload.total_floors, "year_built": payload.year_built, "house_type": payload.house_type, "repair_state": payload.repair_state, "has_balcony": payload.has_balcony, "median_price": result.median_price_rub, "range_low": result.range_low_rub, "range_high": result.range_high_rub, "median_price_per_m2": result.median_price_per_m2, "confidence": result.confidence, "n_analogs": result.n_analogs, "analogs": analogs_json, "actual_deals": deals_json, "expires_at": result.expires_at, }, ) db.commit() logger.info( "trade_in estimate saved id=%s rooms=%d area=%.1f confidence=%s", result.estimate_id, payload.rooms, payload.area_m2, result.confidence, ) return result @router.get("/estimate/{estimate_id}", response_model=AggregatedEstimate) def get_estimate( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], ) -> AggregatedEstimate: """Получить сохранённую оценку по UUID (для генерации PDF). Возвращает 404 если оценка не найдена или TTL истёк. """ row = db.execute( text( """ SELECT id, median_price, range_low, range_high, median_price_per_m2, confidence, n_analogs, analogs, actual_deals, expires_at, address, area_m2, rooms FROM trade_in_estimates WHERE id = CAST(:id AS uuid) AND expires_at > NOW() """ ), {"id": str(estimate_id)}, ).fetchone() if row is None: raise HTTPException(status_code=404, detail="estimate not found or expired") analogs = [AnalogLot(**a) for a in (row.analogs or [])] actual_deals = [AnalogLot(**a) for a in (row.actual_deals or [])] return AggregatedEstimate( estimate_id=row.id, median_price_rub=row.median_price, range_low_rub=row.range_low, range_high_rub=row.range_high, median_price_per_m2=row.median_price_per_m2, confidence=row.confidence, n_analogs=row.n_analogs, period_months=24, analogs=analogs, actual_deals=actual_deals, expires_at=row.expires_at, ) @router.get("/estimate/{estimate_id}/pdf") def estimate_pdf( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], brand: str | None = None, ) -> Response: """Скачать 4-страничный PDF-отчёт для оценки trade-in. Возвращает application/pdf attachment. 404 — оценка не найдена. 410 — оценка просрочена (TTL 24ч). """ row = db.execute( text( """ SELECT id, median_price, range_low, range_high, median_price_per_m2, confidence, confidence_explanation, n_analogs, analogs, actual_deals, sources_used, data_freshness_minutes, expires_at, address, lat, lon, area_m2, rooms, floor, total_floors, year_built, house_type, repair_state, has_balcony FROM trade_in_estimates WHERE id = CAST(:id AS uuid) """ ), {"id": str(estimate_id)}, ).fetchone() if row is None: raise HTTPException(status_code=404, detail="estimate not found") if row.expires_at.replace(tzinfo=UTC) < datetime.now(tz=UTC): raise HTTPException(status_code=410, detail="estimate expired (24h TTL)") analogs = [AnalogLot(**a) for a in (row.analogs or [])] actual_deals = [AnalogLot(**a) for a in (row.actual_deals or [])] estimate = AggregatedEstimate( estimate_id=row.id, median_price_rub=row.median_price, range_low_rub=row.range_low, range_high_rub=row.range_high, median_price_per_m2=row.median_price_per_m2, confidence=row.confidence, confidence_explanation=row.confidence_explanation, n_analogs=row.n_analogs, period_months=24, analogs=analogs, actual_deals=actual_deals, expires_at=row.expires_at, target_address=row.address, target_lat=row.lat, target_lon=row.lon, sources_used=row.sources_used or [], data_freshness_minutes=row.data_freshness_minutes, ) input_snapshot = { "address": row.address, "area_m2": row.area_m2, "rooms": row.rooms, "floor": row.floor, "total_floors": row.total_floors, "year_built": row.year_built, "house_type": row.house_type, "repair_state": row.repair_state, "has_balcony": row.has_balcony, } from app.services.brand import get_brand as _resolve_brand brand_obj = _resolve_brand(brand, db) pdf_bytes = generate_trade_in_pdf(estimate, input_snapshot, brand=brand_obj) filename = f"trade-in-{brand_obj.slug}-{estimate_id}.pdf" logger.info( "PDF generated estimate_id=%s brand=%s size=%d", estimate_id, brand_obj.slug, len(pdf_bytes), ) return Response( content=pdf_bytes, media_type="application/pdf", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, )