feat(monitor): 钉钉机器人关键词与负责人员配置（阶段十五）

钉钉通知增强：
- notification_service.py 新增 _build_dingtalk_markdown（关键词去重补全 + 负责人末尾追加）与 _send_dingtalk_message（统一加签+发送），三个钉钉调用点收敛复用
- DEFAULT_CONFIG.dingtalk 新增 keyword（可选，空=不处理）、responsible_person（默认"陈泽键"，空=不展示）
- 前端 Notification.vue 钉钉卡片新增"自定义关键词/负责人员"表单，DingTalkConfig 类型同步，旧配置自动回填
- 新增 tests/test_notification_service.py 12 用例（全部 mock HTTP，零真实发送）

一并入库（遗留变更）：
- deploy/deploy_update.py：5.60 Docker 容器热更新部署脚本（密钥免密 + tar 进容器 + 健康检查），替代已过时的 upload_to_server.py
- PRD_问题处理/计划执行_监测统计接口404 ×2：8-25 统计接口 404 修复的配套文档补交
- HANDOFF_服务监测：阶段十五进度 + 坑 25-29 + 下一步计划

验证：12 个新单测全绿、vue-tsc 类型检查通过、已部署 5.60（/api/health 正常）
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Docs/需求文档/服务监测/PRD_计划执行_钉钉机器人关键词与负责人员配置.md

+# 计划执行：钉钉机器人关键词与负责人员配置
+
+> 来源：PRD_需求文档_钉钉机器人关键词与负责人员配置.md
+> 日期：2026-09-14
+> 状态：执行中
+
+## 1. 实现范围
+
+### 1.1 后端通知服务
+
+文件：`skill/code/web/service_monitor/services/notification_service.py`
+
+- 在 `DEFAULT_CONFIG.dingtalk` 增加 `keyword` 和 `responsible_person` 默认字段。
+- 增加统一的钉钉 Markdown 文本处理逻辑：
+  - 关键词非空且正文不包含时，补到标题和正文。
+  - 负责人员非空时，追加责任人行。
+- 应用于巡检完成、报告缺失告警、测试消息三类钉钉消息。
+- 保持敏感字段加密逻辑不变，兼容缺少新字段的旧配置。
+
+### 1.2 前端配置页面
+
+文件：`frontend/src/views/service-monitor/Notification.vue`
+
+- 扩展初始钉钉配置对象。
+- 在钉钉机器人区域增加关键词输入框和负责人员输入框。
+- 加载旧配置时通过默认值兼容缺失字段。
+- 保存时沿用现有通知配置接口，不新增 API。
+
+### 1.3 前端类型
+
+文件：`frontend/src/types/service-monitor.ts`
+
+- `DingTalkConfig` 增加 `keyword: string` 和 `responsible_person: string`。
+
+### 1.4 测试
+
+- 增加后端纯函数/消息构造测试，验证关键词去重、责任人追加及空配置兼容。
+- 执行服务监测 pytest。
+- 执行前端 TypeScript 构建或项目既有检查命令。
+- 不调用钉钉测试接口，不发送真实消息。
+
+## 2. 实施步骤
+
+1. 确认当前分支和工作区，避免覆盖已有用户文件。
+2. 修改后端默认配置和三类钉钉消息构造逻辑。
+3. 修改 Vue 页面及 TypeScript 类型。
+4. 编写或更新自动化测试，使用 mock HTTP 请求，不访问真实 Webhook。
+5. 运行后端测试和前端编译检查。
+6. 检查 git diff，确认未修改邮件/企微行为、加签算法和 API 路径。
+7. 汇总验证结果；真实钉钉联调由用户手工完成。
+
+## 3. 风险与处理
+
+| 风险 | 处理方式 |
+|---|---|
+| 旧 notifications.json 缺少字段 | 读取时使用 `.get()` 和默认值，页面初始化提供完整默认结构 |
+| 关键词重复显示 | 仅当正文不含关键词时补充，并同步保证标题含关键词 |
+| 报告缺失告警漏加配置 | 复用统一处理函数，覆盖所有钉钉 Markdown 消息 |
+| 加签逻辑回归 | 不修改 URL 签名代码，仅修改消息内容 |
+| 误发真实通知 | 测试只 mock HTTP 请求，不调用测试发送接口 |
+
+## 4. 验收清单
+
+- [ ] PRD 与本计划文档已生成。
+- [ ] 关键词和负责人员可在页面配置、保存、回显。
+- [ ] 三类钉钉消息按规则包含关键词和责任人。
+- [ ] 空值和旧配置兼容。
+- [ ] 自动化测试通过。
+- [ ] 前端编译检查通过。
+- [ ] 未发送真实钉钉测试消息。

