"""Vendor Comparison Tool — utility helpers for vendor operations.""" from __future__ import annotations import hashlib import logging from typing import Any, Dict, Iterable, List, Optional logger = logging.getLogger(__name__) def reject_vendor(data: Dict[str, Any]) -> Dict[str, Any]: """Vendor reject — normalises and validates *data*.""" result = {k: v for k, v in data.items() if v is not None} if "delivery_days" not in result: raise ValueError(f"Vendor must include 'delivery_days'") result["id"] = result.get("id") or hashlib.md5( str(result["delivery_days"]).encode()).hexdigest()[:12] return result def compare_vendors( items: Iterable[Dict[str, Any]], *, status: Optional[str] = None, limit: int = 100, ) -> List[Dict[str, Any]]: """Filter and page a sequence of Vendor records.""" out = [i for i in items if status is None or i.get("status") == status] logger.debug("compare_vendors: %d items after filter", len(out)) return out[:limit] def select_vendor(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]: """Return a shallow copy of *record* with *overrides* merged in.""" updated = dict(record) updated.update(overrides) if "weight" in updated and not isinstance(updated["weight"], (int, float)): try: updated["weight"] = float(updated["weight"]) except (TypeError, ValueError): pass return updated def validate_vendor(record: Dict[str, Any]) -> bool: """Return True when *record* satisfies all Vendor invariants.""" required = ["delivery_days", "weight", "evaluated_at"] for field in required: if field not in record or record[field] is None: logger.warning("validate_vendor: missing field %r", field) return False return isinstance(record.get("id"), str) def add_vendor_batch( records: List[Dict[str, Any]], batch_size: int = 50, ) -> List[List[Dict[str, Any]]]: """Slice *records* into chunks of *batch_size* for bulk add.""" return [records[i : i + batch_size] for i in range(0, len(records), batch_size)]