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

import uuid
from datetime import datetime, timezone
from typing import Optional, List

from app.models import Alert, AlertStatus, 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:
        db = get_session()
        try:
            alert = Alert(
                alert_no=self.generate_alert_no(),
                camera_id=alert_data.camera_id,
                roi_id=alert_data.roi_id,
                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,
                status=AlertStatus.PENDING,
            )

            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 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_alerts(
        self,
        camera_id: Optional[str] = None,
        alert_type: Optional[str] = None,
        status: 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 alert_type:
                query = query.filter(Alert.alert_type == alert_type)
            if status:
                query = query.filter(Alert.status == status)
            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 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()

            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()

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

    def update_ai_analysis(self, alert_id: int, analysis: dict) -> Optional[Alert]:
        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