部署故障分析及解决助手

当用户提供日志文件（.json .log）或包含 error/错误/异常/故障/报错/failed 等关键词的文字内容时，自动分析故障并生成 Word 文档。分析前必须优先查询 MySQL 故障知识库 fault_knowledge_base.fault_records，用历史案例提高定位准确率；所有新问题继续记录到 Excel 知识库中累积沉淀，并可同步到数据库。

⚠️ 交付规则： Word 报告生成后必须在同一条回复中附带故障分析摘要 + 发送 Word 文件给用户，不能只生成到本地而不发送。

触发条件

满足以下任一条件即触发：

排除：用户只是在陈述中顺带提到"没有错误"/"没问题"/"成功了"时不触发。

故障知识库数据库

数据库信息

故障知识库来自部署排期表第 10 个 sheet「故障库」，由 /data/work/scripts/sync_fault_knowledge_base.py 同步到 MySQL。

Excel 同步能力

当用户说“同步故障库”“更新故障库”“把 Excel 故障库入库”等含义时，执行以下流程：

# 1. 先检查 Excel 可解析和有效行数
python3 /data/work/scripts/sync_fault_knowledge_base.py --dry-run

# 2. 正式同步，重复执行不会重复插入同一条故障
python3 /data/work/scripts/sync_fault_knowledge_base.py

# 3. 验证行数和去重
docker exec resource_pool_mysql mysql -upool_user -ppool_password_2024 fault_knowledge_base \
  -e "SELECT COUNT(*) AS total, COUNT(DISTINCT content_hash) AS unique_hashes FROM fault_records;"

如数据库未初始化，先执行：

docker exec -i resource_pool_mysql mysql -uroot -proot_password_2024 \
  \x3C /data/work/sql/create_fault_knowledge_base.sql

字段含义

核心流程

Step 0 — 保存原始输入

任何输入在处理前必须先存档，防止后续分析覆盖原始数据：

from pathlib import Path
from datetime import datetime
import shutil

RAW_DIR = Path.home() / '.hermes/skills/openclaw-imports/deploy-fault-analyzer/data/raw'
RAW_DIR.mkdir(parents=True, exist_ok=True)

# 文件输入 → 复制到 raw/
ts = datetime.now().strftime('%Y-%m-%d_%H%M%S')
raw_path = RAW_DIR / f'{ts}_{Path(src_file).name}'
shutil.copy2(src_file, raw_path)

# 文本输入 → 写入 raw/
raw_path = RAW_DIR / f'{ts}_user_input.txt'
raw_path.write_text(user_text, encoding='utf-8')

Step 1 — 读取并解析输入

按文件格式选择解析策略：

1A. 结构化 JSON 日志（特征：顶层有 logs 数组，每项有 type/message 字段）

import json
with open(filepath) as f:
    data = json.load(f)

logs = data['logs']  # 日志数组
# 提取字段：
#   log['type']     → 'error'|'warning'|'info'|'success'
#   log['message']  → 日志内容（含时间戳 `[HH:MM:SS]`）
#   log['timestamp']→ ISO 时间（JSON字段，可能缺失）
#   log['task_id']  → 任务ID（如 PREP_UPLOAD_33B3DC）

errors   = [l for l in logs if l['type'] == 'error']
warnings = [l for l in logs if l['type'] == 'warning']
infos    = [l for l in logs if l['type'] == 'info']

1B. 纯文本 .log 文件

# 逐行读取，按关键词提取 ERROR/WARN/CRITICAL/FAIL 行
# 使用正则: ^\d{4}-\d{2}-\d{2}.*(ERROR|WARN|CRITICAL|FAIL)

1C. 用户粘贴文本 → 直接作为分析输入

Step 2 — 错误自动归类与去重

2A. 中文错误模式自动归类

使用关键词正则匹配，按优先级从高到低匹配：

实现：

