""" Delivery History Store ====================== Pattern-first vector rider lookup using recent delivery history. Works on any platform — no faiss-cpu dependency. Data source (config `pattern_source`) -------------------------------------- "csv" (legacy) — built from delivery_details.csv, only refreshed when a human re-exports it and calls POST /ml/reload-history. "db" (default going forward) — built from the nearledb mirror the ETA-sync agent already maintains (delivery_raw), rebuilt automatically every sync cycle via rebuild_from_records(). No manual step, no extra database load. How it works ------------ Records (however sourced) are turned into two structures: 1. PATTERN TABLE (primary, O(1) lookup) City divided into ~1.1 km grid cells (round coords to 2 d.p.). For every (kitchen, zone_cell) pair we count how many times each rider delivered there. A "clear pattern" = one rider owns ≥ 60 % of all deliveries in that cell AND ≥ 3 total deliveries. From the 3 500-row CSV: • 177 unique (kitchen, zone) cells • 121 clear dominant-rider patterns (68 %) • 42 cells where 1 rider owns 100 % 2. VECTOR INDEX (fallback, ~0.5 ms) Pure-NumPy brute-force L2 search over 4D vectors: [pickuplat, pickuplon, deliverylat, deliverylong] Used ONLY when the pattern table has no clear answer. If faiss-cpu is installed it is used instead of NumPy (faster for very large indexes), but NumPy is the default. Disk persistence ---------------- ml_data/faiss_history/ delivery_history_vectors.npy – float32 (N, 4) array delivery_history_records.pkl – list of record dicts delivery_history.meta – JSON: csv_mtime, counts Priority chain in the assignment endpoint: 1. Pattern table (dominant rider for this kitchen + delivery zone) 2. Vector K-NN (nearest-neighbour vote, fallback for new areas) 3. rider_preferences.py hard lock 4. VRP proximity solver """ import csv import json import logging import os import pickle from collections import Counter, defaultdict from threading import Lock from typing import Any, Dict, List, Optional, Tuple import numpy as np logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Paths # --------------------------------------------------------------------------- CSV_PATH = os.getenv("DELIVERY_HISTORY_CSV", "delivery_details.csv") CORRECTIONS_PATH = os.getenv("DELIVERY_CORRECTIONS_CSV", "delivery_corrections.csv") # Each correction record is already written N times in the corrections CSV # (currently 5× per unique delivery) so they naturally outweigh noisy history. # Setting CORRECTION_WEIGHT > 1 here adds a further runtime multiplier on top. _CORRECTION_WEIGHT = int(os.getenv("CORRECTION_WEIGHT", "1")) _STORE_DIR = os.getenv("FAISS_HISTORY_DIR", "ml_data/faiss_history") def _paths_for(source: str) -> Tuple[str, str, str]: """ Disk cache paths, namespaced by source ("csv" or "db"). Namespaced so that flipping `pattern_source` at runtime (e.g. to debug, or to roll back) never clobbers the other mode's cached snapshot — each mode keeps its own independent copy on disk. """ suffix = "" if source == "csv" else f".{source}" return ( os.path.join(_STORE_DIR, f"delivery_history_vectors{suffix}.npy"), os.path.join(_STORE_DIR, f"delivery_history_records{suffix}.pkl"), os.path.join(_STORE_DIR, f"delivery_history{suffix}.meta"), ) # --------------------------------------------------------------------------- # Thresholds # --------------------------------------------------------------------------- _K_NEIGHBORS = 15 _MIN_KNN_CONFIDENCE = 0.50 # vector fallback: top rider wins ≥ 50 % of votes _MIN_PATTERN_DOMINANCE = 0.60 # pattern table: rider owns ≥ 60 % of zone _MIN_PATTERN_VOLUME = 3 # pattern table: at least 3 deliveries in zone def _pattern_source() -> str: """'csv' (legacy, manual refresh) or 'db' (auto-refreshed from nearledb mirror).""" try: from app.config.dynamic_config import get_config return str(get_config().get("pattern_source", "csv")) except Exception: return "csv" # --------------------------------------------------------------------------- # Pure-NumPy L2 index — drop-in replacement for faiss.IndexFlatL2 # --------------------------------------------------------------------------- class _NumpyL2Index: """ Brute-force L2 nearest-neighbour search using NumPy. Same interface as faiss.IndexFlatL2 so both paths share one code. For 3 000–5 000 vectors each query takes < 1 ms — fast enough. """ def __init__(self, d: int): self.d = d self.ntotal = 0 self._vecs: Optional[np.ndarray] = None # shape (N, d) float32 def add(self, vectors: np.ndarray) -> None: self._vecs = vectors.astype(np.float32) self.ntotal = len(self._vecs) def search(self, query: np.ndarray, k: int) -> Tuple[np.ndarray, np.ndarray]: """Return (distances, indices) shaped (1, k), same as faiss.""" if self._vecs is None or self.ntotal == 0: return (np.array([[]], dtype=np.float32), np.array([[-1]], dtype=np.int64)) q = query.astype(np.float32) # (1, d) diff = self._vecs - q # (N, d) broadcasting sq_dists = np.einsum("ij,ij->i", diff, diff) # (N,) squared L2 actual_k = min(k, self.ntotal) if actual_k < self.ntotal: part = np.argpartition(sq_dists, actual_k)[:actual_k] top_idxs = part[np.argsort(sq_dists[part])] else: top_idxs = np.argsort(sq_dists) return (sq_dists[top_idxs].reshape(1, -1), top_idxs.reshape(1, -1).astype(np.int64)) def _build_index(vectors: np.ndarray) -> Any: """ Build the best available index for the given float32 vector array. Tries faiss first; falls back to NumPy silently. """ try: import faiss as _faiss idx = _faiss.IndexFlatL2(vectors.shape[1]) idx.add(vectors) logger.info("[DeliveryHistory] Using faiss-cpu index.") return idx except Exception: idx = _NumpyL2Index(vectors.shape[1]) idx.add(vectors) logger.info("[DeliveryHistory] Using NumPy L2 index (faiss-cpu not available).") return idx # --------------------------------------------------------------------------- # Main store # --------------------------------------------------------------------------- class DeliveryHistoryStore: """ Pattern-first, vector-fallback rider lookup. Public API ---------- store.find_rider(kitchen, plat, plon, dlat, dlon) -> dict | None store.record_count() -> int store.pattern_count() -> int store.get_pattern_stats() -> list store.reload_from_csv() -> int store.rebuild_from_records(records) -> int """ def __init__(self): self._lock = Lock() self._records: List[Dict] = [] self._index = None # Pattern table: (kitchen_lower, zone_lat, zone_lon) → pattern dict self._patterns: Dict[Tuple, Dict] = {} # Zone index: (zone_lat, zone_lon) → [pattern keys at that cell] self._zone_index: Dict[Tuple, List[Tuple]] = defaultdict(list) self._load() # ------------------------------------------------------------------ # Startup # ------------------------------------------------------------------ def _load(self) -> None: os.makedirs(_STORE_DIR, exist_ok=True) if _pattern_source() == "db": # DB-driven: the ETA-sync agent keeps this fresh on its own cadence # by calling rebuild_from_records() after every sync cycle. At # startup we just load whatever was last persisted — no CSV # freshness check applies in this mode. if self._load_from_disk("db"): return logger.info( "[DeliveryHistory] pattern_source=db, no persisted snapshot yet " "— will populate on the next ETA-sync cycle." ) return if self._saved_files_are_current(): if self._load_from_disk("csv"): return logger.warning( "[DeliveryHistory] Saved files corrupt — rebuilding from CSV." ) records = self._parse_csv() if not records: return self._build_and_save(records, source="csv") def _saved_files_are_current(self) -> bool: vectors_path, records_path, meta_path = _paths_for("csv") for path in (vectors_path, records_path, meta_path): if not os.path.isfile(path): return False try: with open(meta_path, "r", encoding="utf-8") as f: meta = json.load(f) if not os.path.isfile(CSV_PATH): return True return abs(os.path.getmtime(CSV_PATH) - meta.get("csv_mtime", 0)) < 1.0 except Exception: return False def _load_from_disk(self, source: str) -> bool: try: vectors_path, records_path, _ = _paths_for(source) vectors = np.load(vectors_path) # (N, 4) float32 with open(records_path, "rb") as f: records = pickle.load(f) if not records or len(vectors) != len(records): return False index = _build_index(vectors) patterns, zone_index = self._compute_patterns(records) with self._lock: self._records = records self._index = index self._patterns = patterns self._zone_index = zone_index clear = sum( 1 for p in patterns.values() if p["dominance"] >= _MIN_PATTERN_DOMINANCE and p["total_deliveries"] >= _MIN_PATTERN_VOLUME ) logger.info( f"[DeliveryHistory] Loaded {len(records)} records from disk. " f"Pattern table: {len(patterns)} zones, {clear} with clear dominant rider." ) return True except Exception as e: logger.warning(f"[DeliveryHistory] Disk load failed: {e}") return False def _parse_csv_file(self, path: str, label: str = "CSV") -> List[Dict]: """Parse any delivery CSV (main history or corrections) into records list.""" records: List[Dict] = [] for encoding in ("utf-8", "latin-1", "cp1252"): try: with open(path, newline="", encoding=encoding) as fh: for row in csv.DictReader(fh): try: plat = float(row.get("pickuplat") or 0) plon = float(row.get("pickuplon") or 0) dlat = float(row.get("deliverylat") or 0) dlon = float(row.get("deliverylong") or 0) uid = int(float(row.get("userid") or 0)) if not plat or not dlat or uid == 0: continue # Skip rows where the delivery destination IS a kitchen dcust = (row.get("deliverycustomer") or "").strip().lower() if "kitchen" in dcust or "selvarani" in dcust: continue records.append({ "kitchen": (row.get("pickupcustomer") or "").strip().lower(), "pickuplat": plat, "pickuplon": plon, "deliverylat": dlat, "deliverylong": dlon, "userid": uid, "ridername": (row.get("ridername") or "").strip(), }) except Exception: continue break except (UnicodeDecodeError, FileNotFoundError) as e: if isinstance(e, FileNotFoundError): logger.warning(f"[DeliveryHistory] {label} not found at '{path}'.") return [] records = [] if records: logger.info(f"[DeliveryHistory] Parsed {len(records)} rows from {label} '{path}'.") else: logger.warning(f"[DeliveryHistory] {label} at '{path}' contained no valid rows.") return records def _parse_csv(self) -> List[Dict]: """Parse main history CSV, then merge manually-corrected records on top.""" main_records = self._parse_csv_file(CSV_PATH, label="Main CSV") if not main_records: return [] # Load manually-corrected assignments and merge with runtime weight multiplier if os.path.isfile(CORRECTIONS_PATH): corr = self._parse_csv_file(CORRECTIONS_PATH, label="Corrections CSV") if corr: weighted = corr * _CORRECTION_WEIGHT # extra amplification if configured main_records = main_records + weighted logger.info( f"[DeliveryHistory] Merged {len(corr)} correction rows " f"(×{_CORRECTION_WEIGHT} weight) into {len(main_records)} total records." ) else: logger.debug( f"[DeliveryHistory] No corrections file at '{CORRECTIONS_PATH}' — " "using main CSV only." ) return main_records def _compute_patterns( self, records: List[Dict] ) -> Tuple[Dict, Dict]: """ Build pattern table from records. Pattern dict keys ----------------- userid, ridername, dominance, total_deliveries, top_deliveries, pattern_score, all_riders """ zone_counts: Dict[Tuple, Dict[int, int]] = defaultdict(lambda: defaultdict(int)) zone_names: Dict[Tuple, Dict[int, str]] = defaultdict(lambda: defaultdict(str)) for rec in records: key = (rec["kitchen"], round(rec["deliverylat"], 2), round(rec["deliverylong"], 2)) zone_counts[key][rec["userid"]] += 1 zone_names[key][rec["userid"]] = rec["ridername"] patterns: Dict[Tuple, Dict] = {} zone_index: Dict[Tuple, List[Tuple]] = defaultdict(list) for key, rider_counts in zone_counts.items(): kitchen, zone_lat, zone_lon = key total = sum(rider_counts.values()) top_rid = max(rider_counts, key=rider_counts.get) top_count = rider_counts[top_rid] dominance = top_count / total # Penalise thin data: full score only at ≥ 5 deliveries pattern_score = dominance * min(1.0, total / 5.0) patterns[key] = { "userid": top_rid, "ridername": zone_names[key].get(top_rid, ""), "dominance": round(dominance, 4), "total_deliveries": total, "top_deliveries": top_count, "pattern_score": round(pattern_score, 4), "all_riders": dict(rider_counts), } zone_index[(zone_lat, zone_lon)].append(key) return patterns, zone_index def _build_and_save(self, records: List[Dict], source: str = "csv") -> None: vectors = np.array( [[r["pickuplat"], r["pickuplon"], r["deliverylat"], r["deliverylong"]] for r in records], dtype=np.float32, ) index = _build_index(vectors) patterns, zone_index = self._compute_patterns(records) with self._lock: self._records = records self._index = index self._patterns = patterns self._zone_index = zone_index clear = sum( 1 for p in patterns.values() if p["dominance"] >= _MIN_PATTERN_DOMINANCE and p["total_deliveries"] >= _MIN_PATTERN_VOLUME ) sole = sum( 1 for p in patterns.values() if p["dominance"] == 1.0 and p["total_deliveries"] >= _MIN_PATTERN_VOLUME ) logger.info( f"[DeliveryHistory] Pattern table: {len(patterns)} zones, " f"{clear} clear patterns (≥60% dominance), {sole} sole-owner zones." ) try: vectors_path, records_path, meta_path = _paths_for(source) np.save(vectors_path, vectors) with open(records_path, "wb") as f: pickle.dump(records, f, protocol=pickle.HIGHEST_PROTOCOL) # csv_mtime only means something for the CSV path's freshness check # (_saved_files_are_current). DB-built snapshots are refreshed by the # ETA-sync agent's own schedule, not a file-mtime comparison. csv_mtime = ( os.path.getmtime(CSV_PATH) if source == "csv" and os.path.isfile(CSV_PATH) else 0 ) with open(meta_path, "w", encoding="utf-8") as f: json.dump({ "source": source, "csv_mtime": csv_mtime, "record_count": len(records), "pattern_zones": len(patterns), "clear_patterns": clear, }, f, indent=2) logger.info( f"[DeliveryHistory] Saved to '{_STORE_DIR}' (source={source}). " "Next startup loads from disk." ) except Exception as e: logger.warning( f"[DeliveryHistory] Could not save to disk (non-fatal): {e}" ) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def find_rider( self, kitchen_name: str, pickup_lat: float, pickup_lon: float, delivery_lat: float, delivery_lon: float, k: int = _K_NEIGHBORS, min_confidence: float = _MIN_KNN_CONFIDENCE, ) -> Optional[Dict[str, Any]]: """ Return the best historical rider, or None. Step 1 — pattern table (O(1)): Clear dominant rider for this kitchen + 1.1 km delivery zone. Step 2 — vector K-NN (fallback): Brute-force 4D search, kitchen-filtered vote. Return keys: userid, ridername, confidence, match_count, top_votes, source """ if not delivery_lat: return None kitchen_lower = (kitchen_name or "").strip().lower() result = self._pattern_lookup(kitchen_lower, delivery_lat, delivery_lon) if result: return result if self._index is None or not self._records: return None if not pickup_lat: return None return self._vector_knn( kitchen_lower, pickup_lat, pickup_lon, delivery_lat, delivery_lon, k, min_confidence ) def _pattern_lookup( self, kitchen_lower: str, delivery_lat: float, delivery_lon: float, ) -> Optional[Dict[str, Any]]: zone_pos = (round(delivery_lat, 2), round(delivery_lon, 2)) with self._lock: candidate_keys = self._zone_index.get(zone_pos, []) best_pattern = None best_score = -1.0 for key in candidate_keys: key_kitchen = key[0] if kitchen_lower and key_kitchen: if (kitchen_lower not in key_kitchen and key_kitchen not in kitchen_lower): continue with self._lock: pat = self._patterns.get(key) if not pat: continue if (pat["dominance"] >= _MIN_PATTERN_DOMINANCE and pat["total_deliveries"] >= _MIN_PATTERN_VOLUME and pat["pattern_score"] > best_score): best_pattern = pat best_score = pat["pattern_score"] if not best_pattern: return None return { "userid": best_pattern["userid"], "ridername": best_pattern["ridername"], "confidence": best_pattern["dominance"], "match_count": best_pattern["total_deliveries"], "top_votes": best_pattern["top_deliveries"], "source": "pattern", } def _vector_knn( self, kitchen_lower: str, pickup_lat: float, pickup_lon: float, delivery_lat: float, delivery_lon: float, k: int, min_confidence: float, ) -> Optional[Dict[str, Any]]: with self._lock: query = np.array( [[pickup_lat, pickup_lon, delivery_lat, delivery_lon]], dtype=np.float32, ) actual_k = min(k, len(self._records)) _, indices = self._index.search(query, actual_k) matched: List[Dict] = [] for idx in indices[0]: idx = int(idx) if idx < 0 or idx >= len(self._records): continue rec = self._records[idx] if kitchen_lower and rec["kitchen"]: if (kitchen_lower not in rec["kitchen"] and rec["kitchen"] not in kitchen_lower): continue matched.append(rec) if not matched: return None votes = Counter(rec["userid"] for rec in matched) top_uid, top_count = votes.most_common(1)[0] confidence = top_count / len(matched) if confidence < min_confidence: return None ridername = next( (r["ridername"] for r in matched if r["userid"] == top_uid), "" ) return { "userid": top_uid, "ridername": ridername, "confidence": round(confidence, 3), "match_count": len(matched), "top_votes": top_count, "source": "vector_knn", } # ------------------------------------------------------------------ # Analytics # ------------------------------------------------------------------ def record_count(self) -> int: return len(self._records) def pattern_count(self) -> int: return sum( 1 for p in self._patterns.values() if p["dominance"] >= _MIN_PATTERN_DOMINANCE and p["total_deliveries"] >= _MIN_PATTERN_VOLUME ) def get_pattern_stats(self) -> List[Dict]: stats = [] with self._lock: for (kitchen, zlat, zlon), pat in self._patterns.items(): if (pat["dominance"] < _MIN_PATTERN_DOMINANCE or pat["total_deliveries"] < _MIN_PATTERN_VOLUME): continue stats.append({ "kitchen": kitchen, "zone_lat": zlat, "zone_lon": zlon, "userid": pat["userid"], "ridername": pat["ridername"], "dominance": pat["dominance"], "total_deliveries": pat["total_deliveries"], "top_deliveries": pat["top_deliveries"], "pattern_score": pat["pattern_score"], "all_riders": pat["all_riders"], }) return sorted(stats, key=lambda x: x["pattern_score"], reverse=True) # ------------------------------------------------------------------ # Rider Efficiency Scores # ------------------------------------------------------------------ def get_rider_efficiency_scores(self) -> Dict[int, Dict]: """ Compute a per-rider efficiency score from the 30-day CSV history. Metrics per rider ----------------- delivery_count — total deliveries in CSV avg_km — average Haversine distance (pickup → delivery) unique_zones — number of distinct 1.1 km delivery cells covered efficiency_score — normalised 0..1 composite score: more deliveries × lower avg km × more zones = higher score A rider who completes many deliveries efficiently across many zones scores highest. Used as a tiebreaker during solo consolidation and Phase-0 host selection. """ from math import radians, cos, sin, asin, sqrt as _sqrt def _hav(la1, lo1, la2, lo2): try: la1, lo1, la2, lo2 = map(radians, [la1, lo1, la2, lo2]) a = sin((la2-la1)/2)**2 + cos(la1)*cos(la2)*sin((lo2-lo1)/2)**2 return 2 * asin(min(1.0, _sqrt(a))) * 6371.0 except Exception: return 0.0 with self._lock: records = list(self._records) # snapshot under lock if not records: return {} stats: Dict[int, Dict] = {} for rec in records: uid = rec["userid"] km = _hav(rec["pickuplat"], rec["pickuplon"], rec["deliverylat"], rec["deliverylong"]) zone = (round(rec["deliverylat"], 2), round(rec["deliverylong"], 2)) if uid not in stats: stats[uid] = { "ridername": rec["ridername"], "delivery_count": 0, "total_km": 0.0, "zones": set(), } s = stats[uid] s["delivery_count"] += 1 s["total_km"] += km s["zones"].add(zone) # Build scores result: Dict[int, Dict] = {} raw_scores: Dict[int, float] = {} for uid, s in stats.items(): n = s["delivery_count"] avg_km = s["total_km"] / n if n else 0.0 zone_count = len(s["zones"]) # Raw: many deliveries, low km per delivery, wide zone coverage raw = n / (1.0 + avg_km) * (1.0 + zone_count ** 0.5) raw_scores[uid] = raw result[uid] = { "ridername": s["ridername"], "delivery_count": n, "avg_km": round(avg_km, 2), "unique_zones": zone_count, } # Normalise 0..1 max_raw = max(raw_scores.values()) if raw_scores else 1.0 for uid in result: result[uid]["efficiency_score"] = round( raw_scores[uid] / max_raw, 4 ) return result def get_rider_score(self, rider_id: int) -> float: """Convenience: return a single rider's efficiency_score (0..1), or 0.5 if unknown.""" scores = self.get_rider_efficiency_scores() return scores.get(rider_id, {}).get("efficiency_score", 0.5) # ------------------------------------------------------------------ # Reload # ------------------------------------------------------------------ def reload_from_csv(self) -> int: """Rebuild index from main CSV + corrections (if present).""" logger.info("[DeliveryHistory] Force-reload from CSV requested.") records = self._parse_csv() # already merges corrections internally if not records: return 0 self._build_and_save(records, source="csv") return len(records) def rebuild_from_records(self, records: List[Dict]) -> int: """ Rebuild the pattern table + vector index directly from pre-shaped records (kitchen/pickuplat/pickuplon/deliverylat/deliverylong/userid/ ridername), bypassing CSV parsing entirely. Called by the ETA-sync agent (delivery_history_service.py) after each sync cycle when pattern_source=db, so this store stays as fresh as the nearledb mirror — no manual CSV re-export/reload needed. """ if not records: logger.warning( "[DeliveryHistory] rebuild_from_records got 0 records — " "keeping the existing store as-is." ) return 0 # Merge manually-verified corrections on top, same as the CSV path, # so the human-override mechanism still works in DB mode. if os.path.isfile(CORRECTIONS_PATH): corr = self._parse_csv_file(CORRECTIONS_PATH, label="Corrections CSV") if corr: records = records + (corr * _CORRECTION_WEIGHT) self._build_and_save(records, source="db") return len(records) def inject_corrections(self, corrections_path: str = CORRECTIONS_PATH, weight: int = 5) -> Dict: """ Hot-inject a correction CSV into the live store without a full CSV reload. Each row in the corrections file is counted `weight` times so manually verified assignments quickly dominate the pattern table for those zones. Returns a summary dict with old/new record counts and pattern changes. """ if not os.path.isfile(corrections_path): raise FileNotFoundError(f"Corrections file not found: {corrections_path}") corr = self._parse_csv_file(corrections_path, label="Corrections") if not corr: return {"status": "error", "message": "No valid rows in corrections file."} weighted = corr * weight with self._lock: old_count = len(self._records) # Remove any existing correction rows for the same zones to avoid # double-injection (identify by source: corrections have no # "source" field — we use a tag approach). # Simplest safe approach: just append (first inject from scratch). merged = list(self._records) + weighted patterns, zone_index = self._compute_patterns(merged) old_clear = sum( 1 for p in self._patterns.values() if p["dominance"] >= _MIN_PATTERN_DOMINANCE and p["total_deliveries"] >= _MIN_PATTERN_VOLUME ) new_clear = sum( 1 for p in patterns.values() if p["dominance"] >= _MIN_PATTERN_DOMINANCE and p["total_deliveries"] >= _MIN_PATTERN_VOLUME ) vectors = np.array( [[r["pickuplat"], r["pickuplon"], r["deliverylat"], r["deliverylong"]] for r in merged], dtype=np.float32, ) index = _build_index(vectors) with self._lock: self._records = merged self._index = index self._patterns = patterns self._zone_index = zone_index # Save to disk (under whichever source is currently active) so next # restart includes corrections try: active_source = _pattern_source() vectors_path, records_path, meta_path = _paths_for(active_source) np.save(vectors_path, vectors) with open(records_path, "wb") as f: pickle.dump(merged, f) csv_mtime = ( os.path.getmtime(CSV_PATH) if active_source == "csv" and os.path.isfile(CSV_PATH) else 0 ) with open(meta_path, "w", encoding="utf-8") as f: json.dump({"source": active_source, "csv_mtime": csv_mtime, "record_count": len(merged), "pattern_count": len(patterns)}, f) logger.info( f"[DeliveryHistory] Corrections injected and saved — " f"records: {old_count} → {len(merged)}, " f"clear patterns: {old_clear} → {new_clear}" ) except Exception as e: logger.warning(f"[DeliveryHistory] Save after inject failed: {e}") return { "status": "ok", "unique_corrections": len(corr), "weight": weight, "records_before": old_count, "records_after": len(merged), "clear_patterns_before": old_clear, "clear_patterns_after": new_clear, "zones_total": len(patterns), } # ------------------------------------------------------------------ # Module-level singleton # ------------------------------------------------------------------ _store: Optional[DeliveryHistoryStore] = None _store_lock: Lock = Lock() def get_delivery_history_store() -> DeliveryHistoryStore: global _store if _store is None: with _store_lock: if _store is None: _store = DeliveryHistoryStore() return _store