feat: 实现知识同步审核发布生命周期

ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py

@@ -50,6 +50,8 @@ def import_knowledge_from_space(
         feishu_space_id=space_id,
         feishu_node_id=node_token,
         status=1,
+        knowledge_type="general",
+        review_mode="manual",
         remark=f"obj_type={info.get('obj_type', '')}",
     )
     db.add(knowledge)
@@ -74,6 +76,8 @@ def create_knowledge(
         feishu_node_id=payload.feishuNodeId,
         status=payload.status,
         remark=payload.remark,
+        knowledge_type=payload.knowledgeType,
+        review_mode=payload.reviewMode,
     )
     db.add(knowledge)
     db.flush()
@@ -96,6 +100,8 @@ def update_knowledge(
     knowledge.feishu_node_id = payload.feishuNodeId
     knowledge.status = payload.status
     knowledge.remark = payload.remark
+    knowledge.knowledge_type = payload.knowledgeType
+    knowledge.review_mode = payload.reviewMode
     db.add(knowledge)
     OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="update", target_id=knowledge.id)
     db.commit()
@@ -137,6 +143,18 @@ def _knowledge_dict(knowledge: Knowledge) -> dict:
         "feishuNodeId": knowledge.feishu_node_id,
         "status": knowledge.status,
         "remark": knowledge.remark,
+        "sourceTitle": knowledge.source_title,
+        "knowledgeType": knowledge.knowledge_type,
+        "reviewMode": knowledge.review_mode,
+        "lifecycleStatus": knowledge.lifecycle_status,
+        "sourceStatus": knowledge.source_status,
+        "sourceError": knowledge.source_error,
+        "manifestConfirmed": knowledge.manifest_confirmed == 1,
+        "currentVersionId": knowledge.current_version_id,
+        "pendingVersionId": knowledge.pending_version_id,
+        "lastSyncedAt": knowledge.last_synced_at,
+        "lastPublishedAt": knowledge.last_published_at,
+        "migrationStatus": knowledge.migration_status,
         "createdAt": knowledge.created_at,
         "updatedAt": knowledge.updated_at,
     }

ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge_lifecycle.py