def categorize_error_message(msg: str) -> str:
    """根据中文关键词自动归类错误"""
    patterns = [
        (r'日期格式不正确|日期格式错误|格式yyyy-mm-dd', '数据异常'),
        (r'资源池名称不正确|名称不一致', '配置错误'),
        (r'未填写|必填项.*未填|必填.*缺失', '配置错误'),
        (r'缺失|不存在|not found|No such file|no such file', '资源不足'),
        (r'格式不正确.*IP|IP地址格式|格式不合法', '配置错误'),
        (r'already installed|is already|冲突|conflict', '服务异常'),
        (r'已终止|用户.*终止|用户.*取消', '用户操作'),
        (r'Connection refused|timeout|unreachable|连接.*拒绝', '网络/连接故障'),
        (r'Permission denied|权限不足|访问被拒', '权限问题'),
        (r'OOM|out of memory|磁盘空间不足|No space', '资源不足'),
        (r'ModuleNotFoundError|ImportError|依赖.*缺失', '依赖缺失'),
    ]
    for pattern, category in patterns:
        if re.search(pattern, msg):
            return category
    return '待分类'

2B. 错误去重归并（关键步骤）

真实日志中同根因错误通常大量重复。必须先去重再分析：

1. 过滤掉 用户操作类（已终止/用户终止）→ 不计入故障，仅在报告中标注
2. 按 错误消息去重：取 message 中 ERROR 关键字后的核心文本去重
3. 按 task_id + 类别 归并：同一任务同一类别的多条错误合并为一个根因
4. 输出：每个根因一个 fault_data 字典

def deduplicate_errors(errors: list) -> list:
    """归并重复错误 → 根因列表"""
    # 1. 分离用户操作
    real_faults = [e for e in errors if '用户已终止' not in e['message'] and '用户终止' not in e['message']]
    
    # 2. 按消息核心去重
    seen = {}
    for e in real_faults:
        msg = e['message']
        # 提取核心：ERROR 后的文本或消息中独特部分
        core = re.sub(r'\[.*?\]', '', msg).strip()[:100]
        cat = categorize_error_message(core)
        key = f"{cat}:{core[:50]}"
        if key not in seen:
            seen[key] = {'count': 1, 'sample': e, 'category': cat, 'task_ids': {e.get('task_id','?')}}
        else:
            seen[key]['count'] += 1
            seen[key]['task_ids'].add(e.get('task_id', '?'))
    
    return list(seen.values())

Step 2.5 — 优先查询 MySQL 故障知识库（必须执行）

在进入根因分析前，必须先用去重后的核心错误到 fault_knowledge_base.fault_records 查询历史案例。查询目标是找到相同或相近的 module_name、issue_type、issue_description，并把命中的 solution_summary 作为解决方案候选，而不是直接凭经验生成结论。

2.5A. 先推断 module_name

优先从日志里的产品名、任务名、服务名、报错路径、部署阶段推断模块；无法确定时再不带模块查询。

高频 module_name 候选（按历史库频次和故障定位价值排序）：

2.5B. 再推断 issue_type

issue_type 用于第二层收窄。优先候选：

2.5C. 查询策略

按“强约束 → 放宽”的顺序查询，避免一开始全库模糊搜索导致误命中：

1. module_name + issue_type + issue_description 关键词查询
2. module_name + issue_description 查询
3. issue_type + issue_description 查询
4. 仅 issue_description 全文/LIKE 查询
5. 如果完全无命中，继续按原流程分析，并在报告中注明“历史故障库未找到高置信匹配”

推荐命令模板：

docker exec resource_pool_mysql mysql -upool_user -ppool_password_2024 fault_knowledge_base -e "
SELECT
  id, module_name, issue_type, found_at, product_version, delivery_branch,
  LEFT(issue_description, 220) AS issue_preview,
  LEFT(solution_summary, 220) AS solution_preview
FROM fault_records
WHERE module_name = '基座'
  AND issue_type = '脚本问题'
  AND (
    issue_description LIKE '%bootstrap%'
    OR solution_summary LIKE '%bootstrap%'
    OR MATCH(issue_description, solution_summary, remark) AGAINST('bootstrap 脚本 失败' IN NATURAL LANGUAGE MODE)
  )
ORDER BY
  (module_name = '基座') DESC,
  (issue_type = '脚本问题') DESC,
  updated_at DESC
LIMIT 5;"

如果无法稳定判断模块，使用关键词全库检索：

docker exec resource_pool_mysql mysql -upool_user -ppool_password_2024 fault_knowledge_base -e "
SELECT id, module_name, issue_type,
       LEFT(issue_description, 200) AS issue_preview,
       LEFT(solution_summary, 200) AS solution_preview
