diff --git a/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py b/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py index b53d9ee..d220e48 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/admin_knowledge.py +++ b/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, } 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 new file mode 100644 index 0000000..bd2c6d3 --- /dev/null +++ b/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 diff --git a/ai_knowledge_base_v2/apps/backend/app/api/router.py b/ai_knowledge_base_v2/apps/backend/app/api/router.py index a325bce..bcb8a4d 100644 --- a/ai_knowledge_base_v2/apps/backend/app/api/router.py +++ b/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"]) diff --git a/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py b/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py index 98c33db..506af45 100644 --- a/ai_knowledge_base_v2/apps/backend/app/schemas/admin.py +++ b/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): diff --git a/ai_knowledge_base_v2/apps/backend/app/schemas/knowledge.py b/ai_knowledge_base_v2/apps/backend/app/schemas/knowledge.py new file mode 100644 index 0000000..c4344bd --- /dev/null +++ b/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) diff --git a/ai_knowledge_base_v2/apps/backend/app/services/knowledge_pipeline_service.py b/ai_knowledge_base_v2/apps/backend/app/services/knowledge_pipeline_service.py new file mode 100644 index 0000000..26697fc --- /dev/null +++ b/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, + )