fix(objective parser): wrap bytes-iter in file-like reader for ijson

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.
This commit is contained in:
lekss361 2026-05-17 19:58:59 +03:00
parent fce48a8924
commit 8d0e67d03b

View file

@ -40,9 +40,10 @@ import json
import logging import logging
import re import re
import sys import sys
from collections.abc import Iterator
from datetime import date, datetime from datetime import date, datetime
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any, BinaryIO
# repo-root → /backend # repo-root → /backend
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent / "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 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 ───────────────────────────────────────────────────────────────────── # ── CLI ─────────────────────────────────────────────────────────────────────