"""知识、排查方案和解决方案的统一版本治理服务。 发布、下线和回滚均保留不可变版本及审计记录,索引更新只在数据库事务提交后触发。 """ from copy import deepcopy from django.db import transaction from django.utils import timezone from apps.audit.models import OperationAuditLog from apps.knowledge.models import ( GovernedAsset, GovernedAssetType, GovernedAssetVersion, GovernedAssetVersionStatus, KnowledgeItem, TroubleshootingPlan, ) from apps.solutions.models import Solution class AssetGovernanceError(ValueError): pass ASSET_MODELS = { GovernedAssetType.KNOWLEDGE_ITEM: KnowledgeItem, GovernedAssetType.TROUBLESHOOTING_PLAN: TroubleshootingPlan, GovernedAssetType.SOLUTION: Solution, } ASSET_FIELDS = { GovernedAssetType.KNOWLEDGE_ITEM: ( "knowledge_base_id", "title", "problem", "applicable_scope", "category_code", "related_system", "environment_scope", "error_code", "root_cause", "troubleshooting_steps", "solution", "risk_note", "source_type", "source_ticket_id", ), GovernedAssetType.TROUBLESHOOTING_PLAN: ( "knowledge_base_id", "title", "problem_pattern", "applicable_scope", "category_code", "related_system", "environment_scope", "decision_rules", "troubleshooting_steps", "risk_note", "source_ticket_id", ), GovernedAssetType.SOLUTION: ( "title", "problem_signature", "related_system", "environment_scope", "category_id", "root_cause", "solution", "risk_note", "allow_auto_resolve", "requires_human_action", "requires_external_confirmation", "requires_production_change", ), } REQUIRED_FIELDS = { GovernedAssetType.KNOWLEDGE_ITEM: ( "knowledge_base_id", "title", "problem", "solution", ), GovernedAssetType.TROUBLESHOOTING_PLAN: ( "knowledge_base_id", "title", "problem_pattern", "troubleshooting_steps", ), GovernedAssetType.SOLUTION: ( "title", "problem_signature", "root_cause", "solution", ), } def snapshot_asset(asset_type: str, instance) -> dict: try: fields = ASSET_FIELDS[asset_type] except KeyError as error: raise AssetGovernanceError("不支持的知识资产类型") from error return {field: deepcopy(getattr(instance, field)) for field in fields} def validate_asset_content(asset_type: str, content: dict) -> dict: if not isinstance(content, dict): raise AssetGovernanceError("资产版本内容必须是对象") allowed = set(ASSET_FIELDS.get(asset_type, ())) if not allowed: raise AssetGovernanceError("不支持的知识资产类型") unknown = sorted(set(content) - allowed) if unknown: raise AssetGovernanceError(f"资产版本包含未知字段:{', '.join(unknown)}") missing = [ field for field in REQUIRED_FIELDS[asset_type] if content.get(field) in {None, ""} ] if missing: raise AssetGovernanceError(f"资产版本缺少必填字段:{', '.join(missing)}") return {field: deepcopy(content.get(field)) for field in ASSET_FIELDS[asset_type]} @transaction.atomic def ensure_governed_asset(asset_type: str, instance, *, actor=None) -> GovernedAsset: if asset_type not in ASSET_MODELS: raise AssetGovernanceError("不支持的知识资产类型") asset, _ = GovernedAsset.objects.select_for_update().get_or_create( asset_type=asset_type, object_id=instance.id, defaults={"created_by": actor or getattr(instance, "created_by", None)}, ) if not asset.versions.exists(): GovernedAssetVersion.objects.create( asset=asset, version=1, status=GovernedAssetVersionStatus.DRAFT, content=snapshot_asset(asset_type, instance), created_by=actor or getattr(instance, "created_by", None), ) return asset def get_asset_instance(asset: GovernedAsset): model = ASSET_MODELS.get(asset.asset_type) if model is None: raise AssetGovernanceError("不支持的知识资产类型") instance = model.objects.filter(id=asset.object_id).first() if instance is None: raise AssetGovernanceError("知识资产对应的业务对象不存在") return instance def get_editable_version(asset: GovernedAsset) -> GovernedAssetVersion | None: return asset.versions.filter( status__in=[ GovernedAssetVersionStatus.DRAFT, GovernedAssetVersionStatus.REJECTED, ] ).first() @transaction.atomic def create_asset_revision( asset: GovernedAsset, *, actor, source_version: GovernedAssetVersion | None = None, change_summary: str = "", ) -> GovernedAssetVersion: """基于当前发布版本创建唯一可编辑修订;已有草稿时直接复用。""" asset = GovernedAsset.objects.select_for_update().get(id=asset.id) editable = get_editable_version(asset) if editable is not None: return editable source = source_version or asset.current_published_version if source is None: content = snapshot_asset(asset.asset_type, get_asset_instance(asset)) else: content = deepcopy(source.content) latest_version = ( asset.versions.order_by("-version") .values_list("version", flat=True) .first() or 0 ) next_version = latest_version + 1 version = GovernedAssetVersion.objects.create( asset=asset, version=next_version, content=content, change_summary=change_summary.strip(), source_version=source, created_by=actor, ) _audit(version, actor, "asset_revision_created") return version @transaction.atomic def update_asset_version( version: GovernedAssetVersion, content: dict, *, actor, change_summary: str | None = None, ) -> GovernedAssetVersion: """执行独立审核、原子切换发布版本,并在提交后刷新检索索引。""" version = GovernedAssetVersion.objects.select_for_update().select_related("asset").get( id=version.id ) if version.status not in { GovernedAssetVersionStatus.DRAFT, GovernedAssetVersionStatus.REJECTED, }: raise AssetGovernanceError("只有草稿或已驳回版本可以编辑") version.content = validate_asset_content(version.asset.asset_type, content) version.status = GovernedAssetVersionStatus.DRAFT version.review_comment = "" version.reviewed_by = None version.reviewed_at = None if change_summary is not None: version.change_summary = change_summary.strip() version.save( update_fields=[ "content", "status", "change_summary", "review_comment", "reviewed_by", "reviewed_at", "updated_at", ] ) _audit(version, actor, "asset_draft_updated") return version @transaction.atomic def submit_asset_version( version: GovernedAssetVersion, *, actor, ) -> GovernedAssetVersion: version = GovernedAssetVersion.objects.select_for_update().select_related("asset").get( id=version.id ) if version.status != GovernedAssetVersionStatus.DRAFT: raise AssetGovernanceError("只有草稿版本可以提交审核") version.content = validate_asset_content(version.asset.asset_type, version.content) version.status = GovernedAssetVersionStatus.PENDING_REVIEW version.submitted_by = actor version.submitted_at = timezone.now() version.save( update_fields=[ "content", "status", "submitted_by", "submitted_at", "updated_at", ] ) from apps.workitems.services import sync_asset_review_work_item sync_asset_review_work_item(version, actor=actor) _audit(version, actor, "asset_submitted") return version def _apply_published_content(version: GovernedAssetVersion): instance = get_asset_instance(version.asset) for field, value in version.content.items(): setattr(instance, field, deepcopy(value)) instance.status = GovernedAssetVersionStatus.PUBLISHED if hasattr(instance, "reviewed_by_id"): instance.reviewed_by = version.reviewed_by update_fields = [*version.content.keys(), "status", "updated_at"] if hasattr(instance, "reviewed_by_id"): update_fields.append("reviewed_by") instance.save(update_fields=update_fields) return instance @transaction.atomic def approve_asset_version( version: GovernedAssetVersion, *, reviewer, comment: str = "", ) -> GovernedAssetVersion: version = ( GovernedAssetVersion.objects.select_for_update() .select_related("asset", "submitted_by") .get(id=version.id) ) if version.status != GovernedAssetVersionStatus.PENDING_REVIEW: raise AssetGovernanceError("只有待审核版本可以通过") if version.submitted_by_id and version.submitted_by_id == reviewer.id: raise AssetGovernanceError("提交人不能审核自己的资产版本") now = timezone.now() previous = version.asset.current_published_version if previous is not None and previous.id != version.id: previous.status = GovernedAssetVersionStatus.OFFLINE previous.offline_at = now previous.save(update_fields=["status", "offline_at", "updated_at"]) version.status = GovernedAssetVersionStatus.PUBLISHED version.reviewed_by = reviewer version.review_comment = comment.strip() version.reviewed_at = now version.published_at = now version.offline_at = None version.save( update_fields=[ "status", "reviewed_by", "review_comment", "reviewed_at", "published_at", "offline_at", "updated_at", ] ) version.asset.current_published_version = version version.asset.save(update_fields=["current_published_version", "updated_at"]) _apply_published_content(version) # 避免索引先于业务事务可见,导致检索读取到未提交或已回滚的版本。 transaction.on_commit( lambda: _dispatch_asset_index(version.asset.asset_type, version.asset.object_id) ) from apps.workitems.services import sync_asset_review_work_item sync_asset_review_work_item(version, actor=reviewer) _audit(version, reviewer, "asset_published") return version @transaction.atomic def reject_asset_version( version: GovernedAssetVersion, *, reviewer, comment: str, ) -> GovernedAssetVersion: """从历史已发布版本复制内容生成新草稿,不直接覆盖当前线上版本。""" version = ( GovernedAssetVersion.objects.select_for_update() .select_related("asset", "submitted_by") .get(id=version.id) ) if version.status != GovernedAssetVersionStatus.PENDING_REVIEW: raise AssetGovernanceError("只有待审核版本可以驳回") if version.submitted_by_id and version.submitted_by_id == reviewer.id: raise AssetGovernanceError("提交人不能审核自己的资产版本") if not comment.strip(): raise AssetGovernanceError("驳回时必须填写原因") version.status = GovernedAssetVersionStatus.REJECTED version.reviewed_by = reviewer version.review_comment = comment.strip() version.reviewed_at = timezone.now() version.save( update_fields=[ "status", "reviewed_by", "review_comment", "reviewed_at", "updated_at", ] ) from apps.workitems.services import sync_asset_review_work_item sync_asset_review_work_item(version, actor=reviewer) _audit(version, reviewer, "asset_rejected") return version @transaction.atomic def offline_asset(asset: GovernedAsset, *, actor, comment: str = ""): asset = ( GovernedAsset.objects.select_for_update() .select_related("current_published_version") .get(id=asset.id) ) version = asset.current_published_version if version is None or version.status != GovernedAssetVersionStatus.PUBLISHED: raise AssetGovernanceError("当前没有可下线的已发布版本") version.status = GovernedAssetVersionStatus.OFFLINE version.offline_at = timezone.now() version.review_comment = comment.strip() version.save( update_fields=["status", "offline_at", "review_comment", "updated_at"] ) instance = get_asset_instance(asset) instance.status = GovernedAssetVersionStatus.OFFLINE instance.save(update_fields=["status", "updated_at"]) asset.current_published_version = None asset.save(update_fields=["current_published_version", "updated_at"]) transaction.on_commit(lambda: _dispatch_asset_index(asset.asset_type, asset.object_id)) _audit(version, actor, "asset_offline") return version @transaction.atomic def rollback_asset( asset: GovernedAsset, source_version: GovernedAssetVersion, *, actor, change_summary: str = "", ) -> GovernedAssetVersion: asset = GovernedAsset.objects.select_for_update().get(id=asset.id) source_version = GovernedAssetVersion.objects.get( id=source_version.id, asset=asset, ) if source_version.status not in { GovernedAssetVersionStatus.PUBLISHED, GovernedAssetVersionStatus.OFFLINE, }: raise AssetGovernanceError("只能从历史发布版本创建回滚草稿") if get_editable_version(asset) is not None: raise AssetGovernanceError("已有可编辑版本,请先处理后再回滚") return create_asset_revision( asset, actor=actor, source_version=source_version, change_summary=change_summary or f"回滚自 v{source_version.version}", ) def compare_asset_versions( left: GovernedAssetVersion, right: GovernedAssetVersion, ) -> dict: if left.asset_id != right.asset_id: raise AssetGovernanceError("只能比较同一资产的版本") fields = sorted(set(left.content) | set(right.content)) return { "left_version": left.version, "right_version": right.version, "changes": [ { "field": field, "before": deepcopy(left.content.get(field)), "after": deepcopy(right.content.get(field)), } for field in fields if left.content.get(field) != right.content.get(field) ], } def _dispatch_asset_index(asset_type: str, object_id: int) -> None: from apps.knowledge.tasks import dispatch_governed_asset_index dispatch_governed_asset_index(asset_type, object_id) def _audit(version: GovernedAssetVersion, actor, action: str) -> None: OperationAuditLog.objects.create( user=actor, module="knowledge", action=action, target_type=version.asset.asset_type, target_id=str(version.asset.object_id), request_snapshot={ "asset_public_id": str(version.asset.public_id), "version": version.version, "status": version.status, "source_version": version.source_version.version if version.source_version_id else None, }, result="success", )