bot-backend
8ffc2cb942
All checks were successful
CI Trade-In / changes (pull_request) Successful in 8s
CI / changes (pull_request) Successful in 8s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 3m6s
Из девяти вызовов save_listings в pipeline.py три не передавали run_id.
У двух — run_avito_city_sweep:1113 и run_avito_newbuilding_sweep:1754 —
run_id лежал в той же функции (keyword-параметр, под которым идёт
runs.mark_done/heartbeat/is_cancelled) и просто не доезжал до вызова.
Передан.
run_avito_pipeline:570 — оставлен без run_id осознанно, решение записано
в код: у функции нет строки в scrape_runs, продуктовых вызывающих нет
(единственные — тесты), привязывать снимок не к чему.
Писатели вне pipeline.py проверены и тоже помечены: admin /scrape (ручной
скрейп без прогона) и scripts/ingest_domclick_jsonl.py (разовый ingest
файла). Второй и есть главная дыра замера — 129 908 снимков domklik без
run_id, а вовсе не Авито: по источникам заполнение yandex 100%, cian 98.8%,
avito 80.3%, domklik 2.9%. Дыра domklik историческая (ежедневный
run_domclick_city_sweep run_id передаёт), а вот у avito пустые строки
приходят КАЖДЫЙ день (944 за 2026-08-06, 635 за 2026-08-05 — все без run_id).
Бэкфилл 150 515 исторических строк невозможен, проверено:
- привязки снимок→прогон в схеме больше нет;
- по времени неоднозначно: на всех 36 днях с осиротевшими avito-снимками
в сутках было >1 пишущего avito-прогона (до 29);
- observed_at не спасает — ON CONFLICT перезаписывает его последним
писателем дня, а не тем, кто строку создал.
Ложная атрибуция хуже пустоты — бэкфилл не делается.
Refs #2701
299 lines
11 KiB
Python
299 lines
11 KiB
Python
"""Ингест 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 scraper_kit.base import ScrapedLot, save_listings
|
||
from scraper_kit.orchestration.pipeline import DEFAULT_REGION_CODE
|
||
from scraper_kit.providers.domclick.detail import (
|
||
DomClickDetailEnrichment,
|
||
save_detail_enrichment,
|
||
)
|
||
from sqlalchemy import text
|
||
|
||
from app.core.db import SessionLocal
|
||
from app.services.scraper_adapters import RealMatcherAdapter
|
||
|
||
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:
|
||
# run_id нет и не будет (#2701): разовый ingest файла — не прогон скрапера,
|
||
# строки в scrape_runs под него не существует. Именно отсюда 129 908 снимков
|
||
# domklik без run_id (2.9% заполнения у источника) — исторические, не текущие:
|
||
# ежедневный run_domclick_city_sweep run_id передаёт.
|
||
inserted, updated = save_listings(
|
||
db, lots, matcher=RealMatcherAdapter(), region_code=DEFAULT_REGION_CODE
|
||
)
|
||
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()
|