feat(knowledge): add batch synchronization

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,

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)

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"