50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
"""Offline-safe product data provider boundary and permanent cache record."""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
from typing import Any, Protocol
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProviderRecord:
|
|
provider: str
|
|
provider_record_version: str
|
|
gtin: str
|
|
retrieved_at: str
|
|
raw_hash: str
|
|
normalized_data: dict[str, Any]
|
|
fiscal_authority: str = "OBSERVATION_ONLY"
|
|
cache_allowed: bool = False
|
|
redistribution_allowed: bool = False
|
|
|
|
|
|
class ProductDataProvider(Protocol):
|
|
def lookup_gtin(self, gtin: str) -> ProviderRecord | None: ...
|
|
|
|
|
|
def normalize_record(provider: str, version: str, gtin: str, data: dict[str, Any]) -> ProviderRecord:
|
|
normalized = {str(key): value for key, value in sorted(data.items()) if value not in (None, "")}
|
|
raw_hash = hashlib.sha256(json.dumps(normalized, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")).hexdigest()
|
|
return ProviderRecord(provider, version, str(gtin), datetime.now(timezone.utc).isoformat(), raw_hash, normalized)
|
|
|
|
|
|
def cache_key(gtin: str, provider: str, provider_record_version: str) -> str:
|
|
return f"{gtin}:{provider}:{provider_record_version}"
|
|
|
|
|
|
def result_cache_key(product: dict[str, Any], candidates: list[str], datasets: dict[str, str], prompt: str, model: str) -> str:
|
|
payload = {"product": product, "candidates": candidates, "datasets": datasets, "prompt": prompt, "model": model}
|
|
return hashlib.sha256(json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")).hexdigest()
|
|
|
|
|
|
class LocalBulkDatasetProvider:
|
|
"""Offline provider backed by explicitly supplied records; never performs I/O."""
|
|
|
|
def __init__(self, records: dict[str, ProviderRecord] | None = None):
|
|
self.records = records or {}
|
|
|
|
def lookup_gtin(self, gtin: str) -> ProviderRecord | None:
|
|
return self.records.get(str(gtin))
|