"""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) def estimate( payload: TradeInEstimateInput, db: Annotated[Session, Depends(get_db)], ) -> AggregatedEstimate: """MOCK реализация оценки квартиры для Trade-In. TODO TI-1b: заменить на реальный SQL aggregation из objective_lots после OBJ-1/2 merge (issue #314). """ 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)], ) -> 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, n_analogs, analogs, actual_deals, expires_at, address, 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, n_analogs=row.n_analogs, period_months=24, analogs=analogs, actual_deals=actual_deals, expires_at=row.expires_at, ) 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, } pdf_bytes = generate_trade_in_pdf(estimate, input_snapshot) filename = f"trade-in-{estimate_id}.pdf" logger.info("PDF generated estimate_id=%s size=%d", estimate_id, len(pdf_bytes)) return Response( content=pdf_bytes, media_type="application/pdf", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, )