# -*- coding: utf-8 -*- """ 短信报警服务 - 液位差异超过阈值报警 - 液位过低/过高报警 - 温度超出范围报警 - 密度超出范围报警 - 当日流入流出累计不平衡超过阈值报警 - 支持多人分组发送，每个分组可以选择接收哪些类型的报警 - 支持每个罐体单独配置阈值 """ import sys from datetime import datetime, date from config import SMS, ALERT # 阿里云短信SDK try: from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_dysmsapi20170525 import models as dysmsapi_models ALIBABA_CLOUD_AVAILABLE = True except ImportError: ALIBABA_CLOUD_AVAILABLE = False print("阿里云短信SDK未安装，短信功能不可用", file=sys.stderr) _sms_client = None _last_alert_time = None # 记录上次报警时间，避免频繁报警 _daily_flow_accumulator = {} # 累计当日流量 {tank_id: {'in': xxx, 'out': xxx}} def init_client(): """初始化短信客户端""" global _sms_client if not ALIBABA_CLOUD_AVAILABLE: return False if not SMS['access_key_id'] or SMS['access_key_id'] == 'your-access-key-id': print("短信配置未填写，短信功能不可用", file=sys.stderr) return False try: config = open_api_models.Config( access_key_id=SMS['access_key_id'], access_key_secret=SMS['access_key_secret'] ) config.endpoint = 'dysmsapi.aliyuncs.com' _sms_client = Dysmsapi20170525Client(config) return True except Exception as e: print(f"初始化短信客户端失败: {str(e)}", file=sys.stderr) return False def send_alert_to_groups(alert_type, alert_name, content): """发送报警短信给所有匹配分组（分组可以过滤报警类型）""" if not ALERT['enabled']: print(f"报警已禁用，[{alert_name}] {content}") return True success = True for group in ALERT['groups']: if not group['enabled']: continue # 分组可以过滤报警类型，只接收指定类型 if 'alarm_types' in group and alert_type not in group['alarm_types']: continue # 这个分组不接收这种报警，跳过 for phone in group['phones']: result = send_single_alert(phone, alert_name, content) if not result: success = False return success def send_single_alert(phone, alert_type, content): """给单个手机号发送报警""" if not _sms_client and not init_client(): print(f"短信客户端初始化失败，无法发送给 {phone}: [{alert_type}] {content}", file=sys.stderr) return False try: template_param = { "type": alert_type, "content": content, "time": datetime.now().strftime('%Y-%m-%d %H:%M:%S') } import json request = dysmsapi_models.SendSmsRequest( phone_numbers=phone, sign_name=SMS['sign_name'], template_code=SMS['template_code'], template_param=json.dumps(template_param) ) response = _sms_client.send_sms(request) if response.body.code == 'OK': print(f"[{datetime.now()}] 短信发送成功: {phone} [{alert_type}] {content}") return True else: print(f"短信发送失败 {phone}: {response.body.message}", file=sys.stderr) return False except Exception as e: print(f"短信发送异常 {phone}: {str(e)}", file=sys.stderr) return False def check_level_alerts(tank_id, tank_config, current_level): """ 检查液位报警 返回报警列表 [(报警类型, 报警内容), ...] """ alerts = [] level_cfg = ALERT['level'] # 获取阈值，优先用罐体单独配置，否则用默认 low_th = tank_config.get('alert', {}).get('low_level', level_cfg['low_threshold_default']) high_th = tank_config.get('alert', {}).get('high_level', level_cfg['high_threshold_default']) diff_th = tank_config.get('alert', {}).get('diff_threshold', level_cfg['diff_threshold_default']) # 液位过低报警 if current_level <= low_th: alerts.append(( 'level_low', f'LNG储罐{tank_id} 液位过低: {current_level:.2f}m，阈值下限 {low_th:.1f}m' )) # 液位过高报警 if current_level >= high_th: alerts.append(( 'level_high', f'LNG储罐{tank_id} 液位过高: {current_level:.2f}m，阈值上限 {high_th:.1f}m' )) return alerts def check_temperature_alert(tank_id, tank_config, current_temp): """检查温度报警""" if not ALERT['temperature']['enabled']: return [] temp_cfg = ALERT['temperature'] low_th = tank_config.get('alert', {}).get('low_temp', temp_cfg['low_threshold_default']) high_th = tank_config.get('alert', {}).get('high_temp', temp_cfg['high_threshold_default']) alerts = [] if current_temp <= low_th: alerts.append(( 'temperature_low', f'LNG储罐{tank_id} 温度过低: {current_temp:.1f}℃，阈值下限 {low_th:.1f}℃' )) if current_temp >= high_th: alerts.append(( 'temperature_high', f'LNG储罐{tank_id} 温度过高: {current_temp:.1f}℃，阈值上限 {high_th:.1f}℃' )) return alerts def check_density_alert(tank_id, tank_config, current_density): """检查密度报警""" if not ALERT['density']['enabled']: return [] dens_cfg = ALERT['density'] low_th = tank_config.get('alert', {}).get('low_density', dens_cfg['low_threshold_default']) high_th = tank_config.get('alert', {}).get('high_density', dens_cfg['high_threshold_default']) alerts = [] if current_density <= low_th: alerts.append(( 'density_low', f'LNG储罐{tank_id} 密度过低: {current_density:.3f}，阈值下限 {low_th:.3f}' )) if current_density >= high_th: alerts.append(( 'density_high', f'LNG储罐{tank_id} 密度过高: {current_density:.3f}，阈值上限 {high_th:.3f}' )) return alerts def send_level_diff_alert(tank_id, diff): """液位差异报警（保留兼容）""" alert_type = "level_diff" alert_name = "液位差异超标" content = f"LNG储罐{ tank_id }液位差异超标，差异值: { diff:.2f}m，请立即处理" return send_alert_to_groups(alert_type, alert_name, content) def check_daily_flow_balance(all_data, all_tanks): """ 检查当日流量平衡 all_data: {tank_id: {flow_in, flow_out, ...}} all_tanks: 罐体配置列表，用于获取每个罐体单独配置 返回报警列表 [(报警类型, 报警内容), ...] """ global _daily_flow_accumulator today = date.today() # 如果是新的一天，重置累计 if 'date' not in _daily_flow_accumulator or _daily_flow_accumulator['date'] != today: _daily_flow_accumulator = { 'date': today, 'total_in': 0.0, 'total_out': 0.0 } # 累加当日总流入总流出 for tank_id_str, data in all_data.items(): tank_id = int(tank_id_str) if data.get('flow_in') is not None: _daily_flow_accumulator['total_in'] += data['flow_in'] if data.get('flow_out') is not None: _daily_flow_accumulator['total_out'] += data['flow_out'] total_in = _daily_flow_accumulator['total_in'] total_out = _daily_flow_accumulator['total_out'] diff = abs(total_in - total_out) print(f"[{datetime.now()}] 当日流量累计: 流入 {total_in:.2f}m³, 流出 {total_out:.2f}m³, 差异 {diff:.2f}m³") if not ALERT['daily_flow']['enabled']: return [] threshold = ALERT['daily_flow'].get('diff_threshold', 100.0) if diff > threshold: return [( 'daily_flow_diff', f'当日流量不平衡: 累计流入 {total_in:.2f}m³，累计流出 {total_out:.2f}m³，差异 {diff:.2f}m³，超过阈值 {threshold:.1f}m³，请核查' )] else: return [] def send_alerts(alerts): """批量发送报警，每个报警单独发""" success = True for alert_type, content in alerts: # 报警类型对应显示名称 type_names = { 'level_low': '液位过低报警', 'level_high': '液位过高报警', 'level_diff': '液位差异超标', 'temperature_low': '温度过低报警', 'temperature_high': '温度过高报警', 'density_low': '密度过低报警', 'density_high': '密度过高报警', 'daily_flow_diff': '流量不平衡报警' } alert_name = type_names.get(alert_type, alert_type) result = send_alert_to_groups(alert_type, alert_name, content) if not result: success = False return success def should_check_alert(): """判断是否需要检查报警（避免频繁报警）""" global _last_alert_time now = datetime.now() if _last_alert_time is None: _last_alert_time = now return True interval_minutes = ALERT.get('check_interval_minutes', 60) delta = (now - _last_alert_time).total_seconds() / 60 if delta >= interval_minutes: _last_alert_time = now return True else: return False def reset_daily_accumulator(): """重置当日累计（可以在每天零点自动调用）""" global _daily_flow_accumulator _daily_flow_accumulator = { 'date': date.today(), 'total_in': 0.0, 'total_out': 0.0 }