@@ -0,0 +1,401 @@
+from __future__ import annotations
+
+import json
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.core.dependencies import get_current_admin
+from app.core.responses import api_success
+from app.models.admin import Admin
+from app.models.knowledge import (
+    Knowledge,
+    KnowledgeCard,
+    KnowledgeChunk,
+    KnowledgeManifest,
+    KnowledgeSection,
+    KnowledgeSyncJob,
+    KnowledgeVersion,
+)
+from app.schemas.knowledge import (
+    KnowledgeLifecycleRequest,
+    KnowledgePublishRequest,
+    KnowledgeReviewRequest,
+    KnowledgeStatusRequest,
+    KnowledgeSyncRequest,
+)
+from app.services.admin_service import OperationLogService
+from app.services.knowledge_pipeline_service import KnowledgePipelineService
+
+router = APIRouter()
+
+
+@router.get("/knowledge/{knowledge_id}/detail")
+def detail(
+    knowledge_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    knowledge = _knowledge(db, knowledge_id)
+    current = db.get(KnowledgeVersion, knowledge.current_version_id) if knowledge.current_version_id else None
+    pending = db.get(KnowledgeVersion, knowledge.pending_version_id) if knowledge.pending_version_id else None
+    manifest = None
+    manifest_version_id = knowledge.pending_version_id or knowledge.current_version_id
+    if manifest_version_id:
+        manifest = db.scalar(select(KnowledgeManifest).where(KnowledgeManifest.version_id == manifest_version_id))
+    return api_success(
+        {
+            "knowledge": _knowledge_detail(knowledge),
+            "currentVersion": _version_dict(current),
+            "pendingVersion": _version_dict(pending),
+            "manifest": _manifest_dict(manifest),
+        }
+    )
+
+
+@router.get("/knowledge/{knowledge_id}/versions")
+def versions(
+    knowledge_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    _knowledge(db, knowledge_id)
+    rows = db.scalars(
+        select(KnowledgeVersion)
+        .where(KnowledgeVersion.knowledge_id == knowledge_id)
+        .order_by(KnowledgeVersion.version_no.desc())
+    ).all()
+    return api_success([_version_dict(item) for item in rows])
+
+
+@router.get("/knowledge/version/{version_id}/content")
+def version_content(
+    version_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    version = _version(db, version_id)
+    sections = db.scalars(
+        select(KnowledgeSection).where(KnowledgeSection.version_id == version.id).order_by(KnowledgeSection.sort_order)
+    ).all()
+    cards = db.scalars(
+        select(KnowledgeCard).where(KnowledgeCard.version_id == version.id).order_by(KnowledgeCard.id)
+    ).all()
+    manifest = db.scalar(select(KnowledgeManifest).where(KnowledgeManifest.version_id == version.id))
+    return api_success(
+        {
+            "version": _version_dict(version),
+            "manifest": _manifest_dict(manifest),
+            "sections": [
+                {
+                    "id": item.id,
+                    "sectionKey": item.section_key,
+                    "title": item.title,
+                    "content": item.content,
+                    "sourceStart": item.source_start,
+                    "sourceEnd": item.source_end,
+                    "contentHash": item.content_hash,
+                }
+                for item in sections
+            ],
+            "cards": [
+                {
+                    "id": item.id,
+                    "sectionId": item.section_id,
+                    "title": item.title,
+                    "summary": item.summary,
+                    "coreConclusion": item.core_conclusion,
+                    "applicableQuestions": item.applicable_questions,
+                    "inapplicableQuestions": item.inapplicable_questions,
+                    "keywords": _json_value(item.keywords),
+                    "synonyms": _json_value(item.synonyms),
+                    "riskLevel": item.risk_level,
+                    "sourceRange": item.source_range,
+                    "reviewStatus": item.review_status,
+                }
+                for item in cards
+            ],
+        }
+    )
+
+
+@router.get("/knowledge/{knowledge_id}/sync-jobs")
+def sync_jobs(
+    knowledge_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    rows = db.scalars(
+        select(KnowledgeSyncJob)
+        .where(KnowledgeSyncJob.knowledge_id == knowledge_id)
+        .order_by(KnowledgeSyncJob.id.desc())
+        .limit(50)
+    ).all()
+    return api_success([_job_dict(item) for item in rows])
+
+
+@router.post("/knowledge/{knowledge_id}/sync")
+async def synchronize(
+    knowledge_id: int,
+    payload: KnowledgeSyncRequest,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    knowledge = _knowledge(db, knowledge_id)
+    if knowledge.lifecycle_status == "archived":
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="归档知识库不能同步")
+    try:
+        job = await KnowledgePipelineService.synchronize(
+            db, knowledge, admin_id=current_admin.id, mode=payload.mode
+        )
+    except Exception as exc:
+        raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"知识同步失败：{exc}") from exc
+    OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="sync", target_id=knowledge.id)
+    db.commit()
+    return api_success(_job_dict(job))
+
+
+@router.post("/knowledge/{knowledge_id}/manifest/{version_id}/confirm")
+def confirm_manifest(
+    knowledge_id: int,
+    version_id: int,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    knowledge, version = _knowledge_version(db, knowledge_id, version_id)
+    try:
+        KnowledgePipelineService.confirm_manifest(db, knowledge, version, current_admin.id)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+    OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="confirm_manifest", target_id=knowledge.id)
+    db.commit()
+    return api_success()
+
+
+@router.post("/knowledge/{knowledge_id}/version/{version_id}/review")
+def review(
+    knowledge_id: int,
+    version_id: int,
+    payload: KnowledgeReviewRequest,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    knowledge, version = _knowledge_version(db, knowledge_id, version_id)
+    if version.quality_status == "blocked" and payload.action == "approve":
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="自动质检存在阻断项，不能审核通过")
+    KnowledgePipelineService.review_version(db, knowledge, version, current_admin.id, payload.action, payload.reason)
+    OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action=f"review_{payload.action}", target_id=knowledge.id)
+    db.commit()
+    return api_success()
+
+
+@router.post("/knowledge/{knowledge_id}/version/{version_id}/publish")
+def publish(
+    knowledge_id: int,
+    version_id: int,
+    payload: KnowledgePublishRequest,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    knowledge, version = _knowledge_version(db, knowledge_id, version_id)
+    try:
+        KnowledgePipelineService.publish_version(db, knowledge, version, current_admin.id, payload.keepOpen)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+    OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="publish", target_id=knowledge.id)
+    db.commit()
+    return api_success(_knowledge_detail(knowledge))
+
+
+@router.post("/knowledge/{knowledge_id}/version/{version_id}/rollback")
+def rollback(
+    knowledge_id: int,
+    version_id: int,
+    payload: KnowledgePublishRequest,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    knowledge, version = _knowledge_version(db, knowledge_id, version_id)
+    try:
+        KnowledgePipelineService.rollback(db, knowledge, version, current_admin.id, payload.keepOpen)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+    OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="rollback", target_id=knowledge.id)
+    db.commit()
+    return api_success(_knowledge_detail(knowledge))
+
+
+@router.put("/knowledge/{knowledge_id}/open-status")
+def open_status(
+    knowledge_id: int,
+    payload: KnowledgeStatusRequest,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    knowledge = _knowledge(db, knowledge_id)
+    if payload.open:
+        if not knowledge.current_version_id:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="知识库没有正式发布版本")
+        if knowledge.manifest_confirmed != 1:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="首次知识库说明卡尚未确认")
+        if knowledge.source_status != "normal":
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="飞书源文章异常，不能开放")
+        if knowledge.lifecycle_status != "active":
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="归档知识库不能开放")
+    knowledge.status = 1 if payload.open else 0
+    db.add(knowledge)
+    OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action="open" if payload.open else "close", target_id=knowledge.id)
+    db.commit()
+    return api_success(_knowledge_detail(knowledge))
+
+
+@router.put("/knowledge/{knowledge_id}/lifecycle")
+def lifecycle(
+    knowledge_id: int,
+    payload: KnowledgeLifecycleRequest,
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    _require_super_admin(current_admin)
+    knowledge = _knowledge(db, knowledge_id)
+    knowledge.lifecycle_status = payload.status
+    if payload.status == "archived":
+        knowledge.status = 0
+    db.add(knowledge)
+    OperationLogService.write(db, admin_id=current_admin.id, module="knowledge", action=payload.status, target_id=knowledge.id)
+    db.commit()
+    return api_success(_knowledge_detail(knowledge))
+
+
+@router.get("/knowledge-migration/summary")
+def migration_summary(
+    db: Session = Depends(get_db), current_admin: Admin = Depends(get_current_admin)
+) -> dict:
+    _require_super_admin(current_admin)
+    rows = db.scalars(select(Knowledge).order_by(Knowledge.id)).all()
+    counts: dict[str, int] = {}
+    for item in rows:
+        counts[item.migration_status] = counts.get(item.migration_status, 0) + 1
+    return api_success({"total": len(rows), "counts": counts, "items": [_knowledge_detail(item) for item in rows]})
+
+
+def _require_super_admin(admin: Admin) -> None:
+    if admin.role is not None and admin.role.code != "super_admin":
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="仅超级管理员可以执行此操作")
+
+
+def _knowledge(db: Session, knowledge_id: int) -> Knowledge:
+    item = db.get(Knowledge, knowledge_id)
+    if item is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="知识库不存在")
+    return item
+
+
+def _version(db: Session, version_id: int) -> KnowledgeVersion:
+    item = db.get(KnowledgeVersion, version_id)
+    if item is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="知识版本不存在")
+    return item
+
+
+def _knowledge_version(db: Session, knowledge_id: int, version_id: int) -> tuple[Knowledge, KnowledgeVersion]:
+    knowledge = _knowledge(db, knowledge_id)
+    version = _version(db, version_id)
+    if version.knowledge_id != knowledge.id:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="版本不属于该知识库")
+    return knowledge, version
+
+
+def _knowledge_detail(item: Knowledge) -> dict:
+    return {
+        "id": item.id,
+        "name": item.name,
+        "sourceTitle": item.source_title,
+        "feishuSpaceId": item.feishu_space_id,
+        "feishuNodeId": item.feishu_node_id,
+        "knowledgeType": item.knowledge_type,
+        "reviewMode": item.review_mode,
+        "open": item.status == 1,
+        "lifecycleStatus": item.lifecycle_status,
+        "sourceStatus": item.source_status,
+        "sourceError": item.source_error,
+        "manifestConfirmed": item.manifest_confirmed == 1,
+        "currentVersionId": item.current_version_id,
+        "pendingVersionId": item.pending_version_id,
+        "lastSyncedAt": item.last_synced_at,
+        "lastPublishedAt": item.last_published_at,
+        "migrationStatus": item.migration_status,
+    }
+
+
+def _version_dict(item: KnowledgeVersion | None) -> dict | None:
+    if item is None:
+        return None
+    return {
+        "id": item.id,
+        "knowledgeId": item.knowledge_id,
+        "snapshotId": item.snapshot_id,
+        "versionNo": item.version_no,
+        "status": item.status,
+        "qualityStatus": item.quality_status,
+        "changeSummary": _json_value(item.change_summary),
+        "qualityReport": _json_value(item.quality_report),
+        "processingRuleVersion": item.processing_rule_version,
+        "generationModel": item.generation_model,
+        "createdAt": item.created_at,
+        "publishedAt": item.published_at,
+    }
+
+
+def _manifest_dict(item: KnowledgeManifest | None) -> dict | None:
+    if item is None:
+        return None
+    return {
+        "id": item.id,
+        "versionId": item.version_id,
+        "purpose": item.purpose,
+        "applicableQuestions": item.applicable_questions,
+        "inapplicableQuestions": item.inapplicable_questions,
+        "coreTopics": item.core_topics,
+        "entities": item.entities,
+        "boundaries": item.boundaries,
+        "changeLevel": item.change_level,
+        "confirmed": item.confirmed == 1,
+    }
+
+
+def _job_dict(item: KnowledgeSyncJob) -> dict:
+    return {
+        "id": item.id,
+        "knowledgeId": item.knowledge_id,
+        "mode": item.mode,
+        "status": item.status,
+        "stage": item.stage,
+        "progress": item.progress,
+        "sourceError": item.source_error,
+        "resultSummary": _json_value(item.result_summary),
+        "createdAt": item.created_at,
+        "startedAt": item.started_at,
+        "finishedAt": item.finished_at,
+    }
+
+
+def _json_value(raw: str | None):
+    if not raw:
+        return None
+    try:
+        return json.loads(raw)
+    except json.JSONDecodeError:
+        return raw

