"""T7: House-level IMV backfill service. Iterates houses with imv_status='pending' (or 'transient_error') that have lat/lon coordinates and at least one linked listing with rooms+area data. For each house: 1. Pick median lot-params from linked listings (rooms, area_m2, floor, total_floors, house_type). 2. Enrich address with region prefix (prevents Avito geocoder ambiguity). 3. Call avito_imv.evaluate_via_imv(). 4. Persist house_imv_evaluations + house_placement_history + house_suggestions. 5. Mark houses.imv_status accordingly (ok/not_found/no_params/error/transient_error). Resumable: re-running picks up from where the previous run stopped. Deliberately NOT wired into scheduler — triggered manually via admin API. """ from __future__ import annotations import asyncio import json import logging import time from collections.abc import Callable from dataclasses import dataclass, field from typing import Literal from sqlalchemy import text from sqlalchemy.orm import Session from app.core.config import settings from app.services.scrapers.avito_imv import ( IMVAddressNotFoundError, IMVAuthError, IMVEvaluation, IMVTransientError, evaluate_via_imv, ) from app.services.scrapers.browser_fetcher import BrowserFetcher logger = logging.getLogger(__name__) # Каждые N домов дёргаем heartbeat-колбэк (если передан) внутри длинного per-house # цикла. Без этого долгая IMV-фаза (десятки/сотни домов × request_delay_sec) не # обновляет scrape_runs.heartbeat_at дольше ZOMBIE_THRESHOLD_HOURS, и scheduler # ложно помечает живой run 'zombie' (после чего терминальный mark_done — no-op). См. #1363. _HEARTBEAT_EVERY_N_HOUSES = 5 # ── house_type normalisation ───────────────────────────────────────────────── _HOUSE_TYPE_TO_IMV: dict[str, str] = { "panel": "panel", "brick": "brick", "monolith": "monolithic", "monolithic": "monolithic", "monolith_brick": "monolithic", # Avito API не принимает гибриды "block": "block", "wood": "wood", } _HOUSE_TYPE_DEFAULT = "panel" # самый распространённый в ЕКБ def _map_house_type(raw: str | None) -> str: if not raw: return _HOUSE_TYPE_DEFAULT return _HOUSE_TYPE_TO_IMV.get(raw.lower().strip(), _HOUSE_TYPE_DEFAULT) # ── Region bbox prefix для Avito geocoder ──────────────────────────────────── _REGION_BBOX: list[tuple[str, float, float, float, float]] = [ # (region_name, lat_min, lat_max, lon_min, lon_max) ("Свердловская область, Екатеринбург", 56.5, 57.1, 59.9, 61.0), ("Свердловская область", 56.0, 61.0, 58.0, 65.0), ("Кировская область, Киров", 58.4, 58.8, 49.4, 49.9), ("Ставропольский край, Ставрополь", 44.9, 45.2, 41.7, 42.2), ("Тюменская область, Тюмень", 57.0, 57.3, 65.3, 65.8), ("Челябинская область, Челябинск", 55.0, 55.4, 61.2, 61.7), ] _REGION_DEFAULT = "Свердловская область, Екатеринбург" def _detect_region_prefix(lat: float | None, lon: float | None) -> str: if lat is None or lon is None: return _REGION_DEFAULT for name, lat_min, lat_max, lon_min, lon_max in _REGION_BBOX: if lat_min <= lat <= lat_max and lon_min <= lon <= lon_max: return name return _REGION_DEFAULT def _enrich_address_for_imv(raw: str, lat: float | None, lon: float | None) -> str: """Prepend region prefix if address lacks a region marker.""" addr = (raw or "").strip() addr_lower = addr.lower() if any( m in addr_lower for m in ("область,", " обл.,", " обл,", "край,", "респ.,", "республика,") ): return addr prefix = _detect_region_prefix(lat, lon) return f"{prefix}, {addr}" # ── DB helpers ──────────────────────────────────────────────────────────────── def pick_lot_params(db: Session, house_id: int) -> dict: """Pick representative lot-params — median values from linked listings.""" row = ( db.execute( text(""" SELECT CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY rooms) AS integer) AS rooms, CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY area_m2) AS numeric(8,2)) AS area_m2, CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY floor) AS integer) AS floor, CAST(percentile_cont(0.5) WITHIN GROUP (ORDER BY total_floors) AS integer) AS total_floors, mode() WITHIN GROUP (ORDER BY house_type) AS house_type FROM listings WHERE house_id_fk = :hid AND rooms IS NOT NULL AND area_m2 IS NOT NULL """), {"hid": house_id}, ) .mappings() .first() ) if row is None or row["rooms"] is None: return {} house = ( db.execute( text("SELECT house_type, total_floors FROM houses WHERE id = :hid"), {"hid": house_id}, ) .mappings() .first() ) rooms_raw = int(row["rooms"]) floor = int(row["floor"] or 1) floor_at_home = int(row["total_floors"] or (house and house["total_floors"]) or 9) # floor и floor_at_home считаются независимо (медиана vs default-9). Если у # всех listings и у дома total_floors=NULL, floor_at_home молча падает на 9, # а медианный floor может быть выше → Avito IMV получает кв. на этаже выше дома. # Консервативный clamp: floor не может превышать этажность дома. floor = min(floor, floor_at_home) return { "rooms": max(rooms_raw, 1), # Avito IMV отклоняет rooms=0 (студия) "area_m2": float(row["area_m2"]), "floor": floor, "floor_at_home": floor_at_home, "house_type": _map_house_type(row["house_type"] or (house and house["house_type"])), "renovation_type": "cosmetic", "has_balcony": True, "has_loggia": False, } def save_imv_result(db: Session, house_id: int, params: dict, result: IMVEvaluation) -> None: """Persist evaluation to house_imv_evaluations, house_placement_history, house_suggestions.""" # 1. House-level evaluation (UPSERT on house_id) db.execute( text(""" INSERT INTO house_imv_evaluations ( house_id, cache_key, recommended_price, lower_price, higher_price, market_count, raw_response, fetched_at, rooms, area_m2, floor, floor_at_home, house_type, renovation_type, has_balcony, has_loggia ) VALUES ( :hid, :ck, :rec, :lo, :hi, :mc, CAST(:raw AS jsonb), NOW(), :rooms, CAST(:area AS numeric), :floor, :fah, :htype, :rtype, :balcony, :loggia ) ON CONFLICT (house_id) DO UPDATE SET cache_key = EXCLUDED.cache_key, 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(), rooms = EXCLUDED.rooms, area_m2 = EXCLUDED.area_m2, floor = EXCLUDED.floor, floor_at_home = EXCLUDED.floor_at_home, house_type = EXCLUDED.house_type, renovation_type = EXCLUDED.renovation_type, has_balcony = EXCLUDED.has_balcony, has_loggia = EXCLUDED.has_loggia """), { "hid": house_id, "ck": result.cache_key, "rec": result.recommended_price, "lo": result.lower_price, "hi": result.higher_price, "mc": result.market_count, "raw": ( json.dumps(result.raw_response, ensure_ascii=False) if result.raw_response else None ), "rooms": params["rooms"], "area": params["area_m2"], "floor": params["floor"], "fah": params["floor_at_home"], "htype": params["house_type"], "rtype": params["renovation_type"], "balcony": params["has_balcony"], "loggia": params["has_loggia"], }, ) # 2. Placement history items for item in result.placement_history: db.execute( text(""" INSERT INTO house_placement_history ( 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, raw_payload, scraped_at ) VALUES ( 'avito_imv', :hid, :ext, :title, :rooms, CAST(:area AS numeric), :floor, :total_floors, :sp, :spd, :lp, :lpd, :rmd, :exp, CAST(:raw AS jsonb), NOW() ) ON CONFLICT (source, ext_item_id) DO UPDATE SET house_id = EXCLUDED.house_id, title = EXCLUDED.title, rooms = EXCLUDED.rooms, area_m2 = EXCLUDED.area_m2, floor = EXCLUDED.floor, total_floors = EXCLUDED.total_floors, start_price = EXCLUDED.start_price, start_price_date = EXCLUDED.start_price_date, 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() """), { "hid": house_id, "ext": item.ext_item_id, "title": item.title, "rooms": item.rooms, "area": item.area_m2, "floor": item.floor, "total_floors": item.total_floors, "sp": item.start_price, "spd": item.start_price_date, "lp": item.last_price, "lpd": item.last_price_date, "rmd": item.removed_date, "exp": item.exposure_days, "raw": ( json.dumps(item.raw_payload, ensure_ascii=False) if item.raw_payload else None ), }, ) # 3. Suggestions for sug in result.suggestions: db.execute( text(""" INSERT INTO house_suggestions ( house_id, ext_item_id, title, address, price_rub, exposure_days, publish_date, item_link, metro_name, metro_distance, has_good_price_badge, raw_payload, fetched_at ) VALUES ( :hid, :ext, :title, :addr, :price, :exp, :pdate, :link, :mname, :mdist, :gpb, CAST(:raw AS jsonb), NOW() ) ON CONFLICT (house_id, ext_item_id) DO UPDATE SET title = EXCLUDED.title, price_rub = EXCLUDED.price_rub, exposure_days = EXCLUDED.exposure_days, publish_date = EXCLUDED.publish_date, item_link = EXCLUDED.item_link, metro_name = EXCLUDED.metro_name, metro_distance = EXCLUDED.metro_distance, has_good_price_badge = EXCLUDED.has_good_price_badge, raw_payload = EXCLUDED.raw_payload, fetched_at = NOW() """), { "hid": house_id, "ext": sug.ext_item_id, "title": sug.title, "addr": sug.address, "price": sug.price_rub, "exp": sug.exposure_days, "pdate": sug.publish_date, "link": sug.item_url, "mname": sug.metro_name, "mdist": sug.metro_distance, "gpb": sug.has_good_price_badge, "raw": ( json.dumps(sug.raw_payload, ensure_ascii=False) if sug.raw_payload else None ), }, ) # 4. Mark house success db.execute( text(""" UPDATE houses SET imv_status = 'ok', last_imv_attempt_at = NOW(), imv_error_reason = NULL WHERE id = :hid """), {"hid": house_id}, ) def _mark_status( db: Session, house_id: int, status: str, reason: str | None = None, ) -> None: db.execute( text(""" UPDATE houses SET imv_status = :s, last_imv_attempt_at = NOW(), imv_error_reason = :r WHERE id = :hid """), {"hid": house_id, "s": status, "r": reason}, ) db.commit() # ── Top-level orchestrator ──────────────────────────────────────────────────── _IMVStatus = Literal[ "ok", "no_params", "no_address", "not_found", "auth_error", "transient", "error" ] @dataclass class HouseIMVBackfillResult: checked: int = 0 saved: int = 0 skipped: int = 0 errors: int = 0 duration_sec: float = field(default=0.0) status_counts: dict[str, int] = field(default_factory=dict) def _beat(heartbeat: Callable[[], None] | None) -> None: """Best-effort вызов heartbeat-колбэка. Сбой heartbeat не должен ронять backfill.""" if heartbeat is None: return try: heartbeat() except Exception: logger.warning("house_imv_backfill: heartbeat callback failed (ignored)", exc_info=True) async def backfill_house_imv( db: Session, *, batch_size: int = 50, request_delay_sec: float = 5.0, only_status: str = "pending", house_id: int | None = None, heartbeat: Callable[[], None] | None = None, ) -> HouseIMVBackfillResult: """Run Avito IMV evaluation for each house in scope, save results. Args: db: SQLAlchemy session. batch_size: max houses to process (ignored when house_id given). request_delay_sec: sleep between Avito API calls (default 5s — anti-bot). only_status: process houses with this imv_status (default 'pending'). Use 'transient_error' to retry failures. house_id: process a single specific house (debug). heartbeat: optional callback дёргается каждые _HEARTBEAT_EVERY_N_HOUSES домов — caller обновляет scrape_runs.heartbeat_at, чтобы reap_zombies не пометил живой долгий run 'zombie' (#1363). Best-effort: исключения из колбэка логируются и не прерывают backfill. Returns: HouseIMVBackfillResult with aggregate counters. """ result = HouseIMVBackfillResult() t0 = time.time() if house_id is not None: rows = ( db.execute( text(""" SELECT id, address, full_address, lat, lon FROM houses WHERE id = CAST(:hid AS bigint) AND lat IS NOT NULL AND lon IS NOT NULL """), {"hid": house_id}, ) .mappings() .all() ) else: rows = ( db.execute( text(""" SELECT id, address, full_address, lat, lon FROM houses WHERE imv_status = :status AND lat IS NOT NULL AND lon IS NOT NULL AND address IS NOT NULL ORDER BY last_imv_attempt_at NULLS FIRST, id LIMIT :batch """), {"status": only_status, "batch": batch_size}, ) .mappings() .all() ) result.checked = len(rows) if not rows: logger.info("house_imv_backfill: nothing to process (status=%r)", only_status) result.duration_sec = time.time() - t0 return result logger.info( "house_imv_backfill: %d houses (status=%r delay=%.1fs)", result.checked, only_status, request_delay_sec, ) async def _run_loop(_bf: BrowserFetcher | None) -> None: for i, house in enumerate(rows): hid: int = house["id"] status_str = await _process_one_house(db, dict(house), browser_fetcher=_bf) result.status_counts[status_str] = result.status_counts.get(status_str, 0) + 1 if status_str == "ok": result.saved += 1 elif status_str in ("no_params", "no_address"): result.skipped += 1 else: result.errors += 1 logger.info( "[%d/%d] house_id=%d status=%s", i + 1, result.checked, hid, status_str, ) # Периодический heartbeat внутри длинного цикла — иначе reap_zombies # может пометить живой run 'zombie' за время IMV-фазы (#1363). if (i + 1) % _HEARTBEAT_EVERY_N_HOUSES == 0: _beat(heartbeat) if i < len(rows) - 1: if status_str == "auth_error": logger.warning("IMV auth error — extra delay 60s before next house") await asyncio.sleep(60) else: await asyncio.sleep(request_delay_sec) # #915 Stage 3: при флаге открываем ОДИН browser на весь батч (warmed-сессия # + прокси переиспользуются всеми домами; обходит datacenter-403, #562/#853). # Флаг OFF → _bf=None → evaluate_via_imv делает свою curl-сессию как раньше # (поведение байт-в-байт идентично доспринтовому). if settings.avito_imv_use_browser_fetcher: async with BrowserFetcher(source="avito") as _bf: await _run_loop(_bf) else: await _run_loop(None) result.duration_sec = time.time() - t0 logger.info( "house_imv_backfill done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s", result.checked, result.saved, result.skipped, result.errors, result.duration_sec, result.status_counts, ) return result async def process_houses_imv_batch( db: Session, house_ids: set[int], *, request_delay_sec: float = 5.0, heartbeat: Callable[[], None] | None = None, ) -> HouseIMVBackfillResult: """Обработать конкретный набор house_id (для sweep-интеграции). Зеркало backfill_house_imv, но работает по явному списку house_id (не по imv_status-фильтру). Предназначен для финальной IMV-фазы run_avito_city_sweep — обрабатываем только дома, тронутые за этот sweep. Дома без lat/lon или address пропускаются (статус no_params/no_address). Дома уже с imv_status='ok' тоже пропускаются — UPSERT всё равно обновит, поэтому лучше не гонять лишних API-запросов, если данные свежие. Для форс-обновления — оставь это решение caller'у (sweep по умолчанию обходит дома с ok, обновляя только pending/not_found/error/transient_error). heartbeat: optional callback дёргается каждые _HEARTBEAT_EVERY_N_HOUSES домов, чтобы scrape_runs.heartbeat_at обновлялся в течение долгой IMV-фазы и reap_zombies не пометил живой run 'zombie' (#1363). Best-effort. """ result = HouseIMVBackfillResult() t0 = time.time() if not house_ids: result.duration_sec = time.time() - t0 return result ids_list = list(house_ids) # Загружаем только дома, у которых есть координаты + адрес + imv_status != 'ok' # (уже оцененные дома пропускаем — не тратим API-квоту повторно) placeholders = ", ".join(f":hid_{i}" for i in range(len(ids_list))) params: dict[str, object] = {f"hid_{i}": hid for i, hid in enumerate(ids_list)} rows = ( db.execute( text(f""" SELECT id, address, full_address, lat, lon FROM houses WHERE id IN ({placeholders}) AND lat IS NOT NULL AND lon IS NOT NULL AND address IS NOT NULL AND COALESCE(imv_status, 'pending') != 'ok' ORDER BY id """), params, ) .mappings() .all() ) result.checked = len(rows) if not rows: logger.info( "process_houses_imv_batch: nothing to process (%d house_ids, all ok/no-coords)", len(house_ids), ) result.duration_sec = time.time() - t0 return result logger.info( "process_houses_imv_batch: %d/%d houses eligible (delay=%.1fs)", result.checked, len(house_ids), request_delay_sec, ) for i, house in enumerate(rows): hid: int = house["id"] status_str = await _process_one_house(db, dict(house)) result.status_counts[status_str] = result.status_counts.get(status_str, 0) + 1 if status_str == "ok": result.saved += 1 elif status_str in ("no_params", "no_address"): result.skipped += 1 else: result.errors += 1 logger.info( "[%d/%d] imv_batch house_id=%d status=%s", i + 1, result.checked, hid, status_str, ) # Периодический heartbeat внутри длинного цикла — иначе reap_zombies # может пометить живой run 'zombie' за время IMV-фазы (#1363). if (i + 1) % _HEARTBEAT_EVERY_N_HOUSES == 0: _beat(heartbeat) if i < len(rows) - 1: if status_str == "auth_error": logger.warning("IMV auth error — extra delay 60s before next house") await asyncio.sleep(60) else: await asyncio.sleep(request_delay_sec) result.duration_sec = time.time() - t0 logger.info( "process_houses_imv_batch done: checked=%d saved=%d skipped=%d errors=%d %.1fs %s", result.checked, result.saved, result.skipped, result.errors, result.duration_sec, result.status_counts, ) return result async def _process_one_house( db: Session, house: dict, *, browser_fetcher: BrowserFetcher | None = None ) -> str: """Process a single house. Returns final status string. browser_fetcher: при #915 Stage 3 backfill пробрасывает один общий BrowserFetcher на весь батч → evaluate_via_imv ходит через /fetch-json sidecar (обходит datacenter-403). None → curl_cffi-путь (как раньше). """ hid: int = house["id"] params = pick_lot_params(db, hid) if not params: _mark_status(db, hid, "no_params", "no listings with rooms+area") return "no_params" address = house.get("address") or house.get("full_address") if not address: _mark_status(db, hid, "no_address", "house.address is NULL") return "no_address" enriched = _enrich_address_for_imv(address, house.get("lat"), house.get("lon")) if enriched != address: logger.debug("house_imv: enriched address house=%d %r -> %r", hid, address, enriched) try: eval_result = await evaluate_via_imv( address=enriched, browser_fetcher=browser_fetcher, **params ) except IMVAddressNotFoundError as exc: _mark_status(db, hid, "not_found", str(exc)[:200]) return "not_found" except IMVAuthError as exc: _mark_status(db, hid, "transient_error", f"auth: {exc!s}"[:200]) return "auth_error" except IMVTransientError as exc: _mark_status(db, hid, "transient_error", str(exc)[:200]) return "transient" except Exception as exc: _mark_status(db, hid, "error", repr(exc)[:200]) logger.error("house_imv: unexpected error house=%d: %r", hid, exc) return "error" # Адрес найден, но рыночной цены нет: avito_imv возвращает recommended_price=0 # ("нет оценки"). Не считаем это успехом — иначе дом залипает в imv_status='ok' # и больше не переоценивается (backfill_house_imv фильтрует pending, # process_houses_imv_batch — != 'ok', estimator — recommended_price>0). # Помечаем not_found (адрес есть, но usable-оценки нет) — дом остаётся # доступным для повторной обработки sweep'ом (imv_status != 'ok'). if eval_result.recommended_price <= 0: logger.info( "house_imv: empty IMV (no market price) house=%d market_count=%r", hid, eval_result.market_count, ) _mark_status( db, hid, "not_found", f"imv_empty: recommended_price=0 market_count={eval_result.market_count}"[:200], ) return "not_found" try: save_imv_result(db, hid, params, eval_result) db.commit() except Exception as exc: logger.error("house_imv: save failed house=%d: %r", hid, exc) try: db.rollback() except Exception: pass _mark_status(db, hid, "error", f"save_failed: {exc!s}"[:200]) return "error" return "ok"