Compressor/3d/backend/services/alarm_service.py

"""压缩机告警（compressor_alarm）查询，仅限压力/振动两类。"""

import json
from typing import Any

from db import connect

ALARM_TYPES = ("振动故障", "润滑油压力故障")


def _parse_info(raw: Any) -> dict[str, Any]:
    if not raw:
        return {}
    try:
        value = json.loads(raw)
    except (TypeError, ValueError):
        return {}
    return value if isinstance(value, dict) else {}


def list_alarms(
    device_code: str | None = None,
    alarm_type: str | None = None,
    page: int = 1,
    page_size: int = 10,
) -> dict[str, Any]:
    where = ["alarm_type IN (%s, %s)"]
    params: list[Any] = list(ALARM_TYPES)
    if device_code:
        where.append("device_code = %s")
        params.append(device_code)
    if alarm_type:
        where.append("alarm_type = %s")
        params.append(alarm_type)
    clause = " AND ".join(where)

    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(f"SELECT COUNT(*) AS c FROM compressor_alarm WHERE {clause}", params)
            total = int(cursor.fetchone()["c"])
            offset = (page - 1) * page_size
            cursor.execute(
                f"""
                SELECT id, device_code, device_part, device_point, alarm_type, alarm_level,
                       alarm_time_start, alarm_time_end, alarm_des, alarm_info_json
                FROM compressor_alarm
                WHERE {clause}
                ORDER BY alarm_time_start DESC, id DESC
                LIMIT %s OFFSET %s
                """,
                params + [page_size, offset],
            )
            rows = cursor.fetchall()
    finally:
        connection.close()

    items = [
        {
            "id": int(row["id"]),
            "device_code": row["device_code"],
            "device_part": row["device_part"],
            "device_point": row["device_point"],
            "alarm_type": row["alarm_type"],
            "alarm_level": row["alarm_level"],
            "alarm_time_start": row["alarm_time_start"],
            "alarm_time_end": row["alarm_time_end"],
            "alarm_des": row["alarm_des"],
            "alarm_info": _parse_info(row["alarm_info_json"]),
        }
        for row in rows
    ]
    return {"total": total, "page": page, "page_size": page_size, "items": items}


def alarm_filters() -> dict[str, Any]:
    connection = connect()
    try:
        with connection.cursor() as cursor:
            cursor.execute(
                """
                SELECT DISTINCT device_code FROM compressor_alarm
                WHERE alarm_type IN (%s, %s) AND device_code IS NOT NULL
                ORDER BY device_code
                """,
                ALARM_TYPES,
            )
            devices = [row["device_code"] for row in cursor.fetchall()]
    finally:
        connection.close()
    return {"devices": devices, "alarm_types": list(ALARM_TYPES)}