From 8d0e67d03bd0d98bad44fdbfff95cabf70669d73 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 17 May 2026 19:58:59 +0300 Subject: [PATCH 1/2] 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 ───────────────────────────────────────────────────────────────────── -- 2.45.3 From c8ac1fa7321fba64dd991a43d751ef44239f719a Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 17 May 2026 20:48:27 +0300 Subject: [PATCH 2/2] fix(objective parser): restore bulk executemany in stream _flush MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer flagged perf regression vs PR #315 — _flush loop'ил per-row через db.execute(stmt, dict) вместо bulk db.execute(stmt, list). Для 324k лотов это 648k round-trips × 2 sync/неделю. Восстановлен SQLAlchemy executemany через list[dict] + commit cadence 5 batch (2500 строк) как в #315. Также убран неиспользуемый импорт datetime (ruff F401). Per https://git.gendsgn.ru/lekss361/gendesign/pulls/318#issuecomment-1589 --- data/sql/70_parse_objective_raw.py | 64 +++++++++++++++++------------- 1 file changed, 37 insertions(+), 27 deletions(-) diff --git a/data/sql/70_parse_objective_raw.py b/data/sql/70_parse_objective_raw.py index f7d214f6..4488af75 100644 --- a/data/sql/70_parse_objective_raw.py +++ b/data/sql/70_parse_objective_raw.py @@ -41,7 +41,7 @@ import logging import re import sys from collections.abc import Iterator -from datetime import date, datetime +from datetime import date from pathlib import Path from typing import Any, BinaryIO @@ -597,26 +597,36 @@ def parse_lots_pf_stream( n_lots = 0 n_history = 0 batch: list[dict] = [] + batches_since_commit = 0 - 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"], + 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": 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"], + "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): @@ -625,20 +635,20 @@ def parse_lots_pf_stream( if not params["objective_lot_id"]: logger.warning("lots_pf_stream: пропуск row без Id") continue + batch.append(params) n_lots += 1 n_history += 1 - if not dry_run: - batch.append(params) - if len(batch) >= batch_size: - _flush(batch) + if len(batch) >= batch_size: + _flush() + batches_since_commit += 1 + if not dry_run and batches_since_commit >= 5: db.commit() - batch.clear() - logger.info("lots_pf_stream: committed batch, n_lots=%d so far", n_lots) + batches_since_commit = 0 + logger.info("lots_pf_stream: committed 5 batches, n_lots=%d so far", n_lots) - if not dry_run and batch: - _flush(batch) + _flush() # финальный неполный batch + if not dry_run and batches_since_commit > 0: db.commit() - batch.clear() logger.info( "lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s", -- 2.45.3