vent_skills/vent-device-daily-report/scripts/device_online_report.py

# -*- coding: utf-8 -*-
"""
设备设施在线率统计日报生成器（通防管控平台）

功能：按设备类型（主通风机/局部通风机/风门/风窗/各类传感器/其他）统计在线情况，
     生成当天设备在线日报（Markdown + JSON），并执行告警规则：
       - 主通风机：有 1 套不在线即红色报警
       - 其余类型：在线率 < 90% 橙色报警

输入（JSON 文件，数据来源为 tf_mcp 的 execute_sql_query 查询结果）：
  --devices  设备明细（含分站连接状态），字段：ID/strName/deviceKind/strType/
             nSubStationID/StationName/monitorFlag/testFlag/linkStatus/isUse
  --fan      可选，主通风机开机状态（fan 表），字段：nFanID/strName/bOn
  --aigate   可选，智能风门状态（ai_gate 表），字段：name/status(1正常0离线2故障)
输出：
  report.md / report.json （写入 --outdir，默认当前目录）

用法示例：
  python device_online_report.py --devices devices.json --outdir ./report
  python device_online_report.py --devices devices.json --fan fan.json --aigate aigate.json --outdir ./report

本脚本仅依赖 Python 标准库，无第三方依赖。
"""

import argparse
import json
import os
from datetime import datetime

# ============================================================
# 一、类型分组配置（可覆盖：通过 --groups 传入自定义 JSON 文件）
#     键为日报分组名，值为该组包含的 deviceKind 编码列表
# ============================================================
DEFAULT_GROUPS = {
    "主通风机": ["fanmain"],
    "局部通风机": ["fanlocal"],
    "风门": ["gate"],
    "风窗": ["window"],
    "传感器": [
        "modelsensor", "safetymonitor", "gasmonitor", "gas",
        "gaspatrol", "dusting", "fiber", "bundletube", "windrect",
        "atomizing", "duaSpeCamera", "imgFireDet",
    ],
    # 其余 deviceKind 自动归入"其他"
}

# 告警阈值（可覆盖）
DEFAULT_THRESHOLD = 0.90       # 其余类型在线率下限
DEFAULT_FANMAIN_OFFLINE_TOL = 0  # 主通风机允许离线台数（0 = 1台离线即报警）


def load_json(path):
    """读取 JSON 文件；支持 tf_mcp 返回的 {result: [...]} 包装结构。"""
    if not path:
        return []
    with open(path, "r", encoding="utf-8") as f:
        data = json.load(f)
    if isinstance(data, dict):
        # 兼容 {success/code/result:[...]} 或 {result:[...]}
        for key in ("result", "data"):
            if key in data and isinstance(data[key], list):
                return data[key]
        return []
    return data


def norm(s):
    """规范化字符串用于匹配（去空白）。"""
    return "" if s is None else str(s).strip()


def get(d, key, default=None):
    v = d.get(key, default)
    return v


# ============================================================
# 二、在线判定（分类型混合判定，优先级链，命中即止）
# ============================================================
def judge_device(dev, fan_map, aigate_map, judge_mode="strict"):
    """
    判定单台设备在线状态，返回 (state, reason, detail)
      state: "online" / "offline" / "unknown"
    """
    kind = norm(dev.get("deviceKind"))
    name = norm(dev.get("strName"))
    monitor_flag = get(dev, "monitorFlag")
    test_flag = get(dev, "testFlag")
    link_status = get(dev, "linkStatus")
    is_use = get(dev, "isUse")
    station_id = get(dev, "nSubStationID")

    # 规则 1：主通风机 —— fan 表开机状态（bOn=1 在线 / 0 离线 / NULL 继续）
    if kind == "fanmain" and fan_map:
        b_on = fan_map.get(name)
        if b_on is not None:
            if int(b_on) == 1:
                return "online", "fan表bOn=1(开机)", "fan.bOn"
            return "offline", f"fan表bOn={b_on}(停机)", "fan.bOn"

    # 规则 2：风门 —— ai_gate 表状态（按名称匹配，status=1 在线 / 0/2 离线 / 无匹配继续）
    if kind == "gate" and aigate_map:
        st = aigate_map.get(name)
        if st is not None:
            if int(st) == 1:
                return "online", "ai_gate状态正常", "ai_gate.status"
            return "offline", f"ai_gate状态={st}(离线/故障)", "ai_gate.status"

    # 规则 3：通用主判定 —— 关联分站连接状态
    #   分站在线 = linkStatus=1 且 isUse=1（开启监控）
    if station_id is not None and str(station_id) != "":
        if link_status is None and is_use is None:
            # 分站 ID 在 sub_station 中不存在
            return "unknown", "关联分站不存在于sub_station", "station"
        if int(link_status or 0) == 1 and int(is_use or 0) == 1:
            return "online", "关联分站在线(linkStatus=1)", "station"
        return "offline", f"关联分站离线(linkStatus={link_status},isUse={is_use})", "station"

    # 设备未关联分站
    return "unknown", "未关联分站", "station"