Docs/需求文档/服务监测/PRD_问题处理_监测统计接口404.md

+# 问题处理：监测统计接口 404
+
+## 文档信息
+
+- 创建时间：2026-08-25
+- 问题来源：用户反馈（5.60 服务界面上监测统计页面报 404）
+- 关联模块：service_monitor / statistics、notification
+- 排查人：czj
+
+---
+
+## 1. 问题描述
+
+### 错误信息
+
+浏览器（监测统计页面）发起以下两个接口请求，均返回 HTTP 404：
+
+```
+GET http://192.168.5.60:8088/api/service-monitor/notification/check-missing-reports
+GET http://192.168.5.60:8088/api/service-monitor/statistics/health-trend?start_date=2026-07-26&end_date=2026-08-25
+```
+
+### 触发条件
+
+打开 `http://192.168.5.60:8088/service-monitor/statistics`（监测统计）页面时触发。
+
+---
+
+## 2. 根因分析
+
+### 结论：后端路由完全没有实现（实现缺失，非路径写错）
+
+对全代码库（`skill/code/web/`、`frontend/src/`、`frontend/dist/`、模板）进行 `check-missing-reports`、`health-trend` 字符串检索：
+
+| 位置 | 结果 |
+|------|------|
+| `service_monitor/routes.py`（677 行，API 路由全集） | ❌ 无 `health-trend` 端点；❌ 无 `check-missing-reports` 端点 |
+| `services/statistics_service.py` | ❌ 无 `get_health_trend()` 函数 |
+| `services/notification_service.py` | ❌ 无 `check_missing_reports()` 函数 |
+| `frontend/src/api/service-monitor/statistics.ts` | ❌ 只有 5 个 API 包装，无 `health-trend` |
+| `frontend/src/api/service-monitor/notification.ts` | ❌ 只有 GET/PUT/3 个测试端点，无 `check-missing-reports` |
+| `frontend/src/views/service-monitor/Statistics.vue` | ❌ 未调用 health-trend |
+| 模板 `templates/service_monitor/statistics.html` | ❌ 未调用 |
+| `frontend/dist/`（生产构建产物） | ❌ 不存在该目录（本地未构建） |
+| Git 全历史（`git log -S`） | ❌ 从未提交过这两个端点 |
+
+### 为什么是 404 而不是 500
+
+Flask 蓝图（Blueprint，前缀 `/api/service-monitor/*`）中未注册的路由，会被 Flask 默认的 404 处理逻辑拦截，返回 `{"success": false, ...}` / HTML 404。这与"服务端逻辑报错"（500）不同——服务端**根本不知道**这两个接口的存在。
+
+### 为什么页面会出现两个"孤儿"请求
+
+监测统计页面（Vue SPA，浏览器加载的 `/assets/index-*.js`）调用了这两个接口，但：
+
+1. **`health-trend`**：这是统计模块 PRD（`PRD_需求文档_监测统计模块.md`）中规划的"查看各目标服务器的健康趋势"功能，设计留有前端调用，但后端一直未实现对应聚合函数与路由。
+2. **`check-missing-reports`**：规划用于"定时巡检是否有缺失报告"的运维提示，同样只停留在前端调用层面。
+
+前端构建版本与后端代码版本不同步（前端 SPA 已包含趋势/缺失检测逻辑，后端从未实现），导致线上 404。
+
+---
+
+## 3. 影响范围
+
+| 影响 | 说明 |
+|------|------|
+| 监测统计页面部分模块加载失败 | 健康趋势图表、缺失报告提示拉取失败，控制台出现 404 |
+| 统计数据不完整 | 页面其余统计（概览/模块/检测项趋势）正常，但趋势模块无数据 |
+| 部署一致性隐患 | 前端已构建、后端未实现，运行版本与源码仓库不一致 |
+
+---
+
+## 4. 修复方案
+
+### 4.1 健康趋势（statistics）
+
+在 `services/statistics_service.py` 新增 `get_health_trend()`：
+
+- 复用现有 `_parse_date_filter()` / `_load_reports_in_range()`（30 天默认、日期校验、报告加载逻辑完全一致）
+- 按自然日（`started_at[:10]`）分组，聚合每日的 `正常 / 警告 / 严重` 数量
+- 返回按日期升序的每日时段数组
+
+在 `routes.py` 新增：
+
+```python
+@bp.route('/api/service-monitor/statistics/health-trend', methods=['GET'])
+def api_statistics_health_trend():
+    # 参数：target_id / start_date / end_date（登录即可查看）
+    # 返回：{"success": True, "data": {...}}
+```
+
+### 4.2 缺失报告检测（notification）
+
+在 `services/notification_service.py` 新增 `check_missing_reports()`：
+
+- 读取 `schedule_service.list_schedules()` 与 `report_service.list_reports()`
+- 对每个**启用中**的定时任务，按 `start_date ~ 今天` 生成应执行日期集合（daily/weekday/weekly 按 cron 语义）
+- 与已生成的报告进行比对，输出缺失日期/时间段列表（含目标名、任务名、缺失日期、最近一次报告时间）
+
+在 `routes.py` 新增：
+
+```python
+@bp.route('/api/service-monitor/notification/check-missing-reports', methods=['GET'])
+def api_check_missing_reports():
+    # 参数：days（默认 7）/ target_id（可选）
+    # 返回：{"success": True, "missing": [...]}
+```
+
+### 4.3 前端同步
+
+- `frontend/src/api/service-monitor/statistics.ts` 增加 `getHealthTrend()` 包装
+- `frontend/src/api/service-monitor/notification.ts` 增加 `checkMissingReports()` 包装
+- 业务逻辑页面（Statistics.vue / 通知页）按需挂载（本仓不含前端构建产物，由部署侧构建 dist 后同步）
+
+### 4.4 部署
+
+后端代码走 `deploy/upload_to_server.py`（`service_monitor` 目录递归上传，自动排除 `data/` 不覆盖运行时数据）。前端需在构建机重新构建 `frontend/dist/` 并同步到服务器 `/opt/troubleshoot/dist/`（避免坑 6：空 volume 覆盖）。
+
+---
+
+## 5. 预防措施
+
+1. **前后端接口对齐检查**：前端新增 API 调用前，先确认后端路由存在（`grep` 路由注册）；后端新增接口后，同步更新前端 API 包装。
+2. **部署清单同步**：新增代码文件后，同步检查 `deploy/upload_to_server.py` 的 `FILES_TO_UPLOAD` / `DIRS_TO_UPLOAD` / `RECURSIVE_DIRS_TO_UPLOAD`（service_monitor 子包已覆盖递归上传，无需新增）。
+3. **部署后接口冒烟**：部署后对新增接口执行 `curl` 冒烟（GET 返回 200 非 404 即通过）。
+4. **版本一致性**：前端 dist 与后端代码必须同批次部署，避免运行版本错位。
+
+---
+
+## 6. 验收标准
+
+| 标准 | 验证方法 |
+|------|---------|
+| 两个接口不再 404 | `curl http://192.168.5.60:8088/api/service-monitor/statistics/health-trend` 返回 200（未登录 401 也可，属于已注册路由） |
+| health-trend 返回结构化数据 | 返回 `data.days[]`，含 date / summary 的 正常/警告/严重 统计 |
+| check-missing-reports 返回检测结果 | 返回 `missing[]`，无缺失时为空数组 |
+| 统计页面趋势模块正常 | 浏览器打开统计页面，控制台无 404，趋势图表有数据 |
+| 单元测试通过 | `cd skill/code && python -m pytest -v`（现有 143 用例不回归） |
\ No newline at end of file

