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;

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:<pwd>@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())