data/sql/70_parse_objective_raw.py

@@ -40,9 +40,10 @@ 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
+from typing import Any, BinaryIO
 
 # repo-root → /backend
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "backend"))
@@ -524,6 +525,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 ─────────────────────────────────────────────────────────────────────