Spaces:
Sleeping
Sleeping
| """Change detection module for Azure Search index synchronization.""" | |
| from dataclasses import dataclass | |
| from typing import Any | |
| from chatassistant_retail.data.models import Product | |
| 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) | |