FROM fault_records
WHERE issue_description LIKE '%关键报错%'
   OR solution_summary LIKE '%关键报错%'
   OR remark LIKE '%关键报错%'
LIMIT 10;"

2.5D. 命中结果使用规则

• 高置信命中：module_name、issue_type、核心报错关键词均匹配，解决方案可作为首选，但仍要结合当前日志证据验证。
• 中置信命中：模块或问题类型匹配，但错误文本只部分相似，只能作为排查方向。
• 低置信命中：仅关键词相似，不能直接引用为结论，只在“历史类似案例”中备注。
• 如果多个历史案例冲突，优先选择同模块、同问题类型、同版本/分支的记录。
• Word 报告中增加“历史故障库匹配”小节，列出命中记录的模块、问题类型、问题摘要、解决摘要和置信度。

Step 3 — 故障信息提取（多故障）

从去重后的错误列表，为每个根因提取结构化信息：

提取方法：

任务ID解析（JSON日志特有）：

# task_id 前缀含义:
#   PREP_UPLOAD_* → 上传检查步骤
#   PREP_TASK_*   → 前置任务
#   CHECK_DEPLOY_* → 部署表检查
#   PREP_DEPLOY_*  → 部署准备

def parse_task_id(task_id: str) -> dict:
    """解析 task_id → 任务类型和阶段"""
    if task_id is None:
        return {'phase': '全局', 'type': '系统'}
    parts = task_id.split('_')
    if len(parts) >= 2:
        return {'phase': parts[0], 'type': parts[1], 'id': parts[-1] if len(parts) > 2 else ''}
    return {'phase': task_id, 'type': 'UNKNOWN'}

Step 4 — 根因分析

基于提取的信息，执行以下分析步骤：

1. 错误分类：归入以下类别之一

   • 🟥 网络/连接故障（ConnectionError, timeout, unreachable）
   • 🟧 配置错误（config file, parameter, permission）
   • 🟨 资源不足（OOM, disk full, CPU throttle, fd limit）
   • 🟩 服务异常（service down, crash, restart loop）
   • 🟦 权限问题（permission denied, unauthorized）
   • 🟪 依赖缺失（module not found, library mismatch）
   • ⬜ 数据异常（data corrupt, schema mismatch）

2. 历史案例对齐：把 Step 2.5 命中的故障库记录与当前日志证据逐项对比，确认模块、问题类型、关键词、版本/分支是否一致

3. 因果关系链：从堆栈自底向上追溯，找到最初触发点

4. 关键证据：摘录日志中能佐证根因的2-3条关键行

Step 5 — 生成解决方案

优先采用高置信历史案例中的 solution_summary，再结合当前日志、环境和标准排查路径细化；没有命中时才完全按通用框架生成方案。

输出解决方案时遵循的结构：

【解决方案】
1. 排查步骤
   - 第一步：xxx
   - 第二步：xxx
2. 修复操作
   - 操作命令（可复制执行）
3. 验证方法
   - 验证命令 + 预期结果
4. 回滚方案（如适用）

Step 6 — 生成交付物

6.1 判断生成模式

if len(root_causes) == 1:
    mode = 'single'   # 单故障 → 标准五段式报告
else:
    mode = 'multi'    # 多故障 → 总览 + 逐项分析 + 综合建议

6.2 Word 文档（单故障模式）

生成 部署故障分析及解决方案_YYYY-MM-DD_HHmmss.docx，标准五段式结构（同原版）。

6.3 Word 文档（多故障模式）

当 mode == 'multi' 时，增加总览页和综合建议：

封面
├── 总体概况表
│   ├── 文档编号 / 分析时间 / 部署目标 / 部署计划
│   ├── 总日志数 / 总错误数 / 归并根因数
│   └── 最高故障级别
├── 错误全景图（表格）
│   ├── # / 故障 / 类别 / 级别 / 出现次数
├── 逐项详细分析
│   ├── 故障 1：...（完整五段式）
│   ├── 故障 2：...
│   └── ...
└── 综合建议与整改清单
    └── 按优先级排列的改进项

