iot-device-management-service/app/services/alert_service.py

"""
告警服务
处理告警 CRUD 和 MQTT 消息
"""
import uuid
from datetime import datetime, timezone
from typing import Optional, List, Dict, Any

from app.models import Alert, AlertStatus, AlertLevel, get_session
from app.schemas import AlertCreate, AlertHandleRequest
from app.services.oss_storage import get_oss_storage
from app.utils.logger import logger


class AlertService:
    """告警服务"""

    def __init__(self):
        self.oss = get_oss_storage()

    def generate_alert_no(self) -> str:
        """生成告警编号"""
        timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
        unique_id = uuid.uuid4().hex[:8].upper()
        return f"ALT{timestamp}{unique_id}"

    def create_alert(
        self,
        alert_data: AlertCreate,
        snapshot_data: Optional[bytes] = None,
    ) -> Alert:
        """创建告警（HTTP 方式）"""
        db = get_session()
        try:
            alert = Alert(
                alert_no=self.generate_alert_no(),
                camera_id=alert_data.camera_id,
                roi_id=alert_data.roi_id,
                bind_id=getattr(alert_data, 'bind_id', None),
                device_id=getattr(alert_data, 'device_id', None),
                alert_type=alert_data.alert_type,
                algorithm=alert_data.algorithm,
                confidence=alert_data.confidence,
                duration_minutes=alert_data.duration_minutes,
                trigger_time=alert_data.trigger_time,
                message=alert_data.message,
                bbox=getattr(alert_data, 'bbox', None),
                status=AlertStatus.PENDING,
                level=AlertLevel.MEDIUM,
            )

            if snapshot_data:
                snapshot_url = self.oss.upload_image(snapshot_data)
                alert.snapshot_url = snapshot_url

            db.add(alert)
            db.commit()
            db.refresh(alert)

            logger.info(f"告警创建成功: {alert.alert_no}")
            return alert

        finally:
            db.close()

    def create_alert_from_mqtt(self, mqtt_data: Dict[str, Any]) -> Optional[Alert]:
        """从 MQTT 消息创建告警"""
        db = get_session()
        try:
            # 解析时间
            trigger_time_str = mqtt_data.get("timestamp")
            if trigger_time_str:
                try:
                    trigger_time = datetime.fromisoformat(trigger_time_str.replace("Z", "+00:00"))
                except ValueError:
                    trigger_time = datetime.now(timezone.utc)
            else:
                trigger_time = datetime.now(timezone.utc)

            # 解析置信度（MQTT 可能是 0-1，需要转为 0-100）
            confidence = mqtt_data.get("confidence")
            if confidence is not None:
                if isinstance(confidence, float) and confidence <= 1:
                    confidence = int(confidence * 100)
                else:
                    confidence = int(confidence)

            # 解析持续时长
            duration_minutes = mqtt_data.get("duration_minutes")
            if duration_minutes is not None:
                duration_minutes = int(float(duration_minutes))

            alert = Alert(
                alert_no=self.generate_alert_no(),
                camera_id=mqtt_data.get("camera_id", "unknown"),
                roi_id=mqtt_data.get("roi_id"),
                bind_id=mqtt_data.get("bind_id"),
                device_id=mqtt_data.get("device_id"),
                alert_type=mqtt_data.get("alert_type", "unknown"),
                algorithm=mqtt_data.get("algorithm", "YOLO"),
                confidence=confidence,
                duration_minutes=duration_minutes,
                trigger_time=trigger_time,
                message=mqtt_data.get("message"),
                bbox=mqtt_data.get("bbox"),
                status=AlertStatus.PENDING,
                level=self._determine_level(mqtt_data),
            )

            db.add(alert)
            db.commit()
            db.refresh(alert)

            logger.info(f"MQTT告警创建成功: {alert.alert_no}, type={alert.alert_type}")
            return alert

        except Exception as e:
            db.rollback()
            logger.error(f"创建MQTT告警失败: {e}")
            return None
        finally:
            db.close()

    def _determine_level(self, data: Dict[str, Any]) -> AlertLevel:
        """根据告警数据确定告警级别"""
        alert_type = data.get("alert_type", "")
        confidence = data.get("confidence", 0)

        # 根据告警类型和置信度确定级别
        if alert_type == "intrusion":
            return AlertLevel.HIGH
        elif alert_type == "leave_post":
            duration = data.get("duration_minutes", 0)
            if duration and duration > 30:
                return AlertLevel.HIGH
            elif duration and duration > 10:
                return AlertLevel.MEDIUM
            return AlertLevel.LOW
        elif confidence and confidence > 0.9:
            return AlertLevel.HIGH
        elif confidence and confidence > 0.7:
            return AlertLevel.MEDIUM

        return AlertLevel.MEDIUM

    def get_alert(self, alert_id: int) -> Optional[Alert]:
        """获取告警详情"""
        db = get_session()
        try:
            return db.query(Alert).filter(Alert.id == alert_id).first()
        finally:
            db.close()

    def get_alert_by_no(self, alert_no: str) -> Optional[Alert]:
        """根据告警编号获取"""
        db = get_session()
        try:
            return db.query(Alert).filter(Alert.alert_no == alert_no).first()
        finally:
            db.close()

    def get_alerts(
        self,
        camera_id: Optional[str] = None,
        device_id: Optional[str] = None,
        alert_type: Optional[str] = None,
        status: Optional[str] = None,
        level: Optional[str] = None,
        start_time: Optional[datetime] = None,
        end_time: Optional[datetime] = None,
        page: int = 1,
        page_size: int = 20,
    ) -> tuple[List[Alert], int]:
        """获取告警列表"""
        db = get_session()
        try:
            query = db.query(Alert)

            if camera_id:
                query = query.filter(Alert.camera_id == camera_id)
            if device_id:
                query = query.filter(Alert.device_id == device_id)
            if alert_type:
                query = query.filter(Alert.alert_type == alert_type)
            if status:
                query = query.filter(Alert.status == status)
            if level:
                query = query.filter(Alert.level == level)
            if start_time:
                query = query.filter(Alert.trigger_time >= start_time)
            if end_time:
                query = query.filter(Alert.trigger_time <= end_time)

            total = query.count()
            alerts = (
                query.order_by(Alert.created_at.desc())
                .offset((page - 1) * page_size)
                .limit(page_size)
                .all()
            )

            return alerts, total
        finally:
            db.close()

    def handle_alert(
        self,
        alert_id: int,
        handle_data: AlertHandleRequest,
        handled_by: Optional[str] = None,
    ) -> Optional[Alert]:
        """处理告警"""
        db = get_session()
        try:
            alert = db.query(Alert).filter(Alert.id == alert_id).first()
            if not alert:
                return None

            alert.status = AlertStatus(handle_data.status)
            alert.handle_remark = handle_data.remark
            alert.handled_by = handled_by
            alert.handled_at = datetime.now(timezone.utc)
            alert.updated_at = datetime.now(timezone.utc)

            db.commit()
            db.refresh(alert)

            logger.info(f"告警处理完成: {alert.alert_no}, 状态: {handle_data.status}")
            return alert

        finally:
            db.close()

    def dispatch_alert(self, alert_id: int, work_order_id: int) -> Optional[Alert]:
        """派发告警（关联工单）"""
        db = get_session()
        try:
            alert = db.query(Alert).filter(Alert.id == alert_id).first()
            if not alert:
                return None

            alert.status = AlertStatus.DISPATCHED
            alert.work_order_id = work_order_id
            alert.updated_at = datetime.now(timezone.utc)

            db.commit()
            db.refresh(alert)

            logger.info(f"告警已派单: {alert.alert_no} -> 工单 {work_order_id}")
            return alert

        finally:
            db.close()

    def delete_alert(self, alert_id: int) -> bool:
        """删除告警"""
        db = get_session()
        try:
            alert = db.query(Alert).filter(Alert.id == alert_id).first()
            if not alert:
                return False

            db.delete(alert)
            db.commit()

            logger.info(f"告警已删除: {alert.alert_no}")
            return True

        except Exception as e:
            db.rollback()
            logger.error(f"删除告警失败: {e}")
            return False
        finally:
            db.close()

    def get_statistics(self) -> dict:
        """获取告警统计"""
        db = get_session()
        try:
            total = db.query(Alert).count()
            pending = db.query(Alert).filter(Alert.status == AlertStatus.PENDING).count()
            confirmed = db.query(Alert).filter(Alert.status == AlertStatus.CONFIRMED).count()
            ignored = db.query(Alert).filter(Alert.status == AlertStatus.IGNORED).count()
            resolved = db.query(Alert).filter(Alert.status == AlertStatus.RESOLVED).count()
            dispatched = db.query(Alert).filter(Alert.status == AlertStatus.DISPATCHED).count()

            by_type = {}
            for alert in db.query(Alert.alert_type).distinct():
                alert_type = alert[0]
                by_type[alert_type] = db.query(Alert).filter(Alert.alert_type == alert_type).count()

            by_level = {}
            for level in AlertLevel:
                by_level[level.value] = db.query(Alert).filter(Alert.level == level).count()

            return {
                "total": total,
                "pending": pending,
                "confirmed": confirmed,
                "ignored": ignored,
                "resolved": resolved,
                "dispatched": dispatched,
                "by_type": by_type,
                "by_level": by_level,
            }
        finally:
            db.close()

    def update_ai_analysis(self, alert_id: int, analysis: dict) -> Optional[Alert]:
        """更新 AI 分析结果"""
        db = get_session()
        try:
            alert = db.query(Alert).filter(Alert.id == alert_id).first()
            if alert:
                alert.ai_analysis = analysis
                db.commit()
                db.refresh(alert)
            return alert
        finally:
            db.close()


# 全局单例
alert_service = AlertService()


def get_alert_service() -> AlertService:
    """获取告警服务单例"""
    return alert_service