RailWise KB已发布

监测预警通知模板

RailWise监测预警通知标准化代码模板，支持多通道预警推送（邮件/短信/钉钉/企业微信/飞书），实现预警分级、通知模板、确认机制与升级策略的完整流程。

复核 2026-07-09入门公开可引用RailWise 技术团队

template-doc

监测预警通知模板

1. 模板概述

1.1 用途

本模板提供监测预警从检测到通知的完整自动化流程，支持：

多渠道推送：邮件、短信、钉钉、企业微信、飞书、Webhook
分级预警：正常、黄色预警、橙色预警、红色预警四级体系
通知模板：不同预警等级对应不同通知内容和接收人
确认机制：通知发送后要求接收人确认，未确认自动升级
升级策略：超时未确认自动升级通知等级和接收人范围

1.2 适用场景

场景	预警等级	通知方式	响应时间
日常监测	正常	日报汇总	24小时
轻微超限	黄色预警	钉钉/企业微信	30分钟
明显超限	橙色预警	短信+钉钉	15分钟
严重超限	红色预警	电话+短信+全渠道	5分钟

1.3 输入输出

输入：监测数据、阈值配置、通知配置、人员通讯录
输出：多渠道通知消息、通知记录、确认状态
中间产物：预警事件记录、升级日志、统计报表

2. 代码示例

2.1 核心预警通知引擎

"""
RailWise 监测预警通知引擎
支持多渠道、分级、可确认的预警通知系统

作者: RailWise技术部
版本: 1.0.0
"""

import asyncio
import json
import logging
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Set, Union
from collections import defaultdict

import aiohttp
import requests
from jinja2 import Environment, BaseLoader

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("RailWise.AlertNotification")


class AlertLevel(Enum):
    """预警等级枚举"""
    NORMAL = "normal"           # 正常
    YELLOW = "yellow"           # 黄色预警
    ORANGE = "orange"           # 橙色预警
    RED = "red"                 # 红色预警


class NotificationChannel(Enum):
    """通知渠道枚举"""
    EMAIL = "email"             # 邮件
    SMS = "sms"                 # 短信
    DINGTALK = "dingtalk"       # 钉钉
    WECHAT = "wechat"           # 企业微信
    FEISHU = "feishu"           # 飞书
    WEBHOOK = "webhook"         # Webhook
    PHONE = "phone"             # 电话


class NotificationStatus(Enum):
    """通知状态枚举"""
    PENDING = "pending"         # 待发送
    SENT = "sent"               # 已发送
    DELIVERED = "delivered"     # 已送达
    CONFIRMED = "confirmed"     # 已确认
    FAILED = "failed"           # 发送失败
    ESCALATED = "escalated"     # 已升级


@dataclass
class AlertEvent:
    """预警事件"""
    event_id: str
    project_id: str
    point_id: str
    point_name: str
    alert_level: AlertLevel
    current_value: float
    threshold_value: float
    control_value: Optional[float] = None
    observation_time: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    message: str = ""
    context: Dict[str, Any] = field(default_factory=dict)


@dataclass
class NotificationRecord:
    """通知记录"""
    record_id: str
    event_id: str
    channel: NotificationChannel
    recipient: str
    status: NotificationStatus
    content: str
    sent_at: Optional[datetime] = None
    confirmed_at: Optional[datetime] = None
    confirmed_by: Optional[str] = None
    retry_count: int = 0
    error_message: Optional[str] = None


@dataclass
class NotificationConfig:
    """通知配置"""
    alert_level: AlertLevel
    channels: List[NotificationChannel]
    recipients: List[str]
    template_name: str
    require_confirmation: bool = False
    confirmation_timeout_minutes: int = 30
    escalation_level: Optional[AlertLevel] = None
    escalation_channels: Optional[List[NotificationChannel]] = None
    escalation_recipients: Optional[List[str]] = None


@dataclass
class Contact:
    """联系人信息"""
    name: str
    email: Optional[str] = None
    phone: Optional[str] = None
    dingtalk_id: Optional[str] = None
    wechat_id: Optional[str] = None
    feishu_id: Optional[str] = None
    roles: List[str] = field(default_factory=list)


