Spaces:
Running
Running
File size: 3,413 Bytes
8b30412 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
"""Change detection module for Azure Search index synchronization."""
from dataclasses import dataclass
from typing import Any
from chatassistant_retail.data.models import Product
@dataclass
class IndexDiff:
"""Results of comparing local products with Azure index."""
inserts: list[Product] # SKUs in local but not in index
updates: list[Product] # SKUs in both, but fields differ
deletes: list[str] # SKUs in index but not in local
unchanged: int # SKUs with identical data
def total_changes(self) -> int:
"""Calculate total number of changes."""
return len(self.inserts) + len(self.updates) + len(self.deletes)
def summary(self) -> str:
"""Generate human-readable summary of changes."""
return (
f"Changes detected:\n"
f" - Inserts: {len(self.inserts)}\n"
f" - Updates: {len(self.updates)}\n"
f" - Deletes: {len(self.deletes)}\n"
f" - Unchanged: {self.unchanged}\n"
f" Total changes: {self.total_changes()}"
)
def _products_differ(local: Product, indexed: dict[str, Any]) -> bool:
"""
Check if local product differs from indexed document.
Compares all mutable fields (excludes id/sku which are keys).
Args:
local: Local Product instance
indexed: Indexed document as dictionary
Returns:
True if products differ, False if identical
"""
fields_to_compare = ["name", "category", "description", "price", "current_stock", "reorder_level", "supplier"]
for field in fields_to_compare:
local_value = getattr(local, field)
indexed_value = indexed.get(field)
# Handle type coercion (JSON may load ints as floats, etc.)
if isinstance(local_value, (int, float)) and isinstance(indexed_value, (int, float)):
if abs(local_value - indexed_value) > 1e-9: # Float comparison tolerance
return True
elif str(local_value) != str(indexed_value):
return True
return False
def calculate_diff(local_products: list[Product], indexed_documents: list[dict[str, Any]]) -> IndexDiff:
"""
Calculate difference between local products and indexed documents.
Args:
local_products: Products from local JSON file
indexed_documents: Documents from Azure Search index
Returns:
IndexDiff object with inserts, updates, deletes, and unchanged counts
"""
# Create lookup dictionaries
local_by_sku = {p.sku: p for p in local_products}
indexed_by_sku = {doc["sku"]: doc for doc in indexed_documents}
inserts = []
updates = []
deletes = []
unchanged = 0
# Find inserts and updates
for sku, product in local_by_sku.items():
if sku not in indexed_by_sku:
# SKU exists locally but not in index → INSERT
inserts.append(product)
else:
# SKU exists in both → check if fields differ
if _products_differ(product, indexed_by_sku[sku]):
updates.append(product)
else:
unchanged += 1
# Find deletes
for sku in indexed_by_sku.keys():
if sku not in local_by_sku:
# SKU exists in index but not locally → DELETE
deletes.append(sku)
return IndexDiff(inserts=inserts, updates=updates, deletes=deletes, unchanged=unchanged)
|