提交 fdb0ee63 authored 作者: 陈泽健's avatar 陈泽健

feat(service-monitor): 通知配置功能(邮件/钉钉/企业微信)

- 左侧菜单新增「通知配置」(仅管理员可见)
- 新增 notification_service.py:三种渠道配置/测试发送/巡检后通知
- 新增 notification.html:配置卡片 + 触发条件 + 测试按钮
- runner_service.py:run_inspection_sync 完成后调用通知服务
- 敏感字段用 Fernet 加密存储
- 新增 PRD 文档
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 28f55365
# 计划执行 — 通知配置功能
> 版本:1.0 | 日期:2026-07-21 | 关联 PRD:PRD_需求文档_通知配置.md
---
## 执行步骤
### 步骤 1:后端 notification_service.py
**文件**`skill/code/web/service_monitor/services/notification_service.py`
**改动**
1. 数据模型:`notifications.json` 结构定义
2. `get_config()` — 读取配置(敏感字段解密)
3. `save_config(data)` — 保存配置(敏感字段加密)
4. `test_email()` — 发送测试邮件(smtplib)
5. `test_dingtalk()` — 发送测试钉钉消息(requests + HMAC 签名)
6. `test_wecom()` — 发送测试企业微信消息(requests)
7. `send_notification(report, report_url)` — 根据配置发送通知
8. `_build_email_content(report)` — 构建邮件内容
9. `_build_dingtalk_content(report)` — 构建钉钉消息
10. `_build_wecom_content(report)` — 构建企业微信消息
**依赖**
- `smtplib` + `email` — 标准库,无需安装
- `requests` — 已有依赖
**验证**:Python 语法检查
---
### 步骤 2:更新 paths.py
**文件**`skill/code/web/service_monitor/utils/paths.py`
**改动**
```python
NOTIFICATIONS_FILE = DATA_DIR / "notifications.json"
```
---
### 步骤 3:路由层 routes.py 新增接口
**文件**`skill/code/web/service_monitor/routes.py`
**改动**
1. 新增 `page_notification()` — 配置页面路由
2. 新增 `api_get_notification()` — 获取配置 API
3. 新增 `api_save_notification()` — 保存配置 API(管理员)
4. 新增 `api_test_email()` — 测试邮件 API(管理员)
5. 新增 `api_test_dingtalk()` — 测试钉钉 API(管理员)
6. 新增 `api_test_wecom()` — 测试企业微信 API(管理员)
---
### 步骤 4:前端 notification.html
**文件**`skill/code/web/templates/service_monitor/notification.html`
**改动**
1. 继承 `base.html`
2. 三个通知渠道卡片:邮件 / 钉钉 / 企业微信
3. 每个卡片有启用开关 + 配置表单 + 测试按钮
4. 触发条件配置区域
5. 保存按钮
6. JS:加载配置、保存配置、发送测试请求
---
### 步骤 5:更新 base.html 左侧菜单
**文件**`skill/code/web/templates/service_monitor/base.html`
**改动**
在"定时任务"和"目标管理"之间增加"通知配置"菜单项:
```html
<a href="/service-monitor/notification" class="nav-item {{ 'active' if active_menu == 'notification' }}">
<span class="nav-icon">🔔</span>
<span>通知配置</span>
</a>
```
---
### 步骤 6:集成到巡检流程
**文件**`skill/code/web/service_monitor/services/runner_service.py`
**改动**
`run_inspection_sync()` 完成后调用通知服务:
```python
# 发送通知(定时任务触发的巡检)
from . import notification_service
report = report_service.get_report(report_id)
report_url = f"http://request.host/service-monitor/report/{report_id}"
notification_service.send_notification(report, report_url)
```
---
### 步骤 7:部署验证
1. 部署到 5.60
2. 浏览器访问通知配置页面
3. 配置邮件通知,发送测试邮件
4. 配置钉钉/企业微信,发送测试消息
5. 创建定时任务,等待执行,验证通知发送
---
## 风险与兜底
| 风险 | 兜底方案 |
|------|----------|
| SMTP 连接失败 | 测试接口返回具体错误信息,帮助用户排查 |
| 钉钉签名错误 | 提供 webhook 和 secret 填写说明,测试验证 |
| 企业微信 webhook 失效 | 测试按钮验证配置有效性 |
| 通知发送阻塞巡检 | 通知发送用 try-except 包裹,失败不影响报告保存 |
---
## 文件清单
| 文件 | 操作 |
|------|------|
| `service_monitor/services/notification_service.py` | 新增 |
| `service_monitor/utils/paths.py` | 修改(增加 NOTIFICATIONS_FILE) |
| `service_monitor/routes.py` | 修改(增加 6 个路由) |
| `templates/service_monitor/notification.html` | 新增 |
| `templates/service_monitor/base.html` | 修改(增加菜单项) |
| `service_monitor/services/runner_service.py` | 修改(集成通知调用) |
\ No newline at end of file
# PRD — 通知配置功能
> 版本:1.0 | 日期:2026-07-21 | 作者:czj
---
## 1. 背景与问题
当前服务监测模块生成巡检报告后,**仅保存在服务器本地**,用户需要主动登录系统查看。存在以下问题:
1. **无法及时感知**:巡检完成后(特别是定时巡检),管理员无法第一时间获知结果
2. **异常响应滞后**:出现严重问题时,需要等用户主动查看才能发现
3. **缺少推送渠道**:没有邮件/钉钉/企业微信等通知通道配置
## 2. 需求目标
新增**通知配置**功能,让管理员配置报告发送通知的相关信息:
- 配置邮件通知(SMTP)
- 配置钉钉机器人通知
- 配置企业微信机器人通知
- 设置通知触发条件(报告完成后 / 出现异常时)
- 设置通知接收人
## 3. 功能设计
### 3.1 左侧菜单新增
在"定时任务"和"目标管理"之间增加:
```
📊 监测目标
📋 巡检报告
⏰ 定时任务
🔔 通知配置 ← 新增
⚙️ 目标管理
```
### 3.2 通知配置页面
#### 3.2.1 整体布局
- **通知渠道卡片**:邮件 / 钉钉 / 企业微信(可多选启用)
- **触发条件**:报告完成后通知 / 仅异常时通知
- **接收人配置**:邮箱列表 / 钉钉群 / 企业微信群
#### 3.2.2 邮件通知配置
| 字段 | 控件 | 说明 |
|------|------|------|
| 启用邮件通知 | 开关 | 是/否 |
| SMTP 服务器 | 文本 | 如 `smtp.qq.com` |
| SMTP 端口 | 数字 | 如 `465`(SSL)或 `25` |
| 发件人邮箱 | 文本 | 如 `admin@example.com` |
| 邮箱授权码 | 密码 | SMTP 认证密码 |
| 使用 SSL | 开关 | 推荐 SSL |
| 收件人列表 | 文本域 | 多个邮箱用英文逗号分隔 |
| 邮件主题模板 | 文本 | 支持 `{target}`, `{suite}`, `{status}` 变量 |
**测试按钮**:发送测试邮件验证配置
#### 3.2.3 钉钉机器人配置
| 字段 | 控件 | 说明 |
|------|------|------|
| 启用钉钉通知 | 开关 | 是/否 |
| Webhook 地址 | 文本 | 钉钉群机器人的 webhook URL |
| 签名密钥 | 密码 | 加签机器人的密钥(可选) |
| @人员列表 | 文本 | 手机号,多个用逗号分隔(可选) |
**测试按钮**:发送测试消息验证配置
#### 3.2.4 企业微信机器人配置
| 字段 | 控件 | 说明 |
|------|------|------|
| 启用企业微信通知 | 开关 | 是/否 |
| Webhook 地址 | 文本 | 企业微信群机器人的 webhook URL |
**测试按钮**:发送测试消息验证配置
#### 3.2.5 触发条件
| 字段 | 控件 | 说明 |
|------|------|------|
| 触发时机 | 单选 | 报告完成后立即通知 / 仅异常(警告/严重)时通知 |
| 通知内容 | 多选 | 包含摘要 / 包含异常详情 / 包含报告链接 |
### 3.3 通知时机
1. **手动巡检完成**:不发送通知(用户正在页面等待)
2. **定时巡检完成**:根据配置发送通知
3. **异常时通知**:summary 中有 WARNING 或 CRITICAL 时触发
### 3.4 通知内容模板
#### 邮件模板
```
主题:【巡检报告】{target} - {suite} - {status}
正文:
巡检目标:{target_name}
巡检套件:{suite_name}
完成时间:{finished_at}
汇总:正常 {normal} 项,警告 {warning} 项,严重 {critical} 项
{if has_abnormal}
异常项:
{for item in abnormal_items}
- {module}: {name} = {value}(阈值:{threshold})
{endfor}
{endif}
报告链接:{report_url}
```
#### 钉钉/企业微信模板
```
【巡检报告】{target_name}
套件:{suite_name}
时间:{finished_at}
结果:正常 {normal} / 警告 {warning} / 严重 {critical}
链接:{report_url}
```
## 4. 数据模型
### notifications.json
```json
{
"email": {
"enabled": false,
"smtp_host": "",
"smtp_port": 465,
"smtp_user": "",
"smtp_password": "",
"use_ssl": true,
"recipients": [],
"subject_template": "【巡检报告】{target} - {status}"
},
"dingtalk": {
"enabled": false,
"webhook": "",
"secret": "",
"at_mobiles": []
},
"wecom": {
"enabled": false,
"webhook": ""
},
"trigger": {
"on_complete": true,
"on_abnormal_only": false,
"include_details": true,
"include_link": true
},
"updated_at": "2026-07-21T10:00:00",
"updated_by": "admin"
}
```
## 5. 后端实现
### 5.1 新增文件
| 文件 | 用途 |
|------|------|
| `service_monitor/services/notification_service.py` | 通知服务:发送邮件/钉钉/企业微信 |
| `service_monitor/templates/service_monitor/notification.html` | 配置页面 |
### 5.2 notification_service.py 核心函数
```python
def get_config() -> dict # 获取配置
def save_config(data: dict) # 保存配置
def test_email() -> dict # 发送测试邮件
def test_dingtalk() -> dict # 发送测试钉钉消息
def test_wecom() -> dict # 发送测试企业微信消息
def send_notification(report: dict, report_url: str) # 根据配置发送通知
```
### 5.3 调用时机
`runner_service.run_inspection_sync()` 完成后调用:
```python
# 定时任务执行完成后
if result.get("success"):
report = report_service.get_report(result["report_id"])
notification_service.send_notification(report, report_url)
```
## 6. API 设计
| 方法 | 路径 | 说明 |
|------|------|------|
| GET | `/service-monitor/notification` | 配置页面 |
| GET | `/api/service-monitor/notification` | 获取配置 |
| PUT | `/api/service-monitor/notification` | 保存配置(管理员) |
| POST | `/api/service-monitor/notification/test-email` | 测试邮件 |
| POST | `/api/service-monitor/notification/test-dingtalk` | 测试钉钉 |
| POST | `/api/service-monitor/notification/test-wecom` | 测试企业微信 |
## 7. 安全考虑
1. **敏感信息加密**:SMTP 密码、钉钉密钥使用 Fernet 加密存储(复用现有 `crypto.py`
2. **权限控制**:仅管理员可配置和查看敏感信息
3. **接口保护**:测试接口加频率限制(防止滥用)
## 8. 不做的事
- 不支持短信通知(需要付费服务商)
- 不支持 Slack 等国外平台
- 不支持通知历史记录(后续可扩展)
- 不支持多套通知配置(如不同目标发不同人)
## 9. 后续扩展
- 支持按目标配置不同通知渠道
- 支持通知历史记录和重发
- 支持通知频率限制(如 1 小时内同类异常只通知一次)
\ No newline at end of file
......@@ -22,7 +22,7 @@ from flask import (
Response, redirect, url_for, stream_with_context,
)
from .services import target_service, report_service, runner_service, schedule_service
from .services import target_service, report_service, runner_service, schedule_service, notification_service
logger = logging.getLogger("service_monitor.routes")
......@@ -113,6 +113,9 @@ def page_schedule():
return redirect(url_for('auth.login'))
user = _current_user()
schedules = schedule_service.list_schedules()
# 为每个 schedule 生成自然语言描述
for s in schedules:
s['description'] = schedule_service._describe_schedule(s)
targets = target_service.list_targets(role=user.get('role', ''))
return render_template(
'service_monitor/schedule.html',
......@@ -122,6 +125,22 @@ def page_schedule():
)
@bp.route('/service-monitor/notification')
def page_notification():
"""通知配置页面(仅管理员)。"""
if 'user' not in session:
return redirect(url_for('auth.login'))
user = _current_user()
if user.get('role') != 'admin':
return redirect(url_for('service-monitor.page_index'))
config = notification_service.get_config_masked()
return render_template(
'service_monitor/notification.html',
user=user, config=config,
is_admin=True, active_menu='notification',
)
@bp.route('/service-monitor/run/<target_id>')
def page_run(target_id):
"""巡检执行页(仅管理员)。"""
......@@ -297,6 +316,22 @@ def api_delete_report(report_id):
return jsonify({"success": ok})
@bp.route('/api/service-monitor/reports/batch', methods=['DELETE'])
def api_batch_delete_reports():
"""批量删除巡检报告(管理员)。"""
guard = _require_admin_json()
if guard:
return guard
ids = (request.get_json(force=True) or {}).get("ids", [])
if not ids:
return jsonify({"success": False, "error": {"code": 400, "message": "未选择报告"}}), 400
deleted = 0
for rid in ids:
if report_service.delete_report(rid):
deleted += 1
return jsonify({"success": True, "deleted": deleted})
@bp.route('/api/service-monitor/reports/<report_id>/export', methods=['GET'])
def api_export_report(report_id):
guard = _require_login_json()
......@@ -400,3 +435,63 @@ def api_toggle_schedule(schedule_id):
return jsonify({"success": True, "schedule": sched})
except ValueError as e:
return jsonify({"success": False, "error": {"code": 400, "message": str(e)}}), 400
# ============================================================
# API:通知配置(管理员)
# ============================================================
@bp.route('/api/service-monitor/notification', methods=['GET'])
def api_get_notification():
"""获取通知配置。"""
guard = _require_admin_json()
if guard:
return guard
config = notification_service.get_config_masked()
return jsonify({"success": True, "config": config})
@bp.route('/api/service-monitor/notification', methods=['PUT'])
def api_save_notification():
"""保存通知配置。"""
guard = _require_admin_json()
if guard:
return guard
try:
data = request.get_json(force=True) or {}
config = notification_service.save_config(
data, updated_by=_current_user().get('username', 'admin')
)
return jsonify({"success": True, "config": config})
except Exception as e:
return jsonify({"success": False, "error": {"code": 400, "message": str(e)}}), 400
@bp.route('/api/service-monitor/notification/test-email', methods=['POST'])
def api_test_email():
"""测试邮件通知。"""
guard = _require_admin_json()
if guard:
return guard
result = notification_service.test_email()
return jsonify(result)
@bp.route('/api/service-monitor/notification/test-dingtalk', methods=['POST'])
def api_test_dingtalk():
"""测试钉钉通知。"""
guard = _require_admin_json()
if guard:
return guard
result = notification_service.test_dingtalk()
return jsonify(result)
@bp.route('/api/service-monitor/notification/test-wecom', methods=['POST'])
def api_test_wecom():
"""测试企业微信通知。"""
guard = _require_admin_json()
if guard:
return guard
result = notification_service.test_wecom()
return jsonify(result)
......@@ -268,6 +268,19 @@ def run_inspection_sync(target_id: str, suite: str) -> dict:
for k in total_summary:
total_summary[k] += ms.get(k, 0)
# 发送通知(定时任务触发的巡检)
try:
from . import notification_service
report = report_service.get_report(report_id)
if report:
# 构建报告链接(需要从配置获取外部访问地址)
import os
host = os.environ.get('EXTERNAL_HOST', 'http://192.168.5.60:8088')
report_url = f"{host}/service-monitor/report/{report_id}"
notification_service.send_notification(report, report_url)
except Exception as e:
logger.warning("发送通知失败: %s", e)
return {
"success": True,
"report_id": report_id,
......
......@@ -23,6 +23,7 @@ DATA_DIR = MODULE_DIR / "data"
REPORTS_DIR = DATA_DIR / "reports"
TARGETS_FILE = DATA_DIR / "targets.json"
SCHEDULES_FILE = DATA_DIR / "schedules.json"
NOTIFICATIONS_FILE = DATA_DIR / "notifications.json"
def ensure_dirs() -> None:
......
......@@ -224,6 +224,10 @@
<span>定时任务</span>
</a>
{% if is_admin %}
<a href="/service-monitor/notification" class="nav-item {{ 'active' if active_menu == 'notification' }}">
<span class="nav-icon">🔔</span>
<span>通知配置</span>
</a>
<a href="/service-monitor/targets" class="nav-item {{ 'active' if active_menu == 'manage' }}">
<span class="nav-icon">⚙️</span>
<span>目标管理</span>
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论