Skip to content

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

fix(security): P0级问题修复 

- P0-1: 移除所有硬编码密码，使用环境变量
- P0-2: 修复 server.py 重复赋值 bug
- P0-3: 消除 deploy/web/ 代码重复
- 新增 .env.example 环境变量模板
- 新增 .gitignore
- 更新 README.md 目录结构说明
- 更新部署脚本从统一源目录复制
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 2dbfa212
展开全部 显示空白字符变更
内嵌 并排
正在显示 包含 449 行增加 和 2966 行删除
.env.example 0 → 100644
浏览文件 @ 023f2e65
# 环境变量配置模板
# 复制此文件为 .env 并填写实际值
# ============================================
# SSH 连接配置（用于远程部署和测试）
# ============================================
SSH_HOST=192.168.5.60
SSH_USER=ubains
SSH_PASSWORD=your_password_here
# ============================================
# Flask Session 密钥
# 生产环境必须修改为随机字符串（至少 32 字符）
# ============================================
SECRET_KEY=your_secret_key_here
# ============================================
# Claude API 配置
# ============================================
CLAUDE_API_BASE=https://office.ubainsyun.com:8400
CLAUDE_API_KEY=your_api_key_here
# ============================================
# 数据库配置（如需要）
# ============================================
# DB_HOST=localhost
# DB_PORT=3306
# DB_USER=root
# DB_PASSWORD=your_db_password
\ No newline at end of file
.gitignore 0 → 100644
浏览文件 @ 023f2e65
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
# 环境变量文件（包含敏感信息）
.env
.env.local
.env.*.local
# IDE
.idea/
.vscode/
*.swp
*.swo
*~
# 系统文件
.DS_Store
Thumbs.db
# 日志文件
*.log
logs/
# 缓存文件
cache/
*.cache
# 临时文件
tmp/
temp/
*.tmp
# 数据文件
*.xlsx
!问题反馈跟踪表.xlsx
# Claude Code
.claude/skills/*/tmp/
.claude/worktrees/
\ No newline at end of file
Docs/PRD_计划执行_P0级问题修复.md 0 → 100644
浏览文件 @ 023f2e65
此差异已折叠。
Docs/PRD_需求文档_P0级问题修复.md 0 → 100644
浏览文件 @ 023f2e65
# PRD_需求文档_P0级问题修复
## 基本信息
| 项目 | 内容 |
|------|------|
| 文档类型 | 需求文档 |
| 创建日期 | 2026-07-12 |
| 负责人 | 研发组 |
| 优先级 | P0（最高优先级） |
| 状态 | 待实施 |
---
## 一、背景与目标
### 1.1 问题背景
经过代码质量分析，发现项目中存在 3 个 P0 级别的严重问题，需要立即修复：
| 问题编号 | 问题描述 | 风险等级 |
|----------|----------|----------|
| P0-1 | 硬编码密码泄露 | 🔴 严重 |
| P0-2 | server.py 重复赋值 Bug | 🔴 严重 |
| P0-3 | skill/ 与 deploy/ 代码重复 | 🟠 高 |
### 1.2 修复目标
1. **消除安全隐患**：移除所有硬编码密码，使用环境变量管理敏感信息
2. **修复代码缺陷**：解决重复赋值 Bug，确保逻辑正确
3. **提升代码质量**：消除代码重复，建立单一数据源，便于维护
---
## 二、需求详情
### 2.1 P0-1：移除硬编码密码
#### 问题描述
以下文件中存在硬编码密码，存在严重安全风险：
| 文件路径 | 行号 | 硬编码内容 |
|----------|------|------------|
| deploy/check_service.py | 10 | PASSWORD = 'Ubains@123' |
| deploy/upload_to_server.py | 14 | PASSWORD = 'Ubains@123' |
| skill/code/test_login_auth.py | 14 | ssh.connect(..., password='Ubains@123') |
| skill/code/test_normal_user.py | 12 | ssh.connect(..., password='Ubains@123') |
| skill/code/web/server.py | 86 | app.secret_key = 'troubleshoot-secret-key-2026-secure' |
#### 需求规格
| 需求项 | 规格 |
|--------|------|
| 密码存储 | 使用 .env 文件或环境变量 |
| 配置读取 | 通过 python-dotenv 或 os.environ.get() 读取 |
| 默认值处理 | 生产环境禁止使用默认密码，开发环境可使用占位符 |
| 文档说明 | 在 README 中说明如何配置环境变量 |
#### 验收标准
- [x] 所有硬编码密码已移除
- [x] 创建 .env.example 模板文件
- [x] .env 已添加到 .gitignore
- [x] 代码中使用环境变量读取敏感配置
- [x] README 中添加配置说明
---
### 2.2 P0-2：修复重复赋值 Bug
#### 问题描述
skill/code/web/server.py 第 796-797 行存在重复赋值：
```python
# 第 796 行
query = request.args.get('query', '').strip()
# 第 797 行（重复，会覆盖上面的值）
query = request.args.get('query', '').strip()
```
此 Bug 会导致：
- 变量被重复赋值，代码逻辑混乱
- 如果未来在第 796 行之后添加了对 query 的处理，第 797 行会覆盖处理结果
#### 需求规格
| 需求项 | 规格 |
|--------|------|
| 代码修复 | 删除第 797 行重复代码 |
| 逻辑验证 | 确认 analyze_stream 函数参数读取逻辑正确 |
| 测试验证 | 手动测试流式 API 功能正常 |
#### 验收标准
- [x] 重复赋值代码已删除
- [x] 流式 API 功能测试通过
- [x] 代码逻辑清晰，无冗余
---
### 2.3 P0-3：消除代码重复
#### 问题描述
skill/code/web/ 与 deploy/web/ 目录下存在大量重复代码：
| 文件 | 相似度 | 说明 |
|------|--------|------|
| server.py | ~95% | 几乎相同 |
| search_engine.py | ~95% | 几乎相同 |
| cache_manager.py | 100% | 完全相同 |
| safety_filter.py | 100% | 完全相同 |
| auth.py | ~95% | 几乎相同 |
| decorators.py | 100% | 完全相同 |
#### 问题影响
1. **维护困难**：修改一处代码需要同步修改另一处，容易遗漏
2. **版本混乱**：两份代码可能存在细微差异，导致行为不一致
3. **存储浪费**：重复代码占用额外空间
#### 需求规格
| 需求项 | 规格 |
|--------|------|
| 目标架构 | 保留 skill/code/web/ 作为开发目录，deploy/ 仅保留部署配置 |
| 共享模块 | 提取公共模块到 skill/code/web/shared/ 或使用符号链接 |
| 部署脚本 | 更新部署脚本，从 skill/code/web/ 复制文件 |
| 文档更新 | 更新 README 说明目录结构 |
#### 验收标准
- [x] 重复代码已消除
- [x] 部署脚本可正确复制源代码
- [x] 部署后服务功能正常
- [x] README 目录结构说明已更新
---
## 三、影响范围
### 3.1 涉及文件
| 文件 | 变更类型 |
|------|----------|
| deploy/check_service.py | 修改 |
| deploy/upload_to_server.py | 修改 |
| skill/code/test_login_auth.py | 修改 |
| skill/code/test_normal_user.py | 修改 |
| skill/code/web/server.py | 修改 |
| .env.example | 新增 |
| .gitignore | 修改 |
| README.md | 修改 |
| deploy/web/ | 删除（使用符号链接或部署脚本复制） |
| deploy/deploy.sh | 修改 |
### 3.2 风险评估
| 风险项 | 等级 | 缓解措施 |
|--------|------|----------|
| 环境变量未配置导致服务启动失败 | 中 | 提供默认值 + 启动检查 |
| 部署脚本复制失败 | 低 | 部署前验证文件存在 |
| 测试文件修改后无法运行 | 低 | 更新测试配置 |
---
## 四、非功能需求
| 需求项 | 说明 |
|--------|------|
| 向后兼容 | 环境变量未配置时使用默认值（开发环境） |
| 文档完善 | README 添加环境变量配置说明 |
| 安全合规 | 敏感信息不得提交到版本控制 |
---
## 五、验收清单
### P0-1 验收
- [x] 所有硬编码密码已移除
- [x] 创建 .env.example 模板
- [x] .env 已添加到 .gitignore
- [x] 服务启动正常，环境变量读取正确
### P0-2 验收
- [x] 重复赋值代码已删除
- [x] 流式 API 功能测试通过
### P0-3 验收
- [x] 重复代码已消除
- [x] 部署脚本正常工作
- [x] 服务部署后功能正常
---
## 六、时间估算
| 任务 | 预计工时 |
|------|----------|
| P0-1：移除硬编码密码 | 0.5 天 |
| P0-2：修复重复赋值 Bug | 0.1 天 |
| P0-3：消除代码重复 | 0.4 天 |
| 测试验证与文档更新 | 0.2 天 |
| **总计** | **1.2 天** |
README.md
浏览文件 @ 023f2e65
......@@ -9,7 +9,7 @@ troubleshoot-ai-assistant/
├── skill/ # Claude Skill 代码
│ ├── SKILL.md # Skill 定义文件
│ └── code/ # 核心代码
│ ├── web/ # Web服务代码
│ ├── web/ # Web服务代码（唯一源）
│ │ ├── server.py # Flask服务器主程序
│ │ ├── search_engine.py # 搜索引擎
│ │ ├── cache_manager.py # 缓存管理
......@@ -22,9 +22,8 @@ troubleshoot-ai-assistant/
│ └── start.bat # Windows启动脚本
│
├── deploy/ # 部署相关文件
│ ├── web/ # Web服务部署版本
│ ├── deploy.sh # Linux部署脚本（从 skill/code/web/ 复制）
│ ├── 问题记录/ # 问题知识库（日常+项目）
│ ├── deploy.sh # Linux部署脚本
│ ├── upload_to_server.py # 上传工具
│ ├── check_service.py # 服务检查脚本
│ └── verify_deployment.py # 验证脚本
......@@ -43,6 +42,8 @@ troubleshoot-ai-assistant/
│ ├── troubleshoot.service # systemd服务定义
│ └── troubleshoot-data.zip # 数据压缩包（如有）
│
├── .env.example # 环境变量模板
├── .gitignore # Git忽略配置
└── README.md # 本说明文件
```
......@@ -65,10 +66,14 @@ troubleshoot-ai-assistant/
### Web服务部署
```bash
# 安装依赖
# 1. 配置环境变量
cp .env.example .env
# 编辑 .env 文件，填写实际配置
# 2. 安装依赖
pip install -r skill/code/requirements.txt
# 启动服务
# 3. 启动服务
python skill/code/web/server.py
# Linux systemd部署
......@@ -77,6 +82,17 @@ sudo systemctl enable troubleshoot
sudo systemctl start troubleshoot
```
## 环境变量配置
| 变量名 | 说明 | 必填 |
|--------|------|------|
| `SSH_HOST` | SSH服务器地址 | 否（测试用） |
| `SSH_USER` | SSH用户名 | 否（测试用） |
| `SSH_PASSWORD` | SSH密码 | 否（测试用） |
| `SECRET_KEY` | Flask Session密钥 | 是（生产环境） |
| `CLAUDE_API_BASE` | Claude API地址 | 是 |
| `CLAUDE_API_KEY` | Claude API密钥 | 是 |
## 相关链接
- [技术实现文档](docs/技术实现文档_问题排查助手Web服务.md)
......@@ -91,3 +107,9 @@ sudo systemctl start troubleshoot
- 原位置: `Troubleshoot-deploy/` → 新位置: `deploy/`
- 原位置: `Docs/PRD/问题知识库/` → 新位置: `docs/`
- 原位置: `troubleshoot.service` → 新位置: `config/`
## 代码规范
- **单一数据源**: `skill/code/web/` 是唯一开发目录，`deploy/web/` 已删除
- **部署方式**: 通过 `deploy/deploy.sh` 从 `skill/code/web/` 复制文件
- **敏感信息**: 使用环境变量，禁止硬编码密码
\ No newline at end of file
deploy/check_service.py
浏览文件 @ 023f2e65
......@@ -4,10 +4,17 @@ check_service.py - 检查服务器上的服务状态
"""
import paramiko
import os
HOST = '192.168.5.60'
USER = 'ubains'
PASSWORD = 'Ubains@123'
# 从环境变量读取配置
HOST = os.environ.get('SSH_HOST', '192.168.5.60')
USER = os.environ.get('SSH_USER', 'ubains')
PASSWORD = os.environ.get('SSH_PASSWORD', '')
if not PASSWORD:
print("错误：SSH_PASSWORD 环境变量未设置")
print("请设置环境变量或创建 .env 文件")
exit(1)
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
......
deploy/deploy.sh
浏览文件 @ 023f2e65
......@@ -2,6 +2,9 @@
# ============================================================
# 问题排查助手 - Linux 部署脚本
# 运行方式: bash deploy.sh
#
# 说明：从 skill/code/web/ 复制源代码到部署目录
# deploy/web/ 目录已删除，统一使用单一数据源
# ============================================================
set -e
......@@ -14,21 +17,28 @@ echo "=================================================="
DEPLOY_DIR="/opt/troubleshoot"
WEB_PORT=8088
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
echo "[1/4] 安装 Python 依赖..."
pip3 install flask openpyxl -q 2>/dev/null || pip install flask openpyxl -q
echo "[1/5] 安装 Python 依赖..."
pip3 install -r "$PROJECT_ROOT/skill/code/requirements.txt" -q 2>/dev/null || \
pip install -r "$PROJECT_ROOT/skill/code/requirements.txt" -q
echo "[2/4] 创建部署目录..."
echo "[2/5] 创建部署目录..."
sudo mkdir -p $DEPLOY_DIR
sudo cp -r "$SCRIPT_DIR/web" "$DEPLOY_DIR/"
sudo cp -r "$SCRIPT_DIR/问题记录" "$DEPLOY_DIR/"
sudo cp "$SCRIPT_DIR/搜索索引.json" "$DEPLOY_DIR/"
sudo cp "$SCRIPT_DIR/build_index.py" "$DEPLOY_DIR/"
sudo mkdir -p $DEPLOY_DIR/问题记录
echo "[3/4] 配置权限..."
echo "[3/5] 复制 Web 服务（从 skill/code/web/）..."
# 从统一源目录 skill/code/web/ 复制
sudo cp -r "$PROJECT_ROOT/skill/code/web" "$DEPLOY_DIR/"
echo "[4/5] 复制问题记录和搜索索引..."
sudo cp -r "$SCRIPT_DIR/问题记录"/* "$DEPLOY_DIR/问题记录/" 2>/dev/null || true
sudo cp "$PROJECT_ROOT/Docs/搜索索引.json" "$DEPLOY_DIR/" 2>/dev/null || true
sudo cp "$PROJECT_ROOT/skill/code/build_index.py" "$DEPLOY_DIR/"
echo "[5/5] 配置权限和 systemd 服务..."
sudo chown -R $USER:$USER $DEPLOY_DIR
echo "[4/4] 创建 systemd 服务..."
sudo tee /etc/systemd/system/troubleshoot.service > /dev/null << EOF
[Unit]
Description=问题排查助手 Web 服务
......@@ -54,7 +64,7 @@ echo ""
echo "=================================================="
echo " 部署完成！"
echo "=================================================="
echo " 访问地址: http://192.168.5.60:$WEB_PORT"
echo " 访问地址: http://$(hostname -I | awk '{print $1}'):$WEB_PORT"
echo " 查看状态: sudo systemctl status troubleshoot"
echo " 查看日志: sudo journalctl -u troubleshoot -f"
echo " 重启服务: sudo systemctl restart troubleshoot"
......
deploy/upload_to_server.py
浏览文件 @ 023f2e65
......@@ -8,12 +8,17 @@ import os
import time
from datetime import datetime
# 配置
HOST = '192.168.5.60'
USER = 'ubains'
PASSWORD = 'Ubains@123'
# 从环境变量读取配置
HOST = os.environ.get('SSH_HOST', '192.168.5.60')
USER = os.environ.get('SSH_USER', 'ubains')
PASSWORD = os.environ.get('SSH_PASSWORD', '')
REMOTE_BASE = '/opt/troubleshoot'
if not PASSWORD:
print("错误：SSH_PASSWORD 环境变量未设置")
print("请设置环境变量或创建 .env 文件")
exit(1)
# 本地部署目录
LOCAL_BASE = os.path.dirname(os.path.abspath(__file__))
......
deploy/web/audit.log deleted 100644 → 0
浏览文件 @ 2dbfa212
{"timestamp": "2026-07-11T11:26:44.326533", "project": "厦门银行", "system_type": "标准版预定2.0", "apk_product": "门口屏5.0", "query": "门口屏绑定失败，提示 MQTT 连接错误", "matched_count": 5, "api_time": 0.0, "removed_lines": 1}
{"timestamp": "2026-07-11T13:24:53.004305", "project": "厦门银行", "system_type": "标准版预定2.0", "apk_product": "门口屏5.0", "query": "门口屏绑定失败，提示MQTT连接错误", "matched_count": 5, "api_time": 0.02, "removed_lines": 1}
{"timestamp": "2026-07-11T14:18:10.910545", "project": "测试项目", "system_type": "标准版预定2.0", "apk_product": "", "query": "测试入库功能验证-门口屏绑定失败", "matched_count": 0, "api_time": 0.01, "removed_lines": 1}
{"timestamp": "2026-07-11T14:21:49.913640", "action": "submit_record", "record_id": "RC-20260711-001", "project": "测试项目", "recorder": "测试用户"}
{"timestamp": "2026-07-11T14:31:46.254810", "project": "招商局", "system_type": "标准版预定2.0", "apk_product": "门口屏5.0", "query": "门口屏无法绑定会议室，提示接口请求超时", "matched_count": 5, "api_time": 0.01, "removed_lines": 1}
{"timestamp": "2026-07-11T14:33:47.134132", "action": "submit_record", "record_id": "RC-20260711-002", "project": "招商局", "recorder": "Claude"}
deploy/web/cache_manager.py deleted 100644 → 0
浏览文件 @ 2dbfa212
# -*- coding: utf-8 -*-
"""
cache_manager.py — 问题排查助手缓存管理模块
功能：
- 基于 JSON 文件的缓存存储
- 支持缓存过期检查
- 支持缓存清理
- 支持缓存统计
用法：
from cache_manager import CacheManager
cache = CacheManager(cache_dir='./cache', expire_hours=24)
# 获取缓存
cached = cache.get(project_name, system_type, apk_product, query)
# 设置缓存
cache.set(project_name, system_type, apk_product, query, response, matched_cases)
# 清理过期缓存
cache.clear_expired()
"""
import json
import hashlib
import time
import os
from pathlib import Path
from datetime import datetime
class CacheManager:
"""问题排查结果缓存管理器"""
def __init__(self, cache_dir, expire_hours=24, max_size_mb=100):
"""
初始化缓存管理器。
参数：
cache_dir: 缓存目录
expire_hours: 缓存过期时间（小时）
max_size_mb: 最大缓存大小（MB），超过时自动清理最旧的缓存
"""
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.expire_seconds = expire_hours * 3600
self.max_size_bytes = max_size_mb * 1024 * 1024
def _get_cache_key(self, project_name, system_type, apk_product, query):
"""生成缓存键（MD5 哈希）"""
content = f"{project_name}|{system_type}|{apk_product}|{query}"
return hashlib.md5(content.encode('utf-8')).hexdigest()
def get(self, project_name, system_type, apk_product, query):
"""
获取缓存结果。
返回：
缓存数据字典，或 None（缓存不存在或已过期）
"""
key = self._get_cache_key(project_name, system_type, apk_product, query)
cache_file = self.cache_dir / f"{key}.json"
if not cache_file.exists():
return None
try:
with open(cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
# 检查过期
if time.time() - data.get('timestamp', 0) > self.expire_seconds:
cache_file.unlink()
return None
return data
except Exception as e:
print(f"[缓存] 读取缓存失败: {e}")
return None
def set(self, project_name, system_type, apk_product, query, response, matched_cases):
"""
设置缓存。
参数：
project_name: 项目名称
system_type: 系统类型
apk_product: APK 产品
query: 问题描述
response: AI 响应内容
matched_cases: 匹配案例列表
"""
key = self._get_cache_key(project_name, system_type, apk_product, query)
cache_file = self.cache_dir / f"{key}.json"
data = {
'timestamp': time.time(),
'datetime': datetime.now().isoformat(),
'project_name': project_name,
'system_type': system_type,
'apk_product': apk_product,
'query': query,
'response': response,
'matched_cases': matched_cases,
}
try:
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}")
def clear_expired(self):
"""清理过期缓存"""
cleared = 0
for cache_file in self.cache_dir.glob("*.json"):
try:
with open(cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
if time.time() - data.get('timestamp', 0) > self.expire_seconds:
cache_file.unlink()
cleared += 1
except:
pass
if cleared > 0:
print(f"[缓存] 已清理 {cleared} 个过期缓存")
return cleared
def clear_all(self):
"""清空所有缓存"""
cleared = 0
for cache_file in self.cache_dir.glob("*.json"):
cache_file.unlink()
cleared += 1
print(f"[缓存] 已清空 {cleared} 个缓存")
return cleared
def get_stats(self):
"""获取缓存统计信息"""
total_files = 0
total_size = 0
oldest_time = None
newest_time = None
for cache_file in self.cache_dir.glob("*.json"):
total_files += 1
total_size += cache_file.stat().st_size
try:
with open(cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
ts = data.get('timestamp', 0)
if oldest_time is None or ts < oldest_time:
oldest_time = ts
if newest_time is None or ts > newest_time:
newest_time = ts
except:
pass
return {
'total_files': total_files,
'total_size_mb': round(total_size / (1024 * 1024), 2),
'oldest': datetime.fromtimestamp(oldest_time).isoformat() if oldest_time else None,
'newest': datetime.fromtimestamp(newest_time).isoformat() if newest_time else None,
}
def check_and_clean_if_needed(self):
"""检查缓存大小，超过限制时清理最旧的缓存"""
total_size = sum(f.stat().st_size for f in self.cache_dir.glob("*.json"))
if total_size > self.max_size_bytes:
print(f"[缓存] 缓存大小 {total_size / (1024*1024):.2f}MB 超过限制，清理最旧的缓存...")
# 获取所有缓存文件及其时间戳
cache_files = []
for cache_file in self.cache_dir.glob("*.json"):
try:
with open(cache_file, 'r', encoding='utf-8') as f:
data = json.load(f)
cache_files.append((cache_file, data.get('timestamp', 0)))
except:
cache_files.append((cache_file, 0))
# 按时间戳排序，删除最旧的 20%
cache_files.sort(key=lambda x: x[1])
delete_count = max(1, len(cache_files) // 5)
for cache_file, _ in cache_files[:delete_count]:
cache_file.unlink()
print(f"[缓存] 已清理 {delete_count} 个最旧缓存")
# 模块级单例
_cache_manager = None
def get_cache_manager():
"""获取缓存管理器单例"""
global _cache_manager
if _cache_manager is None:
from pathlib import Path
cache_dir = Path(__file__).resolve().parent / "cache"
_cache_manager = CacheManager(cache_dir, expire_hours=24)
return _cache_manager
if __name__ == "__main__":
# 测试
cache = CacheManager("./test_cache", expire_hours=1)
# 测试设置缓存
cache.set("厦门银行", "标准版预定2.0", "门口屏5.0", "MQTT连接失败",
"这是测试响应", [{"rank": 1, "score": 0.9}])
# 测试获取缓存
result = cache.get("厦门银行", "标准版预定2.0", "门口屏5.0", "MQTT连接失败")
print(f"缓存命中: {result is not None}")
# 测试统计
stats = cache.get_stats()
print(f"缓存统计: {stats}")
# 清理测试缓存
cache.clear_all()
deploy/web/config.json deleted 100644 → 0
浏览文件 @ 2dbfa212
{
"claude_api_base": "",
"claude_api_key": "",
"claude_model": "claude-sonnet-4-6",
"system_types": [
{"value": "std20", "label": "标准版预定2.0"},
{"value": "ops", "label": "标准版运维集控系统"},
{"value": "new_unified", "label": "新统一平台"},
{"value": "unified", "label": "统一平台"}
]
}
\ No newline at end of file
deploy/web/safety_filter.py deleted 100644 → 0
浏览文件 @ 2dbfa212
此差异已折叠。
deploy/web/search_engine.py deleted 100644 → 0
浏览文件 @ 2dbfa212
# -*- coding: utf-8 -*-
"""
search_engine.py — 问题知识库搜索引擎
功能：
基于关键词匹配 + TF-IDF 相似度，从问题知识库中检索最相关的历史案例。
用法：
from search_engine import SearchEngine
engine = SearchEngine()
results = engine.search("门口屏 MQTT 连接失败", top_k=5)
"""
import os
import re
import json
import math
from pathlib import Path
from collections import defaultdict, Counter
# ============================================================
# 配置
# ============================================================
SCRIPT_DIR = Path(__file__).resolve().parent # .../code/web
CODE_DIR = SCRIPT_DIR.parent # .../code
SKILL_DIR = CODE_DIR.parent # .../Troubleshoot
CLAUDE_SKILLS_DIR = SKILL_DIR.parent # .../skills
CLAUDE_DIR = CLAUDE_SKILLS_DIR.parent # .../.claude
PROJECT_ROOT = CLAUDE_DIR.parent # 项目根 (develop/)
DATA_DIR = PROJECT_ROOT / "Docs" / "PRD" / "问题知识库"
SEARCH_INDEX_FILE = DATA_DIR / "搜索索引.json"
# 停用词（过滤无意义的词）
STOPWORDS = {
'的', '了', '是', '在', '有', '我', '不', '和', '与', '或',
'这个', '那个', '怎么', '为什么', '哪里', '什么', '可以',
'应该', '需要', '请', '帮', '一下', '吗', '呢', '啊', '吧',
'the', 'a', 'an', 'is', 'are', 'was', 'were', 'to', 'of', 'in',
'and', 'or', 'not', 'for', 'on', 'at', 'by', 'be', 'it', 'its',
'as', 'if', 'then', 'than', 'that', 'this', 'these', 'those',
}
def load_search_index():
"""加载搜索索引"""
if not SEARCH_INDEX_FILE.exists():
raise FileNotFoundError(f"搜索索引不存在：{SEARCH_INDEX_FILE}")
with open(SEARCH_INDEX_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
def tokenize(text):
"""中文分词（简单实现：按字切分 + 关键词提取）"""
# 提取中文词组（2-4字）
cn_words = re.findall(r'[一-龥]{2,4}', text)
# 提取英文单词
en_words = re.findall(r'[a-zA-Z]{2,}', text.lower())
# 提取数字
numbers = re.findall(r'\d+', text)
tokens = cn_words + en_words + numbers
# 过滤停用词
tokens = [t for t in tokens if t not in STOPWORDS]
return tokens
def compute_tf(tokens):
"""计算词频 (TF)"""
tf = Counter(tokens)
total = len(tokens) if tokens else 1
return {word: count / total for word, count in tf.items()}
def compute_idf(documents):
"""计算逆文档频率 (IDF)"""
N = len(documents)
df = defaultdict(int)
for doc in documents:
unique_tokens = set(doc)
for token in unique_tokens:
df[token] += 1
idf = {}
for token, count in df.items():
# 平滑处理：避免除零
idf[token] = math.log((N + 1) / (count + 1)) + 1
return idf
def compute_tfidf(tf, idf):
"""计算 TF-IDF 向量"""
return {word: tf_val * idf.get(word, 1) for word, tf_val in tf.items()}
def cosine_similarity(vec1, vec2):
"""计算余弦相似度"""
# 找到共同词
common_words = set(vec1.keys()) & set(vec2.keys())
if not common_words:
return 0.0
# 计算点积
dot_product = sum(vec1[w] * vec2[w] for w in common_words)
# 计算模长
norm1 = math.sqrt(sum(v ** 2 for v in vec1.values()))
norm2 = math.sqrt(sum(v ** 2 for v in vec2.values()))
if norm1 == 0 or norm2 == 0:
return 0.0
return dot_product / (norm1 * norm2)
class SearchEngine:
"""问题知识库搜索引擎"""
def __init__(self):
"""初始化：加载索引并构建 TF-IDF 模型"""
self.index = load_search_index()
self.records = self.index.get('records', [])
# 预处理：为每条记录构建 TF-IDF 向量
self.doc_tokens = []
self.doc_vectors = []
self.idf = {}
if self.records:
# 对所有记录进行分词
self.doc_tokens = [
tokenize(r.get('full_text', '') + ' ' + r.get('title', ''))
for r in self.records
]
# 计算 IDF
self.idf = compute_idf(self.doc_tokens)
# 预计算每条记录的 TF-IDF 向量
self.doc_vectors = [
compute_tfidf(compute_tf(tokens), self.idf)
for tokens in self.doc_tokens
]
print(f"🔍 搜索引擎初始化完成：{len(self.records)} 条记录")
def search(self, query, top_k=5, project_filter=None, category_filter=None):
"""
搜索相似问题记录。
参数：
query: 搜索查询字符串
top_k: 返回结果数量
project_filter: 项目名称过滤（可选）
category_filter: 分类过滤（可选）
返回：
[
{
'rank': 1,
'score': 0.85,
'record': {...} # 原始记录
},
...
]
"""
if not self.records:
return []
# 对查询进行分词和向量化
query_tokens = tokenize(query)
query_tf = compute_tf(query_tokens)
query_vector = compute_tfidf(query_tf, self.idf)
# 计算与每条记录的相似度
scores = []
for i, doc_vector in enumerate(self.doc_vectors):
# 应用过滤器
record = self.records[i]
if project_filter:
if record.get('project', '').lower() != project_filter.lower():
continue
if category_filter:
if category_filter not in record.get('category', []):
continue
# 计算相似度
similarity = cosine_similarity(query_vector, doc_vector)
# 关键词匹配加分
matched_keywords = 0
for kw in record.get('keywords', []):
if kw.lower() in query.lower():
matched_keywords += 1
keyword_bonus = matched_keywords * 0.05 # 每匹配一个关键词加 5%
# 项目匹配加分
project_bonus = 0
if project_filter and record.get('project', '').lower() == project_filter.lower():
project_bonus = 0.1 # 同项目加 10%
# 综合得分
final_score = min(1.0, similarity + keyword_bonus + project_bonus)
if final_score > 0:
scores.append((i, final_score))
# 按得分排序
scores.sort(key=lambda x: x[1], reverse=True)
# 返回 Top K
results = []
for rank, (idx, score) in enumerate(scores[:top_k], start=1):
results.append({
'rank': rank,
'score': round(score, 3),
'record': self.records[idx],
})
return results
def get_projects(self):
"""获取所有项目名称列表"""
return self.index.get('project_names', [])
def get_categories(self):
"""获取所有分类列表"""
return list(self.index.get('categories', {}).keys())
# 模块级单例
_engine = None
def get_engine():
"""获取搜索引擎单例"""
global _engine
if _engine is None:
_engine = SearchEngine()
return _engine
if __name__ == "__main__":
# 测试
engine = SearchEngine()
test_queries = [
"门口屏 MQTT 连接失败",
"redis 连接失败",
"预定系统启动失败",
"配置文件不生效",
]
for q in test_queries:
print(f"\n{'='*50}")
print(f"查询：{q}")
print(f"{'='*50}")
results = engine.search(q, top_k=3)
for r in results:
rec = r['record']
print(f"[{r['rank']}] 得分：{r['score']}")
print(f" 项目：{rec.get('project', '无')}")
print(f" 标题：{rec.get('title', '无')}")
print(f" 分类：{rec.get('category', [])}")
deploy/web/server.py deleted 100644 → 0
浏览文件 @ 2dbfa212
此差异已折叠。
deploy/web/templates/index.html deleted 100644 → 0
浏览文件 @ 2dbfa212
此差异已折叠。
skill/GitCommit/SKILL.md
浏览文件 @ 023f2e65
......@@ -134,14 +134,39 @@ Co-Authored-By: Claude <noreply@anthropic.com>
3. **不要自动 push**：提交成功后告知用户，并询问是否需要 `git push`（push 也需用户明确同意才执行）
**推送规则（🚨 分支安全确认，不可跳过）：**
在执行 `git push` 之前，必须执行以下检查：
1. **获取当前分支信息**：执行 `git branch --show-current` 获取当前分支名
2. **获取远程关联信息**：执行 `git remote -v` 获取远程仓库地址
3. **向用户确认**：使用 `AskUserQuestion` 展示以下信息并让用户确认：
- 当前分支名：`<branch-name>`
- 远程仓库：`<remote-url>`
- 推送目标：`origin <branch-name>`
- 确认选项：✅ 确认推送 / ❌ 取消推送
4. **只有用户明确确认后**，才执行 `git push origin <当前分支名>`（**推送的目标分支必须与当前分支同名，不得推送到其他分支**）
**禁止行为：**
- ❌ 跳过分支确认直接 push
- ❌ 将当前分支推送到非当前分支的远程分支（如从 `feature-a` 推送到 `master`）
- ❌ 使用 `git push` 不带参数（依赖默认行为可能推错分支）
- ❌ 使用 `git push --force` 或 `git push -f`
- ❌ 把"提交 + push"打包执行，两者必须分别确认
**禁止行为：**
- ❌ 跳过任一确认门直接 add/commit
- ❌ 把"未明确拒绝"当作"同意"——必须用户主动确认才执行
- ❌ 一次性把"提交 + push"打包执行，两者必须分别确认
- ❌ 推送前不显示分支信息
- ❌ 推送到与当前分支不同的远程分支
## 注意事项
- **🚨 双重确认门控（最高优先级）**：范围确认（门1）+ message 确认（门2）都必须用户主动点头，未明确同意绝不执行 add/commit/push
- **🚨 推送分支安全确认（最高优先级）**：push 前必须显示当前分支名和推送目标，用户明确确认后才能执行 `git push origin <当前分支名>`
- **🚨 禁止推送到非当前分支**：只能推送 `origin <当前分支>`，不能推送到其他分支
- **🚨 禁止使用 `git push` 不带参数或 `git push --force`**
- **绝不使用 `git add .` 或 `git add -A`**：必须逐文件 add，只添加用户确认的文件
- **不自动 push**：push 需用户明确指令
- **不修改 .gitignore**：如需添加忽略规则，提醒用户自行添加
......
skill/code/test_login_auth.py
浏览文件 @ 023f2e65
......@@ -6,12 +6,23 @@
import paramiko
import json
import time
import os
def test_login_auth():
"""测试登录认证功能"""
# 从环境变量读取配置
ssh_host = os.environ.get('SSH_HOST', '192.168.5.60')
ssh_user = os.environ.get('SSH_USER', 'ubains')
ssh_password = os.environ.get('SSH_PASSWORD', '')
if not ssh_password:
print("错误：SSH_PASSWORD 环境变量未设置")
print("请设置环境变量或创建 .env 文件")
return
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('192.168.5.60', username='ubains', password='Ubains@123')
ssh.connect(ssh_host, username=ssh_user, password=ssh_password)
print('='*60)
print('功能测试验证')
......
skill/code/test_normal_user.py
浏览文件 @ 023f2e65
......@@ -6,10 +6,21 @@
import paramiko
import json
import time
import os
# 从环境变量读取配置
ssh_host = os.environ.get('SSH_HOST', '192.168.5.60')
ssh_user = os.environ.get('SSH_USER', 'ubains')
ssh_password = os.environ.get('SSH_PASSWORD', '')
if not ssh_password:
print("错误：SSH_PASSWORD 环境变量未设置")
print("请设置环境变量或创建 .env 文件")
exit(1)
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('192.168.5.60', username='ubains', password='Ubains@123')
ssh.connect(ssh_host, username=ssh_user, password=ssh_password)
print('='*60)
print('普通用户功能测试')
......
skill/code/upload_login_auth.py
浏览文件 @ 023f2e65
......@@ -7,12 +7,17 @@ import paramiko
import os
import time
# 服务器配置
HOST = '192.168.5.60'
USER = 'ubains'
PASSWORD = 'Ubains@123'
# 从环境变量读取服务器配置
HOST = os.environ.get('SSH_HOST', '192.168.5.60')
USER = os.environ.get('SSH_USER', 'ubains')
PASSWORD = os.environ.get('SSH_PASSWORD', '')
REMOTE_BASE = '/opt/troubleshoot'
if not PASSWORD:
print("错误：SSH_PASSWORD 环境变量未设置")
print("请设置环境变量或创建 .env 文件")
exit(1)
# 本地路径
LOCAL_BASE = r'E:\github\ubains-module-test\develop\.claude\skills\Troubleshoot\code'
......
skill/code/upload_update.py
浏览文件 @ 023f2e65
......@@ -7,12 +7,17 @@ import paramiko
import os
import time
# 服务器配置
HOST = '192.168.5.60'
USER = 'ubains'
PASSWORD = 'Ubains@123'
# 从环境变量读取服务器配置
HOST = os.environ.get('SSH_HOST', '192.168.5.60')
USER = os.environ.get('SSH_USER', 'ubains')
PASSWORD = os.environ.get('SSH_PASSWORD', '')
REMOTE_BASE = '/opt/troubleshoot'
if not PASSWORD:
print("错误：SSH_PASSWORD 环境变量未设置")
print("请设置环境变量或创建 .env 文件")
exit(1)
# 本地路径
LOCAL_BASE = r'E:\github\ubains-module-test\develop\.claude\skills\Troubleshoot\code'
......
skill/code/upload_users.py
浏览文件 @ 023f2e65
......@@ -4,11 +4,22 @@
"""
import paramiko
import os
import time
# 从环境变量读取服务器配置
ssh_host = os.environ.get('SSH_HOST', '192.168.5.60')
ssh_user = os.environ.get('SSH_USER', 'ubains')
ssh_password = os.environ.get('SSH_PASSWORD', '')
if not ssh_password:
print("错误：SSH_PASSWORD 环境变量未设置")
print("请设置环境变量或创建 .env 文件")
exit(1)
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect('192.168.5.60', username='ubains', password='Ubains@123')
ssh.connect(ssh_host, username=ssh_user, password=ssh_password)
# 上传 users.json
print('上传 users.json...')
......
skill/code/web/server.py
浏览文件 @ 023f2e65
......@@ -83,7 +83,12 @@ app = Flask(
# ============================================================
# Session 配置
# ============================================================
app.secret_key = os.environ.get('SECRET_KEY', 'troubleshoot-secret-key-2026-secure')
app.secret_key = os.environ.get('SECRET_KEY')
if not app.secret_key:
import secrets
app.secret_key = secrets.token_hex(32)
print("[警告] SECRET_KEY 环境变量未设置，已生成随机密钥")
print("[提示] 生产环境请设置 SECRET_KEY 环境变量")
app.config['SESSION_TYPE'] = 'filesystem'
app.config['PERMANENT_SESSION_LIFETIME'] = 86400 # 24小时
......@@ -794,7 +799,6 @@ def analyze_stream():
apk_product = request.args.get('apk_product', '').strip()
query = request.args.get('query', '').strip()
model = request.args.get('model', '') or config.get('default_model', 'glm-5.1')
query = request.args.get('query', '').strip()
if not query:
def error_gen():
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论