diff --git a/tradein-mvp/backend/scripts/ingest_domclick_jsonl.py b/tradein-mvp/backend/scripts/ingest_domclick_jsonl.py new file mode 100644 index 00000000..a963818d --- /dev/null +++ b/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()