"""Trade-In Estimator — endpoints (TI-2 PDF, photos, history). Реальная оценка делается через app.services.estimator.estimate_quality(). """ from __future__ import annotations import logging from datetime import UTC, datetime from typing import Annotated, Any from uuid import UUID from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile from sqlalchemy import text from sqlalchemy.orm import Session from app.core.db import get_db from app.schemas.trade_in import ( AggregatedEstimate, AnalogLot, CianPriceChangeStats, HouseAnalyticsKpi, HouseAnalyticsResponse, HouseInfoForEstimate, IMVBenchmarkResponse, PhotoMeta, PlacementHistoryEntry, PriceHistoryYearPoint, RecentSoldEntry, TradeInEstimateInput, ) from app.services.exporters.trade_in_pdf import generate_trade_in_pdf from app.services.image_sanitizer import ImageSanitizationError, sanitize_image logger = logging.getLogger(__name__) router = APIRouter() @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) @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, 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) 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 [])] # ВАЖНО: возвращаем ПОЛНЫЙ набор полей. Раньше эндпоинт отдавал огрызок # без sources_used / confidence_explanation / координат — и при открытии # оценки по ссылке (?id=) карточка источников пустела до «0/7». 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, confidence_explanation=row.confidence_explanation, n_analogs=row.n_analogs, period_months=12, 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, 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, ) @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}"'}, ) # ── Фото квартиры (#394) ───────────────────────────────────────────────────── _MAX_PHOTO_BYTES = 10 * 1024 * 1024 # 10 МБ на фото _MAX_PHOTOS_PER_ESTIMATE = 12 _ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp", "image/heic"} @router.post("/estimate/{estimate_id}/photos", response_model=PhotoMeta) async def upload_photo( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], file: Annotated[UploadFile, File()], ) -> PhotoMeta: """Загрузить фото квартиры к оценке (#394). Хранение в estimate_photos (bytea).""" estimate = db.execute( text("SELECT 1 FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"), {"id": str(estimate_id)}, ).fetchone() if estimate is None: raise HTTPException(status_code=404, detail="estimate not found") ctype = (file.content_type or "").lower() if ctype not in _ALLOWED_IMAGE_TYPES: raise HTTPException( status_code=415, detail=f"unsupported content-type: {ctype or 'unknown'}" ) count = db.execute( text("SELECT count(*) FROM estimate_photos WHERE estimate_id = CAST(:id AS uuid)"), {"id": str(estimate_id)}, ).scalar_one() if count >= _MAX_PHOTOS_PER_ESTIMATE: raise HTTPException( status_code=409, detail=f"photo limit reached ({_MAX_PHOTOS_PER_ESTIMATE})" ) content = await file.read() if not content: raise HTTPException(status_code=400, detail="empty file") if len(content) > _MAX_PHOTO_BYTES: raise HTTPException(status_code=413, detail="file too large (max 10 MB)") # Sanitize: re-encode through Pillow to drop EXIF, kill polyglot payloads, # cap dimensions. Closes finding #6 from 2026-05-24 audit. try: sanitized_bytes, sanitized_ctype = sanitize_image(content) except ImageSanitizationError as e: raise HTTPException(status_code=400, detail=str(e)) from e row = db.execute( text( """ INSERT INTO estimate_photos (estimate_id, filename, content_type, content, size_bytes) VALUES (CAST(:eid AS uuid), :fn, :ct, :content, :sz) RETURNING id, filename, content_type, size_bytes, uploaded_at """ ), { "eid": str(estimate_id), "fn": file.filename, "ct": sanitized_ctype, "content": sanitized_bytes, "sz": len(sanitized_bytes), }, ).mappings().fetchone() db.commit() logger.info( "photo uploaded: estimate=%s photo=%s orig_size=%d sanitized_size=%d ctype=%s", estimate_id, row["id"], len(content), len(sanitized_bytes), sanitized_ctype, ) return PhotoMeta(**row) @router.get("/estimate/{estimate_id}/photos", response_model=list[PhotoMeta]) def list_photos( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], ) -> list[PhotoMeta]: """Список фото оценки — метаданные, без содержимого (#394).""" rows = db.execute( text( """ SELECT id, filename, content_type, size_bytes, uploaded_at FROM estimate_photos WHERE estimate_id = CAST(:id AS uuid) ORDER BY uploaded_at """ ), {"id": str(estimate_id)}, ).mappings().all() return [PhotoMeta(**r) for r in rows] @router.get("/estimate/{estimate_id}/photos/{photo_id}") def get_photo( estimate_id: UUID, photo_id: UUID, db: Annotated[Session, Depends(get_db)], ) -> Response: """Отдать содержимое фото — image bytes (#394).""" row = db.execute( text( """ SELECT content, content_type FROM estimate_photos WHERE id = CAST(:pid AS uuid) AND estimate_id = CAST(:eid AS uuid) """ ), {"pid": str(photo_id), "eid": str(estimate_id)}, ).fetchone() if row is None: raise HTTPException(status_code=404, detail="photo not found") return Response( content=bytes(row.content), media_type=row.content_type, headers={"Cache-Control": "private, max-age=3600"}, ) # ── История и кэш (#399) ───────────────────────────────────────────────────── @router.get("/history") def estimate_history( db: Annotated[Session, Depends(get_db)], limit: int = 50, ) -> list[dict[str, object]]: """История оценок (#399) — последние N записей trade_in_estimates.""" rows = db.execute( text( """ SELECT id, address, rooms, area_m2, median_price, confidence, n_analogs, created_at FROM trade_in_estimates ORDER BY created_at DESC LIMIT :limit """ ), {"limit": min(max(limit, 1), 200)}, ).mappings().all() return [dict(r) for r in rows] @router.get("/cache-stats") def cache_stats(db: Annotated[Session, Depends(get_db)]) -> dict[str, object]: """Состояние данных и кэшей (#399) — для страницы «Кэш».""" row = db.execute( text( """ SELECT (SELECT count(*) FROM geocode_cache) AS geocode_cache, (SELECT count(*) FROM geocode_cache WHERE expires_at > NOW()) AS geocode_cache_fresh, (SELECT count(*) FROM listings WHERE is_active) AS listings_active, (SELECT max(scraped_at) FROM listings) AS listings_last_scraped, (SELECT count(*) FROM deals) AS deals, (SELECT count(*) FROM gendesign_cad_buildings) AS cad_buildings, (SELECT count(*) FROM house_metadata) AS house_metadata, (SELECT count(*) FROM trade_in_estimates) AS estimates_total """ ) ).mappings().fetchone() return dict(row) if row else {} # ── Stage 4a: house info + IMV benchmark для UI ─────────────────────────────── _HOUSE_SELECT_COLS = """ h.id AS house_id, h.source, h.ext_house_id, h.address, h.short_address, h.lat, h.lon, h.year_built, h.total_floors, h.house_type, h.passenger_elevators, h.cargo_elevators, h.has_concierge, h.closed_yard, h.has_playground, h.parking_type, h.developer_name, h.rating, h.reviews_count, COALESCE(h.raw_characteristics, '[]'::jsonb) AS raw_characteristics """ @router.get("/estimate/{estimate_id}/houses", response_model=list[HouseInfoForEstimate]) def get_estimate_houses( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], ) -> list[HouseInfoForEstimate]: """House(s) информация для estimate. Логика (двойной поиск, union-deduplicate): 1. Прямое совпадение по нормализованному адресу (tradein_normalize_short_addr). 2. Geo-nearby — ST_DWithin 500м, любой source (avito/derived/etc.). Возвращаем прямой матч + nearby (dedup by id), up to ~6 домов. Пустой список если нет matches. """ target = db.execute( text( """ SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid) """ ), {"id": str(estimate_id)}, ).fetchone() if target is None: raise HTTPException(status_code=404, detail="estimate not found") # Path 1: прямой матч по нормализованному адресу direct: list[Any] = [] if target.address: direct = list( db.execute( text( f""" SELECT DISTINCT {_HOUSE_SELECT_COLS}, 0 AS distance_m FROM houses h WHERE h.short_address = tradein_normalize_short_addr(:addr) OR tradein_normalize_short_addr(h.address) = tradein_normalize_short_addr(:addr) LIMIT 1 """ ), {"addr": target.address}, ).mappings().all() ) # Path 2: geo-nearby (any source, 500м radius) nearby: list[Any] = [] if target.lat is not None and target.lon is not None: nearby = list( db.execute( text( f""" SELECT DISTINCT {_HOUSE_SELECT_COLS}, ST_Distance( h.geom::geography, ST_MakePoint(:lon, :lat)::geography )::int AS distance_m FROM houses h WHERE h.geom IS NOT NULL AND ST_DWithin( h.geom::geography, ST_MakePoint(:lon, :lat)::geography, 500 ) ORDER BY distance_m LIMIT 5 """ ), {"lat": target.lat, "lon": target.lon}, ).mappings().all() ) # Merge: direct первым, затем nearby (dedup by house_id) seen_ids = {r["house_id"] for r in direct} merged = list(direct) + [r for r in nearby if r["house_id"] not in seen_ids] return [ HouseInfoForEstimate(**{k: v for k, v in row.items() if k != "distance_m"}) for row in merged ] @router.get("/estimate/{estimate_id}/placement-history", response_model=list[PlacementHistoryEntry]) def get_estimate_placement_history( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], ) -> list[PlacementHistoryEntry]: """Историческая продажная активность по дому(ам) target estimate. Возвращает rows из house_placement_history для всех houses связанных с target адресом. Сортировано по last_price_date DESC. """ target = db.execute( text( """ SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid) """ ), {"id": str(estimate_id)}, ).fetchone() if target is None: raise HTTPException(status_code=404, detail="estimate not found") # Поиск house_ids по нормализованному адресу house_ids: list[int] = [] if target.address: rows = db.execute( text( """ SELECT id FROM houses WHERE short_address = tradein_normalize_short_addr(:addr) OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr) """ ), {"addr": target.address}, ).all() house_ids = [r.id for r in rows] if not house_ids and target.lat is not None and target.lon is not None: # Geo fallback: 100м (tight radius чтобы не смешать соседние дома) rows = db.execute( text( """ SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin( geom::geography, ST_MakePoint(:lon, :lat)::geography, 100 ) LIMIT 3 """ ), {"lat": target.lat, "lon": target.lon}, ).all() house_ids = [r.id for r in rows] if not house_ids: return [] history = db.execute( text( """ SELECT id, source, house_id, ext_item_id, title, rooms, area_m2, floor, total_floors, start_price, start_price_date, last_price, last_price_date, removed_date, exposure_days, notes FROM house_placement_history WHERE house_id = ANY(:house_ids) ORDER BY COALESCE(last_price_date, start_price_date) DESC NULLS LAST LIMIT 50 """ ), {"house_ids": house_ids}, ).mappings().all() return [PlacementHistoryEntry(**dict(r)) for r in history] @router.get("/estimate/{estimate_id}/house-analytics", response_model=HouseAnalyticsResponse) def get_estimate_house_analytics( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], ) -> HouseAnalyticsResponse: """House-level analytics from house_placement_history backfill. Resolves target house(s) — если в самом доме <8 hist rows — расширяем поиск до 300м. Возвращает: price-history by year (median ₽/м²), recent sold (12mo), KPI. """ target = db.execute( text("SELECT lat, lon, address FROM trade_in_estimates WHERE id = CAST(:id AS uuid)"), {"id": str(estimate_id)}, ).fetchone() if target is None: raise HTTPException(status_code=404, detail="estimate not found") # 1. Resolve target house_ids (same as placement-history endpoint) house_ids: list[int] = [] if target.address: rows = db.execute( text( "SELECT id FROM houses WHERE short_address = tradein_normalize_short_addr(:addr) " "OR tradein_normalize_short_addr(address) = tradein_normalize_short_addr(:addr)" ), {"addr": target.address}, ).all() house_ids = [r.id for r in rows] if not house_ids and target.lat is not None and target.lon is not None: rows = db.execute( text( "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin(" "geom::geography, ST_MakePoint(:lon, :lat)::geography, 100) LIMIT 3" ), {"lat": target.lat, "lon": target.lon}, ).all() house_ids = [r.id for r in rows] # 2. Expand to 300m if too few hist rows radius_used = 0 n_in_house = 0 if house_ids: n_in_house = ( db.execute( text("SELECT COUNT(*) FROM house_placement_history WHERE house_id = ANY(:ids)"), {"ids": house_ids}, ).scalar() or 0 ) if n_in_house < 8 and target.lat is not None and target.lon is not None: rows = db.execute( text( "SELECT id FROM houses WHERE geom IS NOT NULL AND ST_DWithin(" "geom::geography, ST_MakePoint(:lon, :lat)::geography, 300) LIMIT 30" ), {"lat": target.lat, "lon": target.lon}, ).all() house_ids = sorted(set(house_ids) | {r.id for r in rows}) radius_used = 300 if not house_ids: return HouseAnalyticsResponse( house_ids=[], radius_m=0, price_history=[], recent_sold=[], kpi=HouseAnalyticsKpi( total_lots=0, sold_count=0, sold_rate_pct=0.0, median_exposure_days=None, median_bargain_pct=None, ), ) # 3. Price history by year × source (median ₽/м²) price_history_rows = db.execute( text( """ SELECT EXTRACT(YEAR FROM COALESCE(last_price_date, start_price_date))::int AS year, source, COUNT(*) AS n_lots, percentile_cont(0.5) WITHIN GROUP (ORDER BY last_price / NULLIF(area_m2, 0))::int AS median_price_per_m2, percentile_cont(0.5) WITHIN GROUP (ORDER BY last_price)::int AS median_price_rub FROM house_placement_history WHERE house_id = ANY(:ids) AND last_price IS NOT NULL AND last_price > 100000 AND area_m2 IS NOT NULL AND area_m2 > 10 AND COALESCE(last_price_date, start_price_date) IS NOT NULL GROUP BY year, source ORDER BY year ASC, source ASC """ ), {"ids": house_ids}, ).mappings().all() # 4. Recent sold (12 months, with removed_date) recent_sold_rows = db.execute( text( """ SELECT id, source, rooms, area_m2, floor, start_price, last_price, removed_date, exposure_days, CASE WHEN start_price > 0 AND last_price IS NOT NULL THEN ROUND((start_price - last_price)::numeric / start_price * 100, 1) ELSE NULL END AS discount_pct FROM house_placement_history WHERE house_id = ANY(:ids) AND removed_date IS NOT NULL AND removed_date > (NOW() - INTERVAL '12 months')::date ORDER BY removed_date DESC LIMIT 20 """ ), {"ids": house_ids}, ).mappings().all() # 5. KPI aggregate kpi_row = db.execute( text( """ SELECT COUNT(*) AS total_lots, COUNT(*) FILTER (WHERE removed_date IS NOT NULL) AS sold_count, CASE WHEN COUNT(*) > 0 THEN ROUND( COUNT(*) FILTER (WHERE removed_date IS NOT NULL)::numeric / COUNT(*) * 100, 1 ) ELSE 0 END AS sold_rate_pct, percentile_cont(0.5) WITHIN GROUP (ORDER BY exposure_days) FILTER (WHERE exposure_days IS NOT NULL) AS median_exposure_days, ROUND( percentile_cont(0.5) WITHIN GROUP ( ORDER BY (start_price - last_price)::numeric / NULLIF(start_price, 0) * 100 ) FILTER ( WHERE start_price > 0 AND last_price IS NOT NULL AND last_price != start_price )::numeric, 1 ) AS median_bargain_pct FROM house_placement_history WHERE house_id = ANY(:ids) """ ), {"ids": house_ids}, ).mappings().first() return HouseAnalyticsResponse( house_ids=house_ids, radius_m=radius_used, price_history=[PriceHistoryYearPoint(**dict(r)) for r in price_history_rows], recent_sold=[RecentSoldEntry(**dict(r)) for r in recent_sold_rows], kpi=HouseAnalyticsKpi( total_lots=kpi_row["total_lots"] or 0 if kpi_row else 0, sold_count=kpi_row["sold_count"] or 0 if kpi_row else 0, sold_rate_pct=float(kpi_row["sold_rate_pct"] or 0) if kpi_row else 0.0, median_exposure_days=( int(kpi_row["median_exposure_days"]) if kpi_row and kpi_row["median_exposure_days"] is not None else None ), median_bargain_pct=( float(kpi_row["median_bargain_pct"]) if kpi_row and kpi_row["median_bargain_pct"] is not None else None ), ), ) @router.get( "/estimate/{estimate_id}/cian-price-changes", response_model=list[CianPriceChangeStats], ) def get_estimate_cian_price_changes( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], ) -> list[CianPriceChangeStats]: """История изменений цены для Cian-аналогов из estimate. Для каждого cian-аналога в estimate.analogs: - extract cian_id из URL (/sale/flat//) - JOIN listings (source='cian', source_id IN cian_ids) - JOIN offer_price_history per listing_id - Aggregate: n_changes, last_change_time, last_diff_percent, total_change_pct Возвращает только аналоги с хотя бы одним изменением цены. Пустой список если нет cian-аналогов или нет истории. """ rows = db.execute( text( """ WITH cian_analogs AS ( SELECT DISTINCT substring(a->>'source_url' from '/sale/flat/(\\d+)/') AS cian_id FROM trade_in_estimates, jsonb_array_elements(analogs) a WHERE id = CAST(:eid AS uuid) AND a->>'source' = 'cian' AND a->>'source_url' IS NOT NULL AND substring(a->>'source_url' from '/sale/flat/(\\d+)/') IS NOT NULL ), listings_resolved AS ( SELECT l.id AS listing_id, l.source_id AS cian_id, l.price_rub::int AS current_price FROM listings l JOIN cian_analogs c ON l.source_id = c.cian_id WHERE l.source = 'cian' ), changes_agg AS ( SELECT oph.listing_id, COUNT(*) AS n_changes, MAX(oph.change_time) AS last_change_time, (array_agg(oph.diff_percent ORDER BY oph.change_time DESC))[1] AS last_diff_percent, ( SELECT oph2.price_rub::int FROM offer_price_history oph2 WHERE oph2.listing_id = oph.listing_id ORDER BY oph2.change_time ASC LIMIT 1 ) AS first_seen_price FROM offer_price_history oph WHERE oph.listing_id IN (SELECT listing_id FROM listings_resolved) GROUP BY oph.listing_id ) SELECT lr.cian_id, lr.listing_id, ca.n_changes::int, ca.last_change_time, ca.last_diff_percent::float AS last_diff_percent, ca.first_seen_price, lr.current_price, CASE WHEN ca.first_seen_price IS NOT NULL AND ca.first_seen_price > 0 THEN ROUND( (lr.current_price - ca.first_seen_price)::numeric / ca.first_seen_price * 100, 1 )::float ELSE NULL END AS total_change_pct FROM listings_resolved lr JOIN changes_agg ca ON ca.listing_id = lr.listing_id WHERE ca.n_changes > 0 ORDER BY ca.last_change_time DESC """ ), {"eid": str(estimate_id)}, ).mappings().all() return [CianPriceChangeStats(**dict(r)) for r in rows] @router.get("/estimate/{estimate_id}/imv-benchmark", response_model=IMVBenchmarkResponse) def get_estimate_imv_benchmark( estimate_id: UUID, db: Annotated[Session, Depends(get_db)], ) -> IMVBenchmarkResponse: """Avito IMV benchmark для estimate (для UI badge «наша 6.4М · Avito 6.29М»). Источники lookup: 1. avito_imv_evaluations WHERE estimate_id = :id (если linked в estimator) 2. Если не linked — fallback: most recent IMV для same address (TTL 24h) """ # Сначала пытаемся найти directly linked row = db.execute( text( """ SELECT cache_key, recommended_price, lower_price, higher_price, market_count, fetched_at FROM avito_imv_evaluations WHERE estimate_id = CAST(:id AS uuid) ORDER BY fetched_at DESC LIMIT 1 """ ), {"id": str(estimate_id)}, ).fetchone() if row is None: # Fallback: same address за 24h (на случай если link не успел) est = db.execute( text( """ SELECT address FROM trade_in_estimates WHERE id = CAST(:id AS uuid) """ ), {"id": str(estimate_id)}, ).fetchone() if est is None: raise HTTPException(status_code=404, detail="estimate not found") if est.address: row = db.execute( text( """ SELECT cache_key, recommended_price, lower_price, higher_price, market_count, fetched_at FROM avito_imv_evaluations WHERE address = :address AND fetched_at > NOW() - INTERVAL '24 hours' ORDER BY fetched_at DESC LIMIT 1 """ ), {"address": est.address}, ).fetchone() if row is None: return IMVBenchmarkResponse(available=False) # Get our_median_price для compare our = db.execute( text( """ SELECT median_price FROM trade_in_estimates WHERE id = CAST(:id AS uuid) """ ), {"id": str(estimate_id)}, ).fetchone() our_median = our.median_price if our else None diff_pct = None if our_median and row.recommended_price: diff_pct = round((our_median - row.recommended_price) / row.recommended_price * 100, 1) return IMVBenchmarkResponse( available=True, cache_key=row.cache_key, recommended_price=row.recommended_price, lower_price=row.lower_price, higher_price=row.higher_price, market_count=row.market_count, fetched_at=row.fetched_at, our_median_price=our_median, diff_pct=diff_pct, )