305 lines
12 KiB
Python
305 lines
12 KiB
Python
"""
|
|
Smart action generator for alerts.
|
|
|
|
Generates actionable buttons with deep links, phone numbers,
|
|
and other interactive elements based on alert type and metadata.
|
|
"""
|
|
|
|
from typing import Dict, Any, List
|
|
import structlog
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class SmartActionGenerator:
|
|
"""Generate smart action buttons for alerts"""
|
|
|
|
def generate_actions(
|
|
self,
|
|
event_type: str,
|
|
metadata: Dict[str, Any],
|
|
orchestrator_context: dict
|
|
) -> List[dict]:
|
|
"""
|
|
Generate smart actions for an event.
|
|
|
|
Each action has:
|
|
- action_type: Identifier for frontend handling
|
|
- label_key: i18n key for button label
|
|
- label_params: Parameters for label translation
|
|
- variant: primary/secondary/danger/ghost
|
|
- disabled: Boolean
|
|
- disabled_reason_key: i18n key if disabled
|
|
- consequence_key: i18n key for confirmation dialog
|
|
- url: Deep link or tel: or mailto:
|
|
- metadata: Additional data for action
|
|
"""
|
|
|
|
actions = []
|
|
|
|
# If AI already addressed, show "View Action" button
|
|
if orchestrator_context and orchestrator_context.get("already_addressed"):
|
|
actions.append(self._create_view_action(orchestrator_context))
|
|
return actions
|
|
|
|
# Generate actions based on event type
|
|
if "po_approval" in event_type:
|
|
actions.extend(self._create_po_approval_actions(metadata))
|
|
|
|
elif "stock" in event_type or "shortage" in event_type:
|
|
actions.extend(self._create_stock_actions(metadata))
|
|
|
|
elif "production" in event_type or "delay" in event_type:
|
|
actions.extend(self._create_production_actions(metadata))
|
|
|
|
elif "equipment" in event_type:
|
|
actions.extend(self._create_equipment_actions(metadata))
|
|
|
|
elif "delivery" in event_type or "overdue" in event_type:
|
|
actions.extend(self._create_delivery_actions(metadata))
|
|
|
|
elif "temperature" in event_type:
|
|
actions.extend(self._create_temperature_actions(metadata))
|
|
|
|
# Always add common actions
|
|
actions.extend(self._create_common_actions())
|
|
|
|
return actions
|
|
|
|
def _create_view_action(self, orchestrator_context: dict) -> dict:
|
|
"""Create action to view what AI did"""
|
|
return {
|
|
"action_type": "open_reasoning",
|
|
"label_key": "actions.view_ai_action",
|
|
"label_params": {},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"metadata": {
|
|
"action_id": orchestrator_context.get("action_id"),
|
|
"action_type": orchestrator_context.get("action_type")
|
|
}
|
|
}
|
|
|
|
def _create_po_approval_actions(self, metadata: Dict[str, Any]) -> List[dict]:
|
|
"""Create actions for PO approval alerts"""
|
|
po_id = metadata.get("po_id")
|
|
po_amount = metadata.get("total_amount", metadata.get("po_amount", 0))
|
|
|
|
return [
|
|
{
|
|
"action_type": "approve_po",
|
|
"label_key": "actions.approve_po",
|
|
"label_params": {"amount": po_amount},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"consequence_key": "actions.approve_po_consequence",
|
|
"url": f"/app/procurement/purchase-orders/{po_id}",
|
|
"metadata": {"po_id": po_id, "amount": po_amount}
|
|
},
|
|
{
|
|
"action_type": "reject_po",
|
|
"label_key": "actions.reject_po",
|
|
"label_params": {},
|
|
"variant": "danger",
|
|
"disabled": False,
|
|
"consequence_key": "actions.reject_po_consequence",
|
|
"url": f"/app/procurement/purchase-orders/{po_id}",
|
|
"metadata": {"po_id": po_id}
|
|
},
|
|
{
|
|
"action_type": "modify_po",
|
|
"label_key": "actions.modify_po",
|
|
"label_params": {},
|
|
"variant": "secondary",
|
|
"disabled": False,
|
|
"url": f"/app/procurement/purchase-orders/{po_id}/edit",
|
|
"metadata": {"po_id": po_id}
|
|
}
|
|
]
|
|
|
|
def _create_stock_actions(self, metadata: Dict[str, Any]) -> List[dict]:
|
|
"""Create actions for stock-related alerts"""
|
|
actions = []
|
|
|
|
# If supplier info available, add call button
|
|
if metadata.get("supplier_contact"):
|
|
actions.append({
|
|
"action_type": "call_supplier",
|
|
"label_key": "actions.call_supplier",
|
|
"label_params": {
|
|
"supplier": metadata.get("supplier_name", "Supplier"),
|
|
"phone": metadata.get("supplier_contact")
|
|
},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"url": f"tel:{metadata['supplier_contact']}",
|
|
"metadata": {
|
|
"supplier_name": metadata.get("supplier_name"),
|
|
"phone": metadata.get("supplier_contact")
|
|
}
|
|
})
|
|
|
|
# If PO exists, add view PO button
|
|
if metadata.get("po_id"):
|
|
if metadata.get("po_status") == "pending_approval":
|
|
actions.append({
|
|
"action_type": "approve_po",
|
|
"label_key": "actions.approve_po",
|
|
"label_params": {"amount": metadata.get("po_amount", 0)},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"url": f"/app/procurement/purchase-orders/{metadata['po_id']}",
|
|
"metadata": {"po_id": metadata["po_id"]}
|
|
})
|
|
else:
|
|
actions.append({
|
|
"action_type": "view_po",
|
|
"label_key": "actions.view_po",
|
|
"label_params": {"po_number": metadata.get("po_number", metadata["po_id"])},
|
|
"variant": "secondary",
|
|
"disabled": False,
|
|
"url": f"/app/procurement/purchase-orders/{metadata['po_id']}",
|
|
"metadata": {"po_id": metadata["po_id"]}
|
|
})
|
|
|
|
# Add create PO button if no PO exists
|
|
else:
|
|
actions.append({
|
|
"action_type": "create_po",
|
|
"label_key": "actions.create_po",
|
|
"label_params": {},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"url": f"/app/procurement/purchase-orders/new?ingredient_id={metadata.get('ingredient_id')}",
|
|
"metadata": {"ingredient_id": metadata.get("ingredient_id")}
|
|
})
|
|
|
|
return actions
|
|
|
|
def _create_production_actions(self, metadata: Dict[str, Any]) -> List[dict]:
|
|
"""Create actions for production-related alerts"""
|
|
actions = []
|
|
|
|
if metadata.get("batch_id"):
|
|
actions.append({
|
|
"action_type": "view_batch",
|
|
"label_key": "actions.view_batch",
|
|
"label_params": {"batch_number": metadata.get("batch_number", "")},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"url": f"/app/production/batches/{metadata['batch_id']}",
|
|
"metadata": {"batch_id": metadata["batch_id"]}
|
|
})
|
|
|
|
actions.append({
|
|
"action_type": "adjust_production",
|
|
"label_key": "actions.adjust_production",
|
|
"label_params": {},
|
|
"variant": "secondary",
|
|
"disabled": False,
|
|
"url": f"/app/production/batches/{metadata['batch_id']}/adjust",
|
|
"metadata": {"batch_id": metadata["batch_id"]}
|
|
})
|
|
|
|
return actions
|
|
|
|
def _create_equipment_actions(self, metadata: Dict[str, Any]) -> List[dict]:
|
|
"""Create actions for equipment-related alerts"""
|
|
return [
|
|
{
|
|
"action_type": "view_equipment",
|
|
"label_key": "actions.view_equipment",
|
|
"label_params": {"equipment_name": metadata.get("equipment_name", "")},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"url": f"/app/production/equipment/{metadata.get('equipment_id')}",
|
|
"metadata": {"equipment_id": metadata.get("equipment_id")}
|
|
},
|
|
{
|
|
"action_type": "schedule_maintenance",
|
|
"label_key": "actions.schedule_maintenance",
|
|
"label_params": {},
|
|
"variant": "secondary",
|
|
"disabled": False,
|
|
"url": f"/app/production/equipment/{metadata.get('equipment_id')}/maintenance",
|
|
"metadata": {"equipment_id": metadata.get("equipment_id")}
|
|
}
|
|
]
|
|
|
|
def _create_delivery_actions(self, metadata: Dict[str, Any]) -> List[dict]:
|
|
"""Create actions for delivery-related alerts"""
|
|
actions = []
|
|
|
|
if metadata.get("supplier_contact"):
|
|
actions.append({
|
|
"action_type": "call_supplier",
|
|
"label_key": "actions.call_supplier",
|
|
"label_params": {
|
|
"supplier": metadata.get("supplier_name", "Supplier"),
|
|
"phone": metadata.get("supplier_contact")
|
|
},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"url": f"tel:{metadata['supplier_contact']}",
|
|
"metadata": {
|
|
"supplier_name": metadata.get("supplier_name"),
|
|
"phone": metadata.get("supplier_contact")
|
|
}
|
|
})
|
|
|
|
if metadata.get("po_id"):
|
|
actions.append({
|
|
"action_type": "view_po",
|
|
"label_key": "actions.view_po",
|
|
"label_params": {"po_number": metadata.get("po_number", "")},
|
|
"variant": "secondary",
|
|
"disabled": False,
|
|
"url": f"/app/procurement/purchase-orders/{metadata['po_id']}",
|
|
"metadata": {"po_id": metadata["po_id"]}
|
|
})
|
|
|
|
return actions
|
|
|
|
def _create_temperature_actions(self, metadata: Dict[str, Any]) -> List[dict]:
|
|
"""Create actions for temperature breach alerts"""
|
|
return [
|
|
{
|
|
"action_type": "view_sensor",
|
|
"label_key": "actions.view_sensor",
|
|
"label_params": {"location": metadata.get("location", "")},
|
|
"variant": "primary",
|
|
"disabled": False,
|
|
"url": f"/app/inventory/sensors/{metadata.get('sensor_id')}",
|
|
"metadata": {"sensor_id": metadata.get("sensor_id")}
|
|
},
|
|
{
|
|
"action_type": "acknowledge_breach",
|
|
"label_key": "actions.acknowledge_breach",
|
|
"label_params": {},
|
|
"variant": "secondary",
|
|
"disabled": False,
|
|
"metadata": {"sensor_id": metadata.get("sensor_id")}
|
|
}
|
|
]
|
|
|
|
def _create_common_actions(self) -> List[dict]:
|
|
"""Create common actions available for all alerts"""
|
|
return [
|
|
{
|
|
"action_type": "snooze",
|
|
"label_key": "actions.snooze",
|
|
"label_params": {"hours": 4},
|
|
"variant": "ghost",
|
|
"disabled": False,
|
|
"metadata": {"snooze_hours": 4}
|
|
},
|
|
{
|
|
"action_type": "dismiss",
|
|
"label_key": "actions.dismiss",
|
|
"label_params": {},
|
|
"variant": "ghost",
|
|
"disabled": False,
|
|
"metadata": {}
|
|
}
|
|
]
|