bot-backend
14f3ef2019
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
385 lines
13 KiB
Python
385 lines
13 KiB
Python
"""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.astimezone(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}"'},
|
||
)
|