class NotificationChannelBase(ABC):
    """通知渠道基类"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)

    @abstractmethod
    async def send(self, recipient: str, content: str, **kwargs) -> bool:
        """发送通知"""
        pass

    @abstractmethod
    def format_content(self, template: str, event: AlertEvent, **kwargs) -> str:
        """格式化通知内容"""
        pass


class EmailChannel(NotificationChannelBase):
    """邮件通知渠道"""

    async def send(self, recipient: str, content: str, **kwargs) -> bool:
        """发送邮件通知"""
        try:
            import smtplib
            from email.mime.text import MIMEText
            from email.mime.multipart import MIMEMultipart

            msg = MIMEMultipart()
            msg['From'] = self.config.get('from_addr', 'alert@railwise.com')
            msg['To'] = recipient
            msg['Subject'] = kwargs.get('subject', '监测预警通知')

            body = MIMEText(content, 'html', 'utf-8')
            msg.attach(body)

            server = smtplib.SMTP(self.config.get('smtp_host', 'smtp.railwise.com'))
            server.starttls()
            server.login(
                self.config.get('username', ''),
                self.config.get('password', '')
            )
            server.send_message(msg)
            server.quit()

            self.logger.info(f"邮件已发送至: {recipient}")
            return True

        except Exception as e:
            self.logger.error(f"邮件发送失败: {e}")
            return False

    def format_content(self, template: str, event: AlertEvent, **kwargs) -> str:
        """格式化邮件内容"""
        jinja_env = Environment(loader=BaseLoader())
        template_obj = jinja_env.from_string(template)
        return template_obj.render(event=event, **kwargs)


class DingTalkChannel(NotificationChannelBase):
    """钉钉通知渠道"""

    async def send(self, recipient: str, content: str, **kwargs) -> bool:
        """发送钉钉通知"""
        try:
            webhook_url = self.config.get('webhook_url')
            if not webhook_url:
                self.logger.error("钉钉Webhook未配置")
                return False

            # 构建钉钉消息
            message = {
                "msgtype": "markdown",
                "markdown": {
                    "title": kwargs.get('title', '监测预警'),
                    "text": content
                }
            }

            # 添加@人员
            if recipient:
                message["at"] = {
                    "atUserIds": [recipient],
                    "isAtAll": False
                }

            async with aiohttp.ClientSession() as session:
                async with session.post(webhook_url, json=message) as response:
                    result = await response.json()
                    if result.get('errcode') == 0:
                        self.logger.info(f"钉钉通知已发送: {recipient}")
                        return True
                    else:
                        self.logger.error(f"钉钉发送失败: {result}")
                        return False

        except Exception as e:
            self.logger.error(f"钉钉通知发送失败: {e}")
            return False

    def format_content(self, template: str, event: AlertEvent, **kwargs) -> str:
        """格式化钉钉Markdown内容"""
        jinja_env = Environment(loader=BaseLoader())
        template_obj = jinja_env.from_string(template)
        return template_obj.render(event=event, **kwargs)


class WeChatChannel(NotificationChannelBase):
    """企业微信通知渠道"""

    async def send(self, recipient: str, content: str, **kwargs) -> bool:
        """发送企业微信通知"""
        try:
            webhook_url = self.config.get('webhook_url')
            if not webhook_url:
                self.logger.error("企业微信Webhook未配置")
                return False

            message = {
                "msgtype": "markdown",
                "markdown": {
                    "content": content
                }
            }

            async with aiohttp.ClientSession() as session:
                async with session.post(webhook_url, json=message) as response:
                    result = await response.json()
                    if result.get('errcode') == 0:
                        self.logger.info(f"企业微信通知已发送")
                        return True
                    else:
                        self.logger.error(f"企业微信发送失败: {result}")
                        return False

        except Exception as e:
            self.logger.error(f"企业微信通知发送失败: {e}")
            return False

    def format_content(self, template: str, event: AlertEvent, **kwargs) -> str:
        """格式化企业微信内容"""
        jinja_env = Environment(loader=BaseLoader())
        template_obj = jinja_env.from_string(template)
        return template_obj.render(event=event, **kwargs)


class SMSChannel(NotificationChannelBase):
    """短信通知渠道"""

    async def send(self, recipient: str, content: str, **kwargs) -> bool:
        """发送短信通知"""
        try:
            # 使用第三方短信API（如阿里云、腾讯云）
            api_url = self.config.get('api_url')
            api_key = self.config.get('api_key')

            if not api_url or not api_key:
                self.logger.error("短信API未配置")
                return False

            payload = {
                "phone": recipient,
                "message": content,
                "api_key": api_key
            }

            async with aiohttp.ClientSession() as session:
                async with session.post(api_url, json=payload) as response:
                    result = await response.json()
                    if result.get('code') == 0:
                        self.logger.info(f"短信已发送至: {recipient}")
                        return True
                    else:
                        self.logger.error(f"短信发送失败: {result}")
                        return False

        except Exception as e:
            self.logger.error(f"短信发送失败: {e}")
            return False

    def format_content(self, template: str, event: AlertEvent, **kwargs) -> str:
        """格式化短信内容（限制长度）"""
        jinja_env = Environment(loader=BaseLoader())
        template_obj = jinja_env.from_string(template)
        content = template_obj.render(event=event, **kwargs)

        # 短信限制70字符（中文）
        if len(content) > 70:
            content = content[:67] + "..."
        return content


class WebhookChannel(NotificationChannelBase):
    """Webhook通知渠道"""

    async def send(self, recipient: str, content: str, **kwargs) -> bool:
        """发送Webhook通知"""
        try:
            webhook_url = recipient  # recipient即为webhook URL

            payload = {
                "event": "monitoring_alert",
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "content": content,
                "data": kwargs.get('event_data', {})
            }

            async with aiohttp.ClientSession() as session:
                async with session.post(webhook_url, json=payload) as response:
                    if response.status == 200:
                        self.logger.info(f"Webhook通知已发送: {webhook_url}")
                        return True
                    else:
                        self.logger.error(f"Webhook发送失败: {response.status}")
                        return False

        except Exception as e:
            self.logger.error(f"Webhook通知发送失败: {e}")
            return False

    def format_content(self, template: str, event: AlertEvent, **kwargs) -> str:
        """格式化Webhook内容"""
        return json.dumps({
            "event_id": event.event_id,
            "alert_level": event.alert_level.value,
            "point_id": event.point_id,
            "current_value": event.current_value,
            "threshold": event.threshold_value,
            "message": event.message
        }, ensure_ascii=False)


# 通知渠道工厂
CHANNEL_CLASSES = {
    NotificationChannel.EMAIL: EmailChannel,
    NotificationChannel.SMS: SMSChannel,
    NotificationChannel.DINGTALK: DingTalkChannel,
    NotificationChannel.WECHAT: WeChatChannel,
    NotificationChannel.WEBHOOK: WebhookChannel,
}


def create_channel(channel_type: NotificationChannel, config: Dict[str, Any]) -> NotificationChannelBase:
    """创建通知渠道实例"""
    channel_class = CHANNEL_CLASSES.get(channel_type)
    if not channel_class:
        raise ValueError(f"不支持的通知渠道: {channel_type}")
    return channel_class(config)


class AlertNotificationEngine:
    """预警通知引擎"""

    # 默认通知模板
    DEFAULT_TEMPLATES = {
        AlertLevel.YELLOW: {
            "dingtalk": """## 🟡 黄色预警