Docs/需求文档/服务监测/PRD_需求文档_钉钉机器人关键词与负责人员配置.md

+# PRD：钉钉机器人关键词与负责人员配置
+
+> 创建日期：2026-09-14  
+> 模块：服务监测 / 通知配置  
+> 状态：待实现
+
+## 1. 背景
+
+服务监测当前支持钉钉自定义机器人通知，但通知配置页面仅支持 Webhook、加签密钥和 @人员手机号。钉钉机器人可启用“自定义关键词”安全校验，消息内容必须包含关键词，否则消息会被钉钉拦截。
+
+同时，巡检通知需要明确展示负责人员，当前通知正文没有责任人信息，且责任人应支持按部署环境配置。
+
+## 2. 需求目标
+
+1. 为钉钉机器人增加可选的自定义关键词配置。
+2. 配置关键词后，所有钉钉 Markdown 通知自动包含该关键词，避免因关键词校验导致发送失败。
+3. 为钉钉通知增加“负责人员”配置，默认值为“陈泽键”。
+4. 所有钉钉业务通知正文增加一行“负责人员：{配置值}”。
+5. 兼容已有通知配置；旧配置没有新增字段时不报错。
+6. 本次实现只进行代码和自动化验证，不主动发送钉钉测试消息。
+
+## 3. 使用范围
+
+### 3.1 配置页面
+
+在服务监测“通知配置”的“钉钉机器人”区域增加：
+
+- **自定义关键词（可选）**：单个文本，默认空；为空时不改变现有行为。
+- **负责人员**：单个文本，默认“陈泽键”；为空时不追加责任人行。
+
+配置随现有通知配置一起保存，并在再次打开页面时回显。
+
+### 3.2 通知类型
+
+以下钉钉 Markdown 消息均需要应用关键词和负责人员配置：
+
+- 巡检完成通知。
+- 报告缺失告警通知。
+- 钉钉测试消息模板（保留现有测试接口，但本次不主动调用）。
+
+## 4. 详细规则
+
+### 4.1 关键词
+
+- 关键词字段为可选字符串，去除首尾空白后使用。
+- 未配置关键词时，消息标题和正文保持原有格式。
+- 已配置关键词时，若正文未包含该关键词，则在正文开头自动增加关键词；若正文已经包含，则不得重复添加。
+- 标题也应包含关键词，确保钉钉 Markdown 关键词校验在不同机器人配置下均可通过。
+- 关键词不得进入加密字段处理，不需要脱敏。
+
+### 4.2 负责人员
+
+- 配置字段使用明确的英文键名 `responsible_person`。
+- 默认值为“陈泽键”。
+- 配置值去除首尾空白。
+- 配置非空时，在钉钉正文末尾增加独立一行：
+
+  `负责人员：陈泽键`
+
+- 配置为空时不增加空的责任人行。
+- 责任人配置不是密码，不加密、不脱敏。
+
+### 4.3 兼容性
+
+- 读取旧的 `notifications.json` 时，缺少 `keyword` 或 `responsible_person` 应使用默认值。
+- 保存时保留现有 email、wecom、trigger、report_missing 等配置字段。
+- 不改变现有接口路径、请求方法和测试接口权限。
+
+## 5. 非功能要求
+
+- 不发送真实钉钉测试消息作为本次开发验证步骤。
+- 不在日志中输出 Webhook、加签密钥等敏感信息。
+- 保持现有通知发送失败处理和返回值语义。
+- 代码遵循现有 routes/services/frontend 分层结构。
+
+## 6. 验收标准
+
+1. 旧配置加载成功，缺少新字段时页面显示默认责任人、空关键词。
+2. 页面可以编辑、保存并回显关键词和负责人员。
+3. 巡检完成钉钉消息配置关键词后，标题或正文包含关键词。
+4. 报告缺失钉钉告警配置关键词后，标题或正文包含关键词。
+5. 巡检完成和报告缺失通知在责任人非空时均包含责任人行。
+6. 关键词为空、责任人为空时行为符合规则，不产生多余空行或异常。
+7. Python 单元测试、前端 TypeScript 构建/检查通过。
+8. 开发过程中不调用 `test-dingtalk` 接口，不发送真实消息。
+
+## 7. 不包含内容
+
+- 不新增钉钉机器人创建、删除或连通性校验功能。
+- 不修改企业微信和邮件的正文格式。
+- 不改变钉钉加签算法和 Webhook 请求协议。
+- 不自动为用户发送测试消息。

