"""Phase C: backfill Avito IMV per house. Iterates `houses` with imv_status='pending' AND lat/lon NOT NULL. For each: pick median lot-params from linked listings -> call evaluate_via_imv() -> save house_imv_evaluations + house_placement_history + house_suggestions. Resumable: marks houses.imv_status = 'ok' / 'not_found' / 'transient_error' / 'error'. Usage: # Locally with SSH tunnel (LocalForward 35432 -> tradein-postgres): DATABASE_URL=postgresql+psycopg://tradein:@localhost:35432/tradein \\ python scripts/backfill-house-imv.py --batch 50 --delay 5 # Inside container: docker exec tradein-backend python /app/scripts/backfill-house-imv.py --batch 100 --delay 3 Args: --batch N How many houses to process this run (default 50) --delay SEC Sleep between houses (default 5 to avoid Avito captcha) --limit-total Stop after N successful evaluations (debug; default no limit) --only-status Process only houses with this imv_status (default 'pending'; also accepts 'transient_error') """ from __future__ import annotations import argparse import asyncio import json import logging import os import sys from pathlib import Path # Ensure repo backend in path sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "backend")) from sqlalchemy import create_engine, text from sqlalchemy.orm import Session from app.services.scrapers.avito_imv import ( IMVAddressNotFoundError, IMVAuthError, IMVEvaluation, IMVTransientError, evaluate_via_imv, ) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", ) logger = logging.getLogger("phase-c") DATABASE_URL = os.environ.get("DATABASE_URL") if not DATABASE_URL: raise SystemExit( "Set DATABASE_URL env" " (e.g. postgresql+psycopg://tradein:...@localhost:35432/tradein)" ) def pick_lot_params(db: Session, house_id: int) -> dict: """Pick representative lot-params from linked listings — median values.""" 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 {} # no listings with required fields, can't pick params # Fall back to house's own type/floors if listings don't have them house = db.execute( text("SELECT house_type, total_floors FROM houses WHERE id = :hid"), {"hid": house_id}, ).mappings().first() return { "rooms": int(row["rooms"]), "area_m2": float(row["area_m2"]), "floor": int(row["floor"] or 1), "floor_at_home": int( row["total_floors"] or (house and house["total_floors"]) or 9 ), "house_type": ( row["house_type"] or (house and house["house_type"]) or "panel" ), "renovation_type": "cosmetic", "has_balcony": True, "has_loggia": False, } def save_imv_result(db: Session, house_id: int, params: dict, result: IMVEvaluation) -> None: """Save 3 things: house_imv_evaluations + history items + suggestions. Uses already-parsed dataclass fields from IMVEvaluation (placement_history, suggestions) instead of re-parsing raw_response to avoid duplication and ensure consistency with _parse_placement_item / _parse_suggestion logic. """ # 1. House-level IMV 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, :area, :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 (already parsed into IMVPlacementHistoryItem dataclasses) 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, :area, :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 (already parsed into IMVSuggestion dataclasses) 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}, ) async def process_one(db: Session, house: dict) -> str: """Process a single house: geocode + IMV call + save. Returns final status string: 'ok' / 'no_params' / 'no_address' / 'not_found' / 'auth_error' / 'transient' / 'error' """ params = pick_lot_params(db, house["id"]) if not params: mark_status(db, house["id"], "no_params", "no listings with rooms+area") return "no_params" address = house["address"] or house["full_address"] if not address: mark_status(db, house["id"], "no_address", "house.address is NULL") return "no_address" try: result = await evaluate_via_imv(address=address, **params) except IMVAddressNotFoundError as e: mark_status(db, house["id"], "not_found", str(e)[:200]) return "not_found" except IMVAuthError as e: mark_status(db, house["id"], "transient_error", f"auth: {e!s}"[:200]) logger.warning("AUTH ERROR house=%d — pausing 60s", house["id"]) await asyncio.sleep(60) # long back-off on auth block return "auth_error" except IMVTransientError as e: mark_status(db, house["id"], "transient_error", str(e)[:200]) return "transient" except Exception as e: mark_status(db, house["id"], "error", repr(e)[:200]) logger.error("Unexpected error for house=%d: %r", house["id"], e) return "error" save_imv_result(db, house["id"], params, result) return "ok" async def main() -> None: parser = argparse.ArgumentParser(description="Phase C: backfill Avito IMV per house") parser.add_argument("--batch", type=int, default=50, help="Houses to process this run") parser.add_argument("--delay", type=float, default=5.0, help="Sleep between houses (s)") parser.add_argument( "--limit-total", type=int, default=0, help="Stop after N successful evaluations (0 = no limit)" ) parser.add_argument( "--only-status", default="pending", help="Process only houses with this imv_status (default: pending)" ) args = parser.parse_args() engine = create_engine(DATABASE_URL, future=True) with Session(engine) as db: 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": args.only_status, "batch": args.batch}, ).mappings().all() if not rows: logger.info("nothing to process — all done or no houses match filter") return logger.info( "processing %d houses (status=%r delay=%.1fs)", len(rows), args.only_status, args.delay ) counters: dict[str, int] = { "ok": 0, "not_found": 0, "no_params": 0, "no_address": 0, "transient": 0, "auth_error": 0, "error": 0, } for i, house in enumerate(rows, 1): status = await process_one(db, dict(house)) counters[status] = counters.get(status, 0) + 1 db.commit() logger.info( "[%d/%d] house=%d status=%s addr=%r", i, len(rows), house["id"], status, (house["address"] or "")[:50], ) if args.limit_total and counters["ok"] >= args.limit_total: logger.info("hit --limit-total=%d; stopping", args.limit_total) break if i < len(rows): await asyncio.sleep(args.delay) logger.info("done: %s", counters) if __name__ == "__main__": asyncio.run(main())