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:
+    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,
-                "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,
-            })
+            }
+            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:
+        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()
-                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()
-        batch.clear()
 
     logger.info(
         "lots_pf_stream: done — n_lots=%d n_history=%d dry_run=%s",