from typing import Dict, List, Optional from datetime import datetime from app.models.enums import AlertTypeEnum from app.models.schemas import Alert from app.utils.logger import get_logger import uuid logger = get_logger(__name__) class AlertService: def __init__(self): self._alerts: List[Alert] = [] self._acknowledged: List[str] = [] async def create_alert( self, alert_type: AlertTypeEnum, message: str, data: Optional[Dict] = None, ) -> Alert: alert_id = str(uuid.uuid4()) alert = Alert( alert_id=alert_id, alert_type=alert_type, timestamp=datetime.utcnow(), message=message, data=data or {}, acknowledged=False, ) self._alerts.append(alert) logger.warning(f"Alert created: {alert_id}, type: {alert_type.value}, message: {message}") return alert async def get_alerts( self, alert_type: Optional[AlertTypeEnum] = None, acknowledged: Optional[bool] = None, limit: int = 100, ) -> List[Alert]: filtered = self._alerts if alert_type: filtered = [a for a in filtered if a.alert_type == alert_type] if acknowledged is not None: filtered = [a for a in filtered if a.acknowledged == acknowledged] return filtered[-limit:] async def acknowledge_alert(self, alert_id: str) -> Alert: for alert in self._alerts: if alert.alert_id == alert_id: alert.acknowledged = True logger.info(f"Alert acknowledged: {alert_id}") return alert raise ValueError(f"Alert not found: {alert_id}") async def get_alert_summary(self) -> Dict: total = len(self._alerts) unacknowledged = len([a for a in self._alerts if not a.acknowledged]) by_type = {} for alert in self._alerts: alert_type = alert.alert_type.value by_type[alert_type] = by_type.get(alert_type, 0) + 1 return { "total_alerts": total, "unacknowledged": unacknowledged, "by_type": by_type, "latest_alert": self._alerts[-1].timestamp if self._alerts else None, }