#!/usr/local/bin/python3 import pandas as pd import json import os # 需执行 pip3 install docopt from docopt import docopt # 定义列映射 column_mapping = { "类型": "type", "厂商": "vendor", "型号": "model", "Tag": "tag", "告警编码": "alarmCode", "告警内容": "alarmInfo", "告警类别": "categoryDisplay", "告警级别": "severityDisplay", "实例类型": "entityTypeDisplay", "告警原因编码": "reasonCode", "告警原因": "reason", "告警附加信息": "additionalInfo", "告警附加信息参数": "additionalInfoParas", "排查方案": "solution" } # 定义列映射 column_mapping_en = { "Alarm code": "alarmCode", "Alarm content": "alarmInfo", "Alarm category": "categoryDisplay", "Alarm level": "severityDisplay", "Instance type": "entityTypeDisplay", "Alarm reason code": "reasonCode", "Alarm reason": "reason", "Alarm additional information": "additionalInfo", "Alarm additional information parameter list" : "additionalInfoParas", "Troubleshooting plan": "solution" } # 定义实例类型映射 entity_type_mapping = { '液冷机' :'LIQUID_COOL', '空调':'AIRCOND', '其他设备':'OTHERS', 'EMS控制器':'EMS', 'LC控制器':'LC', 'UPS':'UPS', '温度计':'HUMITURE', '电池簇':'BMS', '双向逆变器':'PCS', 'IO设备':'DIDO', '电表':'METER' } # 定义模板类型 template_type = { "EMS": "EMS", "LC": "LC", "PCS": "PCS", "BMS": "BMS", "CELL":"CELL", "DIDO": "DIDO", "OTHERS": "OTHERS", "METER": "METER", "UPS": "OTHERS", "AIRCOND": "AC", "LIQUID_COOL": "TMS", "HUMITURE":"HUMITURE", "PV": "PV", "MPPT": "MPPT", "TMS": "TMS", } # 定义告警级别 alarm_severity_g = { "清除":"-1", "提示":"0", "一般":"1", "重要":"2", "严重":"3" } # 定义告警状态 alarm_state_g = { "-1":"清除", "0":"新增", "1":"确认" } # 定义告警类别 alarm_category_g = { "通讯":"Communication", "自定义":"Cutomized", "设备":"Equipments", "资源":"Resources", "业务":"Services" } def read_json(file_path): """读取 JSON 文件""" with open(file_path, "r", encoding="utf-8") as file: data = json.load(file) return data def write_json(data, file_path): """写入 JSON 文件""" with open(file_path, "w", encoding="utf-8") as file: json.dump(data, file, indent=4, ensure_ascii=False) def process_alarm(alarm): """处理单个告警信息""" for key, value in alarm.items(): if value is None: alarm[key] = "" if isinstance(value, str): alarm[key] = value.rstrip() return alarm vendor_models = {} def excel_to_json(in_dir, out_dir, zh_file): # 读取 Excel 文件 if not os.path.exists(os.path.join(in_dir, zh_file)): print(f"文件 {zh_file} 不存在") return False wb = pd.read_excel(os.path.join(in_dir, zh_file)) df = wb.iloc[:, :15] df.rename(columns=column_mapping, inplace=True) # 筛选 AlarmCode 为数值的行 df = df[pd.to_numeric(df['alarmCode'], errors='coerce').notnull()] df['alarmCode'] = df['alarmCode'].astype(int) df["reasonCode"] = df["reasonCode"].astype(int) # 转换为 JSON astr = df.to_json(orient="records") alarms = json.loads(astr) # 处理告警信息 alarms = [process_alarm(alarm) for alarm in alarms] # 组织数据 result = [] for alarm in alarms: if alarm["type"] in template_type: type = template_type[alarm["type"]] else: type = alarm["type"] vendor = alarm["vendor"] model = alarm["model"] tag = alarm["tag"] del alarm["type"] del alarm["vendor"] del alarm["model"] res = next((r for r in result if r["type"] == type and r["vendor"] == vendor and r["model"] == model), None) if res is None: res = {"type": type, "vendor": vendor, "model": model, "tags": []} result.append(res) res["tags"].append(alarm) # 写入 JSON 文件 vendor_models["lang"] = "zh_CN" vendor_models["vendor_models"] = result out_path = os.path.join(out_dir, "alarminfo", "zh_CN", "alarm_def.json") os.makedirs(os.path.dirname(out_path), exist_ok=True) write_json(vendor_models, out_path) print (f"zh_CN/alarm_def.json 写入 {out_path}") return True def make_english(in_dir, out_dir, en_file): # 读取 Excel 文件 if not os.path.exists(os.path.join(in_dir, en_file)): print(f"文件 {en_file} 不存在") return wb = pd.read_excel(os.path.join(in_dir, en_file)) df = wb.iloc[:, :10] df.rename(columns=column_mapping_en, inplace=True) # 筛选 AlarmCode 和 ReasonCode 为数值的行 df = df[pd.to_numeric(df['alarmCode'], errors='coerce').notnull()] df = df[pd.to_numeric(df['reasonCode'], errors='coerce').notnull()] df['alarmCode'] = df['alarmCode'].astype(int) df["reasonCode"] = df["reasonCode"].astype(int) # 转换为 JSON astr = df.to_json(orient="records") alarms = json.loads(astr) # 创建告警哈希表,null 的值改为"" alarm_hash = {} for alarm in alarms: for key, value in alarm.items(): if value is None: alarm[key] = "" if isinstance(value, str): alarm[key] = value.rstrip() alarm_hash[f"{alarm['alarmCode']}-{alarm['reasonCode']}"] = alarm # 更新告警信息 for vm in vendor_models["vendor_models"]: vm["tags"] = [tag for tag in vm["tags"] if f"{tag['alarmCode']}-{tag['reasonCode']}" in alarm_hash] for tag in vm["tags"]: key = f"{tag['alarmCode']}-{tag['reasonCode']}" tag.update(alarm_hash[key]) # 写入 JSON 文件 out_path = os.path.join(out_dir, "alarminfo", "en_US", "alarm_def.json") os.makedirs(os.path.dirname(out_path), exist_ok=True) vendor_models["lang"] = "en_US" write_json(vendor_models, out_path) print (f"en_US/alarm_def.json 写入 {out_path}") def make_old_json(out_dir): pcs = {} bms = {} pcs_vendor_models = [] bms_vendor_models = [] for vm in vendor_models["vendor_models"]: old_vm = {"vendor": vm["vendor"], "model": vm["model"], "tags": []} for tag in vm["tags"]: if tag["tag"] == "" : continue severity = "EMMS2_SEVERITY_CLEAR" if tag["severityDisplay"] == "提示": severity = "EMMS2_SEVERITY_WARNING" elif tag["severityDisplay"] == "一般": severity = "EMMS2_SEVERITY_MINOR" elif tag["severityDisplay"] == "重要": severity = "EMMS2_SEVERITY_MAJOR" elif tag["severityDisplay"] == "严重": severity = "EMMS2_SEVERITY_CRITICAL" old_vm["tags"].append({ "name": tag["tag"], "tag_type":"EMMS2_TAG_TYPE_ALARM", "deadZone":-1, "list_alarm_info":{ "alarmCode": tag["alarmCode"], "reasonCode":tag["reasonCode"], "severity":severity, "condition":"CONDITION_GE", "value":1, "valid": True } }) if vm["type"] == "PCS": pcs_vendor_models.append(old_vm) elif vm["type"] == "BMS": bms_vendor_models.append(old_vm) pcs["vendor_models"] = pcs_vendor_models bms["vendor_models"] = bms_vendor_models out_dir = os.path.dirname(os.path.join(out_dir, "alarminfo", "old", "emms2_alarm_def.json")) os.makedirs(out_dir, exist_ok=True) write_json(pcs, os.path.join(out_dir, "emms2_alarm_mapping_PCS.json")) print (f"emms2_alarm_mapping_PCS.json 写入 {out_dir}") write_json(bms, os.path.join(out_dir, "emms2_alarm_mapping_BMS.json")) print (f"emms2_alarm_mapping_BMS.json 写入 {out_dir}") __doc__ = """Usage: ./alarm_excel_to_json.py INPUT_DIR OUTPUT_DIR [--zh FILE] [--en FILE] [--old] Options: INPUT_DIR 输入文件夹路径 OUTPUT_DIR 输出文件夹路径 --zh FILE 中文告警信息文件名 --en FILE 英文告警信息文件名 --old 生成旧版的json文件 如不指定中文告警信息文件名,则默认为 'EMS告警定义模板-导入(v2.0).xlsx' 如不指定英文告警信息文件名,则默认为 'EMS2.0日志模板-导入_en.xlsx' """ def main(): args = docopt(__doc__) in_dir = args["INPUT_DIR"] out_dir = args["OUTPUT_DIR"] zh_file = args["--zh"] or "EMS告警定义模板-导入(v2.0).xlsx" en_file = args["--en"] or "EMS告警定义模板-导入_en.xlsx" if excel_to_json(in_dir, out_dir, zh_file): make_english(in_dir, out_dir, en_file); if args["--old"]: make_old_json(out_dir) if __name__ == "__main__": main()