"""avito_imv.py — Avito IMV evaluation API client (3-request flow, 2026-05+). Stage 2d: geocode + IMV evaluation. Flow (current contract, подтверждён живым capture 2026-05 — fixtures tests/fixtures/avito_imv_geo_position.json + avito_imv_getdata.json): 0) GET /evaluation/realty → warm-up (seed anti-bot cookies) 1) GET /web/1/coords/by_address?address= → normalize + lat/lon 2) POST /js/v2/geo/position → rich JWT (geoFieldsHash) 3) POST /web/1/realty-imv/get-data → price + placementHistory? + suggestions Anti-bot: zerkalim AvitoScraper (avito.py) — chrome120 TLS + document-заголовки + warm-up GET + XHR Sec-Fetch заголовки. Bare-session XHR ловят 403 с server-IP. Таблицы: avito_imv_evaluations (018_avito_imv_evaluations.sql) house_placement_history (017_house_placement_history.sql) """ from __future__ import annotations import hashlib import json import logging import re from dataclasses import dataclass, field from datetime import date from typing import TYPE_CHECKING, Any from urllib.parse import urlencode from uuid import UUID from sqlalchemy import text from sqlalchemy.orm import Session from app.services.scrapers.avito_shared import _unix_to_date if TYPE_CHECKING: # Импорт под TYPE_CHECKING (как в avito_detail.py/avito_houses.py) — избегаем # hard import cycle browser_fetcher → config → ... и лишней зависимости в # offline-тестах парсинга, которым browser не нужен. from app.services.scrapers.browser_fetcher import BrowserFetcher logger = logging.getLogger(__name__) AVITO_BASE = "https://www.avito.ru" COORDS_ENDPOINT = "/web/1/coords/by_address" POSITION_ENDPOINT = "/js/v2/geo/position" IMV_ENDPOINT = "/web/1/realty-imv/get-data" # Тёплая страница для seed anti-bot cookies (srv_id/_avisc/u/...) перед XHR. WARMUP_URL = f"{AVITO_BASE}/evaluation/realty" # Заголовки document-навигации (warm-up GET) — зеркалят production-набор # из avito.py (AvitoScraper.__aenter__). Нужны чтобы пройти TLS+header anti-bot. _DOC_HEADERS = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", "Cache-Control": "max-age=0", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Sec-Fetch-User": "?1", "Upgrade-Insecure-Requests": "1", } # Заголовки XHR/fetch для API-шагов (coords/position/get-data). Имитируют # fetch() со страницы /evaluation/realty: JSON Accept + same-origin Sec-Fetch. _COMMON_HEADERS = { "Accept": "application/json, text/plain, */*", "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", "Referer": "https://www.avito.ru/evaluation/realty", "X-Requested-With": "XMLHttpRequest", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", } # Timeout для всех Avito-вызовов (без него curl_cffi висит на anti-bot stall). _HTTP_TIMEOUT_SEC = 25 # --------------------------------------------------------------------------- # Exceptions # --------------------------------------------------------------------------- class IMVAddressNotFoundError(Exception): """Raised when /coords/by_address returns no geoHash for given address.""" class IMVCityMismatchError(Exception): """Raised when locationId не входит в ожидаемый город (sanity check).""" class IMVAuthError(Exception): """Avito API rejected auth / quota exceeded (HTTP 401 / 403).""" class IMVTransientError(Exception): """Network / 5xx / timeout — retryable.""" # --------------------------------------------------------------------------- # Dataclasses # --------------------------------------------------------------------------- @dataclass class IMVGeo: geo_hash: str # JWT — не сохраняем как отдельную колонку, только в raw_response lat: float | None = None lon: float | None = None avito_address_id: int | None = None avito_location_id: int | None = None avito_metro_id: int | None = None avito_district_id: int | None = None @dataclass class IMVPlacementHistoryItem: ext_item_id: str # str(item['id']) title: str | None = None rooms: int | None = None # парсится из title (опционально) area_m2: float | None = None # парсится из title (опционально) floor: int | None = None # парсится из title (опционально) total_floors: int | None = None # парсится из title (опционально) start_price: int | None = None start_price_date: date | None = None # unix → date last_price: int | None = None last_price_date: date | None = None removed_date: date | None = None # только IMV API (в widget housePlacementHistory нет) exposure_days: int | None = None raw_payload: dict[str, Any] | None = None @dataclass class IMVSuggestion: ext_item_id: str title: str | None = None address: str | None = None price_rub: int | None = None exposure_days: int | None = None publish_date: date | None = None item_url: str | None = None metro_name: str | None = None metro_distance: str | None = None metro_color: str | None = None has_good_price_badge: bool = False raw_payload: dict[str, Any] | None = None @dataclass class IMVEvaluation: cache_key: str # sha256 address: str rooms: int area_m2: float floor: int floor_at_home: int house_type: str renovation_type: str has_balcony: bool has_loggia: bool geo: IMVGeo recommended_price: int lower_price: int higher_price: int market_count: int | None = None placement_history: list[IMVPlacementHistoryItem] = field(default_factory=list) suggestions: list[IMVSuggestion] = field(default_factory=list) search_similar_url: str | None = None raw_response: dict[str, Any] | None = None # --------------------------------------------------------------------------- # Cache key # --------------------------------------------------------------------------- def compute_imv_cache_key( address: str, rooms: int, area_m2: float, floor: int, floor_at_home: int, house_type: str, renovation_type: str, has_balcony: bool, has_loggia: bool, ) -> str: """Детерминированный sha256 ключ для 24h-кэша IMV оценки.""" key_src = ( f"{address}|{rooms}|{area_m2}|{floor}|{floor_at_home}" f"|{house_type}|{renovation_type}|{int(has_balcony)}|{int(has_loggia)}" ) return hashlib.sha256(key_src.encode("utf-8")).hexdigest() # --------------------------------------------------------------------------- # Parse helpers # --------------------------------------------------------------------------- # Паттерн для заголовка вида "2-к. квартира, 42 м², 4/5 эт." _TITLE_RE = re.compile( r"^(?P\d+)-к[.\s].*?(?P[\d,]+)\s*м².*?(?P\d+)/(?P\d+)\s*эт", re.IGNORECASE, ) def _parse_title(title: str | None) -> dict[str, int | float | None]: """Парсит rooms, area_m2, floor, total_floors из Avito-заголовка объявления.""" result: dict[str, int | float | None] = { "rooms": None, "area_m2": None, "floor": None, "total_floors": None, } if not title: return result m = _TITLE_RE.match(title.strip()) if not m: return result try: result["rooms"] = int(m.group("rooms")) result["area_m2"] = float(m.group("area").replace(",", ".")) result["floor"] = int(m.group("floor")) result["total_floors"] = int(m.group("total")) except (ValueError, AttributeError): pass return result def _parse_placement_item(raw: dict[str, Any]) -> IMVPlacementHistoryItem: """Парсит один элемент из placementHistory.items.""" title = raw.get("title") parsed = _parse_title(title) # Фильтруем itemImage из raw_payload (не нужны) clean_raw = {k: v for k, v in raw.items() if k not in ("itemImage",)} return IMVPlacementHistoryItem( ext_item_id=str(raw["id"]), title=title, rooms=parsed["rooms"], # type: ignore[arg-type] area_m2=parsed["area_m2"], # type: ignore[arg-type] floor=parsed["floor"], # type: ignore[arg-type] total_floors=parsed["total_floors"], # type: ignore[arg-type] start_price=raw.get("startPrice"), start_price_date=_unix_to_date(raw.get("startPriceDate")), last_price=raw.get("lastPrice"), last_price_date=_unix_to_date(raw.get("lastPriceDate")), removed_date=_unix_to_date(raw.get("removedDate")), exposure_days=raw.get("exposure"), raw_payload=clean_raw, ) def _parse_suggestion(raw: dict[str, Any]) -> IMVSuggestion: """Парсит один элемент из suggestions.items.""" metro = raw.get("metro") or {} colors: list[str] = metro.get("colors") or [] # Фильтруем imageLink из raw_payload clean_raw = {k: v for k, v in raw.items() if k not in ("imageLink",)} return IMVSuggestion( ext_item_id=str(raw["id"]), title=raw.get("title"), address=raw.get("address"), price_rub=raw.get("price"), exposure_days=raw.get("exposure"), publish_date=_unix_to_date(raw.get("publishDate")), item_url=raw.get("itemLink"), metro_name=metro.get("name"), metro_distance=metro.get("distance"), metro_color=colors[0] if colors else None, has_good_price_badge=bool(raw.get("hasGoodPriceBadge", False)), raw_payload=clean_raw, ) def _parse_geo_position(data_b: dict[str, Any], *, lat: float, lon: float) -> IMVGeo: """Парсит ответ POST /js/v2/geo/position (step B) → IMVGeo. Контракт (подтверждён живым capture 2026-05, fixture avito_imv_geo_position.json): {address, addressId, districtId, geoFieldsHash, latitude, longitude, locationId, parentLocationId, metroId?, errorText, ...} `geoFieldsHash` — JWT, payload которого кодирует {latitude, longitude, address, addressId, locationId, districtId}; это то, что step C ожидает как `geoHash`. metroId присутствует только для метро-городов (None для ЕКБ-окраин). Raises: IMVAddressNotFoundError если geoFieldsHash пустой/отсутствует. """ jwt = data_b.get("geoFieldsHash") if not jwt: raise IMVAddressNotFoundError( f"Avito /js/v2/geo/position не вернул geoFieldsHash для " f"{str(data_b.get('address'))[:60]!r}: errorText={data_b.get('errorText')!r}" ) return IMVGeo( geo_hash=jwt, lat=lat, lon=lon, avito_address_id=data_b.get("addressId"), avito_location_id=data_b.get("locationId"), avito_metro_id=data_b.get("metroId"), avito_district_id=data_b.get("districtId"), ) def _parse_price(data: dict[str, Any]) -> tuple[int, int, int, int | None]: """Парсит top-level `price` блок из realty-imv/get-data (step C). Контракт (fixture avito_imv_getdata.json): price = {recommendedPrice, lowerPrice, higherPrice, count, addItemUrl} Returns: (recommended_price, lower_price, higher_price, market_count). Отсутствующие цены → 0 (caller трактует как "нет оценки"); count → None. """ price_block: dict[str, Any] = data.get("price") or {} recommended_price: int = price_block.get("recommendedPrice") or 0 lower_price: int = price_block.get("lowerPrice") or 0 higher_price: int = price_block.get("higherPrice") or 0 market_count: int | None = price_block.get("count") return recommended_price, lower_price, higher_price, market_count def _parse_placement_history(data: dict[str, Any]) -> list[IMVPlacementHistoryItem]: """Парсит опциональный `placementHistory.items` блок (step C). placementHistory отсутствует для части адресов (живой capture: проспект Ленина вернул только itemParams/price/suggestions) — тогда возвращаем []. Битые элементы пропускаем с warning, не роняя весь ответ. """ history_block: dict[str, Any] = data.get("placementHistory") or {} items: list[IMVPlacementHistoryItem] = [] for raw_item in history_block.get("items") or []: try: items.append(_parse_placement_item(raw_item)) except (KeyError, TypeError) as exc: logger.warning("IMV: не удалось распарсить placementHistory item: %s", exc) return items def _parse_suggestions(data: dict[str, Any]) -> tuple[list[IMVSuggestion], str | None]: """Парсит `suggestions` блок (step C) → (items, searchSimilarUrl).""" suggestions_block: dict[str, Any] = data.get("suggestions") or {} suggestions: list[IMVSuggestion] = [] for raw_sugg in suggestions_block.get("items") or []: try: suggestions.append(_parse_suggestion(raw_sugg)) except (KeyError, TypeError) as exc: logger.warning("IMV: не удалось распарсить suggestion: %s", exc) return suggestions, suggestions_block.get("searchSimilarUrl") # --------------------------------------------------------------------------- # Browser-transport adapter (#915 Stage 3) # --------------------------------------------------------------------------- class _BrowserResponse: """Мимикрия под curl_cffi Response поверх ответа BrowserFetcher.fetch_json. IMV-флоу читает только .status_code / .text / .json() / .raise_for_status() — адаптируем ровно эти четыре (подтверждено по avito_imv.py: warm-up, _geocode, _imv_evaluate, _raise_for_status_categorized). """ def __init__(self, status: int, body: str) -> None: self.status_code = status self.text = body def json(self) -> Any: return json.loads(self.text) def raise_for_status(self) -> None: if self.status_code >= 400: # _raise_for_status_categorized уже обрабатывает 401/403/5xx ДО вызова # raise_for_status(); сюда попадают только "прочие 4xx" (fall-through). # Зеркалим curl_cffi/httpx-семантику ровно настолько, чтобы caller # получил исключение. raise RuntimeError(f"HTTP {self.status_code}") class _BrowserSessionAdapter: """Адаптер: IMV-флоу думает что это curl_cffi-сессия, а ходит через камуфокс in-page fetch (same-origin avito.ru). Куки/JWT живут в контексте браузера провайдера, поэтому 4 запроса IMV переиспользуют warmed-сессию. Прокси и fingerprint — из sidecar (обходит datacenter-403, #562/#853).""" def __init__(self, browser_fetcher: BrowserFetcher, *, origin: str) -> None: self._bf = browser_fetcher self._origin = origin async def get(self, url: str, *, headers: dict | None = None) -> _BrowserResponse: r = await self._bf.fetch_json(url, method="GET", headers=headers, origin=self._origin) return _BrowserResponse(r["status"], r["body"]) async def post( self, url: str, *, headers: dict | None = None, json: Any = None ) -> _BrowserResponse: import json as _json body = _json.dumps(json) if json is not None else None r = await self._bf.fetch_json( url, method="POST", headers=headers, body=body, origin=self._origin ) return _BrowserResponse(r["status"], r["body"]) async def close(self) -> None: # browser принадлежит вызывающему (backfill открывает один на батч) — no-op. return None # --------------------------------------------------------------------------- # Main async function # --------------------------------------------------------------------------- async def evaluate_via_imv( address: str, rooms: int, area_m2: float, floor: int, floor_at_home: int, house_type: str, # panel/brick/monolith/monolith_brick/block/wood renovation_type: str, # required/cosmetic/euro/designer has_balcony: bool, has_loggia: bool, *, cffi_session: Any | None = None, browser_fetcher: BrowserFetcher | None = None, ) -> IMVEvaluation: """Avito IMV — 3 HTTP requests (current contract 2026-05+): 1) GET /web/1/coords/by_address?address= → normalize + lat/lon 2) POST /js/v2/geo/position → rich JWT (geoFieldsHash) 3) POST /web/1/realty-imv/get-data → price + placementHistory + suggestions Returns IMVEvaluation с уже вычисленным cache_key. Raises: IMVAddressNotFoundError если point/normalizedAddress пустые (step 1) или geoFieldsHash не вернулся (step 2) IMVAuthError на 401/403 IMVTransientError на 5xx / network errors """ # #915 Stage 3: browser-транспорт. Если передан browser_fetcher — ходим через # камуфокс in-page fetch (same-origin avito.ru), а не curl_cffi. Прокси + # warmed-cookies + реальный fingerprint из sidecar обходят datacenter-403 # (#562/#853). _own_session=True → finally дёрнет adapter.close() (no-op, # браузер принадлежит caller'у — backfill открывает один на батч). Этот путь # НЕ требует curl_cffi, поэтому импорт пакета остаётся только в else-ветке. if browser_fetcher is not None: cffi_session = _BrowserSessionAdapter(browser_fetcher, origin=WARMUP_URL) _own_session = True else: # Импортируем здесь чтобы избежать циклических зависимостей при тестах без сети try: from curl_cffi.requests import AsyncSession as CffiAsyncSession from app.core.config import settings _own_session = False if cffi_session is None: # Зеркалим production-набор из AvitoScraper.__aenter__ (avito.py:150-162): # chrome120 TLS + document-заголовки + timeout. Затем warm-up GET для # seed anti-bot cookies — bare-session XHR Avito банит на server-IP. # Mobile proxy wiring (#806 follow-up): только для own-session (переданная # сессия уже с прокси от AvitoScraper). proxy=None → прямое подключение. _proxy_url = settings.scraper_proxy_url _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None cffi_session = CffiAsyncSession( impersonate="chrome120", timeout=_HTTP_TIMEOUT_SEC, proxies=_proxies, headers=_DOC_HEADERS, ) _own_session = True except ImportError as exc: raise RuntimeError( "curl_cffi не установлен. Добавь 'curl-cffi>=0.7.0' в pyproject.toml." ) from exc try: if _own_session: await _warmup(cffi_session) geo = await _geocode(cffi_session, address) evaluation = await _imv_evaluate( cffi_session, geo=geo, address=address, rooms=rooms, area_m2=area_m2, floor=floor, floor_at_home=floor_at_home, house_type=house_type, renovation_type=renovation_type, has_balcony=has_balcony, has_loggia=has_loggia, ) finally: if _own_session: await cffi_session.close() return evaluation async def _warmup(session: Any) -> None: """GET /evaluation/realty чтобы seed anti-bot cookies (srv_id/_avisc/u/...). Без warm-up XHR-шаги (coords/position/get-data) c server-IP ловят 403 anti-bot. Best-effort: не роняем оценку если страница недоступна — последующие шаги всё равно попробуют и дадут типизированную ошибку. """ try: resp = await session.get(WARMUP_URL, headers=_DOC_HEADERS) logger.info("IMV warm-up GET /evaluation/realty → HTTP %d", resp.status_code) except Exception as exc: # warm-up опционален: логируем и продолжаем — шаги ниже дадут типизированную # ошибку, если cookies реально нужны. Не re-raise намеренно. logger.warning("IMV warm-up failed (продолжаем без cookies): %s", exc) def _raise_for_status_categorized(resp: Any, context: str) -> None: """Raise typed IMV exception based on HTTP status code. Используется вместо bare raise_for_status() чтобы caller мог различать auth-failure (требует human attention) от transient (retryable). """ status = resp.status_code if status in (401, 403): raise IMVAuthError(f"Avito {context}: HTTP {status} — auth rejected / quota exceeded") if status >= 500: raise IMVTransientError(f"Avito {context}: HTTP {status} — transient server error") if status >= 400: body_preview = (resp.text or "")[:200] logger.warning("Avito %s HTTP %d: body=%r", context, status, body_preview) resp.raise_for_status() # other 4xx → standard HTTPStatusError async def _geocode(session: Any, address: str) -> IMVGeo: """Two-step Avito IMV geocoder (current contract 2026-05+). Step A: GET /web/1/coords/by_address → normalize + lat/lon Step B: POST /js/v2/geo/position → rich JWT (geoFieldsHash) with addressId/locationId/metroId/districtId JWT from step B is what /realty-imv/get-data expects as `geoHash`. """ # Step A — normalize + coords params = urlencode({"address": address}) url_a = f"{AVITO_BASE}{COORDS_ENDPOINT}?{params}" logger.info("IMV geocode A: address=%r", address) try: resp_a = await session.get(url_a, headers=_COMMON_HEADERS) except Exception as exc: raise IMVTransientError(f"Avito geocode A network error: {exc}") from exc _raise_for_status_categorized(resp_a, "geocode-A") data_a: dict[str, Any] = resp_a.json() point = data_a.get("point") or {} lat = point.get("latitude") lon = point.get("longitude") normalized = data_a.get("normalizedAddress") if not lat or not lon or not normalized: raise IMVAddressNotFoundError(f"Avito /coords/by_address не вернул point для {address!r}") # Step B — rich JWT url_b = f"{AVITO_BASE}{POSITION_ENDPOINT}" payload_b = { "address": normalized, "addressId": "", "categoryId": 24, # 24 = недвижимость / квартиры "isRadius": False, "latitude": lat, "longitude": lon, } logger.info("IMV geocode B: position for %r", normalized[:60]) try: resp_b = await session.post( url_b, headers={**_COMMON_HEADERS, "Content-Type": "application/json"}, json=payload_b, ) except Exception as exc: raise IMVTransientError(f"Avito geocode B network error: {exc}") from exc _raise_for_status_categorized(resp_b, "geocode-B") data_b: dict[str, Any] = resp_b.json() geo = _parse_geo_position(data_b, lat=lat, lon=lon) # raises IMVAddressNotFoundError logger.info( "IMV geocode OK: addressId=%s locationId=%s metroId=%s lat=%s lon=%s", geo.avito_address_id, geo.avito_location_id, geo.avito_metro_id, lat, lon, ) return geo async def _imv_evaluate( session: Any, *, geo: IMVGeo, address: str, rooms: int, area_m2: float, floor: int, floor_at_home: int, house_type: str, renovation_type: str, has_balcony: bool, has_loggia: bool, ) -> IMVEvaluation: """Request 2: POST /web/1/realty-imv/get-data → IMVEvaluation.""" url = f"{AVITO_BASE}{IMV_ENDPOINT}" body = { "geoHash": geo.geo_hash, "floor": floor, "floorAtHome": floor_at_home, "rooms": rooms, "area": area_m2, "houseType": house_type, "renovationType": renovation_type, "hasLoggia": has_loggia, "hasBalcony": has_balcony, } headers = {**_COMMON_HEADERS, "Content-Type": "application/json"} logger.info( "IMV evaluate: address=%r rooms=%d area=%.1f floor=%d/%d", address, rooms, area_m2, floor, floor_at_home, ) try: resp = await session.post(url, json=body, headers=headers) except Exception as exc: raise IMVTransientError(f"Avito IMV network error: {exc}") from exc _raise_for_status_categorized(resp, "imv-evaluate") data: dict[str, Any] = resp.json() recommended_price, lower_price, higher_price, market_count = _parse_price(data) placement_history = _parse_placement_history(data) suggestions, search_similar_url = _parse_suggestions(data) logger.info( "IMV evaluate OK: recommended=%d range=(%d, %d) count=%s history=%d suggestions=%d", recommended_price, lower_price, higher_price, market_count, len(placement_history), len(suggestions), ) cache_key = compute_imv_cache_key( address=address, rooms=rooms, area_m2=area_m2, floor=floor, floor_at_home=floor_at_home, house_type=house_type, renovation_type=renovation_type, has_balcony=has_balcony, has_loggia=has_loggia, ) return IMVEvaluation( cache_key=cache_key, address=address, rooms=rooms, area_m2=area_m2, floor=floor, floor_at_home=floor_at_home, house_type=house_type, renovation_type=renovation_type, has_balcony=has_balcony, has_loggia=has_loggia, geo=geo, recommended_price=recommended_price, lower_price=lower_price, higher_price=higher_price, market_count=market_count, placement_history=placement_history, suggestions=suggestions, search_similar_url=search_similar_url, raw_response=data, ) # --------------------------------------------------------------------------- # DB save functions # --------------------------------------------------------------------------- def save_imv_evaluation( db: Session, e: IMVEvaluation, *, estimate_id: UUID | None = None, ) -> int: """INSERT в avito_imv_evaluations ON CONFLICT(cache_key) DO UPDATE. Колонки строго по 018_avito_imv_evaluations.sql DDL. estimate_id — nullable FK на trade_in_estimates. Returns: id вставленной/обновлённой строки. """ row = db.execute( text(""" INSERT INTO avito_imv_evaluations ( cache_key, estimate_id, address, rooms, area_m2, floor, floor_at_home, house_type, renovation_type, has_balcony, has_loggia, geo_hash, lat, lon, avito_address_id, avito_location_id, avito_metro_id, avito_district_id, recommended_price, lower_price, higher_price, market_count, raw_response, fetched_at ) VALUES ( :cache_key, CAST(:estimate_id AS uuid), :address, :rooms, :area_m2, :floor, :floor_at_home, :house_type, :renovation_type, :has_balcony, :has_loggia, :geo_hash, :lat, :lon, :avito_address_id, :avito_location_id, :avito_metro_id, :avito_district_id, :recommended_price, :lower_price, :higher_price, :market_count, CAST(:raw_response AS jsonb), NOW() ) ON CONFLICT (cache_key) DO UPDATE SET estimate_id = COALESCE(EXCLUDED.estimate_id, avito_imv_evaluations.estimate_id), recommended_price = EXCLUDED.recommended_price, lower_price = EXCLUDED.lower_price, higher_price = EXCLUDED.higher_price, market_count = EXCLUDED.market_count, raw_response = EXCLUDED.raw_response, fetched_at = NOW() RETURNING id """), { "cache_key": e.cache_key, "estimate_id": str(estimate_id) if estimate_id else None, "address": e.address, "rooms": e.rooms, "area_m2": e.area_m2, "floor": e.floor, "floor_at_home": e.floor_at_home, "house_type": e.house_type, "renovation_type": e.renovation_type, "has_balcony": e.has_balcony, "has_loggia": e.has_loggia, "geo_hash": e.geo.geo_hash, "lat": e.geo.lat, "lon": e.geo.lon, "avito_address_id": e.geo.avito_address_id, "avito_location_id": e.geo.avito_location_id, "avito_metro_id": e.geo.avito_metro_id, "avito_district_id": e.geo.avito_district_id, "recommended_price": e.recommended_price, "lower_price": e.lower_price, "higher_price": e.higher_price, "market_count": e.market_count, "raw_response": ( json.dumps(e.raw_response, ensure_ascii=False) if e.raw_response else None ), }, ).fetchone() db.commit() return row.id if row else 0 def save_imv_placement_history( db: Session, eval_id: int, items: list[IMVPlacementHistoryItem], ) -> int: """INSERT в house_placement_history (source='avito_imv') с removed_date. Колонки строго по 017_house_placement_history.sql DDL. Колонка scraped_at (не fetched_at) — по DDL таблицы. house_id — NULL (backfill через отдельный процесс). eval_id принимается для логирования (не сохраняется — нет FK в 017). ON CONFLICT (source, ext_item_id) DO UPDATE — refresh dates. Returns: количество вставленных/обновлённых строк. """ count = 0 for item in items: db.execute( text(""" INSERT INTO house_placement_history ( source, ext_item_id, title, rooms, area_m2, floor, total_floors, start_price, start_price_date, last_price, last_price_date, removed_date, exposure_days, raw_payload, scraped_at ) VALUES ( 'avito_imv', :ext_item_id, :title, :rooms, :area_m2, :floor, :total_floors, :start_price, :start_price_date, :last_price, :last_price_date, :removed_date, :exposure_days, CAST(:raw_payload AS jsonb), NOW() ) ON CONFLICT (source, ext_item_id) DO UPDATE SET last_price = EXCLUDED.last_price, last_price_date = EXCLUDED.last_price_date, removed_date = EXCLUDED.removed_date, exposure_days = EXCLUDED.exposure_days, raw_payload = EXCLUDED.raw_payload, scraped_at = NOW() """), { "ext_item_id": item.ext_item_id, "title": item.title, "rooms": item.rooms, "area_m2": item.area_m2, "floor": item.floor, "total_floors": item.total_floors, "start_price": item.start_price, "start_price_date": item.start_price_date, "last_price": item.last_price, "last_price_date": item.last_price_date, "removed_date": item.removed_date, "exposure_days": item.exposure_days, "raw_payload": ( json.dumps(item.raw_payload, ensure_ascii=False) if item.raw_payload else None ), }, ) count += 1 if count: db.commit() logger.info("IMV placement history saved: eval_id=%d items=%d", eval_id, count) return count