重点：每个故障页有明确的 故障 N / 总N 标注，分隔线分隔。

6.4 使用 generate_report.py 脚本

from scripts.generate_report import generate_fault_report, generate_multi_fault_report, append_to_excel

# 单故障
if len(faults) == 1:
    docx_path = generate_fault_report(output_path, faults[0])

# 多故障
else:
    docx_path = generate_multi_fault_report(output_path, faults)

# 批量追加 Excel（所有故障都写一行）
for fault in faults:
    append_to_excel(xlsx_path, {...})

文档保存路径: ~/.hermes/skills/openclaw-imports/deploy-fault-analyzer/output/

Step 7 — 发送交付物（必须执行）

⚠️ 分析完成后必须在同一条回复中做两件事：

1. 贴出故障分析摘要（故障全景表 + 核心结论），让用户无需打开文件即可了解全貌
2. 发送 Word 文件 — 使用 message 工具发送文件给用户

# 发送 .docx 文件给用户（以当前会话的 chat 通道发送）
# 使用 message 工具: action=send, filePath=docx_path, caption='部署故障分析报告'

发送规则：

• 回复消息中贴摘要表（纯文本，不依赖 markdown 表格渲染）
• 同时调用 message 工具发送 Word 文件
• 发送后回复 NO_REPLY 避免重复消息

摘要模板（回复消息中贴出）：

🔍 部署故障分析结果

概览：N条日志，M条错误 → 归并 K 个根因

| # | 故障 | 类别 | 级别 | 次数 | 定位 |
F1 | xxx | 数据异常 | P1 | 44 | xxx
F2 | ...

🎯 核心结论：xxx

📄 Word 报告已同步发送，请查收。

Python 脚本模板

docx 生成脚本

from docx import Document
from docx.shared import Inches, Pt, Cm, RGBColor
from docx.enum.text import WD_ALIGN_PARAGRAPH
from datetime import datetime
import os

def generate_fault_report(output_path, fault_data):
    """
    fault_data = {
        'fault_time': '2026-05-09 10:23:45',
        'error_type': 'ConnectionError',
        'error_message': '...',
        'stack_trace': '...',
        'affected_component': 'nova-api',
        'affected_scope': 'AZ-1 计算节点',
        'related_config': '/etc/nova/nova.conf',
        'error_category': '网络/连接故障',
        'fault_level': 'P1',
        'root_cause': '...',
        'evidence_lines': ['...', '...'],
        'solution_steps': [...],
        'verification': '...',
        'prevention': '...',
    }
    """
    doc = Document()
    
    # 标题
    title = doc.add_heading('部署故障分析及解决方案', level=0)
    title.alignment = WD_ALIGN_PARAGRAPH.CENTER
    
    # 文档信息表
    doc.add_paragraph('')
    info_table = doc.add_table(rows=3, cols=2, style='Light Grid Accent 1')
    info_data = [
        ('文档编号', f"DOC-{datetime.now().strftime('%Y%m%d')}-{fault_data.get('fault_level','P2')}"),
        ('分析时间', datetime.now().strftime('%Y-%m-%d %H:%M:%S')),
        ('故障级别', fault_data.get('fault_level', 'P2')),
    ]
    for i, (k, v) in enumerate(info_data):
        info_table.rows[i].cells[0].text = k
        info_table.rows[i].cells[1].text = v
    
    doc.add_paragraph('')
    
    # 一、故障概述
    doc.add_heading('一、故障概述', level=1)
    doc.add_paragraph(f"故障时间：{fault_data.get('fault_time', '未知')}")
    doc.add_paragraph(f"影响组件：{fault_data.get('affected_component', '未知')}")
    doc.add_paragraph(f"影响范围：{fault_data.get('affected_scope', '未知')}")
    doc.add_paragraph(f"错误类型：{fault_data.get('error_type', '未知')}")
    
    # 二、故障详情
    doc.add_heading('二、故障详情', level=1)
    doc.add_heading('错误消息', level=2)
    doc.add_paragraph(fault_data.get('error_message', ''))
    
    if fault_data.get('stack_trace'):
        doc.add_heading('堆栈跟踪', level=2)
        p = doc.add_paragraph()
        p.style = doc.styles['No Spacing']
        for line in fault_data['stack_trace'].split('\
')[:20]:
            doc.add_paragraph(line, style='No Spacing')
    
    if fault_data.get('related_config'):
        doc.add_paragraph(f"关联配置：{fault_data['related_config']}")
    
    # 三、根因分析
    doc.add_heading('三、根因分析', level=1)
    doc.add_paragraph(f"错误分类：{fault_data.get('error_category', '未分类')}")
    doc.add_paragraph(f"根本原因：{fault_data.get('root_cause', '待进一步分析')}")
    
    if fault_data.get('evidence_lines'):
        doc.add_heading('关键证据', level=2)
        for i, line in enumerate(fault_data['evidence_lines'], 1):
            doc.add_paragraph(f"{i}. {line}")
    
    # 四、解决方案
    doc.add_heading('四、解决方案', level=1)
    for step in fault_data.get('solution_steps', []):
        doc.add_paragraph(step, style='List Bullet')
    
    if fault_data.get('verification'):
        doc.add_heading('验证方法', level=2)
        doc.add_paragraph(fault_data['verification'])
    
    # 五、预防措施
    doc.add_heading('五、预防措施', level=1)
    doc.add_paragraph(fault_data.get('prevention', '—'))
    
    doc.save(output_path)
    return output_path

