diff --git a/data/sql/70_parse_objective_raw.py b/data/sql/70_parse_objective_raw.py index 6dffb6c1..93798332 100644 --- a/data/sql/70_parse_objective_raw.py +++ b/data/sql/70_parse_objective_raw.py @@ -40,7 +40,8 @@ import json import logging import re import sys -from datetime import date, datetime +from collections.abc import Iterator +from datetime import date from pathlib import Path from typing import Any, BinaryIO @@ -608,6 +609,140 @@ def detect_kind(payload: Any) -> str | None: return None +# ── streaming support ─────────────────────────────────────────────────────── + + +class _IterBytesReader: + """File-like wrapper над iterator of bytes chunks (для ijson.items()). + + ijson 3.x ожидает file-like с .read(n) — но httpx.Response.iter_bytes() + возвращает generator байт. Этот класс собирает chunks в буфер и отдаёт + их по требованию через .read(n). + """ + + def __init__(self, chunks: Iterator[bytes]) -> None: + self._chunks = iter(chunks) + self._buf = bytearray() + self._eof = False + + def read(self, n: int = -1) -> bytes: + if n < 0: + # Прочитать всё что осталось + for c in self._chunks: + self._buf.extend(c) + data = bytes(self._buf) + self._buf.clear() + self._eof = True + return data + while len(self._buf) < n and not self._eof: + try: + self._buf.extend(next(self._chunks)) + except StopIteration: + self._eof = True + break + out = bytes(self._buf[:n]) + del self._buf[:n] + return out + + +def parse_lots_pf_stream( + stream: BinaryIO | Iterator[bytes], + raw_id: int | None, + snapshot_date: date, + db: Any, + dry_run: bool, + batch_size: int = 500, +) -> tuple[int, int]: + """Streaming-парсер lots_pf через ijson — для больших ответов (>100 МБ). + + В отличие от parse_lots_pf (читает весь payload в dict), эта функция + обрабатывает JSON инкрементально через ijson.items(), что позволяет + парсить файлы >100 МБ без загрузки всего payload в память. + + Args: + stream: либо file-like с .read(n), либо iterator of bytes chunks + (например httpx.Response.iter_bytes(65536)). Если stream не имеет + метода .read — автоматически оборачивается в _IterBytesReader. + raw_id: ссылка на objective_raw_reports.raw_id (может быть None) + snapshot_date: дата снимка для objective_lots_history + db: SQLAlchemy Session + dry_run: если True — только подсчёт, без записи в БД + batch_size: коммит каждые N лотов (снижает нагрузку на память транзакции) + + Returns: + tuple[n_lots, n_history] + """ + import ijson # type: ignore[import-untyped] + + # Если stream — iterator chunks (без .read), оборачиваем в file-like reader. + if not hasattr(stream, "read"): + stream = _IterBytesReader(stream) # type: ignore[arg-type] + + n_lots = 0 + n_history = 0 + batch: list[dict] = [] + batches_since_commit = 0 + + def _flush() -> None: + """Bulk-upsert текущего batch двумя executemany-вызовами.""" + nonlocal batch + if not batch: + return + history_batch = [ + { + "objective_lot_id": p["objective_lot_id"], + "snapshot_date": snapshot_date, + "status": p["status"], + "is_sold": p["is_sold"], + "price_calculated_total_rub": p["price_calculated_total_rub"], + "price_per_m2_rub": p["price_per_m2_rub"], + "price_offer_total_rub": p["price_offer_total_rub"], + "price_delta_pct": p["price_delta_pct"], + "area_pd": p["area_pd"], + "contract_date": p["contract_date"], + "registration_date": p["registration_date"], + "contract_type": p["contract_type"], + "bank_name": p["bank_name"], + "raw_id": raw_id, + } + for p in batch + ] + if not dry_run: + db.execute(_LOTS_UPSERT, batch) + db.execute(_LOTS_HISTORY_INSERT, history_batch) + batch = [] + + for row in ijson.items(stream, "result.item"): + if not isinstance(row, dict): + continue + params = _row_to_lot(row, raw_id, snapshot_date) + if not params["objective_lot_id"]: + logger.warning("lots_pf_stream: пропуск row без Id") + continue + batch.append(params) + n_lots += 1 + n_history += 1 + if len(batch) >= batch_size: + _flush() + batches_since_commit += 1 + if not dry_run and batches_since_commit >= 5: + db.commit() + batches_since_commit = 0 + logger.info("lots_pf_stream: committed 5 batches, n_lots=%d so far", n_lots) + + _flush() # финальный неполный batch + if not dry_run and batches_since_commit > 0: + db.commit() + + logger.info( + "lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s", + n_lots, + n_history, + dry_run, + ) + return n_lots, n_history + + # ── CLI ─────────────────────────────────────────────────────────────────────