**项目**: {{ event.project_id }}
**测点**: {{ event.point_name }} ({{ event.point_id }})
**监测时间**: {{ event.observation_time.strftime('%Y-%m-%d %H:%M') }}

| 指标 | 数值 |
|------|------|
| 当前值 | {{ event.current_value }}mm |
| 预警阈值 | {{ event.threshold_value }}mm |
| 控制值 | {{ event.control_value or '未设置' }}mm |

**预警说明**: {{ event.message }}

> 请相关责任人关注并确认处理。
""",
            "sms": "【RailWise预警】{{ event.point_name }}黄色预警，当前值{{ event.current_value }}mm，阈值{{ event.threshold_value }}mm。请确认处理。",
            "email": "<h2>🟡 黄色预警通知</h2><p>测点: {{ event.point_name }}</p><p>当前值: {{ event.current_value }}mm</p><p>阈值: {{ event.threshold_value }}mm</p>"
        },
        AlertLevel.ORANGE: {
            "dingtalk": """## 🟠 橙色预警

**项目**: {{ event.project_id }}
**测点**: {{ event.point_name }} ({{ event.point_id }})
**监测时间**: {{ event.observation_time.strftime('%Y-%m-%d %H:%M') }}

| 指标 | 数值 |
|------|------|
| 当前值 | {{ event.current_value }}mm |
| 预警阈值 | {{ event.threshold_value }}mm |
| 控制值 | {{ event.control_value or '未设置' }}mm |

**预警说明**: {{ event.message }}

> ⚠️ 变化明显，请立即安排现场巡查并确认数据真实性。
""",
            "sms": "【RailWise紧急预警】{{ event.point_name }}橙色预警！当前值{{ event.current_value }}mm，阈值{{ event.threshold_value }}mm。请立即处理！",
            "email": "<h2>🟠 橙色预警通知</h2><p>测点: {{ event.point_name }}</p><p>当前值: {{ event.current_value }}mm</p><p>请立即处理！</p>"
        },
        AlertLevel.RED: {
            "dingtalk": """## 🔴 红色预警

**项目**: {{ event.project_id }}
**测点**: {{ event.point_name }} ({{ event.point_id }})
**监测时间**: {{ event.observation_time.strftime('%Y-%m-%d %H:%M') }}

| 指标 | 数值 |
|------|------|
| 当前值 | {{ event.current_value }}mm |
| 预警阈值 | {{ event.threshold_value }}mm |
| 控制值 | {{ event.control_value or '未设置' }}mm |

**预警说明**: {{ event.message }}