边界情况与注意事项

当输入过于模糊时

如果日志/文本中找不到明确的错误信息，不要强行编造分析结果。此时：

• 输出一份简要的"日志初步筛查报告" Word 文档
• 在文档中标注"待补充信息"
• Excel 中错误类别记为"待分类"

当单个输入包含多个错误时（重要）

流程：

1. Step 2A 自动归类 → 按关键词将错误分入不同类别
2. Step 2B 错误去重 → 归并重复错误，统计每个根因出现次数
3. 过滤"用户操作"类（已终止/用户终止）→ 不计入故障清单
4. 对每个根因执行 Step 2.5，按 module_name、issue_type、issue_description 查询 MySQL 故障知识库
5. 按故障级别（P1→P2→P3）和历史命中置信度排序输出
6. Word 文档使用多故障模式 → 生成总览表 + 逐项分析 + 历史故障库匹配
7. Excel 中每个根因占一行，必要时同步到 MySQL 故障库

真实案例参考

以下是从实际部署日志中提取的典型错误模式，用于指导分类：

模式 1 — 部署表日期格式错误（44条 → 1个根因）

[CHECK_DEPLOY_F5B45D] 导入表：日期格式不正确45677.7661111, 格式yyyy-mm-dd hh:mm:ss
→ 类别：数据异常 | 根因：Excel日期序列号未转换 | 级别：P1

模式 2 — 资源池名称不一致（35条 → 1个根因）

[CHECK_DEPLOY_F5B45D] 资源池名称不正确 sheet11:呼和交付验证云池; sheet1:呼和国产化...
→ 类别：配置错误 | 根因：多sheet资源池名未统一 | 级别：P1

模式 3 — 部署包目录缺失（6条 → 1个根因）

[PREP_UPLOAD_33B3DC] /opt/ImageManager: 缺失
→ 类别：资源不足 | 根因：部署包未上传或不在此次部署范围 | 级别：P1

模式 4 — RPM 安装误报（1条 → 不阻塞）

[PREP_TASK_53658B] rpm -ivh → already installed → exit code ≠ 0
→ 类别：服务异常 | 根因：安装脚本未处理"已安装"状态 | 级别：P3

用户操作类不计入故障

以下消息类型不归类为故障，仅在报告中备注：

• 用户已终止部署流程 → 用户主动操作
• 部署已自动暂停，等待用户确认 → 流程机制，非故障本身
• 用户确认继续 → 流程恢复

路径规范

所有输出统一在 skill 目录下：

~/.hermes/skills/openclaw-imports/deploy-fault-analyzer/
├── SKILL.md
├── output/          # Word 文档输出
├── data/
│   ├── problems.xlsx   # Excel 知识库
│   └── raw/             # 原始输入存档

MySQL 故障知识库不在 skill 目录下，固定查询 fault_knowledge_base.fault_records；Excel 到数据库的同步脚本固定为 /data/work/scripts/sync_fault_knowledge_base.py。

快速参考