# ============================================================
# 三、统计与告警
# ============================================================
def build_groups(custom_groups=None):
    groups = dict(DEFAULT_GROUPS)
    if custom_groups:
        groups.update(custom_groups)
    return groups


def group_of(kind, groups):
    for gname, kinds in groups.items():
        if kind in kinds:
            return gname
    return "其他"


def aggregate(devices, fan_map, aigate_map, groups, threshold, fan_tol):
    """统计各分组：总数/在线/离线/未知/在线率/明细。"""
    results = {}
    for gname in list(groups.keys()) + ["其他"]:
        results[gname] = {
            "total": 0, "online": 0, "offline": 0, "unknown": 0,
            "sim_cnt": 0, "online_rate": None,
            "online_list": [], "offline_list": [], "unknown_list": [],
        }

    for dev in devices:
        kind = norm(dev.get("deviceKind"))
        gname = group_of(kind, groups)
        r = results[gname]
        state, reason, detail = judge_device(dev, fan_map, aigate_map)
        item = {
            "id": get(dev, "ID"),
            "name": norm(dev.get("strName")),
            "deviceKind": kind,
            "strType": norm(dev.get("strType")),
            "stationName": norm(dev.get("StationName")),
            "monitorFlag": get(dev, "monitorFlag"),
            "testFlag": get(dev, "testFlag"),
            "state": state,
            "reason": reason,
            "judge_source": detail,
        }
        r["total"] += 1
        if int(get(dev, "testFlag") or 0) == 1:
            r["sim_cnt"] += 1
        if state == "online":
            r["online"] += 1
            r["online_list"].append(item)
        elif state == "offline":
            r["offline"] += 1
            r["offline_list"].append(item)
        else:
            r["unknown"] += 1
            r["unknown_list"].append(item)

    # 在线率：分子=在线数；分母=正常设备（total），unknown 不计入分母并单独说明
    for r in results.values():
        denom = r["total"] - r["unknown"]
        if denom > 0:
            r["online_rate"] = round(r["online"] / denom, 4)
    return results


def judge_alarms(results, threshold, fan_tol):
    """告警规则：
    - 主通风机：offline/unknown 超过容忍台数 -> 红色报警
    - 其他分组：在线率 < threshold -> 橙色报警
    """
    alarms = []
    fan = results.get("主通风机", {})
    bad = (fan.get("offline", 0) or 0) + (fan.get("unknown", 0) or 0)
    if bad > fan_tol:
        alarms.append({
            "level": "RED",
            "group": "主通风机",
            "message": f"主通风机有 {bad} 套不在线（在线 {fan.get('online',0)}/{fan.get('total',0)}），须立即排查！",
        })
    for gname, r in results.items():
        if gname == "主通风机":
            continue
        rate = r.get("online_rate")
        if rate is not None and rate < threshold:
            alarms.append({
                "level": "ORANGE",
                "group": gname,
                "message": f"{gname}在线率 {rate*100:.1f}% 低于 {threshold*100:.0f}%（在线 {r['online']}/{r['total']}，离线 {r['offline']}，未知 {r['unknown']}）",
            })
    if not alarms:
        alarms.append({"level": "GREEN", "group": "-", "message": "全部设备在线率达标"})
    return alarms


