From 8d0e67d03bd0d98bad44fdbfff95cabf70669d73 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 17 May 2026 19:58:59 +0300 Subject: [PATCH] fix(objective parser): wrap bytes-iter in file-like reader for ijson MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ijson.items() ожидает file-like с .read(n), а resp.iter_bytes(65536) возвращает generator bytes → ijson интерпретирует chunks как (prefix, event, value) tuples → ValueError "too many values to unpack (expected 3, got 65536)". Добавлен _IterBytesReader адаптер: собирает chunks в bytearray buffer и отдаёт через .read(n). Добавлена parse_lots_pf_stream — streaming- вариант парсера для больших ответов (>100 МБ) через ijson.items(), с автоматической обёрткой stream если у него нет .read. Glitchtip BACKEND-1K (release 389e141, 3 events) — silent fail в inner try/except → rows_extracted=NULL, lots=0 после "успешных" runs. --- data/sql/70_parse_objective_raw.py | 127 ++++++++++++++++++++++++++++- 1 file changed, 126 insertions(+), 1 deletion(-) diff --git a/data/sql/70_parse_objective_raw.py b/data/sql/70_parse_objective_raw.py index 6cb04a8f..f7d214f6 100644 --- a/data/sql/70_parse_objective_raw.py +++ b/data/sql/70_parse_objective_raw.py @@ -40,9 +40,10 @@ import json import logging import re import sys +from collections.abc import Iterator from datetime import date, datetime 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,130 @@ 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] = [] + + def _flush(batch: list[dict]) -> None: + for params in batch: + db.execute(_LOTS_UPSERT, params) + db.execute(_LOTS_HISTORY_INSERT, { + "objective_lot_id": params["objective_lot_id"], + "snapshot_date": snapshot_date, + "status": params["status"], + "is_sold": params["is_sold"], + "price_calculated_total_rub": params["price_calculated_total_rub"], + "price_per_m2_rub": params["price_per_m2_rub"], + "price_offer_total_rub": params["price_offer_total_rub"], + "price_delta_pct": params["price_delta_pct"], + "area_pd": params["area_pd"], + "contract_date": params["contract_date"], + "registration_date": params["registration_date"], + "contract_type": params["contract_type"], + "bank_name": params["bank_name"], + "raw_id": raw_id, + }) + + 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 + n_lots += 1 + n_history += 1 + if not dry_run: + batch.append(params) + if len(batch) >= batch_size: + _flush(batch) + db.commit() + batch.clear() + logger.info("lots_pf_stream: committed batch, n_lots=%d so far", n_lots) + + if not dry_run and batch: + _flush(batch) + db.commit() + batch.clear() + + 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 ─────────────────────────────────────────────────────────────────────