AI 监管与法律合规全景：从诉讼判例到开发者合规指南

"""
AI 合规自动化检查器
帮助开发者在部署前自动检测常见合规风险
"""
import re
import hashlib
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum

class ComplianceRisk(Enum):
    HIGH = "高风险"
    MEDIUM = "中风险"
    LOW = "低风险"
    INFO = "信息提示"

@dataclass
class ComplianceIssue:
    """合规问题记录"""
    category: str          # 问题类别
    risk_level: ComplianceRisk
    description: str       # 问题描述
    regulation: str        # 相关法规
    recommendation: str    # 修复建议

class AIComplianceChecker:
    """AI 系统合规检查器"""

    def __init__(self, jurisdiction: str = "EU"):
        """
        Args:
            jurisdiction: 目标司法管辖区 (EU/US/CN)
        """
        self.jurisdiction = jurisdiction
        self.issues: List[ComplianceIssue] = []

    def check_training_data(self, data_sources: List[Dict]) -> List[ComplianceIssue]:
        """检查训练数据合规性"""
        issues = []
        for source in data_sources:
            # 检查数据来源许可
            if not source.get("license"):
                issues.append(ComplianceIssue(
                    category="训练数据",
                    risk_level=ComplianceRisk.HIGH,
                    description=f"数据源 '{source.get('name', 'unknown')}' 未声明许可证",
                    regulation="EU AI Act Art. 10 / 中国生成式 AI 暂行办法第 7 条",
                    recommendation="获取数据源的明确许可证或使用已授权数据集"
                ))

            # 检查是否包含个人数据
            if source.get("contains_pii", False):
                issues.append(ComplianceIssue(
                    category="数据隐私",
                    risk_level=ComplianceRisk.HIGH,
                    description=f"数据源 '{source.get('name')}' 包含个人可识别信息",
                    regulation="GDPR Art. 9 / PIPL 第 28 条",
                    recommendation="对 PII 进行脱敏处理或获取用户明确同意"
                ))

            # 检查数据偏差
            if source.get("bias_score", 0) > 0.7:
                issues.append(ComplianceIssue(
                    category="数据偏差",
                    risk_level=ComplianceRisk.MEDIUM,
                    description=f"数据源 '{source.get('name')}' 存在显著偏差 (score={source.get('bias_score')})",
                    regulation="EU AI Act Art. 10(2) / NIST AI RMF",
                    recommendation="重新采样或使用偏差纠正技术"
                ))

        self.issues.extend(issues)
        return issues

    def check_model_output(self, sample_outputs: List[str]) -> List[ComplianceIssue]:
        """检查模型输出合规性"""
        issues = []
        for output in sample_outputs:
            # 检查是否包含 PII
            pii_patterns = [
                (r'\b\d{3}-\d{2}-\d{4}\b', "社会保险号"),
                (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', "邮箱地址"),
                (r'\b\d{16}\b', "信用卡号"),
            ]
            for pattern, pii_type in pii_patterns:
                if re.search(pattern, output):
                    issues.append(ComplianceIssue(
                        category="隐私泄露",
                        risk_level=ComplianceRisk.HIGH,
                        description=f"模型输出包含 {pii_type}",
                        regulation="GDPR Art. 32 / PIPL 第 51 条",
                        recommendation="添加输出过滤器，移除 PII"
                    ))

            # 检查毒性内容
            toxicity_score = self._calculate_toxicity(output)
            if toxicity_score > 0.8:
                issues.append(ComplianceIssue(
                    category="内容安全",
                    risk_level=ComplianceRisk.HIGH,
                    description=f"模型输出毒性评分过高 ({toxicity_score:.2f})",
                    regulation="EU AI Act Annex III / 中国深度合成管理规定",
                    recommendation="优化安全过滤层或调整模型对齐"
                ))

        self.issues.extend(issues)
        return issues

    def check_transparency(self, system_info: Dict) -> List[ComplianceIssue]:
        """检查透明度合规"""
        issues = []

        if not system_info.get("ai_disclosure"):
            issues.append(ComplianceIssue(
                category="透明度",
                risk_level=ComplianceRisk.MEDIUM,
                description="未明确告知用户正在与 AI 交互",
                regulation="EU AI Act Art. 52 / FTC AI 指南",
                recommendation="在用户界面添加 AI 交互提示"
            ))

        if not system_info.get("content_watermark"):
            issues.append(ComplianceIssue(
                category="内容标识",
                risk_level=ComplianceRisk.MEDIUM,
                description="AI 生成内容未添加水印或标识",
                regulation="EU AI Act Art. 50 / 中国深度合成管理规定第 16 条",
                recommendation="为 AI 生成内容添加不可见水印"
            ))

        self.issues.extend(issues)
        return issues

    def _calculate_toxicity(self, text: str) -> float:
        """简化的毒性评分计算（实际应使用专业模型）"""
        toxic_keywords = ["hate", "violence", "discrimination", "abuse"]
        score = sum(1 for kw in toxic_keywords if kw.lower() in text.lower())
        return min(score / len(toxic_keywords), 1.0)

    def generate_report(self) -> str:
        """生成合规报告"""
        if not self.issues:
            return "✅ 未发现合规问题"

        report = "📋 AI 合规检查报告\n"
        report += "=" * 50 + "\n\n"

        # 按风险等级分组
        for risk in ComplianceRisk:
            risk_issues = [i for i in self.issues if i.risk_level == risk]
            if risk_issues:
                report += f"【{risk.value}】({len(risk_issues)} 项)\n"
                for i, issue in enumerate(risk_issues, 1):
                    report += f"  {i}. {issue.category}: {issue.description}\n"
                    report += f"     📜 依据: {issue.regulation}\n"
                    report += f"     💡 建议: {issue.recommendation}\n\n"

        # 风险评分
        risk_score = sum(
            10 if i.risk_level == ComplianceRisk.HIGH else
            5 if i.risk_level == ComplianceRisk.MEDIUM else
            1
            for i in self.issues
        )
        status = "❌ 不合规" if risk_score > 20 else "⚠️ 部分合规" if risk_score > 5 else "✅ 基本合规"
        report += f"\n总体评估: {status} (风险评分: {risk_score}/100)"

        return report

# ===== 使用示例 =====
if __name__ == "__main__":
    checker = AIComplianceChecker(jurisdiction="EU")

    # 检查训练数据
    training_data = [
        {
            "name": "Common Crawl 子集",
            "license": None,
            "contains_pii": True,
            "bias_score": 0.65,
        },
        {
            "name": "维基百科",
            "license": "CC-BY-SA 4.0",
            "contains_pii": False,
            "bias_score": 0.3,
        },
    ]
    checker.check_training_data(training_data)

    # 检查模型输出
    sample_outputs = [
        "用户的邮箱是 test@example.com",
        "我认为这个群体不应该...",
    ]
    checker.check_model_output(sample_outputs)

    # 检查透明度
    system_info = {
        "ai_disclosure": False,
        "content_watermark": False,
    }
    checker.check_transparency(system_info)

    # 生成报告
    print(checker.generate_report())