部署故障分析及解决助手

部署故障分析及解决助手

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

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

触发条件

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

触发方式 判定标准 日志文件 用户上传/拖入 .json、.记录 文件，或指明文件路径 报错文本 用户消息中包含 error/错误/异常/故障/报错/fAIled/失败/异常/失败 等关键词 主动请求 用户说"帮我分析这个故障"/"这是什么错"/"帮我看下日志"等

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

故障知识库数据库 数据库信息

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

项 值 数据库 fault_knowledge_base 主表 fault_records 连接命令 docker exec resource_pool_mysql mysql -upool_user -ppool_password_2024 fault_knowledge_base 同步脚本 /data/work/scripts/同步_fault_knowledge_base.py 默认 Excel /data/work/bom/基础平台部署排期表-2026年度.xlsx Excel 同步能力

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

# 1. 先检查 Excel 可解析和有效行数 python3 /data/work/scripts/同步_fault_knowledge_base.py --dry-运行

# 2. 正式同步，重复执行不会重复插入同一条故障 python3 /data/work/scripts/同步_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_哈希) AS unique_哈希es FROM fault_records;"

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

docker exec -i resource_pool_mysql mysql -uroot -proot_password_2024 \ < /data/work/sql/创建_fault_knowledge_base.sql

字段含义 字段 含义 查询用途 模块_name 模块 第一层收窄范围，优先从报错中的产品/组件/任务名推断 issue_type 问题类型 第二层收窄范围，优先从故障现象推断 issue_description 问题描述 核心相似度匹配字段 solution_summary 解决方法概要 给出历史解决路径 product_version 交付产品集版本 版本相关问题过滤 delivery_branch 交付分支 架构/OS/分支差异过滤 resource_pool 资源池 判断是否为特定现场案例 核心流程 Step 0 — 保存原始输入

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

from pathlib 导入 Path from datetime 导入 datetime 导入 shutil

RAW_DIR = Path.home() / '.hermes/技能s/OpenClaw-导入s/部署-fault-分析器/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_输入.txt' raw_path.write_text(user_text, encoding='utf-8')

Step 1 — 读取并解析输入

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

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

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

记录s = data['记录s'] # 日志数组 # 提取字段： # 记录['type'] → 'error'|'警告'|'信息'|'成功' # 记录['message'] → 日志内容（含时间戳 [HH:MM:SS]） # 记录['timestamp']→ ISO 时间（JSON字段，可能缺失） # 记录['task_id'] → 任务ID（如 PREP_上传_33B3DC）

errors = [l for l in 记录s if l['type'] == 'error'] 警告s = [l for l in 记录s if l['type'] == '警告'] 信息s = [l for l in 记录s if l['type'] == '信息']

1B. 纯文本 .记录 文件

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

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

Step 2 — 错误自动归类与去重 2A. 中文错误模式自动归类

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

优先级 匹配模式（关键词） 类别 1 日期格式不正确 日期格式错误 格式yyyy-mm-dd 数据异常 — 日期格式 2 资源池名称不正确 名称不一致 sheet.≠ 配置错误 — 资源池名 3 未填写 必填项.未 必填.缺失 配置错误 — 必填字段 4 缺失 不存在 not found No such file 资源不足 — 文件/目录 5 格式不正确.IP IP地址格式 格式不合法 配置错误 — IP格式 6 already 安装ed is already 冲突 conflict 服务异常 — 安装冲突 7 已终止 用户.终止 用户.取消 用户操作 — 部署终止 8 验证ing verification 校验失败 验证失败 服务异常 — 验证超时 9 连接.拒绝 Connection refused timeout unreachable 网络/连接故障 10 host.not found DNS.fAIl 解析失败 网络/连接故障 11 bms.pri hostname.invalid 配置错误 — 主机名 12 权限 denied 权限不足 访问被拒 权限问题 13 OOM out of memory 磁盘空间不足 No space 资源不足 — 系统资源 14 模块NotFoundError 导入Error 依赖.缺失 依赖缺失

实现：

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 安装ed|is already|冲突|conflict', '服务异常'), (r'已终止|用户.终止|用户.取消', '用户操作'), (r'Connection refused|timeout|unreachable|连接.拒绝', '网络/连接故障'), (r'权限 denied|权限不足|访问被拒', '权限问题'), (r'OOM|out of memory|磁盘空间不足|No space', '资源不足'), (r'模块NotFoundError|导入Error|依赖.缺失', '依赖缺失'), ] for pattern, category in patterns: if re.搜索(pattern, msg): return category return '待分类'

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

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

过滤掉 用户操作类（已终止/用户终止）→ 不计入故障，仅在报告中标注 按 错误消息去重：取 message 中 ERROR 关键字后的核心文本去重 按 task_id + 类别 归并：同一任务同一类别的多条错误合并为一个根因 输出：每个根因一个 fault_data 字典 def deduplicate_errors(errors: 列出) -> 列出: """归并重复错误 → 根因列表""" # 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