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 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",