bakery-admin/bakery-ia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix few issues

Browse Source
This commit is contained in:
Urtzi Alfaro
2025-09-26 12:12:17 +02:00
parent d573c38621
commit a27f159e24
32 changed files with 2694 additions and 575 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
shared/alerts/base_service.py
View File

@@ -239,15 +239,18 @@ class BaseAlertService:
# Publishing (Updated for type)
async def publish_item(self, tenant_id: UUID, item: Dict[str, Any], item_type: str = 'alert'):
"""Publish alert or recommendation to RabbitMQ with deduplication"""
try:
# Check for duplicate
item_key = f"{tenant_id}:{item_type}:{item['type']}:{item.get('metadata', {}).get('id', '')}"
# Generate proper deduplication key based on alert type and specific identifiers
unique_id = self._generate_unique_identifier(item)
item_key = f"{tenant_id}:{item_type}:{item['type']}:{unique_id}"
if await self.is_duplicate_item(item_key):
logger.debug("Duplicate item skipped",
service=self.config.SERVICE_NAME,
item_type=item_type,
alert_type=item['type'])
logger.debug("Duplicate item skipped",
service=self.config.SERVICE_NAME,
item_type=item_type,
alert_type=item['type'],
dedup_key=item_key)
return False
# Add metadata
@@ -302,12 +305,49 @@ class BaseAlertService:
item_type=item_type)
return False
def _generate_unique_identifier(self, item: Dict[str, Any]) -> str:
"""Generate unique identifier for deduplication based on alert type and content"""
alert_type = item.get('type', '')
metadata = item.get('metadata', {})
# Generate unique identifier based on alert type
if alert_type == 'overstock_warning':
return metadata.get('ingredient_id', '')
elif alert_type == 'critical_stock_shortage' or alert_type == 'low_stock_warning':
return metadata.get('ingredient_id', '')
elif alert_type == 'expired_products':
# For expired products alerts, create hash of all expired item IDs
expired_items = metadata.get('expired_items', [])
if expired_items:
expired_ids = sorted([str(item.get('id', '')) for item in expired_items])
import hashlib
return hashlib.md5(':'.join(expired_ids).encode()).hexdigest()[:16]
return ''
elif alert_type == 'urgent_expiry':
return f"{metadata.get('ingredient_id', '')}:{metadata.get('stock_id', '')}"
elif alert_type == 'temperature_breach':
return f"{metadata.get('sensor_id', '')}:{metadata.get('location', '')}"
elif alert_type == 'stock_depleted_by_order':
return f"{metadata.get('order_id', '')}:{metadata.get('ingredient_id', '')}"
elif alert_type == 'expired_batches_auto_processed':
# Use processing date and total batches as identifier
processing_date = metadata.get('processing_date', '')[:10] # Date only
total_batches = metadata.get('total_batches_processed', 0)
return f"{processing_date}:{total_batches}"
elif alert_type == 'inventory_optimization':
return f"opt:{metadata.get('ingredient_id', '')}:{metadata.get('recommendation_type', '')}"
elif alert_type == 'waste_reduction':
return f"waste:{metadata.get('ingredient_id', '')}"
else:
# Fallback to generic metadata.id or empty string
return metadata.get('id', '')
async def is_duplicate_item(self, item_key: str, window_minutes: int = 15) -> bool:
"""Prevent duplicate items within time window"""
key = f"item_sent:{item_key}"
try:
result = await self.redis.set(
key, "1",
key, "1",
ex=window_minutes * 60,
nx=True
)

135
shared/clients/inventory_client.py
View File

