fix(objective parser): restore bulk executemany in stream _flush

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 #318 (comment)
This commit is contained in:
lekss361 2026-05-17 20:48:27 +03:00
parent 8d0e67d03b
commit c8ac1fa732

View file

@ -41,7 +41,7 @@ import logging
import re import re
import sys import sys
from collections.abc import Iterator from collections.abc import Iterator
from datetime import date, datetime from datetime import date
from pathlib import Path from pathlib import Path
from typing import Any, BinaryIO from typing import Any, BinaryIO
@ -597,26 +597,36 @@ def parse_lots_pf_stream(
n_lots = 0 n_lots = 0
n_history = 0 n_history = 0
batch: list[dict] = [] batch: list[dict] = []
batches_since_commit = 0
def _flush(batch: list[dict]) -> None: def _flush() -> None:
for params in batch: """Bulk-upsert текущего batch двумя executemany-вызовами."""
db.execute(_LOTS_UPSERT, params) nonlocal batch
db.execute(_LOTS_HISTORY_INSERT, { if not batch:
"objective_lot_id": params["objective_lot_id"], return
history_batch = [
{
"objective_lot_id": p["objective_lot_id"],
"snapshot_date": snapshot_date, "snapshot_date": snapshot_date,
"status": params["status"], "status": p["status"],
"is_sold": params["is_sold"], "is_sold": p["is_sold"],
"price_calculated_total_rub": params["price_calculated_total_rub"], "price_calculated_total_rub": p["price_calculated_total_rub"],
"price_per_m2_rub": params["price_per_m2_rub"], "price_per_m2_rub": p["price_per_m2_rub"],
"price_offer_total_rub": params["price_offer_total_rub"], "price_offer_total_rub": p["price_offer_total_rub"],
"price_delta_pct": params["price_delta_pct"], "price_delta_pct": p["price_delta_pct"],
"area_pd": params["area_pd"], "area_pd": p["area_pd"],
"contract_date": params["contract_date"], "contract_date": p["contract_date"],
"registration_date": params["registration_date"], "registration_date": p["registration_date"],
"contract_type": params["contract_type"], "contract_type": p["contract_type"],
"bank_name": params["bank_name"], "bank_name": p["bank_name"],
"raw_id": raw_id, "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"): for row in ijson.items(stream, "result.item"):
if not isinstance(row, dict): if not isinstance(row, dict):
@ -625,20 +635,20 @@ def parse_lots_pf_stream(
if not params["objective_lot_id"]: if not params["objective_lot_id"]:
logger.warning("lots_pf_stream: пропуск row без Id") logger.warning("lots_pf_stream: пропуск row без Id")
continue continue
batch.append(params)
n_lots += 1 n_lots += 1
n_history += 1 n_history += 1
if not dry_run: if len(batch) >= batch_size:
batch.append(params) _flush()
if len(batch) >= batch_size: batches_since_commit += 1
_flush(batch) if not dry_run and batches_since_commit >= 5:
db.commit() db.commit()
batch.clear() batches_since_commit = 0
logger.info("lots_pf_stream: committed batch, n_lots=%d so far", n_lots) 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() db.commit()
batch.clear()
logger.info( logger.info(
"lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s", "lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s",