ai_knowledge_base_v2/apps/backend/app/api/router.py

@@ -6,6 +6,7 @@ from app.api import (
     admin_auth,
     admin_dashboard,
     admin_knowledge,
+    admin_knowledge_lifecycle,
     admin_records,
     admin_settings,
     admin_users,
@@ -24,5 +25,6 @@ api_router.include_router(admin_auth.router, prefix="/admin", tags=["admin"])
 api_router.include_router(admin_dashboard.router, prefix="/admin", tags=["admin"])
 api_router.include_router(admin_users.router, prefix="/admin", tags=["admin"])
 api_router.include_router(admin_knowledge.router, prefix="/admin", tags=["admin"])
+api_router.include_router(admin_knowledge_lifecycle.router, prefix="/admin", tags=["admin-knowledge-lifecycle"])
 api_router.include_router(admin_settings.router, prefix="/admin", tags=["admin"])
 api_router.include_router(admin_records.router, prefix="/admin", tags=["admin"])

ai_knowledge_base_v2/apps/backend/app/schemas/admin.py

@@ -70,6 +70,8 @@ class KnowledgeSaveRequest(BaseModel):
     feishuNodeId: str = Field(min_length=1, max_length=100)
     status: int = Field(default=1, ge=0, le=1)
     remark: str | None = Field(default=None, max_length=255)
+    knowledgeType: str = Field(default="general", pattern="^(course|qa|general)$")
+    reviewMode: str = Field(default="manual", pattern="^(manual|automatic)$")
 
 
 class PromptSaveRequest(BaseModel):

ai_knowledge_base_v2/apps/backend/app/schemas/knowledge.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class KnowledgeSyncRequest(BaseModel):
+    mode: str = Field(default="review", pattern="^(review|replace)$")
+
+
+class KnowledgeReviewRequest(BaseModel):
+    action: str = Field(pattern="^(approve|reject)$")
+    reason: str | None = Field(default=None, max_length=1000)
+
+
+class KnowledgePublishRequest(BaseModel):
+    keepOpen: bool = False
+
+
+class KnowledgeStatusRequest(BaseModel):
+    open: bool
+
+
+class KnowledgeLifecycleRequest(BaseModel):
+    status: str = Field(pattern="^(active|archived)$")
+
+
+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/app/services/knowledge_pipeline_service.py

@@ -0,0 +1,520 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from datetime import UTC, datetime
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.models.knowledge import (
+    Knowledge,
+    KnowledgeCard,
+    KnowledgeChunk,
+    KnowledgeManifest,
+    KnowledgePublishLog,
+    KnowledgeReviewRecord,
+    KnowledgeSection,
+    KnowledgeSourceSnapshot,
+    KnowledgeSyncJob,
+    KnowledgeVersion,
+)
+from app.services.external_errors import ExternalServiceError
+from app.services.feishu_async_service import AsyncFeishuKnowledgeService
+from app.services.feishu_service import _feishu_retrieval_config
+
+PROCESSING_RULE_VERSION = "knowledge-pipeline-v1"
+MAX_SECTION_CHARS = 2800
+MAX_CHUNK_CHARS = 650
+
+
+@dataclass(frozen=True)
+class ParsedSection:
+    title: str
+    content: str
+    source_start: int
+    source_end: int
+
+
+class KnowledgePipelineService:
+    @classmethod
+    async def synchronize(
+        cls,
+        db: Session,
+        knowledge: Knowledge,
+        *,
+        admin_id: int,
+        mode: str,
+    ) -> KnowledgeSyncJob:
+        if mode not in {"review", "replace"}:
+            raise ValueError("同步模式必须为 review 或 replace")
+        job = KnowledgeSyncJob(
+            knowledge_id=knowledge.id,
+            mode=mode,
+            status="running",
+            stage="fetching",
+            progress=5,
+            requested_by=admin_id,
+            started_at=_now(),
+        )
+        db.add(job)
+        db.commit()
+        db.refresh(job)
+        try:
+            config = _feishu_retrieval_config(db)
+            scope = _scope_from_knowledge(knowledge)
+            documents = await AsyncFeishuKnowledgeService._load_documents(scope, config)
+            if not documents:
+                raise ExternalServiceError("飞书文章内容为空或无法读取", provider="feishu")
+            source_title, source_content, _ = documents[0]
+            if len(documents) > 1:
+                source_content = "\n\n".join(f"# {title}\n{content}" for title, content, _ in documents)
+            source_content = source_content.strip()
+            if not source_content:
+                raise ExternalServiceError("飞书文章内容为空", provider="feishu")
+
+            cls._set_job(db, job, "snapshot", 20)
+            digest = _hash(source_content)
+            latest_snapshot = db.scalar(
+                select(KnowledgeSourceSnapshot)
+                .where(KnowledgeSourceSnapshot.knowledge_id == knowledge.id)
+                .order_by(KnowledgeSourceSnapshot.id.desc())
+                .limit(1)
+            )
+            if latest_snapshot and latest_snapshot.content_hash == digest:
+                knowledge.source_status = "normal"
+                knowledge.source_error = None
+                knowledge.source_title = source_title
+                knowledge.last_synced_at = _now()
+                job.status = "unchanged"
+                job.stage = "completed"
+                job.progress = 100
+                job.result_summary = json.dumps({"changed": False}, ensure_ascii=False)
+                job.finished_at = _now()
+                db.add_all([knowledge, job])
+                db.commit()
+                return job
+
+            snapshot = KnowledgeSourceSnapshot(
+                knowledge_id=knowledge.id,
+                sync_job_id=job.id,
+                source_title=source_title,
+                source_content=source_content,
+                content_hash=digest,
+                source_identifier=f"{knowledge.feishu_space_id}/{knowledge.feishu_node_id}",
+                fetched_at=_now(),
+                created_by=admin_id,
+            )
+            db.add(snapshot)
+            db.flush()
+
+            version_no = int(
+                db.scalar(select(func.coalesce(func.max(KnowledgeVersion.version_no), 0)).where(
+                    KnowledgeVersion.knowledge_id == knowledge.id
+                ))
+                or 0
+            ) + 1
+            version = KnowledgeVersion(
+                knowledge_id=knowledge.id,
+                snapshot_id=snapshot.id,
+                version_no=version_no,
+                status="processing",
+                quality_status="pending",
+                processing_rule_version=PROCESSING_RULE_VERSION,
+                created_by=admin_id,
+            )
+            db.add(version)
+            db.flush()
+
+            cls._set_job(db, job, "parsing", 40)
+            sections = parse_sections(source_content)
+            section_rows: list[KnowledgeSection] = []
+            chunk_count = 0
+            for index, parsed in enumerate(sections, start=1):
+                section = KnowledgeSection(
+                    knowledge_id=knowledge.id,
+                    version_id=version.id,
+                    section_key=f"S{index:04d}",
+                    title=parsed.title,
+                    content=parsed.content,
+                    source_start=parsed.source_start,
+                    source_end=parsed.source_end,
+                    sort_order=index,
+                    content_hash=_hash(parsed.content),
+                )
+                db.add(section)
+                db.flush()
+                section_rows.append(section)
+                for chunk_index, (chunk_content, start, end) in enumerate(split_chunks(parsed), start=1):
+                    terms = extract_terms(f"{parsed.title} {chunk_content}")
+                    db.add(
+                        KnowledgeChunk(
+                            knowledge_id=knowledge.id,
+                            version_id=version.id,
+                            section_id=section.id,
+                            title=parsed.title,
+                            content=chunk_content,
+                            normalized_text=" ".join(terms),
+                            keywords=json.dumps(terms[:30], ensure_ascii=False),
+                            synonyms=json.dumps(expand_synonyms(terms[:20]), ensure_ascii=False),
+                            source_start=start,
+                            source_end=end,
+                            sort_order=chunk_index,
+                            content_hash=_hash(chunk_content),
+                        )
+                    )
+                    chunk_count += 1
+
+            cls._set_job(db, job, "cards", 65)
+            for section in section_rows:
+                terms = extract_terms(f"{section.title} {section.content}")
+                db.add(
+                    KnowledgeCard(
+                        knowledge_id=knowledge.id,
+                        version_id=version.id,
+                        section_id=section.id,
+                        title=section.title,
+                        summary=_summary(section.content, 180),
+                        core_conclusion=_summary(section.content, 260),
+                        applicable_questions="、".join(terms[:8]) or section.title,
+                        inapplicable_questions="与本章节主题无直接关系的问题",
+                        keywords=json.dumps(terms[:30], ensure_ascii=False),
+                        synonyms=json.dumps(expand_synonyms(terms[:20]), ensure_ascii=False),
+                        entities=json.dumps(terms[:12], ensure_ascii=False),
+                        risk_level=detect_risk_level(section.content),
+                        source_range=f"{section.source_start}-{section.source_end}",
+                        content_hash=section.content_hash,
+                        generation_model="deterministic-v1",
+                        generation_rule_version=PROCESSING_RULE_VERSION,
+                        review_status="pending",
+                    )
+                )
+
+            manifest_data = build_manifest(source_title, source_content, section_rows)
+            previous_manifest = None
+            if knowledge.current_version_id:
+                previous_manifest = db.scalar(
+                    select(KnowledgeManifest).where(KnowledgeManifest.version_id == knowledge.current_version_id)
+                )
+            change_level = manifest_change_level(previous_manifest, manifest_data)
+            manifest = KnowledgeManifest(
+                knowledge_id=knowledge.id,
+                version_id=version.id,
+                purpose=manifest_data["purpose"],
+                applicable_questions=manifest_data["applicableQuestions"],
+                inapplicable_questions=manifest_data["inapplicableQuestions"],
+                core_topics=manifest_data["coreTopics"],
+                entities=manifest_data["entities"],
+                boundaries=manifest_data["boundaries"],
+                content_hash=_hash(json.dumps(manifest_data, ensure_ascii=False, sort_keys=True)),
+                change_level=change_level,
+                confirmed=1 if previous_manifest and change_level == "minor" else 0,
+            )
+            db.add(manifest)
+
+            cls._set_job(db, job, "quality", 82)
+            quality = quality_check(source_content, sections, chunk_count)
+            version.quality_status = "passed" if not quality["blocking"] else "blocked"
+            version.quality_report = json.dumps(quality, ensure_ascii=False)
+            version.change_summary = json.dumps(
+                {"sections": len(sections), "chunks": chunk_count, "manifestChange": change_level},
+                ensure_ascii=False,
+            )
+            version.status = "pending_review"
+            knowledge.source_title = source_title
+            knowledge.source_status = "normal"
+            knowledge.source_error = None
+            knowledge.pending_version_id = version.id
+            knowledge.last_synced_at = _now()
+            knowledge.migration_status = "pending_review"
+            job.status = "pending_review"
+            job.stage = "completed"
+            job.progress = 100
+            job.result_summary = json.dumps(
+                {"versionId": version.id, "versionNo": version.version_no, **quality}, ensure_ascii=False
+            )
+            job.finished_at = _now()
+            db.add_all([knowledge, version, job])
+            db.commit()
+            return job
+        except Exception as exc:
+            db.rollback()
+            stored_job = db.get(KnowledgeSyncJob, job.id)
+            stored_knowledge = db.get(Knowledge, knowledge.id)
+            if stored_job:
+                stored_job.status = "failed"
+                stored_job.stage = "failed"
+                stored_job.source_error = str(exc)
+                stored_job.finished_at = _now()
+                db.add(stored_job)
+            if stored_knowledge:
+                stored_knowledge.source_status = "error"
+                stored_knowledge.source_error = str(exc)
+                stored_knowledge.migration_status = "failed"
+                db.add(stored_knowledge)
+            db.commit()
+            raise
+
+    @staticmethod
+    def confirm_manifest(db: Session, knowledge: Knowledge, version: KnowledgeVersion, admin_id: int) -> None:
+        manifest = db.scalar(select(KnowledgeManifest).where(KnowledgeManifest.version_id == version.id))
+        if manifest is None:
+            raise ValueError("该版本没有知识库说明卡")
+        manifest.confirmed = 1
+        manifest.confirmed_by = admin_id
+        manifest.confirmed_at = _now()
+        if knowledge.pending_version_id == version.id:
+            knowledge.manifest_confirmed = 1
+        db.add_all([manifest, knowledge])
+
+    @staticmethod
+    def review_version(db: Session, knowledge: Knowledge, version: KnowledgeVersion, admin_id: int, action: str, reason: str | None) -> None:
+        if action not in {"approve", "reject"}:
+            raise ValueError("审核动作不正确")
+        cards = db.scalars(select(KnowledgeCard).where(KnowledgeCard.version_id == version.id)).all()
+        review_status = "approved" if action == "approve" else "rejected"
+        for card in cards:
+            card.review_status = review_status
+            db.add(card)
+        version.status = "approved" if action == "approve" else "rejected"
+        db.add_all(
+            [
+                version,
+                KnowledgeReviewRecord(
+                    knowledge_id=knowledge.id,
+                    version_id=version.id,
+                    action=action,
+                    reason=reason,
+                    reviewed_by=admin_id,
+                ),
+            ]
+        )
+
+    @staticmethod
+    def publish_version(db: Session, knowledge: Knowledge, version: KnowledgeVersion, admin_id: int, keep_open: bool) -> None:
+        if version.status != "approved":
+            raise ValueError("只有审核通过的版本可以发布")
+        manifest = db.scalar(select(KnowledgeManifest).where(KnowledgeManifest.version_id == version.id))
+        if manifest is None or manifest.confirmed != 1:
+            raise ValueError("知识库说明卡尚未确认")
+        previous_version_id = knowledge.current_version_id
+        if previous_version_id:
+            previous = db.get(KnowledgeVersion, previous_version_id)
+            if previous:
+                previous.status = "superseded"
+                db.add(previous)
+        version.status = "published"
+        version.published_by = admin_id
+        version.published_at = _now()
+        knowledge.current_version_id = version.id
+        knowledge.pending_version_id = None
+        knowledge.manifest_confirmed = 1
+        knowledge.last_published_at = version.published_at
+        knowledge.status = 1 if keep_open else 0
+        knowledge.migration_status = "completed"
+        db.add_all(
+            [
+                version,
+                knowledge,
+                KnowledgePublishLog(
+                    knowledge_id=knowledge.id,
+                    from_version_id=previous_version_id,
+                    to_version_id=version.id,
+                    action="publish",
+                    keep_open=knowledge.status,
+                    operated_by=admin_id,
+                ),
+            ]
+        )
+
+    @staticmethod
+    def rollback(db: Session, knowledge: Knowledge, version: KnowledgeVersion, admin_id: int, keep_open: bool) -> None:
+        if version.status not in {"published", "superseded"}:
+            raise ValueError("只能回滚到历史正式版本")
+        previous_id = knowledge.current_version_id
+        knowledge.current_version_id = version.id
+        knowledge.status = 1 if keep_open else 0
+        knowledge.last_published_at = _now()
+        db.add_all(
+            [
+                knowledge,
+                KnowledgePublishLog(
+                    knowledge_id=knowledge.id,
+                    from_version_id=previous_id,
+                    to_version_id=version.id,
+                    action="rollback",
+                    keep_open=knowledge.status,
+                    operated_by=admin_id,
+                ),
+            ]
+        )
+
+    @staticmethod
+    def _set_job(db: Session, job: KnowledgeSyncJob, stage: str, progress: int) -> None:
+        job.stage = stage
+        job.progress = progress
+        db.add(job)
+        db.flush()
+
+
+def parse_sections(content: str) -> list[ParsedSection]:
+    heading = re.compile(r"(?m)^(#{1,6}\s+.+|[^\n]{1,80}\n[-=]{3,})$")
+    matches = list(heading.finditer(content))
+    boundaries: list[tuple[int, str]] = []
+    for match in matches:
+        raw = match.group(0).splitlines()[0].lstrip("#").strip()
+        boundaries.append((match.start(), raw))
+    if not boundaries or boundaries[0][0] > 0:
+        boundaries.insert(0, (0, "正文"))
+    sections: list[ParsedSection] = []
+    for index, (start, title) in enumerate(boundaries):
+        end = boundaries[index + 1][0] if index + 1 < len(boundaries) else len(content)
+        block = content[start:end].strip()
+        if not block:
+            continue
+        cursor = 0
+        while len(block) - cursor > MAX_SECTION_CHARS:
+            cut = block.rfind("\n\n", cursor, cursor + MAX_SECTION_CHARS)
+            if cut <= cursor:
+                cut = cursor + MAX_SECTION_CHARS
+            part = block[cursor:cut].strip()
+            sections.append(ParsedSection(title, part, start + cursor, start + cut))
+            cursor = cut
+        tail = block[cursor:].strip()
+        if tail:
+            sections.append(ParsedSection(title, tail, start + cursor, end))
+    return sections or [ParsedSection("正文", content, 0, len(content))]
+
+
+def split_chunks(section: ParsedSection) -> list[tuple[str, int, int]]:
+    chunks: list[tuple[str, int, int]] = []
+    cursor = 0
+    while cursor < len(section.content):
+        end = min(len(section.content), cursor + MAX_CHUNK_CHARS)
+        if end < len(section.content):
+            boundary = max(section.content.rfind("\n\n", cursor, end), section.content.rfind("。", cursor, end))
+            if boundary > cursor + 120:
+                end = boundary + 1
+        text = section.content[cursor:end].strip()
+        if text:
+            chunks.append((text, section.source_start + cursor, section.source_start + end))
+        cursor = end
+    return chunks
+
+
+def extract_terms(text: str) -> list[str]:
+    normalized = re.sub(r"\s+", "", text.lower())
+    ascii_words = re.findall(r"[a-z0-9_]{2,}", normalized)
+    chinese_runs = re.findall(r"[\u4e00-\u9fff]+", normalized)
+    terms = list(ascii_words)
+    for run in chinese_runs:
+        if len(run) <= 4:
+            terms.append(run)
+        for size in (2, 3, 4):
+            terms.extend(run[i : i + size] for i in range(max(0, len(run) - size + 1)))
+    return list(dict.fromkeys(term for term in terms if term))
+
+
+_SYNONYMS = {
+    "孩子": ["小孩", "学生", "子女"],
+    "家长": ["父母", "妈妈", "爸爸"],
+    "学习": ["功课", "学业", "上课"],
+    "焦虑": ["担心", "紧张", "害怕"],
+    "沟通": ["交流", "对话", "聊天"],
+    "课程": ["训练营", "大本营", "课堂"],
+}
+
+
+def expand_synonyms(terms: list[str]) -> list[str]:
+    values: list[str] = []
+    for term in terms:
+        values.extend(_SYNONYMS.get(term, []))
+    return list(dict.fromkeys(values))
+
+
+def build_manifest(title: str, content: str, sections: list[KnowledgeSection]) -> dict[str, str]:
+    fields = {}
+    for label, key in [
+        ("知识库用途", "purpose"),
+        ("适用问题", "applicableQuestions"),
+        ("不适用问题", "inapplicableQuestions"),
+        ("核心主题", "coreTopics"),
+    ]:
+        match = re.search(rf"{label}[：:]\s*([^\n]+)", content)
+        if match:
+            fields[key] = match.group(1).strip()
+    topics = "、".join(list(dict.fromkeys(section.title for section in sections))[:12])
+    return {
+        "purpose": fields.get("purpose", f"提供与《{title}》相关的课程、答疑和业务知识"),
+        "applicableQuestions": fields.get("applicableQuestions", topics or title),
+        "inapplicableQuestions": fields.get("inapplicableQuestions", "与本文主题无直接关系的一般问题"),
+        "coreTopics": fields.get("coreTopics", topics or title),
+        "entities": topics,
+        "boundaries": "只能围绕具体问题提供必要解释，不输出整篇原文或大段连续课程资料。",
+    }
+
+
+def manifest_change_level(previous: KnowledgeManifest | None, current: dict[str, str]) -> str:
+    if previous is None:
+        return "major"
+    major_fields = ["purpose", "applicableQuestions", "inapplicableQuestions", "coreTopics"]
+    old = [previous.purpose, previous.applicable_questions, previous.inapplicable_questions, previous.core_topics]
+    new = [current[field] for field in major_fields]
+    return "minor" if all(_similar(left, right) >= 0.72 for left, right in zip(old, new, strict=True)) else "major"
+
+
+def quality_check(content: str, sections: list[ParsedSection], chunk_count: int) -> dict:
+    duplicate_hashes = [_hash(item.content) for item in sections]
+    issues: list[dict[str, str]] = []
+    if len(content) < 50:
+        issues.append({"level": "blocking", "code": "content_too_short", "message": "源文章内容过短"})
+    if not sections or chunk_count == 0:
+        issues.append({"level": "blocking", "code": "parse_empty", "message": "未生成章节或切片"})
+    if len(set(duplicate_hashes)) != len(duplicate_hashes):
+        issues.append({"level": "warning", "code": "duplicate_sections", "message": "存在重复章节"})
+    high_risk = detect_risk_level(content) == "high"
+    if high_risk:
+        issues.append({"level": "warning", "code": "high_risk", "message": "包含高风险主题，需要重点审核"})
+    return {
+        "blocking": any(item["level"] == "blocking" for item in issues),
+        "issues": issues,
+        "sectionCount": len(sections),
+        "chunkCount": chunk_count,
+        "highRisk": high_risk,
+    }
+
+
+def detect_risk_level(text: str) -> str:
+    high = ("自杀", "自伤", "伤人", "违法", "医疗", "诊断", "药物", "投资", "法律")
+    return "high" if any(word in text for word in high) else "normal"
+
+
+def _summary(text: str, limit: int) -> str:
+    value = re.sub(r"\s+", " ", text).strip()
+    return value if len(value) <= limit else value[:limit].rstrip() + "…"
+
+
+def _hash(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def _similar(left: str, right: str) -> float:
+    left_terms, right_terms = set(extract_terms(left)), set(extract_terms(right))
+    return len(left_terms & right_terms) / max(1, len(left_terms | right_terms))
+
+
+def _now() -> datetime:
+    return datetime.now(UTC).replace(tzinfo=None)
+
+
+def _scope_from_knowledge(knowledge: Knowledge):
+    from app.services.knowledge_service import KnowledgeScope
+
+    return KnowledgeScope(
+        id=knowledge.id,
+        name=knowledge.name,
+        feishu_space_id=knowledge.feishu_space_id,
+        feishu_node_id=knowledge.feishu_node_id,
+    )