@@ -340,10 +340,143 @@ class InventoryServiceClient(BaseServiceClient):
error=str(e), tenant_id=tenant_id)
return None
# ================================================================
# PRODUCT TRANSFORMATION
# ================================================================
async def create_transformation(
self,
transformation_data: Dict[str, Any],
tenant_id: str
) -> Optional[Dict[str, Any]]:
"""Create a product transformation (e.g., par-baked to fully baked)"""
try:
result = await self.post("transformations", data=transformation_data, tenant_id=tenant_id)
if result:
logger.info("Created product transformation",
transformation_reference=result.get('transformation_reference'),
source_stage=transformation_data.get('source_stage'),
target_stage=transformation_data.get('target_stage'),
tenant_id=tenant_id)
return result
except Exception as e:
logger.error("Error creating transformation",
error=str(e), transformation_data=transformation_data, tenant_id=tenant_id)
return None
async def create_par_bake_transformation(
self,
source_ingredient_id: Union[str, UUID],
target_ingredient_id: Union[str, UUID],
quantity: float,
tenant_id: str,
target_batch_number: Optional[str] = None,
expiration_hours: int = 24,
notes: Optional[str] = None
) -> Optional[Dict[str, Any]]:
"""Convenience method for par-baked to fresh transformation"""
try:
params = {
"source_ingredient_id": str(source_ingredient_id),
"target_ingredient_id": str(target_ingredient_id),
"quantity": quantity,
"expiration_hours": expiration_hours
}
if target_batch_number:
params["target_batch_number"] = target_batch_number
if notes:
params["notes"] = notes
result = await self.post("transformations/par-bake-to-fresh", params=params, tenant_id=tenant_id)
if result:
logger.info("Created par-bake transformation",
transformation_id=result.get('transformation_id'),
quantity=quantity, tenant_id=tenant_id)
return result
except Exception as e:
logger.error("Error creating par-bake transformation",
error=str(e), source_ingredient_id=source_ingredient_id,
target_ingredient_id=target_ingredient_id, tenant_id=tenant_id)
return None
async def get_transformations(
self,
tenant_id: str,
ingredient_id: Optional[Union[str, UUID]] = None,
source_stage: Optional[str] = None,
target_stage: Optional[str] = None,
days_back: Optional[int] = None,
skip: int = 0,
limit: int = 100
) -> List[Dict[str, Any]]:
"""Get product transformations with filtering"""
try:
params = {
"skip": skip,
"limit": limit
}
if ingredient_id:
params["ingredient_id"] = str(ingredient_id)
if source_stage:
params["source_stage"] = source_stage
if target_stage:
params["target_stage"] = target_stage
if days_back:
params["days_back"] = days_back
result = await self.get("transformations", tenant_id=tenant_id, params=params)
transformations = result if isinstance(result, list) else []
logger.info("Retrieved transformations from inventory service",
count=len(transformations), tenant_id=tenant_id)
return transformations
except Exception as e:
logger.error("Error fetching transformations",
error=str(e), tenant_id=tenant_id)
return []
async def get_transformation_by_id(
self,
transformation_id: Union[str, UUID],
tenant_id: str
) -> Optional[Dict[str, Any]]:
"""Get specific transformation by ID"""
try:
result = await self.get(f"transformations/{transformation_id}", tenant_id=tenant_id)
if result:
logger.info("Retrieved transformation by ID",
transformation_id=transformation_id, tenant_id=tenant_id)
return result
except Exception as e:
logger.error("Error fetching transformation by ID",
error=str(e), transformation_id=transformation_id, tenant_id=tenant_id)
return None
async def get_transformation_summary(
self,
tenant_id: str,
days_back: int = 30
) -> Optional[Dict[str, Any]]:
"""Get transformation summary for dashboard"""
try:
params = {"days_back": days_back}
result = await self.get("transformations/summary", tenant_id=tenant_id, params=params)
if result:
logger.info("Retrieved transformation summary",
days_back=days_back, tenant_id=tenant_id)
return result
except Exception as e:
logger.error("Error fetching transformation summary",
error=str(e), tenant_id=tenant_id)
return None
# ================================================================
# UTILITY METHODS
# ================================================================
async def health_check(self) -> bool:
"""Check if inventory service is healthy"""
try: