Skip to content

U
ubains-module-test
提交 bccc2b9a authored 作者: 陈泽健's avatar 陈泽健
浏览文件
操作

refactor(middleware): P1 级代码质量优化 — 日志系统、异常处理、工具模块抽取 

- 新增 utils/ 工具包：统一日志配置（logger）、错误码定义（error_codes）、API 响应封装（response）
- cache_manager.py：笼统 except Exception 拆分为 FileNotFoundError / JSONDecodeError / OSError 精准处理
- server.py：12 处 print / traceback.print_exc 替换为 logger.error / logger.exception
- server.py：新增网络异常（RequestException）和响应解析异常（KeyError/ValueError）捕获
- deploy/upload_to_server.py：数据源统一到 skill/code/web/，新增 DIRS_TO_UPLOAD 目录递归上传
- 新增 .claude/skills/HANDOFF/ 会话交接技能
- CLAUDE.md 注册 /handoff 技能入口
- Docs/ 新增 P1 级代码质量优化相关 PRD 文档
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 088a402c
展开全部 显示空白字符变更
内嵌 并排
正在显示 包含 821 行增加 和 30 行删除
.claude/skills/CreateCMD/_run_claude.bat
浏览文件 @ bccc2b9a
@echo off
cd /d E:\github\ubains-module-test\develop
cd /d E:\github\ubains-module-test\troubleshoot-ai-assistant
E:\nodejs\claude.cmd --permission-mode bypassPermissions
.claude/skills/HANDOFF/SKILL.md 0 → 100644
浏览文件 @ bccc2b9a
---
name: handoff
description: 会话交接文档生成 - 总结当前会话的任务、已完成项、卡点、下一步、踩坑警示，写入 HANDOFF.md 供新会话接手
---
会话结束前生成交接文档，把本次会话的上下文打包成一份完全没有背景的人也能看懂的 HANDOFF.md。
## Usage
```
/handoff # 生成本次会话的交接文档到 HANDOFF.md
/handoff 仅审查 # 仅输出交接摘要到对话，不写文件
```
## 为什么要这个 skill
会话结束后，新会话（或新人）接手时完全没有上下文。如果不主动总结，关键信息会随会话消失：
- 我们在做什么任务、为什么做
- 已经完成了哪些、验证到什么程度
- 当前卡在哪一步、下一步该干什么
- 踩过哪些坑、绝对不要再踩
一份好的交接文档能让接手者 5 分钟内进入状态，而不是花 30 分钟翻对话历史猜上下文。
## 输出位置
- **文件**：项目根目录 `HANDOFF.md`（与 CLAUDE.md 同级，最显眼处）
- **覆盖策略**：如已存在 HANDOFF.md，询问是覆盖、追加还是另存为 `HANDOFF_<日期>.md`
## 交接文档结构（固定六节）
```markdown
# HANDOFF — <任务一句话标题>
> 最后更新：YYYY-MM-DD HH:MM ｜ 分支：<branch> ｜ 负责人：<who>
## 1. 我们在做什么
（任务背景：为什么做、目标是什么、来源是哪个 PRD/需求/issue。2-3 段，让接手者知道"为什么"。）
## 2. 已经完成了什么
（按时间或子任务顺序列出已完成项，每项一行，标注✅。关键文件路径、关键决策、验证结果都要写明。
不要只写"做完了"，要写"做到了什么程度、怎么验证的"。）
## 3. 当前卡在哪
（现在停在哪一步、为什么停。是等用户确认？等外部依赖？遇到报错？还是自然完成等下一步指令？
如果是报错，贴关键错误信息。如果是等确认，写清楚在等什么、给谁。如果已全部完成没卡点，明确写"无卡点，等待下一步指令"。）
## 4. 下一步计划
（接手者打开新会话后第一件事该做什么。按优先级编号列出 2-5 条具体动作，每条可执行。
不要写"继续优化"这种模糊的，要写"执行 P1-2：在 skill/code/tests/ 下用 pytest 覆盖 search_engine.py"。）
## 5. 踩过的坑（绝对不要再踩）
（本次会话踩过的坑、犯过的错、浪费过时间的点。每条写：坑是什么 + 为什么会踩 + 怎么避免。
这是交接文档最有价值的一节——让接手者不重蹈覆辙。）
## 6. 关键文件与命令速查
（本次任务涉及的核心文件路径、常用命令、环境配置。让接手者不用翻目录就能上手。）
```
## 执行步骤
1. **回溯会话上下文**：扫描当前会话已完成的工具调用和结论，提取：任务目标、已完成子任务、当前状态、遇到的问题
2. **补充外部状态**：
- `git branch --show-current` 获取当前分支
- `git status --short` 获取工作区状态（哪些文件改了、哪些未提交）
- `git log --oneline -5` 获取最近提交（判断已提交 vs 待提交）
3. **检查 HANDOFF.md 是否已存在**：存在则用 AskUserQuestion 询问覆盖/追加/另存
4. **按六节结构生成文档**：
- 第 1 节：背景要写"为什么做"，不只是"做什么"
- 第 2 节：每项标✅，写明验证方式（如"HTTP 验证通过：/api/health 返回 ok"）
- 第 3 节：诚实写卡点，没有就写"无卡点"
- 第 4 节：下一步要具体到可执行动作，按优先级编号
- 第 5 节：**这是重点**——所有踩过的坑、走过的弯路、犯过的错，每条写清"坑 + 原因 + 避免方法"
- 第 6 节：列出本次涉及的关键文件路径和命令
5. **写入 HANDOFF.md** 并向用户展示摘要
6. **提醒**：是否要把 HANDOFF.md 加入 git（提交 or 忽略，由用户决定）
## 写作原则
- **写给完全没上下文的人看**：不假设读者知道任何前置信息，所有专有名词第一次出现要解释
- **诚实**：完成就写完成，没完成就写没完成，验证失败就贴失败原因。不要美化状态
- **具体优于抽象**："修改了 server.py 的异常处理"是坏的，"把 server.py 第 695-1245 行的 12 处 `except Exception` 改为 `logger.exception`"是好的
- **踩坑要写透**：不只是"踩了坑"，要写"为什么会踩、下次怎么第一时间避开"
- **可执行**：下一步计划要让接手者能直接照做，不用再决策
- **不堆代码**：交接文档是地图不是代码本，引用文件路径而非贴大段代码
## 注意事项
- **必须包含第 5 节踩坑**：这是交接文档的核心价值，没踩坑就写"本次会话较顺利，无重大踩坑"
- **不要把整个会话复述一遍**：交接文档是摘要，不是流水账。只保留对下一步有用的信息
- **关键决策要留痕**：为什么选 A 不选 B、为什么放弃某个方案——这些对话历史里查不到的决策理由要写明
- **敏感信息脱敏**：密码、密钥不要写进 HANDOFF.md，写"使用环境变量 SSH_PASSWORD（见 .env）"即可
- **文件路径用 repo 相对路径**：如 `skill/code/web/server.py`，不要用绝对路径
- **日期用绝对日期**：写 `2026-07-13` 不写"今天"，交接文档可能跨天看
CLAUDE.md
浏览文件 @ bccc2b9a
......@@ -20,6 +20,7 @@ config/ # 配置文件
- `/CreateCMD` - 创建 CMD 窗口
- `/prd-code` - PRD 代码生成
- `/prd-plan` - PRD 计划执行
- `/handoff` - 会话交接文档生成
## 环境配置
......
Docs/PRD_计划执行_P1级代码质量优化.md 0 → 100644
浏览文件 @ bccc2b9a
此差异已折叠。
Docs/PRD_需求文档_P1级代码质量优化.md 0 → 100644
浏览文件 @ bccc2b9a
# PRD_需求文档_P1级代码质量优化
## 基本信息
| 项目 | 内容 |
|------|------|
| 文档类型 | 需求文档 |
| 创建日期 | 2026-07-12 |
| 负责人 | 研发组 |
| 优先级 | P1（高优先级） |
| 状态 | P1-1 已完成 / P1-2、P1-3 待实施 |
---
## 一、背景与目标
### 1.1 问题背景
代码质量分析发现三个 P1 级别的代码质量问题：
| 问题编号 | 问题描述 | 影响范围 |
|----------|----------|----------|
| P1-1 | 异常处理不规范 | `cache_manager.py`、`server.py` 等多处 |
| P1-2 | 缺少单元测试 | 全项目 |
| P1-3 | 架构单薄，代码臃肿 | `server.py` 1415 行 |
### 1.2 修复目标
1. **规范异常处理**：细化异常类型，添加日志记录，消除静默失败
2. **建立测试框架**：引入 pytest，覆盖核心模块
3. **架构分层**：拆分 server.py，建立三层架构
---
## 二、需求详情
### 2.1 P1-1：异常处理规范化
#### 问题描述
当前代码存在以下不规范写法：
```python
# 问题1：空异常捕获
except:
pass
# 问题2：过宽异常捕获
except Exception as e:
print(f"错误: {e}")
```
#### 问题位置
| 文件 | 行号 | 问题类型 |
|------|------|----------|
| `cache_manager.py` | 79-81, 126-128, 161-162 | 空异常捕获 |
| `server.py` | 668-673, 706-712 | 过宽异常捕获 |
#### 需求规格
| 需求项 | 规格 |
|--------|------|
| 异常类型 | 必须捕获具体异常类型 |
| 日志记录 | 异常必须记录到日志 |
| 错误响应 | API 返回结构化错误信息 |
| 错误码 | 定义统一错误码规范 |
#### 验收标准
- [x] 无空异常捕获 `except:`
- [ ] 所有异常有具体类型（工具函数层已细化；路由层保留 `except Exception` 兜底属合理设计，避免漏捕）
- [x] 异常信息记录到日志文件
- [ ] API 返回统一错误格式（`error_response` 工具已就绪，路由层暂保持原响应结构以向后兼容，待 P1-3 重构时统一接入）
---
### 2.2 P1-2：添加单元测试
#### 问题描述
| 现状 | 问题 |
|------|------|
| 无本地单元测试 | 代码变更无自动化验证 |
| 仅有远程 SSH 测试 | 测试依赖网络和环境 |
| 无法快速回归 | 每次修改需手动测试 |
#### 需求规格
| 需求项 | 规格 |
|--------|------|
| 测试框架 | pytest |
| 覆盖模块 | search_engine、safety_filter、cache_manager |
| 覆盖率目标 | 核心模块 > 80% |
| CI 集成 | 可选，后续添加 |
#### 测试用例规划
| 模块 | 测试项 |
|------|--------|
| `search_engine.py` | 分词、TF-IDF 计算、相似度匹配 |
| `safety_filter.py` | 危险命令检测、敏感信息过滤 |
| `cache_manager.py` | 缓存读写、过期清理 |
#### 验收标准
- [ ] pytest 框架已配置
- [ ] 核心模块测试用例已编写
- [ ] 测试可通过 `pytest` 命令运行
- [ ] 核心模块覆盖率 > 80%
---
### 2.3 P1-3：架构分层重构
#### 问题描述
`server.py` 单文件包含：
- 路由定义（Controller）
- 业务逻辑（Service）
- 数据访问（Repository）
- 配置加载
- 工具函数
**问题**：
- 1415 行代码，难以维护
- 职责不清，耦合度高
- 难以单独测试各层
#### 目标架构
```
skill/code/web/
├── server.py # 入口文件（精简版）
├── routes/ # 路由层（Controller）
│ ├── troubleshoot.py
│ ├── auth.py
│ ├── cache.py
│ └── export.py
├── services/ # 业务层（Service）
│ ├── troubleshoot_service.py
│ ├── auth_service.py
│ └── cache_service.py
├── repositories/ # 数据层（Repository）
│ ├── search_repository.py
│ └── cache_repository.py
├── models/ # 数据模型
├── utils/ # 工具函数
└── templates/ # HTML 模板
```
#### 需求规格
| 需求项 | 规格 |
|--------|------|
| 入口文件 | server.py 仅保留路由注册和启动逻辑 |
| 路由层 | routes/ 目录，每个功能模块一个文件 |
| 业务层 | services/ 目录，封装业务逻辑 |
| 数据层 | repositories/ 目录，封装数据访问 |
| 向后兼容 | API 接口保持不变 |
#### 验收标准
- [ ] server.py 行数 < 100 行
- [ ] 路由、业务、数据三层分离
- [ ] 所有 API 功能正常
- [ ] 代码职责清晰，无循环依赖
---
## 三、影响范围
### 3.1 涉及文件
| 文件/目录 | 变更类型 |
|-----------|----------|
| `skill/code/web/server.py` | 重构 |
| `skill/code/web/cache_manager.py` | 修改 |
| `skill/code/web/search_engine.py` | 修改 |
| `skill/code/web/safety_filter.py` | 修改 |
| `skill/code/web/routes/` | 新增 |
| `skill/code/web/services/` | 新增 |
| `skill/code/web/repositories/` | 新增 |
| `skill/code/tests/` | 新增 |
| `skill/code/requirements.txt` | 修改（添加 pytest） |
### 3.2 风险评估
| 风险项 | 等级 | 缓解措施 |
|--------|------|----------|
| 重构引入 Bug | 中 | 分步重构，每步验证 |
| 测试覆盖不足 | 低 | 优先覆盖核心逻辑 |
| API 接口变化 | 低 | 保持接口签名不变 |
---
## 四、非功能需求
| 需求项 | 说明 |
|--------|------|
| 向后兼容 | API 接口保持不变 |
| 性能影响 | 无性能下降 |
| 可维护性 | 代码结构清晰，易于扩展 |
---
## 五、时间估算
| 任务 | 预计工时 |
|------|----------|
| P1-1：异常处理规范化 | 0.5 天 |
| P1-2：添加单元测试 | 1 天 |
| P1-3：架构分层重构 | 2 天 |
| 测试验证 | 0.5 天 |
| **总计** | **4 天** |
---
## 六、验收清单
### P1-1 验收
- [x] 无空异常捕获
- [x] 异常类型具体化（cache_manager 已细化；server.py 工具函数层已细化，路由层保留兜底属合理设计）
- [x] 日志记录完善（logs/app.log，控制台+文件双输出）
### P1-2 验收
- [ ] pytest 配置完成
- [ ] 核心模块测试覆盖 > 80%
- [ ] 测试命令可执行
### P1-3 验收
- [ ] 三层架构建立
- [ ] server.py < 100 行
- [ ] API 功能正常
Docs/PRD_需求文档_项目优化方向.md 0 → 100644
浏览文件 @ bccc2b9a
# PRD_需求文档_项目优化方向
## 基本信息
| 项目 | 内容 |
|------|------|
| 文档类型 | 优化规划 |
| 创建日期 | 2026-07-12 |
| 最后更新 | 2026-07-12 |
| 负责人 | 研发组 |
| 状态 | 持续跟踪 |
---
## 一、项目现状
### 1.1 已完成功能（v1.2.0）
| 功能模块 | 状态 | 说明 |
|---------|------|------|
| Web 排查服务 | ✅ 完成 | Flask 后端，端口 8088 |
| 智能搜索引擎 | ✅ 完成 | TF-IDF + 余弦相似度，357 条记录 |
| AI 分析集成 | ✅ 完成 | GLM-4-flash/5.1/4-plus 多模型支持 |
| 流式返回 (SSE) | ✅ 完成 | 实时输出排查结果 |
| 安全过滤器 | ✅ 完成 | 75+ 危险命令模式，敏感信息脱敏 |
| 登录认证 | ✅ 完成 | 用户/管理员角色，暴力破解保护 |
| 缓存管理 | ✅ 完成 | JSON 文件缓存，24 小时过期 |
| Word 导出 | ✅ 完成 | python-docx 生成报告 |
| 问题入库 | ✅ 完成 | 提交新问题到知识库 |
| Claude Skills | ✅ 完成 | 5 个技能集成 |
### 1.2 已修复问题
| 问题 | 优先级 | 修复日期 |
|------|--------|----------|
| 硬编码密码泄露 | P0 | 2026-07-12 |
| server.py 重复赋值 Bug | P0 | 2026-07-12 |
| skill/ 与 deploy/ 代码重复 | P0 | 2026-07-12 |
| 登录页泄露默认账号密码 | P1 | 2026-07-12 |
---
## 二、待优化方向
### 2.1 🔴 高优先级（P1）
#### 2.1.1 异常处理规范化
| 项目 | 说明 |
|------|------|
| **问题** | 空异常捕获 `except: pass`，过宽异常捕获 |
| **位置** | `cache_manager.py:126-128`，`server.py` 多处 |
| **风险** | 静默失败隐藏 Bug，调试困难 |
| **方案** | 细化异常类型，添加日志记录 |
| **工时** | 0.5 天 |
#### 2.1.2 添加单元测试
| 项目 | 说明 |
|------|------|
| **现状** | 无本地单元测试，仅有远程 SSH 测试 |
| **方案** | 引入 pytest 框架，覆盖核心模块 |
| **覆盖目标** | 搜索引擎、安全过滤器、缓存管理 |
| **工时** | 1 天 |
#### 2.1.3 架构分层重构
| 项目 | 说明 |
|------|------|
| **现状** | `server.py` 单文件 1415 行，无分层设计 |
| **问题** | 维护困难，职责不清 |
| **方案** | 拆分为 Controller/Service/Repository 三层 |
| **工时** | 2-3 天 |
---
### 2.2 🟠 功能增强（P2）
#### 2.2.1 语义搜索升级
| 项目 | 说明 |
|------|------|
| **现状** | TF-IDF + 关键词匹配 |
| **方案** | 引入 Embedding 向量搜索（如 text-embedding-3-small） |
| **收益** | 匹配准确率提升 20%+，理解语义相似问题 |
| **工时** | 2 天 |
#### 2.2.2 移动端适配
| 项目 | 说明 |
|------|------|
| **现状** | 仅支持桌面端 |
| **方案** | 响应式布局优化，支持手机浏览器访问 |
| **收益** | 扩大使用场景，支持现场移动办公 |
| **工时** | 1 天 |
#### 2.2.3 问题趋势分析
| 项目 | 说明 |
|------|------|
| **功能** | 按项目/时间/类型统计问题分布 |
| **输出** | 可视化图表（问题热力图、趋势折线图） |
| **收益** | 发现高频问题模式，提前预防 |
| **工时** | 2 天 |
#### 2.2.4 批量导入功能
| 项目 | 说明 |
|------|------|
| **功能** | 支持从 Excel/CSV 批量导入历史问题 |
| **收益** | 知识库扩充效率大幅提升 |
| **工时** | 1 天 |
---
### 2.3 🟢 技术优化（P3）
#### 2.3.1 添加类型注解
| 项目 | 说明 |
|------|------|
| **现状** | 所有 Python 文件缺少类型提示 |
| **方案** | 添加 Python 类型注解（Type Hints） |
| **收益** | IDE 支持更好，早期错误检测 |
| **工时** | 2 天 |
#### 2.3.2 API 文档
| 项目 | 说明 |
|------|------|
| **现状** | 仅有代码注释 |
| **方案** | 引入 Swagger/OpenAPI 自动生成文档 |
| **收益** | 接口文档标准化，便于对接 |
| **工时** | 1 天 |
#### 2.3.3 Docker 部署
| 项目 | 说明 |
|------|------|
| **现状** | 手动部署，依赖 shell 脚本 |
| **方案** | Docker Compose 一键部署 |
| **收益** | 部署标准化，环境一致性 |
| **工时** | 1 天 |
#### 2.3.4 监控告警
| 项目 | 说明 |
|------|------|
| **现状** | 仅日志输出 |
| **方案** | Prometheus + Grafana 监控面板 |
| **收益** | 实时监控服务状态，异常告警 |
| **工时** | 2 天 |
---
## 三、实施路线图
### 3.1 短期（本周）
| 任务 | 优先级 | 工时 | 状态 |
|------|--------|------|------|
| 异常处理规范化 | P1 | 0.5天 | 待开始 |
| 提交代码规范检查配置 | P3 | 0.2天 | 待开始 |
### 3.2 中期（下周）
| 任务 | 优先级 | 工时 | 状态 |
|------|--------|------|------|
| 添加 pytest 单元测试 | P1 | 1天 | 待开始 |
| 移动端响应式适配 | P2 | 1天 | 待开始 |
### 3.3 长期（本月）
| 任务 | 优先级 | 工时 | 状态 |
|------|--------|------|------|
| 语义搜索升级 | P2 | 2天 | 待开始 |
| 架构分层重构 | P1 | 2天 | 待开始 |
| Docker 部署方案 | P3 | 1天 | 待开始 |
---
## 四、技术栈规划
### 4.1 当前技术栈
| 层级 | 技术 |
|------|------|
| 前端 | 原生 HTML/CSS/JavaScript |
| 后端 | Python 3.10 + Flask |
| 搜索 | TF-IDF + 余弦相似度 |
| AI | GLM API |
| 存储 | Markdown + JSON |
| 认证 | Session + users.json |
### 4.2 规划升级
| 层级 | 当前 | 升级方向 |
|------|------|----------|
| 搜索 | TF-IDF | Embedding 向量搜索 |
| 认证 | Session | JWT Token（可选） |
| 存储 | JSON | SQLite/PostgreSQL（可选） |
| 部署 | Shell | Docker Compose |
| 监控 | 日志 | Prometheus + Grafana |
---
## 五、维护记录
| 日期 | 更新内容 | 更新人 |
|------|----------|--------|
| 2026-07-12 | 初始版本，记录已完成和待优化项 | Claude |
deploy/upload_to_server.py
浏览文件 @ bccc2b9a
......@@ -20,13 +20,22 @@ if not PASSWORD:
exit(1)
# 本地部署目录
LOCAL_BASE = os.path.dirname(os.path.abspath(__file__))
# P0-3 后统一数据源在 skill/code/web/，deploy/web/ 已删除
LOCAL_BASE = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "skill", "code", "web"))
# 需要上传的文件
# 远程 web 目录（与 deploy.sh 的 DEPLOY_DIR/web 对应）
REMOTE_WEB_DIR = REMOTE_BASE + "/web"
# 需要上传的单个文件（相对 LOCAL_BASE）
FILES_TO_UPLOAD = [
('web/server.py', 'web/server.py'),
('web/safety_filter.py', 'web/safety_filter.py'),
('web/cache_manager.py', 'web/cache_manager.py'),
('server.py', 'web/server.py'),
('cache_manager.py', 'web/cache_manager.py'),
]
# 需要上传的目录（相对 LOCAL_BASE → 远程 web 目录下同名子目录）
# P1-1 新增 utils/ 模块
DIRS_TO_UPLOAD = [
('utils', 'web/utils'),
]
def upload_files():
......@@ -52,6 +61,12 @@ def upload_files():
print("\n[3/4] Uploading files...")
sftp = ssh.open_sftp()
# 确保远程 web 目录存在
ssh.exec_command(f'mkdir -p {REMOTE_WEB_DIR}')[1].channel.recv_exit_status()
# 确保远程 utils 目录存在
ssh.exec_command(f'mkdir -p {REMOTE_BASE}/web/utils')[1].channel.recv_exit_status()
# 上传单个文件
for local_rel, remote_rel in FILES_TO_UPLOAD:
local_path = os.path.join(LOCAL_BASE, local_rel)
remote_path = REMOTE_BASE + "/" + remote_rel
......@@ -63,6 +78,24 @@ def upload_files():
else:
print(" [FAIL] File not found: " + local_path)
# 上传目录（递归）
for local_rel, remote_rel in DIRS_TO_UPLOAD:
local_dir = os.path.join(LOCAL_BASE, local_rel)
remote_dir = REMOTE_BASE + "/" + remote_rel
ssh.exec_command(f'mkdir -p {remote_dir}')[1].channel.recv_exit_status()
if not os.path.isdir(local_dir):
print(" [FAIL] Dir not found: " + local_dir)
continue
print(" Uploading dir: " + local_rel + "/")
for fname in os.listdir(local_dir):
lpath = os.path.join(local_dir, fname)
if os.path.isfile(lpath) and not fname.endswith('.pyc'):
rpath = remote_dir + "/" + fname
sftp.put(lpath, rpath)
print(" [OK] " + remote_rel + "/" + fname)
sftp.close()
print("\n[OK] Files uploaded")
......
skill/code/web/cache_manager.py
浏览文件 @ bccc2b9a
......@@ -30,6 +30,10 @@ import os
from pathlib import Path
from datetime import datetime
from utils.logger import get_logger
logger = get_logger(__name__)
class CacheManager:
"""问题排查结果缓存管理器"""
......@@ -76,8 +80,14 @@ class CacheManager:
return None
return data
except Exception as e:
print(f"[缓存] 读取缓存失败: {e}")
except FileNotFoundError:
# 缓存文件已被删除（并发场景），视为未命中
return None
except json.JSONDecodeError as e:
logger.error(f"缓存文件 JSON 解析失败: {cache_file} - {e}")
return None
except OSError as e:
logger.error(f"读取缓存文件失败: {cache_file} - {e}")
return None
def set(self, project_name, system_type, apk_product, query, response, matched_cases):
......@@ -110,8 +120,10 @@ class CacheManager:
with open(cache_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"[缓存] 已缓存: {key[:8]}...")
except Exception as e:
print(f"[缓存] 写入缓存失败: {e}")
except (TypeError, ValueError) as e:
logger.error(f"缓存数据序列化失败: {cache_file} - {e}")
except OSError as e:
logger.error(f"写入缓存文件失败: {cache_file} - {e}")
def clear_expired(self):
"""清理过期缓存"""
......@@ -123,8 +135,9 @@ class CacheManager:
if time.time() - data.get('timestamp', 0) > self.expire_seconds:
cache_file.unlink()
cleared += 1
except:
pass
except (json.JSONDecodeError, OSError, KeyError) as e:
# 单个缓存文件损坏或并发删除，跳过该文件，不中断清理
logger.debug(f"清理缓存时跳过损坏文件 {cache_file}: {e}")
if cleared > 0:
print(f"[缓存] 已清理 {cleared} 个过期缓存")
......@@ -158,8 +171,9 @@ class CacheManager:
oldest_time = ts
if newest_time is None or ts > newest_time:
newest_time = ts
except:
pass
except (json.JSONDecodeError, OSError, KeyError) as e:
# 统计场景：损坏文件不影响整体统计，跳过
logger.debug(f"统计缓存时跳过损坏文件 {cache_file}: {e}")
return {
'total_files': total_files,
......@@ -182,7 +196,9 @@ class CacheManager:
with open(cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
cache_files.append((cache_file, data.get('timestamp', 0)))
except:
except (json.JSONDecodeError, OSError, KeyError) as e:
# 损坏文件时间戳按 0 处理（视为最旧，优先清理）
logger.debug(f"按时间戳排序时跳过损坏文件 {cache_file}: {e}")
cache_files.append((cache_file, 0))
# 按时间戳排序，删除最旧的 20%
......
skill/code/web/server.py
浏览文件 @ bccc2b9a
......@@ -33,6 +33,15 @@ from io import BytesIO
from auth import user_manager
from decorators import login_required, admin_required, page_login_required
# ============================================================
# 日志与统一响应工具
# ============================================================
from utils.logger import get_logger
from utils.error_codes import ErrorCodes
from utils.response import error_response, success_response
logger = get_logger(__name__)
# ============================================================
# 配置
# ============================================================
......@@ -344,9 +353,13 @@ def call_claude_api(prompt):
result = resp.json()
return result['choices'][0]['message']['content']
else:
print(f"API 错误: {resp.status_code} - {resp.text[:200]}")
logger.error(f"API 错误: {resp.status_code} - {resp.text[:200]}")
except requests.exceptions.RequestException as e:
logger.error(f"HTTP API 调用失败（网络）: {e}")
except (KeyError, ValueError) as e:
logger.error(f"HTTP API 响应解析失败: {e}")
except Exception as e:
print(f"HTTP API 调用失败: {e}")
logger.exception(f"HTTP API 调用未知异常: {e}")
# 方式2: CLI（降级）
import subprocess
......@@ -372,14 +385,17 @@ def call_claude_api(prompt):
if result.returncode == 0:
return result.stdout.strip()
else:
print(f"Claude CLI 错误：{result.stderr}")
logger.error(f"Claude CLI 错误：{result.stderr}")
return generate_mock_response(prompt)
except FileNotFoundError:
print("未找到 claude CLI，降级为模拟响应")
logger.warning("未找到 claude CLI，降级为模拟响应")
return generate_mock_response(prompt)
except subprocess.TimeoutExpired as e:
logger.error(f"Claude CLI 调用超时: {e}")
return generate_mock_response(prompt)
except Exception as e:
print(f"Claude 调用失败：{e}")
logger.exception(f"Claude 调用失败: {e}")
return generate_mock_response(prompt)
......@@ -475,16 +491,22 @@ def call_claude_api_stream(prompt, model=None):
content = delta.get('content', '')
if content:
yield content
except json.JSONDecodeError:
pass
except json.JSONDecodeError as e:
# 单个 chunk 解析失败不中断整个流，仅记录 debug
logger.debug(f"流式响应 chunk 解析失败（已跳过）: {e}")
else:
print(f"API 错误: {resp.status_code}")
logger.error(f"API 错误: {resp.status_code}")
# 降级：返回模拟响应
for chunk in split_mock_response(generate_mock_response(prompt)):
yield chunk
except requests.exceptions.RequestException as e:
logger.error(f"流式 API 调用失败（网络）: {e}")
# 降级：返回模拟响应
for chunk in split_mock_response(generate_mock_response(prompt)):
yield chunk
except Exception as e:
print(f"流式 API 调用失败: {e}")
logger.exception(f"流式 API 调用失败: {e}")
# 降级：返回模拟响应
for chunk in split_mock_response(generate_mock_response(prompt)):
yield chunk
......@@ -671,7 +693,7 @@ def troubleshoot():
})
except Exception as e:
traceback.print_exc()
logger.exception("排查接口服务异常")
return jsonify({
'success': False,
'error': f'服务异常：{str(e)}',
......@@ -710,7 +732,7 @@ def search_cases():
})
except Exception as e:
traceback.print_exc()
logger.exception("搜索接口异常")
return jsonify({
'success': False,
'error': f'搜索异常：{str(e)}',
......@@ -767,7 +789,7 @@ def analyze():
})
except Exception as e:
traceback.print_exc()
logger.exception("分析接口异常")
return jsonify({
'success': False,
'error': f'分析异常：{str(e)}',
......@@ -885,7 +907,7 @@ def analyze_stream():
})
except Exception as e:
traceback.print_exc()
logger.exception("流式分析接口异常")
yield f'data: {json.dumps({"type": "error", "message": f"分析异常：{str(e)}"}, ensure_ascii=False)}\n\n'
return Response(
......@@ -948,6 +970,7 @@ def health_check():
}
})
except Exception as e:
logger.exception("健康检查接口异常")
return jsonify({
'status': 'error',
'timestamp': datetime.now().isoformat(),
......@@ -1000,6 +1023,7 @@ def get_cache_stats():
'stats': stats,
})
except Exception as e:
logger.exception("获取缓存统计接口异常")
return jsonify({
'success': False,
'error': f'获取缓存统计失败：{str(e)}',
......@@ -1029,6 +1053,7 @@ def clear_cache():
'cleared_count': cleared_count,
})
except Exception as e:
logger.exception("清空缓存接口异常")
return jsonify({
'success': False,
'error': f'清空缓存失败：{str(e)}',
......@@ -1216,7 +1241,7 @@ def submit_record():
})
except Exception as e:
traceback.print_exc()
logger.exception("问题入库接口异常")
return jsonify({
'success': False,
'error': f'提交失败：{str(e)}',
......@@ -1377,7 +1402,7 @@ def rebuild_search_index():
cwd=str(PROJECT_ROOT),
)
if result.returncode != 0:
print(f"索引重建失败：{result.stderr}")
logger.error(f"索引重建失败：{result.stderr}")
# 重新加载搜索引擎
global search_engine
......
skill/code/web/utils/__init__.py 0 → 100644
浏览文件 @ bccc2b9a
# -*- coding: utf-8 -*-
"""
utils 包 — 问题排查助手的通用工具模块
子模块：
- logger 通用日志配置
- error_codes 统一错误码定义
- response 统一 API 响应封装
"""
skill/code/web/utils/error_codes.py 0 → 100644
浏览文件 @ bccc2b9a
# -*- coding: utf-8 -*-
"""
error_codes.py — 统一错误码定义
每个错误码为 (code, default_message) 元组，便于统一响应封装。
分类：
- 1xxx 通用错误
- 2xxx 搜索相关
- 3xxx 缓存相关
- 4xxx AI 相关
"""
class ErrorCodes:
"""统一错误码常量"""
# 通用错误 1xxx
UNKNOWN_ERROR = (1000, "未知错误")
INVALID_PARAM = (1001, "参数错误")
PERMISSION_DENIED = (1002, "权限不足")
NOT_FOUND = (1003, "资源不存在")
# 搜索相关 2xxx
SEARCH_INDEX_ERROR = (2000, "搜索索引加载失败")
SEARCH_QUERY_ERROR = (2001, "搜索查询失败")
# 缓存相关 3xxx
CACHE_READ_ERROR = (3000, "缓存读取失败")
CACHE_WRITE_ERROR = (3001, "缓存写入失败")
# AI 相关 4xxx
AI_API_ERROR = (4000, "AI API 调用失败")
AI_RESPONSE_ERROR = (4001, "AI 响应解析失败")
AI_TIMEOUT = (4002, "AI 调用超时")
skill/code/web/utils/logger.py 0 → 100644
浏览文件 @ bccc2b9a
# -*- coding: utf-8 -*-
"""
logger.py — 通用日志配置
功能：
- 统一配置 logging，输出到控制台 + 文件
- 文件日志按大小轮转（单文件 1MB，保留 3 份）
- 各模块通过 get_logger(__name__) 获取子 logger
用法：
from utils.logger import get_logger
logger = get_logger(__name__)
logger.info("...")
logger.exception("...") # 自动带堆栈
"""
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
# 日志根 logger 名称
ROOT_LOGGER_NAME = "troubleshoot"
# 日志文件位置：与本文件同级的 ../logs/app.log
_LOG_DIR = Path(__file__).resolve().parent.parent / "logs"
_LOG_FILE = _LOG_DIR / "app.log"
# 日志格式
_LOG_FORMAT = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
_DATE_FORMAT = "%Y-%m-%d %H:%M:%S"
# 是否已完成初始化（避免重复添加 handler）
_initialized = False
def _init_root_logger():
"""初始化根 logger（仅执行一次）"""
global _initialized
if _initialized:
return
root = logging.getLogger(ROOT_LOGGER_NAME)
root.setLevel(logging.DEBUG)
# 避免重复添加 handler
if root.handlers:
_initialized = True
return
formatter = logging.Formatter(_LOG_FORMAT, datefmt=_DATE_FORMAT)
# 控制台 handler（与原 print 行为兼容，INFO 级别）
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)
root.addHandler(console_handler)
# 文件 handler（DEBUG 级别，轮转）
try:
_LOG_DIR.mkdir(parents=True, exist_ok=True)
file_handler = RotatingFileHandler(
_LOG_FILE,
maxBytes=1 * 1024 * 1024, # 1MB
backupCount=3,
encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
root.addHandler(file_handler)
except OSError as e:
# 文件 handler 初始化失败时，至少保证控制台日志可用
root.error(f"无法初始化文件日志 handler: {e}")
# 不向上传播到 root logger，避免重复输出
root.propagate = False
_initialized = True
def get_logger(name=None):
"""
获取子 logger。
参数：
name: 模块名，通常传 __name__。None 时返回根 logger。
"""
_init_root_logger()
if name is None:
return logging.getLogger(ROOT_LOGGER_NAME)
# 不在 troubleshoot 命名空间下的，挂到根 logger 下
if not name.startswith(ROOT_LOGGER_NAME):
name = f"{ROOT_LOGGER_NAME}.{name}"
return logging.getLogger(name)
skill/code/web/utils/response.py 0 → 100644
浏览文件 @ bccc2b9a
# -*- coding: utf-8 -*-
"""
response.py — 统一 API 响应封装
提供 success_response / error_response，规范 API 返回结构：
成功：{"success": True, ...data}
失败：{"success": False, "error": {"code": int, "message": str}}
"""
from utils.error_codes import ErrorCodes
def success_response(**data):
"""统一成功响应"""
result = {"success": True}
result.update(data)
return result
def error_response(error_code, message=None):
"""
统一错误响应。
参数：
error_code: ErrorCodes 中的 (code, default_msg) 元组
message: 自定义错误信息，None 时用默认信息
"""
code, default_msg = error_code
return {
"success": False,
"error": {
"code": code,
"message": message or default_msg,
},
}
# 便捷别名
INTERNAL_ERROR = lambda msg=None: error_response(ErrorCodes.UNKNOWN_ERROR, msg)
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论