feat(tradein/domclick): JSONL ingest script for local detail-backfill runner

scripts/ingest_domclick_jsonl.py — idempotent ingest of domclick_local_runner
output: filter ok+price, upsert via save_listings (fault-tolerant house-match
hook in SAVEPOINT — non-matched listings still persist), deep-detail enrichment
via save_detail_enrichment (owners/encumbrances/kitchen/views/wall/floor/
price-history). --limit / --dry-run. psycopg v3 ANY(:ids) list-bind.
Per-record try/except in enrichment loop so one bad row never aborts the tail.

Live-tested in tradein-backend on 30 records: updated=30, house_matched=30,
detail_applied=30, price_history idempotent (75->75), enrich_failed=0.

tradein-mvp/backend/scripts/ingest_domclick_jsonl.py

@@ -0,0 +1,291 @@
+"""Ингест JSONL от локального DomClick-раннера (`scripts/domclick_local_runner.py`) в БД tradein.
+
+Запускается ВНУТРИ tradein-контейнера (импортит `app`). Идемпотентно.
+
+Каждая ok-запись (status=="ok" и валидный price_rub) →
+  1. upsert в `listings` через save_listings() (house-match хук — fault-tolerant, в SAVEPOINT);
+  2. deep-detail enrichment через save_detail_enrichment()
+     (owners / encumbrances / kitchen / views / wall / floor / price-history).
+
+Listing.id не возвращается из save_listings — достаём отдельным SELECT по (source, source_id).
+
+Usage:
+    python -m scripts.ingest_domclick_jsonl --jsonl /tmp/domclick.jsonl [--limit N] [--dry-run]
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+from typing import Any
+
+from sqlalchemy import text
+
+from app.core.db import SessionLocal
+from app.services.scrapers.base import ScrapedLot, save_listings
+from app.services.scrapers.domclick_detail import (
+    DomClickDetailEnrichment,
+    save_detail_enrichment,
+)
+
+logger = logging.getLogger(__name__)
+
+SOURCE = "domklik"
+
+
+def _to_int(v: Any) -> int | None:
+    """int() None-safe: возвращает None на None / нечисловом мусоре (не крэшит прогон)."""
+    if v is None:
+        return None
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _to_float(v: Any) -> float | None:
+    """float() None-safe: возвращает None на None / нечисловом мусоре."""
+    if v is None:
+        return None
+    try:
+        return float(v)
+    except (TypeError, ValueError):
+        return None
+
+
+def _build_lot(rec: dict[str, Any]) -> ScrapedLot:
+    """Построить ScrapedLot из JSONL-записи. price_rub уже провалидирован caller'ом."""
+    raw_extra = rec.get("raw_extra") or {}
+    return ScrapedLot(
+        source=SOURCE,
+        source_url=rec["source_url"],
+        source_id=str(rec["id"]),
+        address=rec.get("address"),
+        lat=_to_float(rec.get("lat")),
+        lon=_to_float(rec.get("lon")),
+        rooms=_to_int(rec.get("rooms")),
+        area_m2=_to_float(rec.get("area_m2")),
+        floor=_to_int(rec.get("floor")),
+        total_floors=_to_int(rec.get("total_floors")),
+        year_built=_to_int(rec.get("year_built")),
+        price_rub=int(rec["price_rub"]),
+        listing_segment="vtorichka",
+        repair_state=rec.get("repair_state"),
+        raw_payload={
+            "is_sber_collateral": rec.get("is_sber_collateral"),
+            "square_price": raw_extra.get("square_price"),
+            "avm": rec.get("avm") or {},
+        },
+    )
+
+
+def _build_enrichment(rec: dict[str, Any]) -> DomClickDetailEnrichment:
+    """Построить deep-detail enrichment из JSONL-записи."""
+    raw_extra_in = rec.get("raw_extra") or {}
+    raw_extra: dict[str, Any] = {
+        "wall_type": rec.get("wall_type"),
+        "floor_type": rec.get("floor_type"),
+        "calls": rec.get("calls"),
+        "is_sber_collateral": rec.get("is_sber_collateral"),
+        "avm": rec.get("avm") or {},
+        **raw_extra_in,
+    }
+    return DomClickDetailEnrichment(
+        item_id=str(rec["id"]),
+        source_url=rec["source_url"],
+        repair_state=rec.get("repair_state"),
+        repair_type=rec.get("repair_raw"),
+        living_area_m2=_to_float(rec.get("living_area_m2")),
+        kitchen_area_m2=_to_float(rec.get("kitchen_area_m2")),
+        sale_type=rec.get("sale_type"),
+        owners_count=_to_int(rec.get("owners_count")),
+        encumbrances_clean=rec.get("encumbrances_clean"),
+        year_built=_to_int(rec.get("year_built")),
+        views_total=_to_int(rec.get("views")),
+        price_changes=rec.get("price_changes") or [],
+        raw_extra=raw_extra,
+    )
+
+
+def run(jsonl_path: str, limit: int | None, dry_run: bool) -> dict[str, int]:
+    """Прочитать JSONL, отфильтровать ok+price, upsert listings + enrichment.
+
+    Returns: dict со счётчиками для итогового лога / тестов.
+    """
+    lots: list[ScrapedLot] = []
+    valid_records: list[dict[str, Any]] = []  # (parallel to lots) для enrichment-шага
+
+    skipped_status = 0
+    skipped_no_price = 0
+    parse_failed = 0
+
+    with open(jsonl_path, encoding="utf-8") as fh:
+        for line in fh:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                rec = json.loads(line)
+            except json.JSONDecodeError as exc:
+                logger.warning("битая JSON-строка пропущена: %s", exc)
+                parse_failed += 1
+                continue
+
+            if rec.get("status") != "ok":
+                skipped_status += 1
+                continue
+            price = _to_int(rec.get("price_rub"))
+            if price is None or price <= 0:
+                skipped_no_price += 1
+                continue
+
+            try:
+                lot = _build_lot(rec)
+            except Exception as exc:
+                logger.warning("не удалось построить lot для id=%s: %s", rec.get("id"), exc)
+                parse_failed += 1
+                continue
+
+            lots.append(lot)
+            valid_records.append(rec)
+
+            if limit is not None and len(lots) >= limit:
+                break
+
+    total_valid = len(lots)
+    logger.info(
+        "прочитано: valid=%d skipped_status=%d skipped_no_price=%d parse_failed=%d",
+        total_valid,
+        skipped_status,
+        skipped_no_price,
+        parse_failed,
+    )
+
+    inserted = 0
+    updated = 0
+    house_matched = 0
+    match_failed = 0
+    detail_applied = 0
+    price_history_attempted = 0
+    enrich_failed = 0
+
+    if dry_run:
+        logger.info("[dry-run] %d записей распарсено, запись в БД пропущена", total_valid)
+        return {
+            "inserted": inserted,
+            "updated": updated,
+            "house_matched": house_matched,
+            "match_failed": match_failed,
+            "detail_applied": detail_applied,
+            "price_history_attempted": price_history_attempted,
+            "enrich_failed": enrich_failed,
+            "skipped_status": skipped_status,
+            "skipped_no_price": skipped_no_price,
+            "parse_failed": parse_failed,
+        }
+
+    db = SessionLocal()
+    try:
+        if lots:
+            inserted, updated = save_listings(db, lots)
+            logger.info("save_listings: inserted=%d updated=%d", inserted, updated)
+
+        # Достаём listing.id + house_id_fk одним SELECT по source_id.
+        source_ids = [lot.source_id for lot in lots if lot.source_id is not None]
+        id_map: dict[str, tuple[int, Any]] = {}
+        if source_ids:
+            rows = db.execute(
+                text(
+                    "SELECT source_id, id, house_id_fk FROM listings "
+                    "WHERE source = :source AND source_id = ANY(:ids)"
+                ),
+                {"source": SOURCE, "ids": source_ids},
+            ).fetchall()
+            for sid, lid, house_id_fk in rows:
+                id_map[str(sid)] = (int(lid), house_id_fk)
+
+        total_saved = len(id_map)
+        house_matched = sum(1 for _, hid in id_map.values() if hid is not None)
+        match_failed = total_saved - house_matched
+
+        # Deep-detail enrichment по каждой ok-записи.
+        for rec in valid_records:
+            sid = str(rec["id"])
+            entry = id_map.get(sid)
+            if entry is None:
+                logger.warning("listing id не найден для source_id=%s — enrichment пропущен", sid)
+                continue
+            listing_id, _ = entry
+            # Per-record изоляция: save_detail_enrichment коммитит сама, поэтому
+            # падение одной записи иначе оборвало бы остаток valid_records трейсбэком.
+            # Уже закоммиченные строки персистят → повторный прогон добьёт хвост.
+            try:
+                enrichment = _build_enrichment(rec)
+                price_history_attempted += len(enrichment.price_changes)
+                if save_detail_enrichment(db, listing_id, enrichment):
+                    detail_applied += 1
+            except Exception as exc:
+                db.rollback()
+                enrich_failed += 1
+                logger.warning("enrichment упал для source_id=%s: %s", sid, exc)
+                continue
+    finally:
+        db.close()
+
+    logger.info(
+        "ИТОГО: saved(inserted=%d/updated=%d) house_matched=%d match_failed=%d "
+        "detail_applied=%d price_history_attempted=%d enrich_failed=%d "
+        "skipped_status=%d skipped_no_price=%d parse_failed=%d",
+        inserted,
+        updated,
+        house_matched,
+        match_failed,
+        detail_applied,
+        price_history_attempted,
+        enrich_failed,
+        skipped_status,
+        skipped_no_price,
+        parse_failed,
+    )
+
+    return {
+        "inserted": inserted,
+        "updated": updated,
+        "house_matched": house_matched,
+        "match_failed": match_failed,
+        "detail_applied": detail_applied,
+        "price_history_attempted": price_history_attempted,
+        "enrich_failed": enrich_failed,
+        "skipped_status": skipped_status,
+        "skipped_no_price": skipped_no_price,
+        "parse_failed": parse_failed,
+    }
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Ингест DomClick JSONL (от domclick_local_runner) в БД tradein.",
+    )
+    parser.add_argument("--jsonl", required=True, help="путь к JSONL-файлу раннера")
+    parser.add_argument(
+        "--limit",
+        type=int,
+        default=None,
+        help="обработать первые N валидных (ok+price) записей",
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="парсить + логировать, но НЕ писать в БД",
+    )
+    args = parser.parse_args()
+    run(args.jsonl, args.limit, args.dry_run)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    main()