deploy/deploy_update.py

+# -*- coding: utf-8 -*-
+"""
+deploy_update.py - 快速更新 5.60（不重建镜像，只传包重启容器）
+
+步骤：
+  1. docker inspect 找出容器 /data/dist 实际挂载的宿主目录
+  2. 上传后端 skill/code/web -> 服务器暂存目录（排除 tests/data/__pycache__）
+  3. 上传前端 frontend/dist -> 容器实际挂载目录（volume 热更新）
+  4. tar 管道 docker exec 把后端代码同步进容器 /app/web（排除 service_monitor/data 运行时数据）
+  5. docker restart troubleshoot
+  6. 验证 /api/health 与两个新接口路由（非 404 即部署成功）
+
+用法：
+  SSH_PASSWORD='xxx' python deploy_update.py     # 密码认证
+  python deploy_update.py                        # 自动尝试 ~/.ssh/id_ed25519 密钥认证
+"""
+
+import os
+import time
+
+import paramiko
+
+HOST = os.environ.get('SSH_HOST', '192.168.5.60')
+USER = os.environ.get('SSH_USER', 'ubains')
+PASSWORD = os.environ.get('SSH_PASSWORD', '')
+CONTAINER = 'troubleshoot'
+REMOTE_BASE = '/data/third_party/monitor-platform'
+
+# 密钥认证候选（优先级从高到低）
+_SSH_KEY_CANDIDATES = [
+    os.path.expanduser('~/.ssh/id_ed25519'),
+    os.path.expanduser('~/.ssh/id_rsa'),
+]
+
+# 本地路径
+REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
+LOCAL_WEB = os.path.join(REPO_ROOT, "skill", "code", "web")
+LOCAL_DIST = os.path.join(REPO_ROOT, "frontend", "dist")
+STAGE_WEB = f"{REMOTE_BASE}/stage_web"        # 后端暂存目录（tar 源）
+DIST_DIR = f"{REMOTE_BASE}/frontend/dist"     # 前端目录（deploy_docker.py 约定）
+
+# 递归上传排除
+_EXCLUDE_DIRS = {'__pycache__', 'tests', 'data', '.pytest_cache', 'logs', 'cache'}
+_EXCLUDE_SUFFIXES = ('.pyc', '.pyo')
+
+
+def run(ssh, cmd):
+    stdin, stdout, stderr = ssh.exec_command(cmd)
+    out = stdout.read().decode()
+    err = stderr.read().decode()
+    rc = stdout.channel.recv_exit_status()
+    return rc, out, err
+
+
+def upload_recursive(sftp, ssh, local_dir, remote_dir, remote_rel):
+    """递归上传，排除 tests/data/__pycache__/pyc。"""
+    ssh.exec_command(f'mkdir -p "{remote_dir}"')[1].channel.recv_exit_status()
+    count = 0
+    for dirpath, dirnames, filenames in os.walk(local_dir):
+        dirnames[:] = [d for d in dirnames if d not in _EXCLUDE_DIRS]
+        rel = os.path.relpath(dirpath, local_dir)
+        target = remote_dir if rel == '.' else f'{remote_dir}/{rel.replace(os.sep, "/")}'
+        if rel != '.':
+            ssh.exec_command(f'mkdir -p "{target}"')[1].channel.recv_exit_status()
+        for fname in filenames:
+            if fname.endswith(_EXCLUDE_SUFFIXES):
+                continue
+            lpath = os.path.join(dirpath, fname)
+            rpath = f'{target}/{fname}'
+            try:
+                sftp.put(lpath, rpath)
+                count += 1
+            except Exception as e:
+                print(f"    [FAIL] {fname}: {e}")
+    print(f"    [OK] {remote_rel}: {count} 个文件")
+
+
+def main():
+    print("=" * 60)
+    print("  Troubleshoot - Quick Update (no rebuild)")
+    print("=" * 60)
+    print(f"Server: {HOST}  Container: {CONTAINER}")
+
+    ssh = paramiko.SSHClient()
+    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+
+    try:
+        print("\n[1/6] Connecting...")
+        # 密码优先；未设密码则尝试密钥认证（~/.ssh/id_ed25519 / id_rsa）
+        connected = False
+        if PASSWORD:
+            ssh.connect(HOST, username=USER, password=PASSWORD, timeout=10)
+            connected = True
+            print("[OK] Connected via password")
+        else:
+            for keypath in _SSH_KEY_CANDIDATES:
+                if not os.path.isfile(keypath):
+                    continue
+                try:
+                    ssh.connect(HOST, username=USER, timeout=10, key_filename=keypath)
+                    connected = True
+                    print(f"[OK] Connected via key: {keypath}")
+                    break
+                except Exception as e:
+                    print(f"[INFO] Key auth failed ({keypath}): {e}")
+            if not connected:
+                print("[FAIL] 无可用认证方式：SSH_PASSWORD 未设置且密钥认证失败")
+                return
+
+        # 检查容器存在
+        rc, out, _ = run(ssh, f"sudo docker ps -a --filter name={CONTAINER} --format '{{{{.Names}}}}'")
+        if CONTAINER not in out:
+            print(f"[FAIL] 容器 {CONTAINER} 不存在，请先完成初始部署")
+            return
+
+        # 先 inspect 挂载点，确定前端实际目录
+        rc, out, _ = run(ssh,
+            f"sudo docker inspect {CONTAINER} --format "
+            "'{{range .Mounts}}{{.Destination}}|{{.Source}}{{println}}{{end}}'")
+        mounts = {}
+        for line in (out or "").splitlines():
+            if '|' in line:
+                dest, src = line.split('|', 1)
+                mounts[dest.strip()] = src.strip()
+        print("[OK] Mounts:")
+        for dest, src in mounts.items():
+            print(f"    {dest} <- {src}")
+
+        # 前端目标目录：优先容器 /data/dist 实际挂载源
+        dist_src = mounts.get('/data/dist')
+        if dist_src and dist_src.startswith('/'):
+            frontend_target = dist_src.rstrip('/')
+        else:
+            frontend_target = DIST_DIR
+        print(f"[INFO] 前端将上传到: {frontend_target}")
+
+        sftp = ssh.open_sftp()
+
+        print("\n[2/6] Uploading backend (skill/code/web -> stage_web)...")
+        upload_recursive(sftp, ssh, LOCAL_WEB, STAGE_WEB, "stage_web")
+
+        print(f"\n[3/6] Uploading frontend (frontend/dist -> {frontend_target})...")
+        upload_recursive(sftp, ssh, LOCAL_DIST, frontend_target, os.path.basename(frontend_target))
+
+        print("\n[4/6] Syncing backend into container (tar via docker exec)...")
+        # tar 管道把暂存目录同步进容器 /app/web
+        # 排除挂载点（写入会穿透到宿主或直接失败）：
+        #   ./users.json            — 绑定挂载的生产用户数据，绝不能覆盖
+        #   ./service_monitor/data  — monitor-data 运行时卷
+        tar_cmd = (
+            f'cd {STAGE_WEB} && '
+            f"tar --exclude='./users.json' --exclude='./service_monitor/data' -cf - . | "
+            f'sudo docker exec -i {CONTAINER} tar -xf - -C /app/web'
+        )
+        rc, out, err = run(ssh, tar_cmd)
+        if rc != 0:
+            print(f"[FAIL] 后端同步失败: {err.strip() or out.strip()}")
+            return
+        print("[OK] Backend synced into container /app/web")
+
+        print("\n[5/6] Restarting container...")
+        rc, out, err = run(ssh, f"sudo docker restart {CONTAINER}")
+        if rc != 0:
+            print(f"[FAIL] restart: {err.strip()}")
+            return
+        print("[OK] Container restarted")
+
+        # 等待服务加载（357 条知识库索引加载较慢）
+        print("[INFO] Waiting 10s for service...")
+        time.sleep(10)
+
+        print("\n[6/6] Verifying...")
+        rc, out, _ = run(ssh, "curl -s http://localhost:8088/api/health")
+        if out and '"status":"ok"' in out:
+            print("[OK] Health: " + out[:200])
+        else:
+            print(f"[FAIL] Health 未返回 ok: {out[:200]}")
+            rc, log, _ = run(ssh, f"sudo docker logs {CONTAINER} --tail 30")
+            print(log)
+
+        # 验证两个新接口（未登录返回 JSON success:false 而非 404）
+        for name, path in [
+            ("health-trend",
+             "/api/service-monitor/statistics/health-trend?start_date=2026-07-26&end_date=2026-08-25"),
+            ("check-missing-reports",
+             "/api/service-monitor/notification/check-missing-reports?days=7"),
+        ]:
+            rc, code, _ = run(ssh,
+                f'curl -s -o /dev/null -w "%{{http_code}}" "http://localhost:8088{path}"')
+            rc, body, _ = run(ssh,
+                f'curl -s "http://localhost:8088{path}" | head -c 200')
+            status = "OK" if code.strip() != "404" else "FAIL"
+            print(f"  [{status}] {name}: HTTP {code.strip()}  body: {(body or '')[:120]}")
+
+        print("\n" + "=" * 60)
+        print("  Update Complete!")
+        print(f"  URL: http://{HOST}:8088")
+        print("=" * 60)
+
+    except Exception as e:
+        print(f"\n[ERROR] {e}")
+        import traceback
+        traceback.print_exc()
+    finally:
+        try:
+            ssh.close()
+        except Exception:
+            pass
+
+
+if __name__ == '__main__':
+    main()
\ No newline at end of file

