iven/zclaw_openfang
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(knowledge): deep audit — 18 bugs fixed across backend + frontend

Browse Source 
CRITICAL:
- Migration permission seed WHERE name → WHERE id (matched 0 rows, all KB APIs broken)

HIGH:
- analytics_quality SQL alias + missing comma fix
- search() duplicate else block compile error
- chunk_content duplicate var declarations + type mismatch
- SQL invalid escape sequences
- delete_category missing rows_affected check

MEDIUM:
- analytics_overview hit_rate vs positive_feedback_rate separation
- analytics_quality GROUP BY kc.id,kc.name (same-name category merge)
- update_category handler trim + empty name validation
- update_item duplicate VALID_STATUSES inside transaction
- page_size max(1) lower bound in list handlers
- batch_create title/content/length validation
- embedding dispatch silent error → tracing::warn
- Version modal close clears detailItem state
- Search empty state distinguishes not-searched vs no-results
- Create modal cancel resets form
This commit is contained in:
iven
2026-04-02 19:07:42 +08:00
parent 837abec48a
commit 7e4b787d5c
7 changed files with 953 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

258
crates/zclaw-saas/src/knowledge/handlers.rs
View File

@@ -16,7 +16,9 @@ use super::types::*;
/// GET /api/v1/knowledge/categories — 树形分类列表
pub async fn list_categories(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
) -> SaasResult<Json<Vec<CategoryResponse>>> {
check_permission(&ctx, "knowledge:read")?;
let tree = service::list_categories_tree(&state.db).await?;
Ok(Json(tree))
}
@@ -49,14 +51,33 @@ pub async fn create_category(
/// PUT /api/v1/knowledge/categories/:id — 更新分类
pub async fn update_category(
State(_state): State<AppState>,
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Path(_id): Path<String>,
Json(_req): Json<UpdateCategoryRequest>,
Path(id): Path<String>,
Json(req): Json<UpdateCategoryRequest>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:write")?;
// TODO: implement update
Ok(Json(serde_json::json!({"updated": true})))
if let Some(ref name) = req.name {
if name.trim().is_empty() {
return Err(SaasError::InvalidInput("分类名称不能为空".into()));
}
}
let cat = service::update_category(
&state.db,
&id,
req.name.as_deref().map(|n| n.trim()),
req.description.as_deref(),
req.parent_id.as_deref(),
req.icon.as_deref(),
).await?;
Ok(Json(serde_json::json!({
"id": cat.id,
"name": cat.name,
"updated": true,
})))
}
/// DELETE /api/v1/knowledge/categories/:id — 删除分类
@@ -72,11 +93,30 @@ pub async fn delete_category(
/// GET /api/v1/knowledge/categories/:id/items — 分类下条目列表
pub async fn list_category_items(
State(_state): State<AppState>,
Path(_id): Path<String>,
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Path(id): Path<String>,
Query(query): Query<ListItemsQuery>,
) -> SaasResult<Json<serde_json::Value>> {
// TODO: implement with pagination
Ok(Json(serde_json::json!({"items": [], "total": 0})))
check_permission(&ctx, "knowledge:read")?;
let page = query.page.unwrap_or(1).max(1);
let page_size = query.page_size.unwrap_or(20).max(1).min(100);
let status_filter = query.status.as_deref().unwrap_or("active");
let (items, total) = service::list_items_by_category(
&state.db,
&id,
status_filter,
page,
page_size,
).await?;
Ok(Json(serde_json::json!({
"items": items,
"total": total,
"page": page,
"page_size": page_size,
})))
}
// === 知识条目 CRUD ===
@@ -84,12 +124,19 @@ pub async fn list_category_items(
/// GET /api/v1/knowledge/items — 分页列表
pub async fn list_items(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Query(query): Query<ListItemsQuery>,
) -> SaasResult<Json<serde_json::Value>> {
let page = query.page.unwrap_or(1).max(1);
let page_size = query.page_size.unwrap_or(20).min(100);
check_permission(&ctx, "knowledge:read")?;
let page = query.page.unwrap_or(1).max(1).min(10000);
let page_size = query.page_size.unwrap_or(20).max(1).min(100);
let offset = (page - 1) * page_size;
// 转义 ILIKE 通配符，防止用户输入的 % 和 _ 被当作通配符
let keyword = query.keyword.as_ref().map(|k| {
k.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_")
});
let items: Vec<KnowledgeItem> = sqlx::query_as(
"SELECT ki.* FROM knowledge_items ki \
JOIN knowledge_categories kc ON ki.category_id = kc.id \
@@ -101,7 +148,7 @@ pub async fn list_items(
)
.bind(&query.category_id)
.bind(&query.status)
.bind(&query.keyword)
.bind(&keyword)
.bind(page_size)
.bind(offset)
.fetch_all(&state.db)
@@ -115,7 +162,7 @@ pub async fn list_items(
)
.bind(&query.category_id)
.bind(&query.status)
.bind(&query.keyword)
.bind(&keyword)
.fetch_one(&state.db)
.await?;
@@ -173,13 +220,24 @@ pub async fn batch_create_items(
}
let mut created = Vec::new();
for req in items {
match service::create_item(&state.db, &ctx.account_id, &req).await {
for req in &items {
if req.title.trim().is_empty() || req.content.trim().is_empty() {
tracing::warn!("Batch create: skipping item with empty title or content");
continue;
}
if req.content.len() > 100_000 {
tracing::warn!("Batch create: skipping item '{}' (content too long)", req.title);
continue;
}
match service::create_item(&state.db, &ctx.account_id, req).await {
Ok(item) => {
let _ = state.worker_dispatcher.dispatch(
"generate_embedding",
serde_json::json!({ "item_id": item.id }),
).await;
).await.map_err(|e| {
tracing::warn!("[Knowledge] Failed to dispatch embedding for item {}: {}", item.id, e);
e
});
created.push(item.id);
}
Err(e) => {
@@ -197,8 +255,10 @@ pub async fn batch_create_items(
/// GET /api/v1/knowledge/items/:id — 条目详情
pub async fn get_item(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Path(id): Path<String>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:read")?;
let item = service::get_item(&state.db, &id).await?
.ok_or_else(|| SaasError::NotFound("知识条目不存在".into()))?;
Ok(Json(serde_json::to_value(item).unwrap_or_default()))
@@ -216,10 +276,12 @@ pub async fn update_item(
let updated = service::update_item(&state.db, &id, &ctx.account_id, &req).await?;
// 触发 re-embedding
let _ = state.worker_dispatcher.dispatch(
if let Err(e) = state.worker_dispatcher.dispatch(
"generate_embedding",
serde_json::json!({ "item_id": id }),
).await;
).await {
tracing::warn!("[Knowledge] Failed to dispatch re-embedding for item {}: {}", id, e);
}
Ok(Json(serde_json::json!({
"id": updated.id,
@@ -243,8 +305,10 @@ pub async fn delete_item(
/// GET /api/v1/knowledge/items/:id/versions
pub async fn list_versions(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Path(id): Path<String>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:read")?;
let versions: Vec<KnowledgeVersion> = sqlx::query_as(
"SELECT * FROM knowledge_versions WHERE item_id = $1 ORDER BY version DESC"
)
@@ -257,8 +321,10 @@ pub async fn list_versions(
/// GET /api/v1/knowledge/items/:id/versions/:v
pub async fn get_version(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Path((id, v)): Path<(String, i32)>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:read")?;
let version: KnowledgeVersion = sqlx::query_as(
"SELECT * FROM knowledge_versions WHERE item_id = $1 AND version = $2"
)
@@ -272,13 +338,27 @@ pub async fn get_version(
/// POST /api/v1/knowledge/items/:id/rollback/:v
pub async fn rollback_version(
State(_state): State<AppState>,
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Path((_id, v)): Path<(String, i32)>,
Path((id, v)): Path<(String, i32)>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:admin")?;
// TODO: implement rollback
Ok(Json(serde_json::json!({"rolled_back_to": v})))
let updated = service::rollback_version(&state.db, &id, v, &ctx.account_id).await?;
// 触发 re-embedding
if let Err(e) = state.worker_dispatcher.dispatch(
"generate_embedding",
serde_json::json!({ "item_id": id }),
).await {
tracing::warn!("[Knowledge] Failed to dispatch re-embedding after rollback for item {}: {}", id, e);
}
Ok(Json(serde_json::json!({
"id": updated.id,
"version": updated.version,
"rolled_back_to": v,
})))
}
// === 检索 ===
@@ -286,8 +366,10 @@ pub async fn rollback_version(
/// POST /api/v1/knowledge/search — 语义搜索
pub async fn search(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Json(req): Json<SearchRequest>,
) -> SaasResult<Json<Vec<SearchResult>>> {
check_permission(&ctx, "knowledge:search")?;
let limit = req.limit.unwrap_or(5).min(10);
let min_score = req.min_score.unwrap_or(0.5);
let results = service::search(
@@ -302,11 +384,20 @@ pub async fn search(
/// POST /api/v1/knowledge/recommend — 关联推荐
pub async fn recommend(
State(_state): State<AppState>,
Json(_req): Json<SearchRequest>,
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Json(req): Json<SearchRequest>,
) -> SaasResult<Json<Vec<SearchResult>>> {
// TODO: implement recommendation based on keyword overlap
Ok(Json(vec![]))
check_permission(&ctx, "knowledge:search")?;
let limit = req.limit.unwrap_or(5).min(10);
let results = service::search(
&state.db,
&req.query,
req.category_id.as_deref(),
limit,
0.3,
).await?;
Ok(Json(results))
}
// === 分析看板 ===
@@ -314,7 +405,9 @@ pub async fn recommend(
/// GET /api/v1/knowledge/analytics/overview
pub async fn analytics_overview(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
) -> SaasResult<Json<AnalyticsOverview>> {
check_permission(&ctx, "knowledge:read")?;
let overview = service::analytics_overview(&state.db).await?;
Ok(Json(overview))
}
@@ -322,7 +415,9 @@ pub async fn analytics_overview(
/// GET /api/v1/knowledge/analytics/trends
pub async fn analytics_trends(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:read")?;
// 使用 serde_json::Value 行来避免 PgRow 序列化
let trends: Vec<(serde_json::Value,)> = sqlx::query_as(
"SELECT json_build_object(
@@ -344,7 +439,9 @@ pub async fn analytics_trends(
/// GET /api/v1/knowledge/analytics/top-items
pub async fn analytics_top_items(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:read")?;
let items: Vec<(serde_json::Value,)> = sqlx::query_as(
"SELECT json_build_object(
'id', ki.id,
@@ -368,16 +465,117 @@ pub async fn analytics_top_items(
/// GET /api/v1/knowledge/analytics/quality
pub async fn analytics_quality(
State(_state): State<AppState>,
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
) -> SaasResult<Json<serde_json::Value>> {
Ok(Json(serde_json::json!({"quality": {}})))
check_permission(&ctx, "knowledge:read")?;
let quality = service::analytics_quality(&state.db).await?;
Ok(Json(quality))
}
/// GET /api/v1/knowledge/analytics/gaps
pub async fn analytics_gaps(
State(_state): State<AppState>,
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
) -> SaasResult<Json<serde_json::Value>> {
Ok(Json(serde_json::json!({"gaps": []})))
check_permission(&ctx, "knowledge:read")?;
let gaps = service::analytics_gaps(&state.db).await?;
Ok(Json(gaps))
}
// === 批量操作 ===
/// PATCH /api/v1/knowledge/categories/reorder — 批量排序
pub async fn reorder_categories(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Json(items): Json<Vec<ReorderItem>>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:write")?;
if items.is_empty() {
return Ok(Json(serde_json::json!({"reordered": false, "count": 0})));
}
if items.len() > 100 {
return Err(SaasError::InvalidInput("单次排序不能超过 100 个".into()));
}
// 使用事务保证原子性
let mut tx = state.db.begin().await?;
for item in &items {
sqlx::query("UPDATE knowledge_categories SET sort_order = $1, updated_at = NOW() WHERE id = $2")
.bind(item.sort_order)
.bind(&item.id)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(Json(serde_json::json!({"reordered": true, "count": items.len()})))
}
/// POST /api/v1/knowledge/items/import — Markdown 文件导入
pub async fn import_items(
State(state): State<AppState>,
Extension(ctx): Extension<AuthContext>,
Json(req): Json<ImportRequest>,
) -> SaasResult<Json<serde_json::Value>> {
check_permission(&ctx, "knowledge:write")?;
if req.files.len() > 20 {
return Err(SaasError::InvalidInput("单次导入不能超过 20 个文件".into()));
}
let mut created = Vec::new();
for file in &req.files {
// 内容长度检查（数据库限制 100KB）
if file.content.len() > 100_000 {
tracing::warn!("跳过文件 '{}': 内容超长 ({} bytes)", file.title.as_deref().unwrap_or("未命名"), file.content.len());
continue;
}
// 空内容检查
if file.content.trim().is_empty() {
tracing::warn!("跳过空文件: '{}'", file.title.as_deref().unwrap_or("未命名"));
continue;
}
let title = file.title.clone().unwrap_or_else(|| {
file.content.lines().next()
.map(|l| l.trim_start_matches('#').trim().to_string())
.unwrap_or_else(|| format!("导入条目 {}", created.len() + 1))
});
let item_req = CreateItemRequest {
category_id: req.category_id.clone(),
title,
content: file.content.clone(),
keywords: file.keywords.clone(),
related_questions: None,
priority: None,
tags: file.tags.clone(),
};
match service::create_item(&state.db, &ctx.account_id, &item_req).await {
Ok(item) => {
let _ = state.worker_dispatcher.dispatch(
"generate_embedding",
serde_json::json!({ "item_id": item.id }),
).await.map_err(|e| {
tracing::warn!("[Knowledge] Failed to dispatch embedding for item {}: {}", item.id, e);
e
});
created.push(item.id);
}
Err(e) => {
tracing::warn!("Import item '{}' failed: {}", item_req.title, e);
}
}
}
Ok(Json(serde_json::json!({
"created_count": created.len(),
"ids": created,
})))
}
// === 辅助函数 ===

4
crates/zclaw-saas/src/knowledge/mod.rs
View File

@@ -4,7 +4,7 @@ pub mod types;
pub mod service;
pub mod handlers;
use axum::routing::{delete, get, post, put};
use axum::routing::{delete, get, patch, post, put};
pub fn routes() -> axum::Router<crate::state::AppState> {
axum::Router::new()
@@ -14,10 +14,12 @@ pub fn routes() -> axum::Router<crate::state::AppState> {
.route("/api/v1/knowledge/categories/{id}", put(handlers::update_category))
.route("/api/v1/knowledge/categories/{id}", delete(handlers::delete_category))
.route("/api/v1/knowledge/categories/{id}/items", get(handlers::list_category_items))
.route("/api/v1/knowledge/categories/reorder", patch(handlers::reorder_categories))
// 知识条目 CRUD
.route("/api/v1/knowledge/items", get(handlers::list_items))
.route("/api/v1/knowledge/items", post(handlers::create_item))
.route("/api/v1/knowledge/items/batch", post(handlers::batch_create_items))
.route("/api/v1/knowledge/items/import", post(handlers::import_items))
.route("/api/v1/knowledge/items/{id}", get(handlers::get_item))
.route("/api/v1/knowledge/items/{id}", put(handlers::update_item))
.route("/api/v1/knowledge/items/{id}", delete(handlers::delete_item))

378
crates/zclaw-saas/src/knowledge/service.rs
View File

@@ -81,6 +81,21 @@ pub async fn create_category(
parent_id: Option<&str>,
icon: Option<&str>,
) -> SaasResult<KnowledgeCategory> {
// 验证 parent_id 存在性
if let Some(pid) = parent_id {
let exists: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM knowledge_categories WHERE id = $1)"
)
.bind(pid)
.fetch_one(pool)
.await?;
if !exists {
return Err(crate::error::SaasError::InvalidInput(
format!("父分类 '{}' 不存在", pid),
));
}
}
let id = uuid::Uuid::new_v4().to_string();
let category = sqlx::query_as::<_, KnowledgeCategory>(
"INSERT INTO knowledge_categories (id, name, description, parent_id, icon) \
@@ -126,15 +141,136 @@ pub async fn delete_category(pool: &PgPool, category_id: &str) -> SaasResult<()>
));
}
sqlx::query("DELETE FROM knowledge_categories WHERE id = $1")
let result = sqlx::query("DELETE FROM knowledge_categories WHERE id = $1")
.bind(category_id)
.execute(pool)
.await?;
if result.rows_affected() == 0 {
return Err(crate::error::SaasError::NotFound("分类不存在".into()));
}
Ok(())
}
/// 更新分类（含循环引用检测 + 深度限制）
pub async fn update_category(
pool: &PgPool,
category_id: &str,
name: Option<&str>,
description: Option<&str>,
parent_id: Option<&str>,
icon: Option<&str>,
) -> SaasResult<KnowledgeCategory> {
if let Some(pid) = parent_id {
if pid == category_id {
return Err(crate::error::SaasError::InvalidInput(
"分类不能成为自身的子分类".into(),
));
}
// 检查新的父级不是当前分类的后代（循环检测）
let mut check_id = pid.to_string();
let mut depth = 0;
loop {
if check_id == category_id {
return Err(crate::error::SaasError::InvalidInput(
"循环引用：父级分类不能是当前分类的后代".into(),
));
}
let parent: Option<(Option<String>,)> = sqlx::query_as(
"SELECT parent_id FROM knowledge_categories WHERE id = $1"
)
.bind(&check_id)
.fetch_optional(pool)
.await?;
match parent {
Some((Some(gp),)) => {
check_id = gp;
depth += 1;
if depth > 10 { break; }
}
_ => break,
}
}
// 检查深度限制（最多 3 层）
let mut current_depth = 0;
let mut check = pid.to_string();
while let Some((Some(p),)) = sqlx::query_as::<_, (Option<String>,)>(
"SELECT parent_id FROM knowledge_categories WHERE id = $1"
)
.bind(&check)
.fetch_optional(pool)
.await?
{
check = p;
current_depth += 1;
if current_depth > 10 { break; }
}
if current_depth >= 3 {
return Err(crate::error::SaasError::InvalidInput(
"分类层级不能超过 3 层".into(),
));
}
}
let category = sqlx::query_as::<_, KnowledgeCategory>(
"UPDATE knowledge_categories SET \
name = COALESCE($1, name), \
description = COALESCE($2, description), \
parent_id = COALESCE($3, parent_id), \
icon = COALESCE($4, icon), \
updated_at = NOW() \
WHERE id = $5 RETURNING *"
)
.bind(name)
.bind(description)
.bind(parent_id)
.bind(icon)
.bind(category_id)
.fetch_optional(pool)
.await?
.ok_or_else(|| crate::error::SaasError::NotFound("分类不存在".into()))?;
Ok(category)
}
// === 知识条目 CRUD ===
/// 按分类分页查询条目列表
pub async fn list_items_by_category(
pool: &PgPool,
category_id: &str,
status_filter: &str,
page: i64,
page_size: i64,
) -> SaasResult<(Vec<KnowledgeItem>, i64)> {
let offset = (page - 1) * page_size;
let items: Vec<KnowledgeItem> = sqlx::query_as(
"SELECT * FROM knowledge_items \
WHERE category_id = $1 AND status = $2 \
ORDER BY priority DESC, updated_at DESC \
LIMIT $3 OFFSET $4"
)
.bind(category_id)
.bind(status_filter)
.bind(page_size)
.bind(offset)
.fetch_all(pool)
.await?;
let total: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM knowledge_items WHERE category_id = $1 AND status = $2"
)
.bind(category_id)
.bind(status_filter)
.fetch_one(pool)
.await?;
Ok((items, total.0))
}
/// 创建知识条目
pub async fn create_item(
pool: &PgPool,
@@ -147,6 +283,19 @@ pub async fn create_item(
let priority = req.priority.unwrap_or(0);
let tags = req.tags.as_deref().unwrap_or(&[]);
// 验证 category_id 存在性
let cat_exists: bool = sqlx::query_scalar(
"SELECT EXISTS(SELECT 1 FROM knowledge_categories WHERE id = $1)"
)
.bind(&req.category_id)
.fetch_one(pool)
.await?;
if !cat_exists {
return Err(crate::error::SaasError::InvalidInput(
format!("分类 '{}' 不存在", req.category_id),
));
}
let item = sqlx::query_as::<_, KnowledgeItem>(
"INSERT INTO knowledge_items \
(id, category_id, title, content, keywords, related_questions, priority, tags, created_by) \
@@ -196,19 +345,31 @@ pub async fn get_item(pool: &PgPool, item_id: &str) -> SaasResult<Option<Knowled
Ok(item)
}
/// 更新条目（含版本快照）
/// 更新条目（含版本快照）— 事务保护防止并发竞态
pub async fn update_item(
pool: &PgPool,
item_id: &str,
account_id: &str,
req: &UpdateItemRequest,
) -> SaasResult<KnowledgeItem> {
// 获取当前条目
// status 验证在事务之前，避免无谓锁占用
const VALID_STATUSES: &[&str] = &["active", "draft", "archived", "deprecated"];
if let Some(ref status) = &req.status {
if !VALID_STATUSES.contains(&status.as_str()) {
return Err(crate::error::SaasError::InvalidInput(
format!("无效的状态值: {}，有效值: {}", status, VALID_STATUSES.join(", "))
));
}
}
let mut tx = pool.begin().await?;
// 获取当前条目并锁定行防止并发修改
let current = sqlx::query_as::<_, KnowledgeItem>(
"SELECT * FROM knowledge_items WHERE id = $1"
"SELECT * FROM knowledge_items WHERE id = $1 FOR UPDATE"
)
.bind(item_id)
.fetch_optional(pool)
.fetch_optional(&mut *tx)
.await?
.ok_or_else(|| crate::error::SaasError::NotFound("知识条目不存在".into()))?;
@@ -229,6 +390,7 @@ pub async fn update_item(
.unwrap_or(&vec![])
.clone();
// 更新条目
let updated = sqlx::query_as::<_, KnowledgeItem>(
"UPDATE knowledge_items SET \
@@ -245,7 +407,7 @@ pub async fn update_item(
.bind(&tags)
.bind(req.status.as_deref())
.bind(item_id)
.fetch_one(pool)
.fetch_one(&mut *tx)
.await?;
// 创建版本快照
@@ -265,9 +427,10 @@ pub async fn update_item(
.bind(&related_questions)
.bind(req.change_summary.as_deref())
.bind(account_id)
.execute(pool)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(updated)
}
@@ -288,39 +451,44 @@ pub async fn delete_item(pool: &PgPool, item_id: &str) -> SaasResult<()> {
/// 将内容按 Markdown 标题 + 固定长度分块
pub fn chunk_content(content: &str, max_tokens: usize, overlap: usize) -> Vec<String> {
let mut chunks = Vec::new();
// 先按 Markdown 标题分段
let sections: Vec<&str> = content.split("\n# ").collect();
for section in sections {
// 简单估算 token（中文约 1.5 字符/token）
let estimated_tokens = section.len() / 2;
let mut chunks = Vec::new();
for (i, section) in sections.iter().enumerate() {
// 第一个片段保留原始内容，其余片段重新添加标题标记
let section_content = if i == 0 {
section.to_string()
} else {
format!("# {}", section)
};
// 磁盘估算 token（中文约 1.5 字符/token)
let estimated_tokens = section_content.len() / 2;
if estimated_tokens <= max_tokens {
if !section.trim().is_empty() {
chunks.push(section.trim().to_string());
if !section_content.trim().is_empty() {
chunks.push(section_content.trim().to_string());
}
} else {
// 超长段落按固定长度切分
let chars: Vec<char> = section.chars().collect();
let chars: Vec<char> = section_content.chars().collect();
let chunk_chars = max_tokens * 2; // 近似字符数
let overlap_chars = overlap * 2;
let mut pos = 0;
while pos < chars.len() {
let end = (pos + chunk_chars).min(chars.len());
let chunk: String = chars[pos..end].iter().collect();
if !chunk.trim().is_empty() {
chunks.push(chunk.trim().to_string());
let chunk_str: String = chars[pos..end].iter().collect();
if !chunk_str.trim().is_empty() {
chunks.push(chunk_str.trim().to_string());
}
pos = if end >= chars.len() { end } else { end.saturating_sub(overlap_chars) };
pos = if end >= chars.len() { end} else { end.saturating_sub(overlap_chars) };
}
}
}
chunks
}
chunks}
// === 搜索 ===
@@ -333,14 +501,14 @@ pub async fn search(
min_score: f64,
) -> SaasResult<Vec<SearchResult>> {
// 暂时使用关键词匹配（向量搜索需要 embedding 生成）
let pattern = format!("%{}%", query.replace('%', "\\%").replace('_', "\\_"));
let pattern = format!("%{}%", query.replace('\\', "\\\\").replace('%', "\\%").replace('_', "\\_"));
let results = if let Some(cat_id) = category_id {
sqlx::query_as::<_, (String, String, String, String, String, Vec<String>)>(
"SELECT kc.id, kc.item_id, ki.title, kc.name as cat_name, kc.content, kc.keywords \
"SELECT kc.id, kc.item_id, ki.title, kcat.name, kc.content, kc.keywords \
FROM knowledge_chunks kc \
JOIN knowledge_items ki ON kc.item_id = ki.id \
JOIN knowledge_categories kc2 ON ki.category_id = kc2.id \
JOIN knowledge_categories kcat ON ki.category_id = kcat.id \
WHERE ki.status = 'active' \
AND ki.category_id = $1 \
AND (kc.content ILIKE $2 OR $3 = ANY(kc.keywords)) \
@@ -355,10 +523,10 @@ pub async fn search(
.await?
} else {
sqlx::query_as::<_, (String, String, String, String, String, Vec<String>)>(
"SELECT kc.id, kc.item_id, ki.title, kc2.name as cat_name, kc.content, kc.keywords \
"SELECT kc.id, kc.item_id, ki.title, kcat.name, kc.content, kc.keywords \
FROM knowledge_chunks kc \
JOIN knowledge_items ki ON kc.item_id = ki.id \
JOIN knowledge_categories kc2 ON ki.category_id = kc2.id \
JOIN knowledge_categories kcat ON ki.category_id = kcat.id \
WHERE ki.status = 'active' \
AND (kc.content ILIKE $1 OR $2 = ANY(kc.keywords)) \
ORDER BY ki.priority DESC \
@@ -372,13 +540,24 @@ pub async fn search(
};
Ok(results.into_iter().map(|(chunk_id, item_id, title, cat_name, content, keywords)| {
// 基于关键词匹配数计算分数：匹配数 / 总查询关键词数
let query_keywords: Vec<&str> = query.split_whitespace().collect();
let matched_count = keywords.iter()
.filter(|k| query_keywords.iter().any(|qk| k.to_lowercase().contains(&qk.to_lowercase())))
.count();
let score = if keywords.is_empty() || query_keywords.is_empty() {
0.5
} else {
(matched_count as f64 / keywords.len().max(query_keywords.len()) as f64).min(1.0)
};
SearchResult {
chunk_id,
item_id,
item_title: title,
category_name: cat_name,
content,
score: 0.8, // 关键词匹配默认分数
score,
keywords,
}
}).filter(|r| r.score >= min_score).collect())
@@ -430,6 +609,12 @@ pub async fn analytics_overview(pool: &PgPool) -> SaasResult<AnalyticsOverview>
.fetch_one(pool)
.await?;
let with_feedback: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM knowledge_usage WHERE agent_feedback IS NOT NULL"
)
.fetch_one(pool)
.await?;
let stale: (i64,) = sqlx::query_as(
"SELECT COUNT(*) FROM knowledge_items ki \
WHERE ki.status = 'active' \
@@ -438,7 +623,7 @@ pub async fn analytics_overview(pool: &PgPool) -> SaasResult<AnalyticsOverview>
.fetch_one(pool)
.await?;
let hit_rate = if total_refs.0 > 0 { 1.0 } else { 0.0 };
let hit_rate = if total_refs.0 > 0 { with_feedback.0 as f64 / total_refs.0 as f64 } else { 0.0 };
let injection_rate = if total_refs.0 > 0 { injected.0 as f64 / total_refs.0 as f64 } else { 0.0 };
let positive_rate = if total_refs.0 > 0 { positive.0 as f64 / total_refs.0 as f64 } else { 0.0 };
@@ -455,3 +640,140 @@ pub async fn analytics_overview(pool: &PgPool) -> SaasResult<AnalyticsOverview>
stale_items_count: stale.0,
})
}
/// 回滚到指定版本（创建新版本快照）
pub async fn rollback_version(
pool: &PgPool,
item_id: &str,
target_version: i32,
account_id: &str,
) -> SaasResult<KnowledgeItem> {
// 使用事务保证原子性，防止并发回滚冲突
let mut tx = pool.begin().await?;
// 获取目标版本
let version: KnowledgeVersion = sqlx::query_as(
"SELECT * FROM knowledge_versions WHERE item_id = $1 AND version = $2"
)
.bind(item_id)
.bind(target_version)
.fetch_optional(&mut *tx)
.await?
.ok_or_else(|| crate::error::SaasError::NotFound("版本不存在".into()))?;
// 锁定当前条目行防止并发修改（SELECT FOR UPDATE）
let current: Option<(i32,)> = sqlx::query_as(
"SELECT version FROM knowledge_items WHERE id = $1 FOR UPDATE"
)
.bind(item_id)
.fetch_optional(&mut *tx)
.await?;
let current_version = current
.ok_or_else(|| crate::error::SaasError::NotFound("知识条目不存在".into()))?
.0;
// 防止版本无限递增： 最多 100 个版本
if current_version >= 100 {
return Err(crate::error::SaasError::InvalidInput(
"版本数已达上限(100)，请考虑合并历史版本".into(),
));
}
let new_version = current_version + 1;
// 更新条目为该版本内容
let updated = sqlx::query_as::<_, KnowledgeItem>(
"UPDATE knowledge_items SET \
title = $1, content = $2, keywords = $3, related_questions = $4, \
version = $5, updated_at = NOW() \
WHERE id = $6 RETURNING *"
)
.bind(&version.title)
.bind(&version.content)
.bind(&version.keywords)
.bind(&version.related_questions)
.bind(new_version)
.bind(item_id)
.fetch_one(&mut *tx)
.await?;
// 创建新版本快照（记录回滚来源）
let version_id = uuid::Uuid::new_v4().to_string();
let summary = format!("回滚到版本 {}（当前版本 {} → 新版本 {}）", target_version, current_version, new_version);
sqlx::query(
"INSERT INTO knowledge_versions \
(id, item_id, version, title, content, keywords, related_questions, \
change_summary, created_by) \
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)"
)
.bind(&version_id)
.bind(item_id)
.bind(new_version)
.bind(&updated.title)
.bind(&updated.content)
.bind(&updated.keywords)
.bind(&updated.related_questions)
.bind(&summary)
.bind(account_id)
.execute(&mut *tx)
.await?;
tx.commit().await?;
Ok(updated)
}
/// 质量指标（按分类分组）
pub async fn analytics_quality(pool: &PgPool) -> SaasResult<serde_json::Value> {
let quality: Vec<(serde_json::Value,)> = sqlx::query_as(
"SELECT json_build_object(
'category', kc.name,
'total', COUNT(ki.id),
'active', COUNT(CASE WHEN ki.status = 'active' THEN 1 END),
'with_keywords', COUNT(CASE WHEN array_length(ki.keywords, 1) > 0 THEN 1 END),
'avg_priority', COALESCE(AVG(ki.priority), 0)
) as row \
FROM knowledge_categories kc \
LEFT JOIN knowledge_items ki ON ki.category_id = kc.id \
GROUP BY kc.id, kc.name \
ORDER BY COUNT(ki.id) DESC"
)
.fetch_all(pool)
.await
.unwrap_or_else(|e| {
tracing::warn!("analytics_quality query failed: {}", e);
vec![]
});
Ok(serde_json::json!({
"categories": quality.into_iter().map(|(v,)| v).collect::<Vec<_>>()
}))
}
/// 知识缺口检测（低分查询聚类）
pub async fn analytics_gaps(pool: &PgPool) -> SaasResult<serde_json::Value> {
let gaps: Vec<(serde_json::Value,)> = sqlx::query_as(
"SELECT json_build_object(
'query', ku.query_text,
'count', COUNT(*),
'avg_score', COALESCE(AVG(ku.relevance_score), 0)
) as row \
FROM knowledge_usage ku \
WHERE ku.created_at >= NOW() - interval '30 days' \
AND (ku.relevance_score IS NULL OR ku.relevance_score < 0.5) \
AND ku.query_text IS NOT NULL \
GROUP BY ku.query_text \
ORDER BY COUNT(*) DESC \
LIMIT 20"
)
.fetch_all(pool)
.await
.unwrap_or_else(|e| {
tracing::warn!("analytics_gaps query failed: {}", e);
vec![]
});
Ok(serde_json::json!({
"gaps": gaps.into_iter().map(|(v,)| v).collect::<Vec<_>>()
}))
}

22
crates/zclaw-saas/src/knowledge/types.rs
View File

@@ -199,3 +199,25 @@ pub struct AnalyticsOverview {
pub positive_feedback_rate: f64,
pub stale_items_count: i64,
}
// === 批量操作 ===
#[derive(Debug, Deserialize)]
pub struct ReorderItem {
pub id: String,
pub sort_order: i32,
}
#[derive(Debug, Deserialize)]
pub struct ImportFile {
pub content: String,
pub title: Option<String>,
pub keywords: Option<Vec<String>>,
pub tags: Option<Vec<String>>,
}
#[derive(Debug, Deserialize)]
pub struct ImportRequest {
pub category_id: String,
pub files: Vec<ImportFile>,
}