> 🚨 严重超限！请立即启动应急预案，停止相关作业，组织专家会商。
""",
            "sms": "【RailWise紧急预警】{{ event.point_name }}红色预警！！当前值{{ event.current_value }}mm，阈值{{ event.threshold_value }}mm。请立即启动应急预案！",
            "email": "<h2>🔴 红色预警通知</h2><p>测点: {{ event.point_name }}</p><p>当前值: {{ event.current_value }}mm</p><p>🚨 请立即启动应急预案！</p>"
        }
    }

    def __init__(self, channel_configs: Dict[NotificationChannel, Dict[str, Any]]):
        self.channel_configs = channel_configs
        self.channels: Dict[NotificationChannel, NotificationChannelBase] = {}
        self.notification_records: Dict[str, NotificationRecord] = {}
        self.confirmation_callbacks: Dict[str, callable] = {}

        # 初始化通知渠道
        for channel_type, config in channel_configs.items():
            self.channels[channel_type] = create_channel(channel_type, config)

    async def process_alert(self, event: AlertEvent, config: NotificationConfig) -> List[NotificationRecord]:
        """处理预警事件，发送通知"""
        records = []

        for channel_type in config.channels:
            channel = self.channels.get(channel_type)
            if not channel:
                logger.warning(f"通知渠道未配置: {channel_type}")
                continue

            # 获取模板
            template = self._get_template(config.alert_level, channel_type)

            # 格式化内容
            content = channel.format_content(template, event)

            # 发送通知
            for recipient in config.recipients:
                record = NotificationRecord(
                    record_id=f"{event.event_id}_{channel_type.value}_{recipient}",
                    event_id=event.event_id,
                    channel=channel_type,
                    recipient=recipient,
                    status=NotificationStatus.PENDING,
                    content=content
                )

                success = await channel.send(recipient, content, event=event)

                if success:
                    record.status = NotificationStatus.SENT
                    record.sent_at = datetime.now(timezone.utc)
                    logger.info(f"通知已发送: {channel_type.value} -> {recipient}")
                else:
                    record.status = NotificationStatus.FAILED
                    record.retry_count += 1
                    logger.error(f"通知发送失败: {channel_type.value} -> {recipient}")

                self.notification_records[record.record_id] = record
                records.append(record)

        # 如果需要确认，启动确认超时检查
        if config.require_confirmation:
            asyncio.create_task(
                self._check_confirmation_timeout(event.event_id, config)
            )

        return records

    def _get_template(self, alert_level: AlertLevel, channel: NotificationChannel) -> str:
        """获取通知模板"""
        level_templates = self.DEFAULT_TEMPLATES.get(alert_level, {})
        return level_templates.get(channel.value, "监测预警: {{ event.message }}")

    async def _check_confirmation_timeout(self, event_id: str, config: NotificationConfig):
        """检查确认超时，执行升级"""
        await asyncio.sleep(config.confirmation_timeout_minutes * 60)

        # 检查是否已确认
        confirmed = any(
            r.status == NotificationStatus.CONFIRMED
            for r in self.notification_records.values()
            if r.event_id == event_id
        )

        if not confirmed and config.escalation_level:
            logger.warning(f"预警事件 {event_id} 未确认，执行升级")

            # 创建升级事件
            escalation_event = AlertEvent(
                event_id=f"{event_id}_escalated",
                project_id="",  # 从原事件获取
                point_id="",
                point_name="",
                alert_level=config.escalation_level,
                current_value=0,
                threshold_value=0,
                message=f"预警事件 {event_id} 未确认，已升级"
            )

            # 升级通知配置
            escalation_config = NotificationConfig(
                alert_level=config.escalation_level,
                channels=config.escalation_channels or config.channels,
                recipients=config.escalation_recipients or config.recipients,
                template_name=config.template_name,
                require_confirmation=False
            )

            await self.process_alert(escalation_event, escalation_config)

    def confirm_notification(self, record_id: str, confirmed_by: str) -> bool:
        """确认通知"""
        record = self.notification_records.get(record_id)
        if not record:
            return False

        record.status = NotificationStatus.CONFIRMED
        record.confirmed_at = datetime.now(timezone.utc)
        record.confirmed_by = confirmed_by

        logger.info(f"通知已确认: {record_id} by {confirmed_by}")
        return True

    def get_notification_history(self, event_id: Optional[str] = None) -> List[NotificationRecord]:
        """获取通知历史"""
        records = list(self.notification_records.values())
        if event_id:
            records = [r for r in records if r.event_id == event_id]
        return records

    def get_pending_confirmations(self) -> List[NotificationRecord]:
        """获取待确认的通知"""
        return [
            r for r in self.notification_records.values()
            if r.status == NotificationStatus.SENT
        ]


class AlertRuleEngine:
    """预警规则引擎"""

    def __init__(self):
        self.rules: List[Dict[str, Any]] = []
        self.notification_configs: Dict[AlertLevel, NotificationConfig] = {}

    def add_rule(
        self,
        point_id: str,
        point_type: str,
        thresholds: Dict[AlertLevel, float],
        control_value: Optional[float] = None
    ):
        """添加预警规则"""
        self.rules.append({
            "point_id": point_id,
            "point_type": point_type,
            "thresholds": thresholds,
            "control_value": control_value
        })

    def set_notification_config(self, alert_level: AlertLevel, config: NotificationConfig):
        """设置通知配置"""
        self.notification_configs[alert_level] = config

    def evaluate(self, point_id: str, current_value: float) -> Optional[AlertEvent]:
        """评估监测值，触发预警"""
        rule = next((r for r in self.rules if r["point_id"] == point_id), None)
        if not rule:
            return None

        thresholds = rule["thresholds"]

        # 确定预警等级
        alert_level = AlertLevel.NORMAL
        for level in [AlertLevel.RED, AlertLevel.ORANGE, AlertLevel.YELLOW]:
            if abs(current_value) >= thresholds.get(level, float('inf')):
                alert_level = level
                break

        if alert_level == AlertLevel.NORMAL:
            return None

        # 创建预警事件
        event = AlertEvent(
            event_id=f"ALT-{point_id}-{datetime.now().strftime('%Y%m%d%H%M%S')}",
            project_id="",  # 从上下文获取
            point_id=point_id,
            point_name=rule["point_type"],
            alert_level=alert_level,
            current_value=current_value,
            threshold_value=thresholds[alert_level],
            control_value=rule.get("control_value"),
            message=f"测点 {point_id} {alert_level.value}预警，当前值 {current_value}mm"
        )

        return event

    def get_notification_config(self, alert_level: AlertLevel) -> Optional[NotificationConfig]:
        """获取通知配置"""
        return self.notification_configs.get(alert_level)

2.2 通知配置管理

"""通知配置管理模块"""

import json
from pathlib import Path
from typing import Dict, List


class NotificationConfigManager:
    """通知配置管理器"""

    def __init__(self, config_path: Path = Path("notification_config.json")):
        self.config_path = config_path
        self.configs: Dict[str, Any] = {}
        self._load()

    def _load(self):
        """加载配置"""
        if self.config_path.exists():
            with open(self.config_path, 'r', encoding='utf-8') as f:
                self.configs = json.load(f)

    def _save(self):
        """保存配置"""
        with open(self.config_path, 'w', encoding='utf-8') as f:
            json.dump(self.configs, f, ensure_ascii=False, indent=2)

    def set_project_config(self, project_id: str, config: Dict[str, Any]):
        """设置项目通知配置"""
        self.configs[project_id] = config
        self._save()

    def get_project_config(self, project_id: str) -> Dict[str, Any]:
        """获取项目通知配置"""
        return self.configs.get(project_id, self._get_default_config())

    def _get_default_config(self) -> Dict[str, Any]:
        """获取默认配置"""
        return {
            "alert_levels": {
                "yellow": {
                    "channels": ["dingtalk", "email"],
                    "recipients": ["monitoring@railwise.com"],
                    "require_confirmation": True,
                    "confirmation_timeout": 30,
                    "escalation_level": "orange"
                },
                "orange": {
                    "channels": ["sms", "dingtalk", "email"],
                    "recipients": ["manager@railwise.com", "monitoring@railwise.com"],
                    "require_confirmation": True,
                    "confirmation_timeout": 15,
                    "escalation_level": "red"
                },
                "red": {
                    "channels": ["phone", "sms", "dingtalk", "email"],
                    "recipients": ["director@railwise.com", "manager@railwise.com"],
                    "require_confirmation": False,
                    "escalation_level": None
                }
            },
            "channel_configs": {
                "email": {
                    "smtp_host": "smtp.railwise.com",
                    "from_addr": "alert@railwise.com"
                },
                "dingtalk": {
                    "webhook_url": "https://oapi.dingtalk.com/robot/send"
                },
                "sms": {
                    "api_url": "https://sms-api.railwise.com/send",
                    "api_key": ""
                }
            }
        }

    def add_recipient(self, project_id: str, alert_level: str, recipient: str):
        """添加通知接收人"""
        config = self.get_project_config(project_id)
        level_config = config["alert_levels"].get(alert_level, {})
        recipients = level_config.get("recipients", [])

        if recipient not in recipients:
            recipients.append(recipient)
            level_config["recipients"] = recipients
            self.set_project_config(project_id, config)

    def remove_recipient(self, project_id: str, alert_level: str, recipient: str):
        """移除通知接收人"""
        config = self.get_project_config(project_id)
        level_config = config["alert_levels"].get(alert_level, {})
        recipients = level_config.get("recipients", [])

        if recipient in recipients:
            recipients.remove(recipient)
            level_config["recipients"] = recipients
            self.set_project_config(project_id, config)

3. 参数说明

3.1 AlertEvent 预警事件参数

参数名	类型	必填	说明
`event_id`	str	是	事件唯一标识
`project_id`	str	是	项目编号
`point_id`	str	是	测点编号
`point_name`	str	是	测点名称
`alert_level`	AlertLevel	是	预警等级
`current_value`	float	是	当前监测值(mm)
`threshold_value`	float	是	触发阈值(mm)
`control_value`	float	否	控制值(mm)
`observation_time`	datetime	否	观测时间
`message`	str	否	预警说明
`context`	dict	否	上下文信息

3.2 NotificationConfig 通知配置参数

参数名	类型	必填	默认值	说明
`alert_level`	AlertLevel	是	-	适用的预警等级
`channels`	List[NotificationChannel]	是	-	通知渠道列表
`recipients`	List[str]	是	-	接收人列表
`template_name`	str	是	-	模板名称
`require_confirmation`	bool	否	False	是否需要确认
`confirmation_timeout_minutes`	int	否	30	确认超时时间(分钟)
`escalation_level`	AlertLevel	否	None	升级后的预警等级
`escalation_channels`	List[NotificationChannel]	否	None	升级后的渠道
`escalation_recipients`	List[str]	否	None	升级后的接收人

3.3 通知渠道配置参数

渠道	必需配置	可选配置
邮件	`smtp_host`, `username`, `password`	`from_addr`, `port`
钉钉	`webhook_url`	`secret`
企业微信	`webhook_url`	`corp_id`, `agent_id`
短信	`api_url`, `api_key`	`signature`, `template_id`
Webhook	-	`headers`, `timeout`

4. 使用示例

4.1 基本预警通知流程

import asyncio
from template_alert_notification import (
    AlertNotificationEngine, AlertRuleEngine, AlertEvent, AlertLevel,
    NotificationConfig, NotificationChannel, NotificationStatus
)

# 1. 配置通知渠道
channel_configs = {
    NotificationChannel.DINGTALK: {
        "webhook_url": "https://oapi.dingtalk.com/robot/send?access_token=xxx"
    },
    NotificationChannel.SMS: {
        "api_url": "https://sms-api.railwise.com/send",
        "api_key": "your-api-key"
    },
    NotificationChannel.EMAIL: {
        "smtp_host": "smtp.railwise.com",
        "username": "alert@railwise.com",
        "password": "your-password"
    }
}

# 2. 创建通知引擎
engine = AlertNotificationEngine(channel_configs)

# 3. 配置预警规则
rule_engine = AlertRuleEngine()
rule_engine.add_rule(
    point_id="JC-01",
    point_type="水平位移",
    thresholds={
        AlertLevel.YELLOW: 5.0,
        AlertLevel.ORANGE: 8.0,
        AlertLevel.RED: 10.0
    },
    control_value=15.0
)

# 4. 配置通知规则
rule_engine.set_notification_config(
    AlertLevel.YELLOW,
    NotificationConfig(
        alert_level=AlertLevel.YELLOW,
        channels=[NotificationChannel.DINGTALK],
        recipients=["user1"],
        template_name="yellow_alert",
        require_confirmation=True,
        confirmation_timeout_minutes=30,
        escalation_level=AlertLevel.ORANGE,
        escalation_channels=[NotificationChannel.SMS, NotificationChannel.DINGTALK],
        escalation_recipients=["manager1", "user1"]
    )
)

rule_engine.set_notification_config(
    AlertLevel.ORANGE,
    NotificationConfig(
        alert_level=AlertLevel.ORANGE,
        channels=[NotificationChannel.SMS, NotificationChannel.DINGTALK],
        recipients=["manager1", "user1"],
        template_name="orange_alert",
        require_confirmation=True,
        confirmation_timeout_minutes=15,
        escalation_level=AlertLevel.RED,
        escalation_channels=[NotificationChannel.PHONE, NotificationChannel.SMS],
        escalation_recipients=["director1", "manager1"]
    )
)

# 5. 模拟监测数据并触发预警
async def main():
    # 模拟监测值超过黄色阈值
    event = rule_engine.evaluate("JC-01", 6.5)

    if event:
        print(f"触发预警: {event.alert_level.value}")
        print(f"测点: {event.point_name}, 当前值: {event.current_value}mm")

        # 获取通知配置
        config = rule_engine.get_notification_config(event.alert_level)

        if config:
            # 发送通知
            records = await engine.process_alert(event, config)

            for record in records:
                print(f"通知状态: {record.channel.value} -> {record.recipient}: {record.status.value}")
    else:
        print("监测值正常，未触发预警")

# 运行
asyncio.run(main())

4.2 确认通知与升级

# 模拟接收人确认通知
async def confirm_and_escalate_example():
    # ... 发送通知后 ...

    # 接收人确认通知
    record_id = "ALT-JC-01-20240708120000_dingtalk_user1"
    success = engine.confirm_notification(record_id, confirmed_by="张三")

    if success:
        print("通知已确认，不会升级")
    else:
        print("确认失败，通知将在超时后升级")

    # 查看待确认通知
    pending = engine.get_pending_confirmations()
    print(f"待确认通知数: {len(pending)}")

    # 查看通知历史
    history = engine.get_notification_history(event_id="ALT-JC-01-20240708120000")
    for record in history:
        print(f"{record.channel.value}: {record.status.value}")

asyncio.run(confirm_and_escalate_example())

4.3 批量测点预警监控

async def monitor_multiple_points():
    """监控多个测点"""

    # 定义多个测点规则
    points_config = [
        {
            "point_id": "JC-01",
            "point_type": "水平位移",
            "thresholds": {AlertLevel.YELLOW: 5.0, AlertLevel.ORANGE: 8.0, AlertLevel.RED: 10.0}
        },
        {
            "point_id": "JC-02",
            "point_type": "竖向位移",
            "thresholds": {AlertLevel.YELLOW: 3.0, AlertLevel.ORANGE: 5.0, AlertLevel.RED: 8.0}
        },
        {
            "point_id": "JC-03",
            "point_type": "收敛",
            "thresholds": {AlertLevel.YELLOW: 10.0, AlertLevel.ORANGE: 15.0, AlertLevel.RED: 20.0}
        }
    ]

    # 添加规则
    for config in points_config:
        rule_engine.add_rule(
            point_id=config["point_id"],
            point_type=config["point_type"],
            thresholds=config["thresholds"]
        )

    # 模拟监测数据
    mock_data = {
        "JC-01": 6.5,   # 黄色预警
        "JC-02": 2.1,   # 正常
        "JC-03": 16.2   # 橙色预警
    }

    # 评估所有测点
    for point_id, value in mock_data.items():
        event = rule_engine.evaluate(point_id, value)

        if event:
            config = rule_engine.get_notification_config(event.alert_level)
            if config:
                records = await engine.process_alert(event, config)
                print(f"{point_id}: 触发{event.alert_level.value}预警，已发送{len(records)}条通知")
        else:
            print(f"{point_id}: 正常")

asyncio.run(monitor_multiple_points())

5. 输出示例

5.1 钉钉通知示例

## 🟡 黄色预警

**项目**: RW-2024-001
**测点**: 3号线K12+350左线 (JC-01)
**监测时间**: 2024-07-08 12:00

| 指标 | 数值 |
|------|------|
| 当前值 | 6.5mm |
| 预警阈值 | 5.0mm |
| 控制值 | 15.0mm |

**预警说明**: 测点 JC-01 黄色预警，当前值 6.5mm

> 请相关责任人关注并确认处理。

5.2 短信通知示例

【RailWise预警】3号线K12+350左线黄色预警，当前值6.5mm，阈值5.0mm。请确认处理。

5.3 通知记录示例

{
    "record_id": "ALT-JC-01-20240708120000_dingtalk_user1",
    "event_id": "ALT-JC-01-20240708120000",
    "channel": "dingtalk",
    "recipient": "user1",
    "status": "confirmed",
    "content": "## 🟡 黄色预警...",
    "sent_at": "2024-07-08T12:00:05+00:00",
    "confirmed_at": "2024-07-08T12:15:30+00:00",
    "confirmed_by": "张三",
    "retry_count": 0
}

6. 扩展指南

6.1 添加新的通知渠道

from template_alert_notification import NotificationChannelBase

class SlackChannel(NotificationChannelBase):
    """Slack通知渠道"""

    async def send(self, recipient: str, content: str, **kwargs) -> bool:
        """发送Slack通知"""
        try:
            webhook_url = self.config.get('webhook_url')
            payload = {
                "text": content,
                "channel": recipient
            }

            async with aiohttp.ClientSession() as session:
                async with session.post(webhook_url, json=payload) as response:
                    return response.status == 200

        except Exception as e:
            self.logger.error(f"Slack发送失败: {e}")
            return False

    def format_content(self, template: str, event: AlertEvent, **kwargs) -> str:
        """格式化Slack内容"""
        return f"*预警通知*\n测点: {event.point_name}\n等级: {event.alert_level.value}\n当前值: {event.current_value}mm"

# 注册到工厂
from template_alert_notification import CHANNEL_CLASSES, NotificationChannel

CHANNEL_CLASSES[NotificationChannel.SLACK] = SlackChannel

6.2 自定义通知模板

# 自定义模板
custom_templates = {
    AlertLevel.YELLOW: {
        "dingtalk": """## 🟡 监测预警提醒