frontend/src/types/service-monitor.ts

@@ -258,6 +258,8 @@ export interface DingTalkConfig {
   webhook_url: string
   secret: string
   at_mobiles: string[]
+  keyword: string
+  responsible_person: string
 }
 
 export interface WeComConfig {

frontend/src/views/service-monitor/Notification.vue

@@ -21,6 +21,8 @@ const config = ref<NotificationConfig>({
     webhook_url: '',
     secret: '',
     at_mobiles: [],
+    keyword: '',
+    responsible_person: '陈泽键',
   },
   wecom: {
     enabled: false,
@@ -69,7 +71,7 @@ async function loadConfig() {
       config.value = {
         ...res.config,
         email: res.config.email || config.value.email,
-        dingtalk: res.config.dingtalk || config.value.dingtalk,
+        dingtalk: { ...config.value.dingtalk, ...(res.config.dingtalk || {}) },
         wecom: res.config.wecom || config.value.wecom,
         trigger: res.config.trigger || config.value.trigger,
         alert_on_consecutive: res.config.alert_on_consecutive || config.value.alert_on_consecutive,
@@ -257,6 +259,14 @@ onMounted(loadConfig)
         <el-form-item label="@人员手机号（可选）">
           <el-input v-model="dingtalkAtMobiles" placeholder="多个手机号用英文逗号分隔" />
         </el-form-item>
+        <el-form-item label="自定义关键词（可选）">
+          <el-input v-model="config.dingtalk.keyword" placeholder="钉钉机器人安全设置中的自定义关键词" />
+          <div class="form-hint">机器人开启"自定义关键词"校验时应填写，通知消息将自动包含该关键词</div>
+        </el-form-item>
+        <el-form-item label="负责人员">
+          <el-input v-model="config.dingtalk.responsible_person" placeholder="默认：陈泽键" />
+          <div class="form-hint">通知正文末尾展示一行"负责人员：xxx"</div>
+        </el-form-item>
         <div class="test-section">
           <el-button :loading="testingDingtalk" @click="handleTestDingtalk">发送测试消息</el-button>
           <span v-if="dingtalkTestResult" class="test-msg" :class="dingtalkTestResult.success ? 'success' : 'error'">

skill/code/web/service_monitor/services/notification_service.py

@@ -45,7 +45,11 @@ DEFAULT_CONFIG = {
         "enabled": False,
         "webhook_url": "",
         "secret": "",
-        "at_mobiles": []
+        "at_mobiles": [],
+        # 自定义关键词（机器人安全校验用，消息自动包含；空 = 不处理）
+        "keyword": "",
+        # 负责人员（通知正文末尾展示；空 = 不展示）
+        "responsible_person": "陈泽键"
     },
     "wecom": {
         "enabled": False,
@@ -189,9 +193,6 @@ def test_email() -> dict:
 
 def test_dingtalk() -> dict:
     """发送测试钉钉消息（Markdown 格式）。"""
-    import requests
-    import base64
-
     config = _load_config()
     ding_cfg = config.get("dingtalk", {})
 
@@ -211,34 +212,13 @@ def test_dingtalk() -> dict:
         f"这是一条测试消息，用于验证钉钉通知配置是否正确。\n\n"
         f"⏰ 发送时间：{now_str}\n"
     )
-    content = {
-        "msgtype": "markdown",
-        "markdown": {"title": "钉钉通知测试", "text": text}
-    }
 
-    # 签名（如果有）
-    url = webhook_url
-    if secret:
-        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).decode())
-        url = f"{webhook_url}&timestamp={timestamp}&sign={sign}"
-
-    try:
-        resp = requests.post(url, json=content, timeout=10)
-        result = resp.json()
-        if result.get("errcode") == 0:
-            logger.info("钉钉测试消息发送成功")
-            return {"success": True, "message": "测试消息已发送到钉钉群"}
-        else:
-            return {"success": False, "message": f"钉钉返回错误: {result.get('errmsg', '未知错误')}"}
-    except Exception as e:
-        return {"success": False, "message": f"发送失败: {e}"}
+    title = "钉钉通知测试"
+    md = _build_dingtalk_markdown(title, text, ding_cfg)
+    ok, errmsg = _send_dingtalk_message(webhook_url, secret, md["title"], md["text"])
+    if ok:
+        return {"success": True, "message": "测试消息已发送到钉钉群"}
+    return {"success": False, "message": f"发送失败: {errmsg}"}
 
 
 def test_wecom() -> dict:
@@ -416,34 +396,17 @@ def send_missing_report_notification(
     # 发送钉钉
     if "dingtalk" in channels and config.get("dingtalk", {}).get("enabled"):
         try:
-            import requests, base64
             ding_cfg = config.get("dingtalk", {})
             webhook_url = ding_cfg.get("webhook_url", "").strip()
             secret = ding_cfg.get("secret", "").strip()
 
             if webhook_url:
-                content = {
-                    "msgtype": "markdown",
-                    "markdown": {
-                        "title": f"报告缺失告警 - {target_count} 个目标",
-                        "text": md_msg,
-                    }
-                }
-                url = webhook_url
-                if secret:
-                    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).decode())
-                    url = f"{webhook_url}&timestamp={timestamp}&sign={sign}"
-
-                resp = requests.post(url, json=content, timeout=10)
-                result = resp.json()
-                if result.get("errcode") == 0:
+                title = f"报告缺失告警 - {target_count} 个目标"
+                md = _build_dingtalk_markdown(title, md_msg, ding_cfg)
+                ok, _ = _send_dingtalk_message(
+                    webhook_url, secret, md["title"], md["text"]
+                )
+                if ok:
                     success = True
         except Exception as e:
             logger.warning("报告缺失告警钉钉发送失败: %s", e)
@@ -551,9 +514,6 @@ def _send_email_notification(report: dict, report_url: str, config: dict) -> boo
 
 def _send_dingtalk_notification(report: dict, report_url: str, config: dict) -> bool:
     """发送钉钉通知（Markdown 格式，突出异常项）。"""
-    import requests
-    import base64
-
     ding_cfg = config.get("dingtalk", {})
     trigger_cfg = config.get("trigger", {})
 
@@ -626,42 +586,16 @@ def _send_dingtalk_notification(report: dict, report_url: str, config: dict) ->
         lines.append("")
         lines.append(f"📎 [查看完整报告]({report_url})")
 
-    content = {
-        "msgtype": "markdown",
-        "markdown": {
-            "title": f"{title_prefix} - {report.get('target_name', '-')}",
-            "text": "\n".join(lines)
-        }
-    }
-
-    if at_mobiles:
-        content["at"] = {"atMobiles": at_mobiles, "isAtAll": False}
-
-    # 签名
-    url = webhook_url
-    if secret:
-        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).decode())
-        url = f"{webhook_url}&timestamp={timestamp}&sign={sign}"
+    title = f"{title_prefix} - {report.get('target_name', '-')}"
+    md = _build_dingtalk_markdown(title, "\n".join(lines), ding_cfg)
 
