diff --git a/tradein-mvp/backend/data/sql/064_house_imv_phase_c.sql b/tradein-mvp/backend/data/sql/064_house_imv_phase_c.sql new file mode 100644 index 00000000..29b7bad1 --- /dev/null +++ b/tradein-mvp/backend/data/sql/064_house_imv_phase_c.sql @@ -0,0 +1,79 @@ +-- 064: Phase C — Avito IMV per-house backfill schema. +-- Idempotent: re-run safe (IF NOT EXISTS). +BEGIN; + +-- Track IMV scrape state per house (resumable backfill) +ALTER TABLE houses + ADD COLUMN IF NOT EXISTS imv_status text NOT NULL DEFAULT 'pending', + ADD COLUMN IF NOT EXISTS last_imv_attempt_at timestamptz, + ADD COLUMN IF NOT EXISTS imv_error_reason text; + +CREATE INDEX IF NOT EXISTS houses_imv_status_idx + ON houses(imv_status, last_imv_attempt_at NULLS FIRST) + WHERE imv_status IN ('pending', 'transient_error'); + +-- House-level IMV evaluation result (1:1 with houses, latest fetch) +CREATE TABLE IF NOT EXISTS house_imv_evaluations ( + id bigserial PRIMARY KEY, + house_id bigint NOT NULL REFERENCES houses(id) ON DELETE CASCADE, + cache_key text, + recommended_price bigint, + lower_price bigint, + higher_price bigint, + market_count integer, + raw_response jsonb, + fetched_at timestamptz NOT NULL DEFAULT NOW(), + -- Lot-params used (median from listings or default) + rooms integer, + area_m2 numeric(8,2), + floor integer, + floor_at_home integer, + house_type text, + renovation_type text, + has_balcony boolean, + has_loggia boolean +); + +CREATE UNIQUE INDEX IF NOT EXISTS house_imv_eval_house_uniq_idx + ON house_imv_evaluations(house_id); +CREATE INDEX IF NOT EXISTS house_imv_eval_fetched_idx + ON house_imv_evaluations(fetched_at DESC); + +-- Avito suggestions[] per house — 8 similar lots nearby +CREATE TABLE IF NOT EXISTS house_suggestions ( + id bigserial PRIMARY KEY, + house_id bigint NOT NULL REFERENCES houses(id) ON DELETE CASCADE, + ext_item_id text NOT NULL, -- Avito item id + title text, + address text, + price_rub bigint, + area_m2 numeric(8,2), + rooms integer, + floor integer, + total_floors integer, + exposure_days integer, + publish_date date, + item_link text, + image_link text, + metro_name text, + metro_distance text, + has_good_price_badge boolean, + raw_payload jsonb, + fetched_at timestamptz NOT NULL DEFAULT NOW() +); + +CREATE UNIQUE INDEX IF NOT EXISTS house_suggestions_uniq_idx + ON house_suggestions(house_id, ext_item_id); +CREATE INDEX IF NOT EXISTS house_suggestions_house_idx + ON house_suggestions(house_id, fetched_at DESC); + +-- Verify +DO $$ +DECLARE + pending_n bigint; +BEGIN + SELECT count(*) INTO pending_n FROM houses WHERE imv_status = 'pending'; + RAISE NOTICE 'migration 064: % houses pending IMV backfill', pending_n; +END $$; + +COMMIT; diff --git a/tradein-mvp/scripts/backfill-house-imv.py b/tradein-mvp/scripts/backfill-house-imv.py new file mode 100644 index 00000000..3b58658f --- /dev/null +++ b/tradein-mvp/scripts/backfill-house-imv.py @@ -0,0 +1,391 @@ +"""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())