ems-core/backend/app/api/v1/alarms.py

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, and_, String
from datetime import datetime, timezone
from pydantic import BaseModel
from app.core.database import get_db
from app.core.deps import get_current_user, require_roles
from app.models.alarm import AlarmRule, AlarmEvent
from app.models.user import User
from app.services.audit import log_audit

router = APIRouter(prefix="/alarms", tags=["告警管理"])


class AlarmRuleCreate(BaseModel):
    name: str
    device_id: int | None = None
    device_type: str | None = None
    data_type: str
    condition: str
    threshold: float | None = None
    threshold_high: float | None = None
    threshold_low: float | None = None
    duration: int = 0
    severity: str = "warning"
    notify_channels: list[str] | None = None
    notify_targets: list[str] | None = None
    silence_start: str | None = None
    silence_end: str | None = None


@router.get("/rules")
async def list_rules(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
    result = await db.execute(select(AlarmRule).order_by(AlarmRule.id.desc()))
    return [_rule_to_dict(r) for r in result.scalars().all()]


@router.post("/rules")
async def create_rule(data: AlarmRuleCreate, db: AsyncSession = Depends(get_db), user: User = Depends(require_roles("admin", "energy_manager"))):
    rule = AlarmRule(**data.model_dump(), created_by=user.id)
    db.add(rule)
    await db.flush()
    return _rule_to_dict(rule)


@router.put("/rules/{rule_id}")
async def update_rule(rule_id: int, data: AlarmRuleCreate, db: AsyncSession = Depends(get_db), user: User = Depends(require_roles("admin", "energy_manager"))):
    result = await db.execute(select(AlarmRule).where(AlarmRule.id == rule_id))
    rule = result.scalar_one_or_none()
    if not rule:
        raise HTTPException(status_code=404, detail="规则不存在")
    for k, v in data.model_dump(exclude_unset=True).items():
        setattr(rule, k, v)
    return _rule_to_dict(rule)


@router.delete("/rules/{rule_id}")
async def delete_rule(rule_id: int, db: AsyncSession = Depends(get_db), user: User = Depends(require_roles("admin", "energy_manager"))):
    result = await db.execute(select(AlarmRule).where(AlarmRule.id == rule_id))
    rule = result.scalar_one_or_none()
    if not rule:
        raise HTTPException(status_code=404, detail="规则不存在")
    rule.is_active = False
    return {"message": "已删除"}


@router.get("/events")
async def list_events(
    status: str | None = None,
    severity: str | None = None,
    device_id: int | None = None,
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    query = select(AlarmEvent)
    if status:
        query = query.where(AlarmEvent.status == status)
    if severity:
        query = query.where(AlarmEvent.severity == severity)
    if device_id:
        query = query.where(AlarmEvent.device_id == device_id)

    count_q = select(func.count()).select_from(query.subquery())
    total = (await db.execute(count_q)).scalar()

    query = query.order_by(AlarmEvent.triggered_at.desc()).offset((page - 1) * page_size).limit(page_size)
    result = await db.execute(query)
    return {
        "total": total,
        "items": [{
            "id": e.id, "rule_id": e.rule_id, "device_id": e.device_id, "severity": e.severity,
            "title": e.title, "description": e.description, "value": e.value, "threshold": e.threshold,
            "status": e.status, "triggered_at": str(e.triggered_at),
            "acknowledged_at": str(e.acknowledged_at) if e.acknowledged_at else None,
            "resolved_at": str(e.resolved_at) if e.resolved_at else None,
        } for e in result.scalars().all()]
    }


@router.post("/events/{event_id}/acknowledge")
async def acknowledge_event(event_id: int, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
    result = await db.execute(select(AlarmEvent).where(AlarmEvent.id == event_id))
    event = result.scalar_one_or_none()
    if not event:
        raise HTTPException(status_code=404, detail="告警不存在")
    event.status = "acknowledged"
    event.acknowledged_by = user.id
    event.acknowledged_at = datetime.now(timezone.utc)
    await log_audit(db, user.id, "acknowledge", "alarm", detail=f"确认告警 #{event_id}")
    return {"message": "已确认"}


@router.post("/events/{event_id}/resolve")
async def resolve_event(event_id: int, note: str = "", db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
    result = await db.execute(select(AlarmEvent).where(AlarmEvent.id == event_id))
    event = result.scalar_one_or_none()
    if not event:
        raise HTTPException(status_code=404, detail="告警不存在")
    event.status = "resolved"
    event.resolved_at = datetime.now(timezone.utc)
    event.resolve_note = note
    await log_audit(db, user.id, "resolve", "alarm", detail=f"解决告警 #{event_id}")
    return {"message": "已解决"}


@router.get("/stats")
async def alarm_stats(db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
    result = await db.execute(
        select(AlarmEvent.severity, AlarmEvent.status, func.count(AlarmEvent.id))
        .group_by(AlarmEvent.severity, AlarmEvent.status)
    )
    stats = {}
    for severity, status, count in result.all():
        if severity not in stats:
            stats[severity] = {}
        stats[severity][status] = count
    return stats


@router.get("/analytics")
async def get_alarm_analytics(
    start_date: str | None = None,
    end_date: str | None = None,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """告警分析 - Alarm trends and patterns"""
    from datetime import timedelta
    if not end_date:
        end_dt = datetime.now(timezone.utc)
    else:
        end_dt = datetime.fromisoformat(end_date)
    if not start_date:
        start_dt = end_dt - timedelta(days=30)
    else:
        start_dt = datetime.fromisoformat(start_date)

    # Daily alarm count by severity
    from app.core.config import get_settings
    settings = get_settings()
    if settings.is_sqlite:
        date_col = func.strftime('%Y-%m-%d', AlarmEvent.triggered_at).label('date')
    else:
        date_col = func.date_trunc('day', AlarmEvent.triggered_at).cast(String).label('date')

    query = select(
        date_col,
        AlarmEvent.severity,
        func.count(AlarmEvent.id).label('count'),
    ).where(
        and_(AlarmEvent.triggered_at >= start_dt, AlarmEvent.triggered_at <= end_dt)
    ).group_by('date', AlarmEvent.severity).order_by('date')

    result = await db.execute(query)
    rows = result.all()

    daily_map: dict[str, dict] = {}
    totals = {"critical": 0, "major": 0, "warning": 0}
    for date_val, severity, count in rows:
        d = str(date_val)[:10]
        if d not in daily_map:
            daily_map[d] = {"date": d, "critical": 0, "major": 0, "warning": 0}
        daily_map[d][severity] = count
        if severity in totals:
            totals[severity] += count

    return {
        "daily_trend": list(daily_map.values()),
        "totals": totals,
    }


@router.get("/top-devices")
async def get_top_alarm_devices(
    limit: int = 10,
    start_date: str | None = None,
    end_date: str | None = None,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """告警设备排名 - Devices with most alarms"""
    from app.models.device import Device

    query = select(
        AlarmEvent.device_id,
        Device.name.label('device_name'),
        func.count(AlarmEvent.id).label('alarm_count'),
        func.max(AlarmEvent.triggered_at).label('last_alarm_time'),
    ).join(Device, AlarmEvent.device_id == Device.id)

    if start_date:
        query = query.where(AlarmEvent.triggered_at >= start_date)
    if end_date:
        query = query.where(AlarmEvent.triggered_at <= end_date)

    query = query.group_by(AlarmEvent.device_id, Device.name).order_by(
        func.count(AlarmEvent.id).desc()
    ).limit(limit)

    result = await db.execute(query)
    return [{
        "device_id": r.device_id,
        "device_name": r.device_name,
        "alarm_count": r.alarm_count,
        "last_alarm_time": str(r.last_alarm_time) if r.last_alarm_time else None,
    } for r in result.all()]


@router.get("/mttr")
async def get_alarm_mttr(
    start_date: str | None = None,
    end_date: str | None = None,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """平均修复时间 - Mean Time To Resolve by severity"""
    from app.core.config import get_settings
    settings = get_settings()

    base_query = select(AlarmEvent).where(
        and_(AlarmEvent.status == "resolved", AlarmEvent.resolved_at.isnot(None))
    )
    if start_date:
        base_query = base_query.where(AlarmEvent.triggered_at >= start_date)
    if end_date:
        base_query = base_query.where(AlarmEvent.triggered_at <= end_date)

    result = await db.execute(base_query)
    events = result.scalars().all()

    mttr_data: dict[str, dict] = {}
    for e in events:
        if not e.resolved_at or not e.triggered_at:
            continue
        hours = (e.resolved_at - e.triggered_at).total_seconds() / 3600
        sev = e.severity
        if sev not in mttr_data:
            mttr_data[sev] = {"total_hours": 0, "count": 0}
        mttr_data[sev]["total_hours"] += hours
        mttr_data[sev]["count"] += 1

    return {
        sev: {
            "avg_hours": round(d["total_hours"] / d["count"], 2) if d["count"] > 0 else 0,
            "count": d["count"],
        }
        for sev, d in mttr_data.items()
    }


@router.put("/rules/{rule_id}/toggle")
async def toggle_rule(
    rule_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_roles("admin", "energy_manager")),
):
    """快速启用/禁用规则"""
    result = await db.execute(select(AlarmRule).where(AlarmRule.id == rule_id))
    rule = result.scalar_one_or_none()
    if not rule:
        raise HTTPException(status_code=404, detail="规则不存在")
    rule.is_active = not rule.is_active
    return {"id": rule.id, "is_active": rule.is_active}


@router.get("/rules/{rule_id}/history")
async def get_rule_history(
    rule_id: int,
    page: int = 1,
    page_size: int = 20,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """规则触发历史"""
    query = select(AlarmEvent).where(AlarmEvent.rule_id == rule_id)
    count_q = select(func.count()).select_from(query.subquery())
    total = (await db.execute(count_q)).scalar()

    query = query.order_by(AlarmEvent.triggered_at.desc()).offset((page - 1) * page_size).limit(page_size)
    result = await db.execute(query)
    return {
        "total": total,
        "items": [{
            "id": e.id, "device_id": e.device_id, "severity": e.severity,
            "title": e.title, "value": e.value, "threshold": e.threshold,
            "status": e.status, "triggered_at": str(e.triggered_at),
            "resolved_at": str(e.resolved_at) if e.resolved_at else None,
        } for e in result.scalars().all()]
    }


def _rule_to_dict(r: AlarmRule) -> dict:
    return {
        "id": r.id, "name": r.name, "device_id": r.device_id, "device_type": r.device_type,
        "data_type": r.data_type, "condition": r.condition, "threshold": r.threshold,
        "threshold_high": r.threshold_high, "threshold_low": r.threshold_low,
        "duration": r.duration, "severity": r.severity, "is_active": r.is_active,
        "notify_channels": r.notify_channels, "silence_start": r.silence_start, "silence_end": r.silence_end,
    }