-    try:
-        resp = requests.post(url, json=content, timeout=10)
-        result = resp.json()
-        if result.get("errcode") == 0:
-            logger.info("钉钉通知已发送")
-            return True
-        else:
-            logger.error("钉钉通知失败: %s", result.get("errmsg"))
-            return False
-    except Exception as e:
-        logger.error("钉钉通知发送失败: %s", e)
-        return False
+    ok, _ = _send_dingtalk_message(
+        webhook_url, secret, md["title"], md["text"], at_mobiles=at_mobiles
+    )
+    if ok:
+        logger.info("钉钉通知已发送")
+        return True
+    return False
 
 
 def _send_wecom_notification(report: dict, report_url: str, config: dict) -> bool:
@@ -706,3 +640,90 @@ def _send_wecom_notification(report: dict, report_url: str, config: dict) -> boo
     except Exception as e:
         logger.error("企业微信通知发送失败: %s", e)
         return False
+
+
+# ============================================================
+# 钉钉消息统一构造（关键词 + 负责人员）
+# ============================================================
+
+def _build_dingtalk_markdown(
+    title: str,
+    text: str,
+    ding_cfg: dict,
+) -> dict:
+    """按钉钉配置构造 Markdown 消息内容。
+
+    规则：
+      1. keyword 非空且正文不含关键词时，标题和正文开头各补一次（标题加粗）；
+         正文已含关键词则不重复添加。
+      2. responsible_person 非空时，正文末尾追加一行"负责人员：xxx"。
+      3. 空配置不改变原有消息格式。
+
+    Returns:
+        {"title": str, "text": str}
+    """
+    keyword = (ding_cfg.get("keyword") or "").strip()
+    person = (ding_cfg.get("responsible_person") or "").strip()
+
+    # 关键词：正文不含才补，避免重复
+    if keyword and keyword not in text:
+        title = f"[{keyword}] {title}"
+        text = f"{keyword}\n\n{text}"
+    elif keyword:
+        # 正文已含关键词时，标题不再重复添加，正文保持不变
+        pass
+
+    # 负责人员
+    if person:
+        text = f"{text.rstrip()}\n\n负责人员：{person}\n"
+
+    return {"title": title, "text": text}
+
+
+def _send_dingtalk_message(
+    webhook_url: str,
+    secret: str,
+    title: str,
+    text: str,
+    at_mobiles: list = None,
+) -> tuple:
+    """发送钉钉 Markdown 消息（含加签），统一封装请求发送。
+
+    Returns:
+        (是否成功, 失败原因 errmsg；成功时为空字符串)
+    """
+    import requests
+    import base64
+
+    content = {
+        "msgtype": "markdown",
+        "markdown": {"title": title, "text": text}
+    }
+    if at_mobiles:
+        content["at"] = {"atMobiles": at_mobiles, "isAtAll": False}
+
+    # 加签
+    url = webhook_url
+    if secret:
+        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).decode())
+        url = f"{webhook_url}&timestamp={timestamp}&sign={sign}"
+
+    try:
+        resp = requests.post(url, json=content, timeout=10)
+        result = resp.json()
+        if result.get("errcode") == 0:
+            logger.info("钉钉消息已发送")
+            return True, ""
+        errmsg = result.get("errmsg", "未知错误")
+        logger.error("钉钉消息失败: %s", errmsg)
+        return False, errmsg
+    except Exception as e:
+        logger.error("钉钉消息发送失败: %s", e)
+        return False, str(e)