gendesign/tradein-mvp/backend/scripts/ingest_domclick_jsonl.py

"""Ингест 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()