From f7076569a7cc4506fa6f01dab7d7ec0209207cbf Mon Sep 17 00:00:00 2001 From: Nelson <1475262689@qq.com> Date: Fri, 17 Jul 2026 13:20:41 +0800 Subject: [PATCH] feat(knowledge): add batch synchronization --- .../components/KnowledgeManagementView.vue | 67 +++++++++++++++- .../apps/admin-web/src/services/api.ts | 6 ++ .../apps/admin-web/src/types/api.ts | 14 ++++ .../app/api/admin_knowledge_lifecycle.py | 76 ++++++++++++++++++- .../apps/backend/app/schemas/knowledge.py | 4 + .../backend/tests/test_knowledge_admin.py | 57 ++++++++++++-- 6 files changed, 217 insertions(+), 7 deletions(-) diff --git a/ai_knowledge_base_v2/apps/admin-web/src/components/KnowledgeManagementView.vue b/ai_knowledge_base_v2/apps/admin-web/src/components/KnowledgeManagementView.vue index aabc192..31cd95e 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/components/KnowledgeManagementView.vue +++ b/ai_knowledge_base_v2/apps/admin-web/src/components/KnowledgeManagementView.vue @@ -8,6 +8,7 @@ import TableRowActions from "./TableRowActions.vue"; import AdminPagination from "./AdminPagination.vue"; import type { KnowledgeDetail, + KnowledgeBatchSyncResult, KnowledgeItem, KnowledgeSyncJob, MigrationSummary, @@ -29,6 +30,9 @@ const editOpen = ref(false); const editSaving = ref(false); const batchOpen = ref(false); const batchSaving = ref(false); +const batchSyncing = ref(false); +const batchSyncResultOpen = ref(false); +const batchSyncResult = ref(null); const resolvingNode = ref(false); const nodeResolved = ref(false); const sourceTitle = ref(""); @@ -130,6 +134,32 @@ async function saveBatchMetadata() { } } +async function syncSelected() { + if (!selectedRows.value.length) { + ElMessage.warning("请先勾选需要同步的知识库"); + return; + } + if (!await confirmAction( + `将依次同步已选择的 ${selectedRows.value.length} 个知识库。同步成功后最新内容立即生效，单个失败不会中断其他知识库。`, + "确认批量同步", + { confirmButtonText: "开始同步", cancelButtonText: "取消", type: "warning" }, + )) return; + batchSyncing.value = true; + try { + const result = await api.batchSyncKnowledge(selectedRows.value.map((item) => item.id)); + batchSyncResult.value = result; + batchSyncResultOpen.value = true; + ElMessage[result.failed ? "warning" : "success"]( + `批量同步完成：成功 ${result.success} 个，失败 ${result.failed} 个`, + ); + await load(); + } catch (error) { + ElMessage.error(errorMessage(error, "批量同步失败")); + } finally { + batchSyncing.value = false; + } +} + async function handleMore(command: string, item: KnowledgeItem) { try { if (command === "detail") return await openDetail(item); @@ -353,7 +383,10 @@ async function confirmAction(
已选择 {{ selectedRows.length }} 个知识库 - 批量修改类型 +
+ 批量同步 + 批量修改类型 +
@@ -404,6 +437,28 @@ async function confirmAction( + + + + + + + + + + + + + + + + diff --git a/ai_knowledge_base_v2/apps/admin-web/src/services/api.ts b/ai_knowledge_base_v2/apps/admin-web/src/services/api.ts index 7b97ef4..79c7e13 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/services/api.ts +++ b/ai_knowledge_base_v2/apps/admin-web/src/services/api.ts @@ -15,6 +15,7 @@ import type { KnowledgeSyncJob, KnowledgeVersionContent, MigrationSummary, + KnowledgeBatchSyncResult, KnowledgeBatchUpdateResult, RetrievalLogItem, AttentionItem, @@ -142,6 +143,11 @@ export const api = { request(`/admin/knowledge/${id}/metadata`, { method: "PATCH", body: JSON.stringify(payload) }), batchUpdateKnowledgeMetadata: (payload: { knowledgeIds: number[]; knowledgeType?: string }) => request("/admin/knowledge/batch/metadata", { method: "PATCH", body: JSON.stringify(payload) }), + batchSyncKnowledge: (knowledgeIds: number[]) => + request("/admin/knowledge/batch/sync", { + method: "POST", + body: JSON.stringify({ knowledgeIds }), + }), deleteKnowledge: (id: number) => request(`/admin/knowledge/${id}`, { method: "DELETE" }), importKnowledgeFromSpace: (body: { node_token: string }) => request<{ imported: number; skipped: number; id: number; name: string }>( diff --git a/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts b/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts index b9d0997..1c76280 100644 --- a/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts +++ b/ai_knowledge_base_v2/apps/admin-web/src/types/api.ts @@ -222,6 +222,20 @@ export interface KnowledgeBatchUpdateResult { results: Array<{ knowledgeId: number; ok: boolean; error?: string }>; } +export interface KnowledgeBatchSyncResult { + total: number; + success: number; + failed: number; + results: Array<{ + knowledgeId: number; + name?: string; + ok: boolean; + status?: string; + job?: KnowledgeSyncJob; + error?: string; + }>; +} + export interface RetrievalLogItem { id: number; sessionId?: number | null; diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py index 24775f5..9f96ea0 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py +++ b/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py @@ -20,8 +20,9 @@ from app.models.knowledge import ( KnowledgeVersion, ) from app.schemas.knowledge import ( - KnowledgeLifecycleRequest, KnowledgeBatchMetadataUpdateRequest, + KnowledgeBatchSyncRequest, + KnowledgeLifecycleRequest, KnowledgeMetadataUpdateRequest, KnowledgeStatusRequest, ) @@ -180,6 +181,79 @@ def sync_jobs( return api_success([_job_dict(item) for item in rows]) +@router.post("/knowledge/batch/sync") +async def batch_sync( + payload: KnowledgeBatchSyncRequest, + db: Session = Depends(get_db), + current_admin: Admin = Depends(get_current_admin), +) -> dict: + _require_super_admin(current_admin) + knowledge_ids = list(dict.fromkeys(payload.knowledgeIds)) + rows = { + item.id: item + for item in db.scalars(select(Knowledge).where(Knowledge.id.in_(knowledge_ids))).all() + } + results: list[dict] = [] + for knowledge_id in knowledge_ids: + knowledge = rows.get(knowledge_id) + if knowledge is None: + results.append({"knowledgeId": knowledge_id, "ok": False, "error": "知识库不存在"}) + continue + if knowledge.lifecycle_status == "archived": + results.append( + { + "knowledgeId": knowledge_id, + "name": knowledge.name, + "ok": False, + "error": "归档知识库不能同步", + } + ) + continue + try: + job = await KnowledgePipelineService.synchronize( + db, knowledge, admin_id=current_admin.id + ) + OperationLogService.write( + db, + admin_id=current_admin.id, + module="knowledge", + action="batch_sync", + target_id=knowledge.id, + ) + db.commit() + results.append( + { + "knowledgeId": knowledge.id, + "name": knowledge.name, + "ok": True, + "status": job.status, + "job": _job_dict(job), + } + ) + except Exception as exc: + OperationLogService.write( + db, + admin_id=current_admin.id, + module="knowledge", + action="batch_sync", + target_id=knowledge.id, + result="FAILED", + ) + db.commit() + results.append( + { + "knowledgeId": knowledge.id, + "name": knowledge.name, + "ok": False, + "error": str(exc), + } + ) + success = sum(1 for item in results if item["ok"]) + return api_success( + {"total": len(results), "success": success, "failed": len(results) - success, "results": results} + ) + + @router.post("/knowledge/{knowledge_id}/sync") async def synchronize( knowledge_id: int, diff --git a/ai_knowledge_base_v2/apps/backend/app/schemas/knowledge.py b/ai_knowledge_base_v2/apps/backend/app/schemas/knowledge.py index 6b53e54..8c86d92 100644 --- a/ai_knowledge_base_v2/apps/backend/app/schemas/knowledge.py +++ b/ai_knowledge_base_v2/apps/backend/app/schemas/knowledge.py @@ -21,6 +21,10 @@ class KnowledgeBatchMetadataUpdateRequest(BaseModel): knowledgeType: str | None = Field(default=None, pattern="^(course|qa|general)$") +class KnowledgeBatchSyncRequest(BaseModel): + knowledgeIds: list[int] = Field(min_length=1, max_length=500) + + class AttentionUpdateRequest(BaseModel): status: str = Field(pattern="^(pending|processing|resolved|ignored)$") note: str | None = Field(default=None, max_length=2000) diff --git a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py index f23e28b..ed35262 100644 --- a/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py +++ b/ai_knowledge_base_v2/apps/backend/tests/test_knowledge_admin.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from datetime import datetime from unittest.mock import AsyncMock, patch import pytest @@ -9,7 +10,7 @@ from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool -from app.api.admin_knowledge_lifecycle import batch_update_metadata, update_metadata +from app.api.admin_knowledge_lifecycle import batch_sync, batch_update_metadata, update_metadata from app.models import Base from app.models.admin import Admin, Role from app.models.knowledge import ( @@ -18,9 +19,14 @@ from app.models.knowledge import ( KnowledgeManifest, KnowledgePublishLog, KnowledgeSourceSnapshot, + KnowledgeSyncJob, KnowledgeVersion, ) -from app.schemas.knowledge import KnowledgeBatchMetadataUpdateRequest, KnowledgeMetadataUpdateRequest +from app.schemas.knowledge import ( + KnowledgeBatchMetadataUpdateRequest, + KnowledgeBatchSyncRequest, + KnowledgeMetadataUpdateRequest, +) from app.services.knowledge_agent_service import KnowledgeAgentService from app.services.knowledge_pipeline_service import KnowledgePipelineService @@ -40,8 +46,8 @@ def _knowledge(db: Session, knowledge_id: int = 1) -> Knowledge: item = Knowledge( id=knowledge_id, name="原名称", - feishu_space_id="space-1", - feishu_node_id="node-1", + feishu_space_id=f"space-{knowledge_id}", + feishu_node_id=f"node-{knowledge_id}", status=1, knowledge_type="general", review_mode="manual", @@ -55,7 +61,7 @@ def _knowledge(db: Session, knowledge_id: int = 1) -> Knowledge: source_title="源标题", source_content="源内容", content_hash=f"hash-{knowledge_id}", - source_identifier="space-1/node-1", + source_identifier=f"space-{knowledge_id}/node-{knowledge_id}", ) db.add(snapshot) db.flush() @@ -128,6 +134,47 @@ def test_batch_metadata_update_reports_missing_items(): assert db.scalar(select(Knowledge).where(Knowledge.id == item.id)).knowledge_type == "qa" +def test_batch_sync_continues_after_an_item_fails_and_reports_each_result(): + with _database() as db: + first = _knowledge(db, 1) + second = _knowledge(db, 2) + successful_job = KnowledgeSyncJob( + id=101, + knowledge_id=first.id, + mode="replace", + status="published", + stage="completed", + progress=100, + requested_by=1, + created_at=datetime.now(), + ) + + async def synchronize(_db, knowledge, *, admin_id): + assert admin_id == 1 + if knowledge.id == second.id: + raise RuntimeError("飞书接口暂时不可用") + return successful_job + + with patch.object(KnowledgePipelineService, "synchronize", side_effect=synchronize) as mocked: + response = asyncio.run( + batch_sync( + KnowledgeBatchSyncRequest(knowledgeIds=[first.id, first.id, second.id, 999]), + db, + _admin(), + ) + ) + + result = response["data"] + assert result["total"] == 3 + assert result["success"] == 1 + assert result["failed"] == 2 + assert [item["knowledgeId"] for item in result["results"]] == [first.id, second.id, 999] + assert result["results"][0]["status"] == "published" + assert result["results"][1]["error"] == "飞书接口暂时不可用" + assert result["results"][2]["error"] == "知识库不存在" + assert mocked.call_count == 2 + + def test_sync_automatically_replaces_current_version_and_removes_history(): source_content = ( "# 一、课程说明\n这是更新后的课程内容，同步成功后应当立即生效。\n\n"