feat(knowledge): auto-publish successful syncs

ai_knowledge_base_v2/apps/backend/app/services/knowledge_pipeline_service.py

@@ -15,7 +15,6 @@ from app.models.knowledge import (
     KnowledgeChunk,
     KnowledgeManifest,
     KnowledgePublishLog,
-    KnowledgeReviewRecord,
     KnowledgeSection,
     KnowledgeSourceSnapshot,
     KnowledgeSyncJob,
@@ -25,7 +24,7 @@ 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"
+PROCESSING_RULE_VERSION = "knowledge-pipeline-v2-auto-publish"
 MAX_SECTION_CHARS = 2800
 MAX_CHUNK_CHARS = 650
 
@@ -46,13 +45,10 @@ class KnowledgePipelineService:
         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,
+            mode="replace",
             status="running",
             stage="fetching",
             progress=5,
@@ -88,10 +84,38 @@ class KnowledgePipelineService:
                 knowledge.source_error = None
                 knowledge.source_title = source_title
                 knowledge.last_synced_at = _now()
-                job.status = "unchanged"
+                latest_version = db.scalar(
+                    select(KnowledgeVersion)
+                    .where(KnowledgeVersion.snapshot_id == latest_snapshot.id)
+                    .order_by(KnowledgeVersion.id.desc())
+                    .limit(1)
+                )
+                if latest_version is not None and latest_version.id != knowledge.current_version_id:
+                    if latest_version.quality_status == "blocked":
+                        raise ValueError("最新同步版本质检未通过，已保留当前正式版本")
+                    cls._activate_version(
+                        db,
+                        knowledge,
+                        latest_version,
+                        admin_id,
+                        keep_open=knowledge.status == 1,
+                        action="sync_publish",
+                    )
+                    job.status = "published"
+                    job.result_summary = json.dumps(
+                        {
+                            "changed": False,
+                            "activatedExistingVersion": True,
+                            "versionId": latest_version.id,
+                            "versionNo": latest_version.version_no,
+                        },
+                        ensure_ascii=False,
+                    )
+                else:
+                    job.status = "unchanged"
+                    job.result_summary = json.dumps({"changed": False}, ensure_ascii=False)
                 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()
@@ -188,7 +212,7 @@ class KnowledgePipelineService:
                         content_hash=section.content_hash,
                         generation_model="deterministic-v1",
                         generation_rule_version=PROCESSING_RULE_VERSION,
-                        review_status="pending",
+                        review_status="approved",
                     )
                 )
 
@@ -210,7 +234,9 @@ class KnowledgePipelineService:
                 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 previous_manifest.confirmed == 1 and change_level == "minor" else 0,
+                confirmed=1,
+                confirmed_by=admin_id,
+                confirmed_at=_now(),
             )
             db.add(manifest)
 
@@ -222,40 +248,32 @@ class KnowledgePipelineService:
                 {"sections": len(sections), "chunks": chunk_count, "manifestChange": change_level},
                 ensure_ascii=False,
             )
-            version.status = "pending_review"
+            if quality["blocking"]:
+                issue_text = "；".join(
+                    item["message"] for item in quality["issues"] if item["level"] == "blocking"
+                )
+                raise ValueError(f"同步内容质检未通过：{issue_text or '存在阻断项'}")
+
             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"
+            cls._activate_version(
+                db,
+                knowledge,
+                version,
+                admin_id,
+                keep_open=knowledge.status == 1,
+                action="sync_publish",
+            )
+            job.status = "published"
             job.stage = "completed"
             job.progress = 100
             job.result_summary = json.dumps(
-                {"versionId": version.id, "versionNo": version.version_no, **quality}, ensure_ascii=False
+                {"versionId": version.id, "versionNo": version.version_no, "autoPublished": True, **quality},
+                ensure_ascii=False,
             )
             job.finished_at = _now()
-            if (
-                knowledge.review_mode == "automatic"
-                and version.quality_status == "passed"
-                and not quality["highRisk"]
-                and manifest.confirmed == 1
-            ):
-                cls.review_version(
-                    db,
-                    knowledge,
-                    version,
-                    admin_id,
-                    "approve",
-                    "自动审核方式：质检通过且说明卡可继承，自动发布。",
-                )
-                cls.publish_version(db, knowledge, version, admin_id, keep_open=knowledge.status == 1)
-                job.status = "published"
-                job.result_summary = json.dumps(
-                    {"versionId": version.id, "versionNo": version.version_no, "autoPublished": True, **quality},
-                    ensure_ascii=False,
-                )
             db.add_all([knowledge, version, job])
             db.commit()
             return job
@@ -278,57 +296,34 @@ class KnowledgePipelineService:
             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("知识库说明卡尚未确认")
+    def _activate_version(
+        db: Session,
+        knowledge: Knowledge,
+        version: KnowledgeVersion,
+        admin_id: int,
+        *,
+        keep_open: bool,
+        action: str,
+    ) -> None:
+        """Atomically activate a validated version and preserve rollback history."""
+        if version.quality_status == "blocked":
+            raise ValueError("质检未通过的版本不能生效")
         if keep_open and knowledge.lifecycle_status != "active":
-            raise ValueError("归档知识库不能在发布后开放")
+            raise ValueError("归档知识库不能在版本替换后开放")
         if keep_open and knowledge.source_status != "normal":
-            raise ValueError("飞书源文章异常，不能在发布后开放")
+            raise ValueError("飞书源文章异常，不能替换正式版本")
         previous_version_id = knowledge.current_version_id
         if previous_version_id:
             previous = db.get(KnowledgeVersion, previous_version_id)
-            if previous:
+            if previous and previous.id != version.id:
                 previous.status = "superseded"
                 db.add(previous)
+        previous_pending_id = knowledge.pending_version_id
+        if previous_pending_id and previous_pending_id not in {previous_version_id, version.id}:
+            previous_pending = db.get(KnowledgeVersion, previous_pending_id)
+            if previous_pending and previous_pending.status not in {"published", "superseded"}:
+                previous_pending.status = "discarded"
+                db.add(previous_pending)
         version.status = "published"
         version.published_by = admin_id
         version.published_at = _now()
@@ -346,7 +341,7 @@ class KnowledgePipelineService:
                     knowledge_id=knowledge.id,
                     from_version_id=previous_version_id,
                     to_version_id=version.id,
-                    action="publish",
+                    action=action,
                     keep_open=knowledge.status,
                     operated_by=admin_id,
                 ),
@@ -510,7 +505,7 @@ def quality_check(content: str, sections: list[ParsedSection], chunk_count: int)
         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": "包含高风险主题，需要重点审核"})
+        issues.append({"level": "warning", "code": "high_risk", "message": "包含高风险主题，请重点关注内容准确性"})
     return {
         "blocking": any(item["level"] == "blocking" for item in issues),
         "issues": issues,