# ============================================================
# 四、输出渲染
# ============================================================
def render_markdown(report_date, results, alarms, threshold, fan_tol, total_summary):
    lines = []
    lines.append(f"# 设备设施在线情况日报（{report_date}）")
    lines.append("")
    lines.append("## 一、总体概况")
    lines.append("")
    lines.append(f"- 正常接入设备：**{total_summary['total']}** 台")
    lines.append(f"- 在线：**{total_summary['online']}** 台（{total_summary['online_rate']*100:.1f}%）")
    lines.append(f"- 离线：**{total_summary['offline']}** 台")
    lines.append(f"- 状态未知（未关联分站/无法判定）：**{total_summary['unknown']}** 台（不计入在线率分母）")
    lines.append(f"- 模拟数据设备：**{total_summary['sim']}** 台（testFlag=1，仅供测试参考）")
    lines.append("")
    lines.append("## 二、分类型在线统计")
    lines.append("")
    lines.append("| 设备类型 | 总数 | 在线 | 离线 | 未知 | 在线率 | 告警 |")
    lines.append("|---|---|---|---|---|---|---|")
    order = list(results.keys())
    for gname in order:
        r = results[gname]
        rate = f"{r['online_rate']*100:.1f}%" if r["online_rate"] is not None else "-"
        lvl = next((a["level"] for a in alarms if a["group"] == gname), "")
        flag = {"RED": "🔴", "ORANGE": "🟠", "GREEN": "🟢"}.get(lvl, "")
        lines.append(f"| {gname} | {r['total']} | {r['online']} | {r['offline']} | {r['unknown']} | {rate} | {flag} |")
    lines.append("")
    lines.append("## 三、告警信息")
    lines.append("")
    for a in alarms:
        icon = {"RED": "🔴", "ORANGE": "🟠", "GREEN": "🟢"}[a["level"]]
        lines.append(f"- {icon} **[{a['level']}]** {a['message']}")
    lines.append("")
    lines.append("## 四、离线/异常设备明细")
    lines.append("")
    for gname in order:
        r = results[gname]
        if not r["offline_list"] and not r["unknown_list"]:
            continue
        lines.append(f"### {gname}（离线 {r['offline']} / 未知 {r['unknown']}）")
        lines.append("")
        lines.append("| 设备名称 | 类型 | 分站 | 状态 | 原因 | 模拟 |")
        lines.append("|---|---|---|---|---|---|")
        for item in r["offline_list"] + r["unknown_list"]:
            state_cn = "离线" if item["state"] == "offline" else "未知"
            sim = "是" if item["testFlag"] == 1 else "否"
            lines.append(f"| {item['name']} | {item['strType']} | {item['stationName'] or '-'} | {state_cn} | {item['reason']} | {sim} |")
        lines.append("")
    lines.append("---")
    lines.append(f"*统计口径：分母为正常接入设备（status=1）；在线率 = 在线 / (总数 - 未知)；"
                 f"其余类型在线率阈值 {threshold*100:.0f}%；主通风机离线 {fan_tol} 台即报警*")
    return "\n".join(lines)


def main():
    parser = argparse.ArgumentParser(description="设备设施在线率统计日报生成器")
    parser.add_argument("--devices", required=True, help="设备明细 JSON（tf_mcp execute_sql_query 结果）")
    parser.add_argument("--fan", default=None, help="可选：fan 表开机状态 JSON")
    parser.add_argument("--aigate", default=None, help="可选：ai_gate 风门状态 JSON")
    parser.add_argument("--groups", default=None, help="可选：自定义分组 JSON {分组名:[deviceKind,...]}")
    parser.add_argument("--threshold", type=float, default=DEFAULT_THRESHOLD, help="在线率阈值，默认0.90")
    parser.add_argument("--fan-tol", type=int, default=DEFAULT_FANMAIN_OFFLINE_TOL, help="主通风机容忍离线台数，默认0")
    parser.add_argument("--outdir", default=".", help="输出目录，默认当前目录")
    parser.add_argument("--date", default=None, help="日报日期，默认当天")
    args = parser.parse_args()

    devices = load_json(args.devices)
    fan_map = {}
    for row in load_json(args.fan):
        fan_map.setdefault(norm(row.get("strName")), row.get("bOn"))
    aigate_map = {}
    for row in load_json(args.aigate):
        aigate_map.setdefault(norm(row.get("name")), row.get("status"))

    groups = build_groups(load_json(args.groups) if args.groups else None)
    results = aggregate(devices, fan_map, aigate_map, groups, args.threshold, args.fan_tol)
    alarms = judge_alarms(results, args.threshold, args.fan_tol)

    total = {"total": 0, "online": 0, "offline": 0, "unknown": 0, "sim": 0}
    for r in results.values():
        total["total"] += r["total"]
        total["online"] += r["online"]
        total["offline"] += r["offline"]
        total["unknown"] += r["unknown"]
        total["sim"] += r["sim_cnt"]
    total["online_rate"] = round(total["online"] / (total["total"] - total["unknown"]), 4) \
        if (total["total"] - total["unknown"]) > 0 else None

    report_date = args.date or datetime.now().strftime("%Y-%m-%d")
    os.makedirs(args.outdir, exist_ok=True)

    md = render_markdown(report_date, results, alarms, args.threshold, args.fan_tol, total)
    with open(os.path.join(args.outdir, "report.md"), "w", encoding="utf-8") as f:
        f.write(md)

    report = {
        "report_date": report_date,
        "generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "summary": total,
        "alarms": alarms,
        "threshold": args.threshold,
        "fanmain_offline_tolerance": args.fan_tol,
        "groups": groups,
        "detail": results,
    }
    with open(os.path.join(args.outdir, "report.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)

    print(f"✅ 已生成日报：{os.path.join(args.outdir, 'report.md')} / report.json")
    for a in alarms:
        print(f"  [{a['level']}] {a['message']}")


if __name__ == "__main__":
    main()