AI决策审计追踪本体 — 供应链自动化决策的完整记录、回溯与合规证明

"""
AI决策审计追踪本体
功能：审计记录生成 / 哈希链完整性 / 合规查询 / 异常决策检测
"""
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
import warnings
warnings.filterwarnings('ignore')


@dataclass
class AuditRecord:
    record_id: str
    timestamp: str
    # 5W
    who_agent_id: str
    who_approval_level: str     # AUTO / MANAGER / VP / LEGAL
    what_action_type: str
    what_parameters: dict
    why_trigger_tags: dict      # 触发此Action的Tag状态
    why_signal_scores: dict     # 信号分数
    how_algorithm: str
    how_algorithm_version: str
    entity_id: str
    estimated_impact_yuan: float
    execution_result: Optional[dict] = None
    prev_record_hash: str = ""
    record_hash: str = ""

    def compute_hash(self) -> str:
        content = json.dumps({
            "record_id": self.record_id,
            "timestamp": self.timestamp,
            "who_agent_id": self.who_agent_id,
            "what_action_type": self.what_action_type,
            "entity_id": self.entity_id,
            "prev_hash": self.prev_record_hash,
        }, sort_keys=True)
        return hashlib.sha256(content.encode()).hexdigest()[:16]


class DecisionAuditTrail:

    def __init__(self):
        self.records: list = []
        self._last_hash = "GENESIS"

    def record_decision(self, agent_id: str, action_type: str, entity_id: str,
                         parameters: dict, trigger_tags: dict, signal_scores: dict,
                         algorithm: str, approval_level: str,
                         impact_yuan: float, result: dict = None) -> AuditRecord:
        record = AuditRecord(
            record_id=f"AUD-{len(self.records)+1:06d}",
            timestamp=datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:23],
            who_agent_id=agent_id,
            who_approval_level=approval_level,
            what_action_type=action_type,