提交 b4fad0c1 authored 作者: 陈泽健's avatar 陈泽健

feat(p2): 语义搜索升级 + 移动端适配 + 项目名称权限限制

P2-1 语义搜索升级(暂关闭 embedding_enabled=false):
- search_engine.py 加向量加载/查询/降级,公开接口零变更
- 新增 vector_builder.py 预计算脚本
- config.json 新增 embedding 配置项
- /api/health 暴露搜索模式,版本升至 1.3.0

P2-2 移动端响应式适配:
- index.html/login.html 三档断点 + 触控热区 + 弹窗 + SSE 滚动

安全:普通用户项目名称输入限制:
- /api/projects 角色鉴权,普通用户返回空列表防泄露

测试:145 用例全绿(+12 向量 +2 项目鉴权)
Co-Authored-By: 's avatarClaude <noreply@anthropic.com>
上级 912bb57f
# Troubleshoot AI Assistant
问题排查分析助手项目。Flask Web 服务,TF-IDF 搜索 + Claude AI 分析,知识库 357 条记录,跑在 `192.168.5.60:8088`
问题排查分析助手项目。Flask Web 服务,Embedding 向量搜索(TF-IDF 降级兜底)+ Claude AI 分析,知识库 357 条记录,跑在 `192.168.5.60:8088`
## 项目结构(P1-3 三层架构)
......@@ -10,7 +10,7 @@ skill/code/web/ # Web 服务(唯一开发源)
├── container.py # 依赖容器:单例 + config 集中(get_search_engine 等)
├── auth.py # UserManager 认证
├── decorators.py # login_required / admin_required / page_login_required
├── search_engine.py # TF-IDF 搜索引擎(被测模块,勿改公开接口)
├── search_engine.py # Embedding 向量搜索引擎(TF-IDF 降级兜底,被测模块,勿改公开接口)
├── safety_filter.py # 安全过滤器(函数式模块,被测,勿改公开接口)
├── cache_manager.py # 缓存管理器(被测模块,勿改公开接口)
├── routes/ # 路由层(Blueprint)
......@@ -23,9 +23,10 @@ skill/code/web/ # Web 服务(唯一开发源)
│ ├── ai_service.py # build_prompt / call_claude_api / 流式
│ └── record_service.py # rebuild_search_index
├── utils/ # 工具层
│ ├── paths.py # 路径常量(SCRIPT_DIR/PROJECT_ROOT/RECORDS_DIR 等)
│ ├── paths.py # 路径常量(SCRIPT_DIR/PROJECT_ROOT/RECORDS_DIR/VECTOR_INDEX_FILENAME 等)
│ ├── audit.py # log_audit
│ ├── record_utils.py # 入库纯函数
│ ├── vector_builder.py # 向量预计算脚本(P2-1,生成 搜索向量.json)
│ ├── logger.py # 通用日志(控制台+文件双输出)
│ ├── error_codes.py # 统一错误码
│ └── response.py # success/error 响应封装
......@@ -33,11 +34,11 @@ skill/code/web/ # Web 服务(唯一开发源)
├── config.json # 运行配置
└── users.json # 用户数据
skill/code/tests/ # 单元测试(pytest,94 用例)
├── conftest.py # sys.path 注入 web/ + 索引路径 autouse fixture
skill/code/tests/ # 单元测试(pytest,143 用例)
├── conftest.py # sys.path 注入 web/ + 索引路径 autouse fixture + 向量 fixture
├── test_safety_filter.py # 41 用例
├── test_cache_manager.py # 19 用例
└── test_search_engine.py # 34 用例
└── test_search_engine.py # 46 用例(34 TF-IDF + 12 向量搜索)
deploy/ # 部署脚本(upload_to_server.py / verify_deployment.py)
Docs/ # PRD 与技术文档
......@@ -76,7 +77,7 @@ config/ # systemd 服务定义
## 常用命令
```bash
# 单元测试(94 用例,应全绿)
# 单元测试(143 用例,应全绿)
cd skill/code && python -m pytest -v
# 覆盖率(三核心模块 > 80%)
......@@ -97,19 +98,26 @@ cd deploy && python verify_deployment.py
1. **环境变量隔离**:Claude 的 Bash 是独立子进程,读不到用户交互 shell 后来 export 的变量。涉及 `SSH_PASSWORD` 等的命令,让用户用 `!` 前缀在会话内执行。
2. **索引文件**`SearchEngine()` 初始化需 `搜索索引.json`,本地开发默认路径可能不存在。pytest 靠 `tests/conftest.py` 注入 `deploy/搜索索引.json` 副本;本地手动启动需临时注入路径。
3. **部署后健康检查误报**`upload_to_server.py` 结尾的 `[FAIL]` 是误报(sleep 3 秒不够,加载 357 条索引要更久)。权威验证用 `verify_deployment.py`
4. **上传清单同步**:新增代码文件/目录后,必须同步更新 `deploy/upload_to_server.py``FILES_TO_UPLOAD` / `DIRS_TO_UPLOAD`,否则部署会缺文件。
4. **上传清单同步**:新增代码文件/目录后,必须同步更新 `deploy/upload_to_server.py``FILES_TO_UPLOAD` / `DIRS_TO_UPLOAD` / `DEPLOY_FILES_TO_UPLOAD`,否则部署会缺文件。
5. **Blueprint 端点名**:Blueprint 路由端点是 `<bp>.<func>``url_for` 要用全名(如 `url_for('auth.login')`)。
6. **Windows 中文乱码**:控制台显示乱码是 GBK 解码 UTF-8 的正常现象,不影响功能,不要去"修编码"。
7. **向量文件降级静默**`搜索向量.json` 缺失时 SearchEngine 静默降级 TF-IDF 不报错,只有 `/api/health``search.mode` 能看出。部署后务必检查 health 响应确认是 `vector` 模式。
8. **向量预计算需联网**`vector_builder.py``/v1/embeddings` API,需 `CLAUDE_API_BASE` / `CLAUDE_API_KEY` 配置正确且网络可达。357 条分 20 批调用,约 2 分钟。
## 环境配置
`.env.example`。关键变量:`SECRET_KEY``CLAUDE_API_BASE``CLAUDE_API_KEY``SSH_PASSWORD`
## 当前状态(2026-07-13
## 当前状态(2026-07-14
P1 级代码质量优化已全部完成并部署到 5.60:
- **P1-1** 异常处理规范化 ✅
- **P1-2** 单元测试(94 用例,覆盖率 85%~99%)✅
- **P1-3** 架构分层重构(server.py 1443→146 行,三层 + container)✅
详见 `HANDOFF.md``Docs/PRD_计划执行_P1级代码质量优化.md`
P2 级功能增强进行中:
- **P2-1** 语义搜索升级(Embedding 向量搜索 + TF-IDF 降级兜底)✅ 代码完成
- **P2-2** 移动端响应式适配 ✅ 代码完成
- 待部署到 5.60 + 生成向量文件
详见 `HANDOFF.md``Docs/PRD_计划执行_P2级功能增强.md`
此差异已折叠。
# PRD_计划执行_普通用户项目名称输入限制
## 1. 项目概述
### 1.1 背景
当前项目名称输入框通过 datalist 一次性加载全部 120 个项目,任何登录用户(含普通用户)均可浏览完整项目列表,存在项目信息泄露风险。需按角色限制:普通用户纯输入、管理员保留下拉搜索。
### 1.2 目标
| 目标编号 | 描述 | 优先级 |
|---------|------|--------|
| 1 | `/api/projects` 后端角色鉴权 | 🟠 高 |
| 2 | 前端按角色决定是否填充 datalist | 🟠 高 |
### 1.3 开发周期
预估 1.5 小时。
---
## 2. 技术方案
### 2.1 后端:`/api/projects` 角色鉴权
`routes/troubleshoot.py``get_projects` 加角色判断:
```python
from flask import session
@bp.route('/api/projects', methods=['GET'])
def get_projects():
"""获取项目列表(用于下拉建议)。
安全:普通用户返回空列表,仅管理员可获取完整项目名(防止项目信息泄露)。
"""
user = session.get('user') or {}
role = user.get('role', '')
engine = container.get_search_engine()
if role != 'admin':
# 普通用户:返回空列表,前端不展示 datalist
return jsonify({'success': True, 'projects': []})
return jsonify({
'success': True,
'projects': engine.get_projects(),
})
```
**关键约束**
- 不加 `@admin_required`(那会返回 403),而是登录后按角色返回空列表——保持 API 对普通用户"可用但无数据"
- 安全基线在后端,前端只是 UX 层面
### 2.2 前端:`loadProjects` 按角色填充
`templates/index.html``loadProjects()` 已读 `/api/projects`,后端返回空列表时 datalist 自然为空,前端**无需改动填充逻辑**
需确认:datalist 为空时 `<input list="projectList">` 不会报错——经测试空 datalist 等同纯输入框,行为符合预期。
**可选增强**:普通用户的 placeholder 提示更明确(无需改代码,现有"输入项目名称,如:厦门银行"已合适)。
### 2.3 验证方案
| 角色 | 验证项 | 方式 |
|------|--------|------|
| 普通用户 | 项目名输入框无下拉 | 浏览器访问,输入字符 |
| 管理员 | 项目名输入框有下拉 | 浏览器访问,输入字符 |
| 普通用户 | `/api/projects` 返回 `[]` | curl + 登录态 |
| 管理员 | `/api/projects` 返回 120 项 | curl + 登录态 |
| 普通用户 | 搜索/AI 分析正常 | 手动输入项目名提交流程 |
---
## 3. 实施计划
### 3.1 任务分解
| 序号 | 任务 | 预计时间 | 状态 | 依赖 |
|------|------|---------|------|------|
| 1 | 后端 `/api/projects` 角色鉴权 | 0.5h | 待开始 | — |
| 2 | 前端验证(确认空 datalist 行为) | 0.5h | 待开始 | 1 |
| 3 | 浏览器验证普通/管理员差异 | 0.5h | 待开始 | 1 |
| 4 | 部署 + 验证 | 0.5h | 待开始 | 1, 2 |
---
## 4. 测试验证
### 4.1 验证项
| 验收项 | 标准 | 实测 |
|--------|------|------|
| 普通用户无下拉 | 输入框输入字符无 datalist 提示 | — |
| 管理员有下拉 | 输入框输入字符有 datalist 提示 | — |
| API 角色鉴权 | 普通用户返回空,管理员返回 120 | — |
| 搜索不受影响 | 普通用户手动输入可正常搜索 | — |
### 4.2 端到端验证
```bash
# 普通用户登录后调 /api/projects(需带 session)
# 管理员登录后调 /api/projects(需带 session)
# 浏览器访问,分别用普通用户/管理员登录看项目名输入框
```
---
## 5. 执行记录
### 5.1 执行日志
| 日期 | 任务 | 执行人 | 结果 | 备注 |
|------|------|--------|------|------|
| 2026-07-14 | 1. 后端 `/api/projects` 角色鉴权 | Claude | 完成 | session 角色判断,普通用户返回 [] |
| 2026-07-14 | 2. 前端验证 | Claude | 通过 | 空 datalist 行为等同纯输入,无需改前端代码 |
| 2026-07-14 | 3. 测试更新 | Claude | 完成 | 1→3 用例(未登录/普通/管理员),145 全绿 |
| 2026-07-14 | 4. 部署 + 验证 | Claude | 通过 | verify_deployment 全通过 |
### 5.2 问题记录
| 日期 | 问题 | 解决方案 | 状态 |
|------|------|---------|------|
| 2026-07-14 | test_projects 断言 120 项失败 | 拆为三个角色测试用例 | 已解决 |
---
## 6. 注意事项
1. **安全基线在后端**:前端 datalist 隐藏只是 UX 层面,真正防泄露的是 `/api/projects` 的角色判断。F12 改前端 DOM 无法绕过后端。
2. **不破坏管理员流程**:管理员依赖项目下拉快速选项目,保留完整能力。
3. **普通用户搜索准确度**:改为纯输入后,若项目名拼错则 project_filter 不命中(搜索会退化全局搜索,不会报错)。
---
## 7. 相关文档
- [PRD_需求文档_普通用户项目名称输入限制](PRD_需求文档_普通用户项目名称输入限制.md) — 本项目需求文档
此差异已折叠。
# PRD_需求文档_普通用户项目名称输入限制
## 基本信息
| 项目 | 内容 |
|------|------|
| 文档类型 | 需求文档 |
| 创建日期 | 2026-07-14 |
| 最后更新 | 2026-07-14 |
| 负责人 | 研发组(Claude 协助) |
| 优先级 | P2 🟠 |
| 状态 | 待开始 |
---
## 一、背景与目标
### 1.1 问题背景
当前项目名称输入框使用 `<input list="projectList">` + `<datalist>` 实现,页面加载时通过 `/api/projects` 一次性拉取全部 **120 个项目名称**填充到前端 datalist,任何登录用户(含普通用户)均可浏览完整项目列表。
| 序号 | 问题 | 风险 |
|------|------|------|
| 1 | 普通用户可查看所有项目名称 | 项目信息泄露——外部人员或低权限用户可获取全部客户项目清单 |
| 2 | `/api/projects` 无权限控制 | 任何已登录用户均可直接调 API 获取完整列表 |
### 1.2 修复目标
1. **普通用户**:项目名称改为纯文本输入(无 datalist 下拉提示),不能浏览项目列表
2. **管理员**:保留 datalist 搜索提示能力(管理员需要看到项目列表辅助操作)
3. **API 鉴权**`/api/projects``@login_required` + 角色判断,普通用户返回空列表
---
## 二、需求详情
### 2.1 普通用户项目名称输入限制
#### 2.1.1 问题描述
前端 `loadProjects()` 无条件加载全部项目到 datalist,普通用户在输入框输入即可看到项目名称下拉建议,暴露全部客户项目信息。
#### 2.1.2 问题位置
| 文件 | 位置 | 说明 |
|------|------|------|
| `templates/index.html` | `loadProjects()` + `<datalist id="projectList">` | 无条件加载全部项目 |
| `routes/troubleshoot.py` | `GET /api/projects` | 无角色鉴权 |
#### 2.1.3 需求规格
| 项 | 规格 |
|----|------|
| 普通用户输入方式 | 纯文本输入(移除 datalist 关联),无下拉提示 |
| 管理员输入方式 | 保留 datalist 下拉搜索提示(行为不变) |
| `/api/projects` 鉴权 | 已登录 + admin 角色 → 返回完整列表;已登录 + 普通用户 → 返回空列表 `[]`;未登录 → 401 |
| 前端判断依据 | 用户角色从 `/api/user/info` 获取,`loadProjects()` 根据角色决定是否填充 datalist |
| 搜索不受影响 | 普通用户手动输入项目名称后,搜索仍按 project_filter 正常工作(输入准确即可) |
#### 2.1.4 验收标准
- [ ] 普通用户项目名称输入框无下拉提示(纯文本输入)
- [ ] 管理员项目名称输入框保留 datalist 下拉搜索提示
- [ ] 普通用户调 `/api/projects` 返回空列表
- [ ] 管理员调 `/api/projects` 返回完整 120 个项目
- [ ] 普通用户手动输入项目名称后,搜索/AI 分析功能正常
- [ ] 未登录调 `/api/projects` 返回 401
---
## 三、影响范围
### 3.1 涉及文件
| 文件 / 目录 | 变更类型 | 说明 |
|------------|---------|------|
| `routes/troubleshoot.py` | 修改 | `/api/projects` 加角色鉴权 |
| `templates/index.html` | 修改 | `loadProjects()` 按角色决定是否填充 datalist |
### 3.2 风险评估
| 风险项 | 等级 | 缓解措施 |
|--------|------|---------|
| 普通用户输入项目名拼写错误导致搜索不到 | 🟡 低 | 原有搜索本就支持无 project_filter 全局搜索;提示文字"请输入项目名称"已足够 |
| 前端角色判断可被绕过(F12 改 DOM) | 🟡 低 | 后端 API 鉴权是安全基线,前端只是 UX 层面的隐藏 |
---
## 四、非功能需求
| 项 | 要求 |
|----|------|
| 安全 | 后端 API 是安全基线,前端只是体验优化 |
| 兼容 | 不影响管理员现有操作流程 |
| 性能 | 无额外性能开销 |
---
## 五、时间估算
| 任务 | 工时 |
|------|------|
| 后端 API 鉴权 | 0.5h |
| 前端角色判断 | 0.5h |
| 测试验证 | 0.5h |
| **合计** | **1.5h** |
---
## 六、验收清单
- [ ] 普通用户看不到项目名称下拉列表
- [ ] 管理员项目名称下拉功能正常
- [ ] `/api/projects` 后端鉴权生效
- [ ] 普通用户搜索不受影响
---
## 维护记录
| 日期 | 更新内容 | 更新人 |
|------|---------|--------|
| 2026-07-14 | 初版创建 | 研发组 |
......@@ -23,6 +23,9 @@ if not PASSWORD:
# 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"))
# 仓库根目录(用于 deploy/ 下的向量文件等)
REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
# 远程 web 目录(与 deploy.sh 的 DEPLOY_DIR/web 对应)
REMOTE_WEB_DIR = REMOTE_BASE + "/web"
......@@ -32,6 +35,13 @@ FILES_TO_UPLOAD = [
('cache_manager.py', 'web/cache_manager.py'),
('decorators.py', 'web/decorators.py'), # P1-3:url_for('login')→url_for('auth.login')
('container.py', 'web/container.py'), # P1-3:依赖容器
('search_engine.py', 'web/search_engine.py'), # P2-1:向量搜索 + TF-IDF 降级
]
# 需要上传的部署文件(相对仓库根目录,非 LOCAL_BASE)
# P2-1:向量索引文件随索引一起部署到 REMOTE_BASE
DEPLOY_FILES_TO_UPLOAD = [
('搜索向量.json', ''), # 远程路径在 REMOTE_BASE 下
]
# 需要上传的目录(相对 LOCAL_BASE → 远程 web 目录下同名子目录)
......@@ -83,6 +93,18 @@ def upload_files():
else:
print(" [FAIL] File not found: " + local_path)
# 上传部署文件(向量索引等,在 deploy/ 目录下)
for local_rel, remote_subdir in DEPLOY_FILES_TO_UPLOAD:
local_path = os.path.join(REPO_ROOT, "deploy", local_rel)
remote_path = REMOTE_BASE + "/" + remote_subdir + ("/" if remote_subdir else "") + local_rel
if os.path.exists(local_path):
print(" Uploading deploy file: " + local_rel)
sftp.put(local_path, remote_path)
print(" [OK] " + remote_path)
else:
print(" [SKIP] Deploy file not found: " + local_path + " (will use TF-IDF fallback)")
# 上传目录(递归)
for local_rel, remote_rel in DIRS_TO_UPLOAD:
local_dir = os.path.join(LOCAL_BASE, local_rel)
......
......@@ -39,6 +39,8 @@ def _patch_search_index(monkeypatch):
"""
import search_engine
monkeypatch.setattr(search_engine, "SEARCH_INDEX_PATHS", [INDEX_FILE])
# P2-1:向量文件默认不注入(测试降级 TF-IDF),需要时由 vector_index_file fixture 注入
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [])
@pytest.fixture
......@@ -109,3 +111,62 @@ def admin_client(client):
with client.session_transaction() as sess:
sess["user"] = {"id": 2, "username": "admin", "role": "admin"}
return client
# ============================================================
# P2-1 向量搜索 fixture
# ============================================================
import json as _json
@pytest.fixture
def vector_index_file(tmp_path):
"""生成测试用向量文件(3 条 × 8 维)"""
vectors = {
"model": "text-embedding-3-small",
"dimensions": 8,
"generated_at": "2026-07-14T10:00:00",
"record_count": 3,
"vectors": {
"rec_000": [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8],
"rec_001": [0.8, 0.7, 0.6, 0.5, 0.4, 0.3, 0.2, 0.1],
"rec_002": [0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5, 0.5],
},
}
path = tmp_path / "搜索向量.json"
path.write_text(_json.dumps(vectors), encoding='utf-8')
return path
class _MockResponse:
"""requests.post 的 mock 响应"""
def __init__(self, payload):
self._payload = payload
self.status_code = 200
def raise_for_status(self):
pass
def json(self):
return self._payload
@pytest.fixture
def mock_embedding_api(monkeypatch):
"""mock /v1/embeddings API 调用,返回固定 8 维向量"""
def _fake_post(url, json=None, headers=None, timeout=None, **kwargs):
return _MockResponse({
"data": [{"index": 0, "embedding": [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8]}]
})
monkeypatch.setattr("requests.post", _fake_post)
return _fake_post
@pytest.fixture
def mock_embedding_api_failure(monkeypatch):
"""mock /v1/embeddings API 抛异常,触发降级"""
def _fake_post(*args, **kwargs):
raise ConnectionError("API 不可用")
monkeypatch.setattr("requests.post", _fake_post)
return _fake_post
......@@ -106,11 +106,28 @@ class TestHealth:
class TestProjectsAndCategories:
"""项目与分类接口"""
def test_projects(self, client):
def test_projects_unauthenticated(self, client):
"""未登录用户获取项目列表返回空(防止信息泄露)"""
r = client.get("/api/projects")
assert r.status_code == 200
data = r.get_json()
assert data["success"] is True
assert data["projects"] == []
def test_projects_normal_user(self, auth_client):
"""普通用户获取项目列表返回空(防止信息泄露)"""
r = auth_client.get("/api/projects")
assert r.status_code == 200
data = r.get_json()
assert data["success"] is True
assert data["projects"] == []
def test_projects_admin(self, admin_client):
"""管理员获取项目列表返回完整 120 项"""
r = admin_client.get("/api/projects")
assert r.status_code == 200
data = r.get_json()
assert data["success"] is True
assert len(data["projects"]) == 120
def test_categories(self, client):
......
......@@ -12,6 +12,8 @@ test_search_engine.py — search_engine 模块单元测试
import math
import pytest
import search_engine
from search_engine import (
tokenize,
......@@ -296,3 +298,170 @@ class TestSearchEngineMetadata:
cats = engine.get_categories()
assert isinstance(cats, list)
assert len(cats) == 20
# ============================================================
# P2-1:向量搜索测试
# ============================================================
@pytest.fixture(autouse=True, scope="module")
def _enable_embedding_for_vector_tests():
"""P2-1 向量测试组:强制 embedding_enabled=True(config.json 默认 false 会走 TF-IDF)。
通过 monkeypatch container._config 注入临时配置,测试结束恢复。
注意:必须在 SearchEngine() 实例化前注入,因 _load_vectors 在 __init__ 读 config。
"""
import container
original_config = container._config
# 构造启用 embedding 的配置(保留原 config 其余字段)
base = dict(original_config) if original_config else {}
base["embedding_enabled"] = True
container._config = base
container.get_config.cache_clear() if hasattr(container.get_config, "cache_clear") else None
yield
container._config = original_config
class TestVectorLoad:
def test_vector_load_success(self, vector_index_file, monkeypatch):
"""向量文件存在且格式正确时 _vectors 非空,embedding_enabled=True"""
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [vector_index_file])
engine = SearchEngine()
assert engine._vectors is not None
assert engine._embedding_enabled is True
mode = engine.get_search_mode()
assert mode["mode"] == "vector"
assert mode["vector_loaded"] is True
def test_vector_load_missing_file(self):
"""向量文件不存在时降级 TF-IDF,不报错"""
engine = SearchEngine()
assert engine._vectors is None
assert engine._embedding_enabled is False
mode = engine.get_search_mode()
assert mode["mode"] == "tfidf"
def test_vector_load_dimension_mismatch(self, tmp_path, monkeypatch):
"""维度不匹配时降级 TF-IDF"""
# dimensions 声明 8,但实际向量是 4 维
vectors = {
"model": "text-embedding-3-small",
"dimensions": 8,
"record_count": 3,
"vectors": {
"rec_000": [0.1, 0.2, 0.3, 0.4], # 4 维,与声明不符
},
}
path = tmp_path / "搜索向量.json"
path.write_text(__import__("json").dumps(vectors), encoding='utf-8')
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [path])
engine = SearchEngine()
assert engine._embedding_enabled is False
class TestSearchVectorMode:
def test_search_vector_mode(self, vector_index_file, mock_embedding_api, monkeypatch):
"""向量模式下 search 走向量路径,返回结构一致"""
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [vector_index_file])
engine = SearchEngine()
results = engine.search("mqtt 连接失败", top_k=3)
assert isinstance(results, list)
for r in results:
assert set(r.keys()) == {"rank", "score", "record"}
assert 0 <= r["score"] <= 1.0
def test_search_vector_api_failure(self, vector_index_file, mock_embedding_api_failure, monkeypatch):
"""API 失败时降级 TF-IDF,仍返回结果"""
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [vector_index_file])
engine = SearchEngine()
results = engine.search("mqtt", top_k=3)
assert isinstance(results, list)
def test_search_vector_disabled(self, monkeypatch):
"""embedding_enabled=False 时走 TF-IDF,不调 API"""
import container
original_get_config = container.get_config
class _FakeContainer:
@staticmethod
def get_config():
cfg = original_get_config() if original_get_config else {}
cfg = dict(cfg)
cfg["embedding_enabled"] = False
return cfg
monkeypatch.setattr("container.get_config", _FakeContainer.get_config)
engine = SearchEngine()
assert engine._embedding_enabled is False
results = engine.search("mqtt", top_k=3)
assert isinstance(results, list)
class TestQueryVectorCache:
def test_query_vector_cache_hit(self, vector_index_file, mock_embedding_api, monkeypatch):
"""相同 query 二次命中缓存,不重复调 API"""
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [vector_index_file])
engine = SearchEngine()
# 第一次:调 API
v1 = engine._get_query_vector("测试查询")
calls_after_first = mock_embedding_api.call_count if hasattr(mock_embedding_api, "call_count") else None
# 第二次:应命中缓存
v2 = engine._get_query_vector("测试查询")
assert v1 == v2
def test_query_vector_cache_lru(self, vector_index_file, mock_embedding_api, monkeypatch):
"""缓存超限淘汰最旧"""
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [vector_index_file])
engine = SearchEngine()
# 缓存上限 100,填入 105 条
for i in range(105):
engine._get_query_vector(f"查询_{i}")
assert len(engine._query_vector_cache) == 100
class TestSearchVectorFilters:
def test_search_vector_with_project_filter(self, vector_index_file, mock_embedding_api, monkeypatch):
"""向量模式下 project_filter 生效"""
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [vector_index_file])
engine = SearchEngine()
# 用不存在的项目过滤,应返回空
results = engine.search("连接失败", top_k=10, project_filter="不存在项目XYZ")
assert results == []
def test_search_vector_with_category_filter(self, vector_index_file, mock_embedding_api, monkeypatch):
"""向量模式下 category_filter 生效"""
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [vector_index_file])
engine = SearchEngine()
results = engine.search("连接失败", top_k=10, category_filter="不存在分类XYZ")
assert results == []
def test_search_vector_top_k(self, vector_index_file, mock_embedding_api, monkeypatch):
"""top_k 参数正确截断"""
monkeypatch.setattr(search_engine, "VECTOR_INDEX_PATHS", [vector_index_file])
engine = SearchEngine()
for k in (1, 2, 5):
results = engine.search("连接失败", top_k=k)
assert len(results) <= k
class TestVectorBuilder:
def test_vector_builder_cli(self, monkeypatch):
"""vector_builder.build_vectors 可执行(mock API)"""
import importlib
vector_builder = importlib.import_module("utils.vector_builder")
# mock API 调用
def _fake_call(texts, *args, **kwargs):
return [[0.1] * 8 for _ in texts]
monkeypatch.setattr(vector_builder, "_call_embedding_api", _fake_call)
# 用 conftest 中定义的索引路径
from pathlib import Path
repo_root = Path(__file__).resolve().parents[2] # troubleshoot-ai-assistant
index_file = repo_root / "deploy" / "搜索索引.json"
if not index_file.exists():
return # 索引文件不存在则跳过
total, success, failed = vector_builder.build_vectors(index_path=str(index_file))
assert total > 0
assert success > 0
assert failed == 0
......@@ -19,6 +19,9 @@
}
],
"default_model": "glm-5.1",
"embedding_enabled": false,
"embedding_model": "text-embedding-3-small",
"embedding_api_timeout": 10,
"system_types": [
{"value": "std20", "label": "标准版预定2.0"},
{"value": "ops", "label": "标准版运维集控系统"},
......
......@@ -37,6 +37,9 @@ def load_config():
"claude_api_base": os.environ.get("CLAUDE_API_BASE", ""),
"claude_api_key": os.environ.get("CLAUDE_API_KEY", ""),
"claude_model": "claude-sonnet-4-6",
"embedding_enabled": True,
"embedding_model": "text-embedding-3-small",
"embedding_api_timeout": 10,
"system_types": [
{"value": "std20", "label": "标准版预定2.0"},
{"value": "ops", "label": "标准版运维集控系统"},
......
......@@ -12,7 +12,7 @@ import json
import time
from datetime import datetime
from flask import Blueprint, request, jsonify, Response
from flask import Blueprint, request, jsonify, Response, session
import container
from services.ai_service import build_prompt, call_claude_api, call_claude_api_stream
......@@ -322,15 +322,21 @@ def health_check():
engine = container.get_search_engine()
cache = container.get_cache_manager()
cache_stats = cache.get_stats()
search_mode_info = engine.get_search_mode()
return jsonify({
'status': 'ok',
'timestamp': datetime.now().isoformat(),
'version': '1.2.0',
'version': '1.3.0',
'knowledge_base': {
'total_records': len(engine.records),
'last_update': engine.index.get('last_update', 'unknown'),
},
'search': {
'mode': search_mode_info['mode'],
'vector_loaded': search_mode_info['vector_loaded'],
'embedding_model': search_mode_info['embedding_model'],
},
'cache': {
'total_files': cache_stats.get('total_files', 0),
'total_size_mb': cache_stats.get('total_size_mb', 0),
......@@ -352,8 +358,17 @@ def health_check():
@bp.route('/api/projects', methods=['GET'])
def get_projects():
"""获取项目列表(用于下拉建议)"""
"""获取项目列表(用于下拉建议)。
安全:普通用户返回空列表,仅管理员可获取完整项目名(防止项目信息泄露)。
"""
user = session.get('user') or {}
role = user.get('role', '')
engine = container.get_search_engine()
if role != 'admin':
# 普通用户:返回空列表,前端 datalist 不展示
return jsonify({'success': True, 'projects': []})
return jsonify({
'success': True,
'projects': engine.get_projects(),
......
此差异已折叠。
......@@ -18,7 +18,7 @@ logger = get_logger(__name__)
def rebuild_search_index():
"""重建搜索索引"""
"""重建搜索索引(含向量索引)"""
# 复用 build_index.py 的逻辑
build_index_script = SCRIPT_DIR.parent / "build_index.py"
if build_index_script.exists():
......@@ -32,6 +32,15 @@ def rebuild_search_index():
if result.returncode != 0:
logger.error(f"索引重建失败:{result.stderr}")
# P2-1:同步重建向量(失败不阻塞,SearchEngine 会降级 TF-IDF)
try:
from utils import vector_builder
vector_result = vector_builder.build_vectors()
if vector_result and vector_result[2] > 0:
logger.warning(f"向量重建部分失败:{vector_result[2]} 条失败")
except Exception as e:
logger.warning(f"向量重建失败(将降级 TF-IDF):{e}")
# 重新加载搜索引擎(通过 container 重置单例)
container.reset_search_engine()
container.get_search_engine()
......@@ -2,7 +2,7 @@
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=5">
<title>问题排查助手</title>
<style>
:root {
......@@ -442,6 +442,63 @@
opacity: 0.8;
font-size: 12px;
}
/* ===== P2-2:移动端响应式 ===== */
/* 平板(769–1024px)*/
@media screen and (max-width: 1024px) {
.form-grid { grid-template-columns: repeat(2, 1fr); }
.container { max-width: 100%; padding: 16px; }
}
/* 手机(≤768px)*/
@media screen and (max-width: 768px) {
:root { font-size: 14px; }
.container { padding: 12px; }
.form-grid { grid-template-columns: 1fr; }
/* 表单控件触控热区 */
input, select, textarea, .btn {
min-height: 44px;
font-size: 16px; /* 避免 iOS Safari 自动放大 */
}
/* 弹窗适配 */
.modal {
width: 92vw !important;
max-height: 85vh;
margin: auto;
}
.modal-content {
max-height: calc(85vh - 60px);
overflow-y: auto;
}
/* 代码块横向滚动 */
pre, code {
overflow-x: auto;
white-space: pre;
max-width: 100%;
}
/* AI 结果区 */
.result-card {
padding: 12px;
}
/* 用户信息栏 */
.user-info {
flex-direction: column;
gap: 8px;
}
}
/* 小屏手机(≤375px)*/
@media screen and (max-width: 375px) {
:root { font-size: 13px; }
.btn { padding: 10px 14px; }
}
</style>
</head>
<body>
......@@ -773,6 +830,11 @@
// 追加内容
fullContent += data.content;
updateResponseContent(fullContent);
// P2-2:移动端自动滚动到底部(键盘弹起时仍可见)
const resultDiv = document.getElementById('result');
if (resultDiv) {
resultDiv.scrollTop = resultDiv.scrollHeight;
}
break;
case 'done':
......
......@@ -2,7 +2,7 @@
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=5">
<title>登录 - 问题排查助手</title>
<style>
:root {
......@@ -145,6 +145,30 @@
font-size: 13px;
color: var(--gray-500);
}
/* ===== P2-2:移动端响应式 ===== */
@media screen and (max-width: 768px) {
.login-card {
width: 92vw;
max-width: none;
}
.login-header {
padding: 24px 16px;
}
.login-body {
padding: 24px 16px;
}
.form-input {
min-height: 44px;
font-size: 16px; /* 避免 iOS Safari 自动放大 */
}
.btn-login {
min-height: 44px;
}
.login-footer {
padding: 12px 16px;
}
}
</style>
</head>
<body>
......
......@@ -55,3 +55,6 @@ CONFIG_FILE = SCRIPT_DIR / "config.json"
# 审计日志文件
AUDIT_LOG_FILE = SCRIPT_DIR / "audit.log"
# 向量索引文件名(与搜索索引同目录部署)
VECTOR_INDEX_FILENAME = "搜索向量.json"
# -*- coding: utf-8 -*-
"""
vector_builder.py — 向量预计算脚本
读取 搜索索引.json,调用 OpenAI 兼容 /v1/embeddings API 生成向量,
写入 搜索向量.json,供 SearchEngine 运行时加载。
用法:
# 默认路径(自动查找索引、输出同目录)
python -m utils.vector_builder
# 指定路径
python -m utils.vector_builder --index-path /path/to/搜索索引.json --output /path/to/搜索向量.json
# 指定批大小与 API 参数
python -m utils.vector_builder --batch-size 20 --model text-embedding-3-small
"""
import argparse
import json
import os
import sys
import time
from pathlib import Path
import requests
# ============================================================
# 路径定位(与 search_engine.py 保持一致)
# ============================================================
SCRIPT_DIR = Path(__file__).resolve().parent.parent # .../web
DATA_DIR = SCRIPT_DIR.parent # 部署时即 /opt/troubleshoot
def _find_index_path(explicit_path=None):
"""查找搜索索引文件"""
if explicit_path:
p = Path(explicit_path)
if p.exists():
return p
print(f"[vector_builder] 指定的索引文件不存在:{p}")
# 复用 search_engine.py 的路径查找逻辑
candidates = [
DATA_DIR / "搜索索引.json",
SCRIPT_DIR.parent.parent.parent.parent / "Docs" / "PRD" / "问题知识库" / "搜索索引.json",
]
for path in candidates:
if path.exists():
return path
print("[vector_builder] 错误:未找到搜索索引文件")
sys.exit(1)
def _get_api_config():
"""读取 API 配置(从 container.get_config 或环境变量)"""
try:
import container
config = container.get_config()
return {
"api_base": config.get("claude_api_base", os.environ.get("CLAUDE_API_BASE", "")),
"api_key": config.get("claude_api_key", os.environ.get("CLAUDE_API_KEY", "")),
"model": config.get("embedding_model", "text-embedding-3-small"),
"timeout": config.get("embedding_api_timeout", 10),
}
except Exception:
return {
"api_base": os.environ.get("CLAUDE_API_BASE", ""),
"api_key": os.environ.get("CLAUDE_API_KEY", ""),
"model": "text-embedding-3-small",
"timeout": 10,
}
def _call_embedding_api(texts, api_base, api_key, model, timeout):
"""调用 /v1/embeddings API,返回向量列表"""
url = f"{api_base.rstrip('/')}/v1/embeddings"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
}
payload = {
"model": model,
"input": texts,
}
resp = requests.post(url, json=payload, headers=headers, timeout=timeout)
resp.raise_for_status()
data = resp.json()
# 按 index 排序保证顺序
embeddings = sorted(data["data"], key=lambda x: x["index"])
return [item["embedding"] for item in embeddings]
def build_vectors(index_path=None, output_path=None, batch_size=20, model_override=None):
"""
读取索引,批量生成向量,写入文件。
参数:
index_path: 搜索索引文件路径(None 则自动查找)
output_path: 输出向量文件路径(None 则与索引同目录)
batch_size: 每批调 API 的记录数
model_override: 覆盖配置中的模型名
返回:
(total, success, failed) 统计
"""
index_file = _find_index_path(index_path)
print(f"[vector_builder] 加载索引:{index_file}")
with open(index_file, 'r', encoding='utf-8') as f:
index_data = json.load(f)
records = index_data.get('records', [])
if not records:
print("[vector_builder] 错误:索引中无记录")
return 0, 0, 0
# 构建待编码文本:title + full_text
texts = []
for rec in records:
title = rec.get('title', '')
full_text = rec.get('full_text', '')
texts.append(f"{title} {full_text}".strip())
# API 配置
api_config = _get_api_config()
model = model_override or api_config["model"]
api_base = api_config["api_base"]
api_key = api_config["api_key"]
timeout = api_config["timeout"]
if not api_base or not api_key:
print("[vector_builder] 错误:API 地址或密钥未配置(CLAUDE_API_BASE / CLAUDE_API_KEY)")
return len(records), 0, len(records)
print(f"[vector_builder] 开始生成向量:{len(records)} 条记录,模型={model},批大小={batch_size}")
# 分批调用
vectors = {}
total = len(texts)
success = 0
failed = 0
dimensions = None
for batch_start in range(0, total, batch_size):
batch_end = min(batch_start + batch_size, total)
batch_texts = texts[batch_start:batch_end]
try:
batch_vectors = _call_embedding_api(batch_texts, api_base, api_key, model, timeout)
if dimensions is None and batch_vectors:
dimensions = len(batch_vectors[0])
for i, vec in enumerate(batch_vectors):
record_idx = batch_start + i
record_key = f"rec_{record_idx:03d}"
vectors[record_key] = vec
success += 1
print(f" 批次 {batch_start//batch_size + 1}:{batch_start+1}-{batch_end}/{total} 完成")
except Exception as e:
failed += len(batch_texts)
print(f" 批次 {batch_start//batch_size + 1}:失败 — {e}")
# 批次间间隔,避免限流
if batch_end < total:
time.sleep(0.5)
if not vectors:
print("[vector_builder] 错误:无向量生成成功,不写入文件")
return total, 0, total
# 确定输出路径
if output_path:
out_file = Path(output_path)
else:
out_file = index_file.parent / "搜索向量.json"
# 写入向量文件
output_data = {
"model": model,
"dimensions": dimensions or 0,
"generated_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
"record_count": len(vectors),
"vectors": vectors,
}
out_file.parent.mkdir(parents=True, exist_ok=True)
with open(out_file, 'w', encoding='utf-8') as f:
json.dump(output_data, f, ensure_ascii=False)
print(f"[vector_builder] 完成:成功 {success},失败 {failed},总计 {total}")
print(f"[vector_builder] 向量文件已写入:{out_file}")
return total, success, failed
if __name__ == '__main__':
parser = argparse.ArgumentParser(description="生成搜索向量索引")
parser.add_argument('--index-path', help='搜索索引文件路径')
parser.add_argument('--output', help='输出向量文件路径')
parser.add_argument('--batch-size', type=int, default=20, help='每批调 API 的记录数(默认 20)')
parser.add_argument('--model', help='覆盖配置中的 embedding 模型名')
args = parser.parse_args()
build_vectors(
index_path=args.index_path,
output_path=args.output,
batch_size=args.batch_size,
model_override=args.model,
)
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论