**测点**: {{ event.point_name }}
**时间**: {{ event.observation_time.strftime('%H:%M') }}
**当前值**: {{ event.current_value }}mm (阈值: {{ event.threshold_value }}mm)

请登录系统查看详情: [WorkWise](https://workwise.railwise.com)
"""
    }
}

# 设置自定义模板
engine.DEFAULT_TEMPLATES.update(custom_templates)

6.3 集成到监测工作流

# 与数据导入流程集成
from template_data_import import MonitoringRecord

async def on_data_imported(record: MonitoringRecord):
    """数据导入后的回调"""

    # 评估是否触发预警
    event = rule_engine.evaluate(record.point_id, record.value_d or 0)

    if event:
        config = rule_engine.get_notification_config(event.alert_level)
        if config:
            await engine.process_alert(event, config)

6.4 定时检查未确认通知

from apscheduler.schedulers.background import BackgroundScheduler

def check_pending_alerts():
    """定时检查待确认预警"""
    pending = engine.get_pending_confirmations()

    for record in pending:
        # 计算已等待时间
        if record.sent_at:
            elapsed = (datetime.now(timezone.utc) - record.sent_at).total_seconds()

            # 发送提醒
            if elapsed > 600:  # 10分钟未确认
                print(f"提醒: {record.record_id} 已等待 {elapsed/60:.0f} 分钟未确认")

# 每5分钟检查一次
scheduler = BackgroundScheduler()
scheduler.add_job(check_pending_alerts, 'interval', minutes=5)
scheduler.start()

7. 常见问题

Q1: 钉钉机器人发送失败怎么办？

A: 检查以下几点：

Webhook URL是否正确
是否配置了安全设置（IP白名单/加签）
消息内容是否包含敏感词
机器人是否被禁言

# 加签验证示例
import hmac
import hashlib
import base64
import urllib.parse

def sign_dingtalk(secret: str) -> str:
    timestamp = str(round(time.time() * 1000))
    string_to_sign = f"{timestamp}\n{secret}"
    hmac_code = hmac.new(secret.encode('utf-8'), string_to_sign.encode('utf-8'), digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    return f"&timestamp={timestamp}&sign={sign}"

Q2: 如何限制短信发送频率？

A: 使用频率限制器：

from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, max_requests: int = 10, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = window_seconds
        self.requests = defaultdict(list)

    def is_allowed(self, key: str) -> bool:
        now = time.time()
        self.requests[key] = [t for t in self.requests[key] if now - t < self.window]

        if len(self.requests[key]) < self.max_requests:
            self.requests[key].append(now)
            return True
        return False

# 使用
limiter = RateLimiter(max_requests=5, window_seconds=60)
if limiter.is_allowed("sms_13800138000"):
    await sms_channel.send("13800138000", content)

Q3: 如何实现通知去重？

A: 使用事件指纹去重：

def get_event_fingerprint(event: AlertEvent) -> str:
    """生成事件指纹"""
    return f"{event.point_id}_{event.alert_level.value}_{event.observation_time.strftime('%Y%m%d%H')}"

# 检查是否已发送
sent_fingerprints = set()

fingerprint = get_event_fingerprint(event)
if fingerprint not in sent_fingerprints:
    await engine.process_alert(event, config)
    sent_fingerprints.add(fingerprint)

Q4: 如何记录通知日志？

A: 使用数据库存储：

import sqlite3
from datetime import datetime

def log_notification(record: NotificationRecord):
    """记录通知日志"""
    conn = sqlite3.connect('notifications.db')
    conn.execute("""
        INSERT INTO notification_logs
        (record_id, event_id, channel, recipient, status, sent_at, content)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    """, (
        record.record_id, record.event_id, record.channel.value,
        record.recipient, record.status.value,
        record.sent_at.isoformat() if record.sent_at else None,
        record.content
    ))
    conn.commit()
    conn.close()

Q5: 如何支持语音电话通知？

A: 集成第三方语音API：

class PhoneChannel(NotificationChannelBase):
    async def send(self, recipient: str, content: str, **kwargs) -> bool:
        api_url = self.config.get('api_url')
        api_key = self.config.get('api_key')

        payload = {
            "phone": recipient,
            "template_id": self.config.get('template_id'),
            "template_params": {
                "point_name": kwargs.get('event').point_name,
                "alert_level": kwargs.get('event').alert_level.value
            }
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(api_url, json=payload, headers={"Authorization": f"Bearer {api_key}"}) as response:
                return response.status == 200

元数据标签

ai_tags:
  domain: ["工程监测", "预警系统", "通知推送"]
  technology: ["Python", "asyncio", "aiohttp", "Jinja2"]
  standard: ["GB 50497", "CJJ/T 202"]
  product: ["RAILWISE-TSM", "RAILWISE-CLI", "WorkWise"]
  difficulty: "advanced"
  role: ["监测工程师", "系统管理员", "项目经理"]
  task_type: ["预警通知", "实时告警", "消息推送"]

监测预警通知模板

监测预警通知模板

1. 模板概述

1.1 用途

1.2 适用场景

1.3 输入输出

2. 代码示例

2.1 核心预警通知引擎

2.2 通知配置管理

3. 参数说明

3.1 AlertEvent 预警事件参数

3.2 NotificationConfig 通知配置参数

3.3 通知渠道配置参数

4. 使用示例

4.1 基本预警通知流程

4.2 确认通知与升级

4.3 批量测点预警监控

5. 输出示例

5.1 钉钉通知示例

5.2 短信通知示例

5.3 通知记录示例

6. 扩展指南

6.1 添加新的通知渠道

6.2 自定义通知模板

6.3 集成到监测工作流

6.4 定时检查未确认通知

7. 常见问题

Q1: 钉钉机器人发送失败怎么办？

Q2: 如何限制短信发送频率？

Q3: 如何实现通知去重？

Q4: 如何记录通知日志？

Q5: 如何支持语音电话通知？

相关文档链接

元数据标签