245 lines
8.1 KiB
Python
245 lines
8.1 KiB
Python
"""
|
|
Message generator for creating i18n message keys and parameters.
|
|
|
|
Converts minimal event metadata into structured i18n format for frontend translation.
|
|
"""
|
|
|
|
from typing import Dict, Any
|
|
from datetime import datetime
|
|
from app.utils.message_templates import ALERT_TEMPLATES, NOTIFICATION_TEMPLATES, RECOMMENDATION_TEMPLATES
|
|
import structlog
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class MessageGenerator:
|
|
"""Generates i18n message keys and parameters from event metadata"""
|
|
|
|
def generate_message(self, event_type: str, metadata: Dict[str, Any], event_class: str = "alert") -> dict:
|
|
"""
|
|
Generate i18n structure for frontend.
|
|
|
|
Args:
|
|
event_type: Alert/notification/recommendation type
|
|
metadata: Event metadata dictionary
|
|
event_class: One of: alert, notification, recommendation
|
|
|
|
Returns:
|
|
Dictionary with title_key, title_params, message_key, message_params
|
|
"""
|
|
|
|
# Select appropriate template collection
|
|
if event_class == "notification":
|
|
templates = NOTIFICATION_TEMPLATES
|
|
elif event_class == "recommendation":
|
|
templates = RECOMMENDATION_TEMPLATES
|
|
else:
|
|
templates = ALERT_TEMPLATES
|
|
|
|
template = templates.get(event_type)
|
|
|
|
if not template:
|
|
logger.warning("no_template_found", event_type=event_type, event_class=event_class)
|
|
return self._generate_fallback(event_type, metadata)
|
|
|
|
# Build parameters from metadata
|
|
title_params = self._build_params(template["title_params"], metadata)
|
|
message_params = self._build_params(template["message_params"], metadata)
|
|
|
|
# Select message variant based on context
|
|
message_key = self._select_message_variant(
|
|
template["message_variants"],
|
|
metadata
|
|
)
|
|
|
|
return {
|
|
"title_key": template["title_key"],
|
|
"title_params": title_params,
|
|
"message_key": message_key,
|
|
"message_params": message_params
|
|
}
|
|
|
|
def _generate_fallback(self, event_type: str, metadata: Dict[str, Any]) -> dict:
|
|
"""Generate fallback message structure when template not found"""
|
|
return {
|
|
"title_key": "alerts.generic.title",
|
|
"title_params": {},
|
|
"message_key": "alerts.generic.message",
|
|
"message_params": {
|
|
"event_type": event_type,
|
|
"metadata_summary": self._summarize_metadata(metadata)
|
|
}
|
|
}
|
|
|
|
def _summarize_metadata(self, metadata: Dict[str, Any]) -> str:
|
|
"""Create human-readable summary of metadata"""
|
|
# Take first 3 fields
|
|
items = list(metadata.items())[:3]
|
|
summary_parts = [f"{k}: {v}" for k, v in items]
|
|
return ", ".join(summary_parts)
|
|
|
|
def _build_params(self, param_mapping: dict, metadata: dict) -> dict:
|
|
"""
|
|
Extract and transform parameters from metadata.
|
|
|
|
param_mapping format: {"display_param_name": "metadata_key"}
|
|
"""
|
|
params = {}
|
|
|
|
for param_key, metadata_key in param_mapping.items():
|
|
if metadata_key in metadata:
|
|
value = metadata[metadata_key]
|
|
|
|
# Apply transformations based on parameter suffix
|
|
if param_key.endswith("_kg"):
|
|
value = round(float(value), 1)
|
|
elif param_key.endswith("_eur"):
|
|
value = round(float(value), 2)
|
|
elif param_key.endswith("_percentage"):
|
|
value = round(float(value), 1)
|
|
elif param_key.endswith("_date"):
|
|
value = self._format_date(value)
|
|
elif param_key.endswith("_day_name"):
|
|
value = self._format_day_name(value)
|
|
elif param_key.endswith("_datetime"):
|
|
value = self._format_datetime(value)
|
|
|
|
params[param_key] = value
|
|
|
|
return params
|
|
|
|
def _select_message_variant(self, variants: dict, metadata: dict) -> str:
|
|
"""
|
|
Select appropriate message variant based on metadata context.
|
|
|
|
Checks for specific conditions in priority order.
|
|
"""
|
|
|
|
# Check for PO-related variants
|
|
if "po_id" in metadata:
|
|
if metadata.get("po_status") == "pending_approval":
|
|
variant = variants.get("with_po_pending")
|
|
if variant:
|
|
return variant
|
|
else:
|
|
variant = variants.get("with_po_created")
|
|
if variant:
|
|
return variant
|
|
|
|
# Check for time-based variants
|
|
if "hours_until" in metadata:
|
|
variant = variants.get("with_hours")
|
|
if variant:
|
|
return variant
|
|
|
|
if "production_date" in metadata or "planned_date" in metadata:
|
|
variant = variants.get("with_date")
|
|
if variant:
|
|
return variant
|
|
|
|
# Check for customer-related variants
|
|
if "customer_names" in metadata and metadata.get("customer_names"):
|
|
variant = variants.get("with_customers")
|
|
if variant:
|
|
return variant
|
|
|
|
# Check for order-related variants
|
|
if "affected_orders" in metadata and metadata.get("affected_orders", 0) > 0:
|
|
variant = variants.get("with_orders")
|
|
if variant:
|
|
return variant
|
|
|
|
# Check for supplier contact variants
|
|
if "supplier_contact" in metadata:
|
|
variant = variants.get("with_supplier")
|
|
if variant:
|
|
return variant
|
|
|
|
# Check for batch-related variants
|
|
if "affected_batches" in metadata and metadata.get("affected_batches", 0) > 0:
|
|
variant = variants.get("with_batches")
|
|
if variant:
|
|
return variant
|
|
|
|
# Check for product names list variants
|
|
if "product_names" in metadata and metadata.get("product_names"):
|
|
variant = variants.get("with_names")
|
|
if variant:
|
|
return variant
|
|
|
|
# Check for time duration variants
|
|
if "hours_overdue" in metadata:
|
|
variant = variants.get("with_hours")
|
|
if variant:
|
|
return variant
|
|
|
|
if "days_overdue" in metadata:
|
|
variant = variants.get("with_days")
|
|
if variant:
|
|
return variant
|
|
|
|
# Default to generic variant
|
|
return variants.get("generic", variants[list(variants.keys())[0]])
|
|
|
|
def _format_date(self, date_value: Any) -> str:
|
|
"""
|
|
Format date for display.
|
|
|
|
Accepts:
|
|
- ISO string: "2025-12-10"
|
|
- datetime object
|
|
- date object
|
|
|
|
Returns: ISO format "YYYY-MM-DD"
|
|
"""
|
|
if isinstance(date_value, str):
|
|
# Already a string, might be ISO format
|
|
try:
|
|
dt = datetime.fromisoformat(date_value.replace('Z', '+00:00'))
|
|
return dt.date().isoformat()
|
|
except:
|
|
return date_value
|
|
|
|
if isinstance(date_value, datetime):
|
|
return date_value.date().isoformat()
|
|
|
|
if hasattr(date_value, 'isoformat'):
|
|
return date_value.isoformat()
|
|
|
|
return str(date_value)
|
|
|
|
def _format_day_name(self, date_value: Any) -> str:
|
|
"""
|
|
Format day name with date.
|
|
|
|
Example: "miércoles 10 de diciembre"
|
|
|
|
Note: Frontend will handle localization.
|
|
For now, return ISO date and let frontend format.
|
|
"""
|
|
iso_date = self._format_date(date_value)
|
|
|
|
try:
|
|
dt = datetime.fromisoformat(iso_date)
|
|
# Frontend will use this to format in user's language
|
|
return iso_date
|
|
except:
|
|
return iso_date
|
|
|
|
def _format_datetime(self, datetime_value: Any) -> str:
|
|
"""
|
|
Format datetime for display.
|
|
|
|
Returns: ISO 8601 format with timezone
|
|
"""
|
|
if isinstance(datetime_value, str):
|
|
return datetime_value
|
|
|
|
if isinstance(datetime_value, datetime):
|
|
return datetime_value.isoformat()
|
|
|
|
if hasattr(datetime_value, 'isoformat'):
|
|
return datetime_value.isoformat()
|
|
|
|
return str(datetime_value)
|