不可接受风险（禁止）
├── 政府对公民的社会信用评分
├── 实时远程生物特征识别（执法场景）
└── 利用弱点的潜意识操控

高风险（严格监管）
├── 关键基础设施（交通、能源）
├── 教育和职业训练评估
├── 就业、工人管理
├── 基本私人和公共服务（信贷评分、医疗）
└── 执法和司法

有限风险（透明度义务）
├── 与人交互的AI系统（须告知用户）
└── 生成式AI内容（须标注AI生成）

最低风险（自愿遵循）
└── 推荐系统、垃圾邮件过滤器等

import os
import base64
import json
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC

# 安装: pip install cryptography

class SecureDataStore:
    """
    安全数据存储类
    
    功能：
    - AES-128（通过Fernet实现）对称加密
    - 基于口令的密钥派生（PBKDF2）
    - 加密数据的安全存储和读取
    """
    
    def __init__(self, password: str, salt: bytes = None):
        """
        参数：
            password: 用于派生密钥的口令（应从环境变量读取）
            salt: 盐值（首次创建后必须保存，用于后续还原密钥）
        """
        self.salt = salt or os.urandom(16)
        self._key = self._derive_key(password, self.salt)
        self._fernet = Fernet(self._key)
    
    def _derive_key(self, password: str, salt: bytes) -> bytes:
        """使用PBKDF2从口令派生加密密钥"""
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=480000,  # NIST推荐值
        )
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        return key
    
    def encrypt(self, data: str) -> bytes:
        """加密字符串数据"""
        return self._fernet.encrypt(data.encode("utf-8"))
    
    def decrypt(self, encrypted_data: bytes) -> str:
        """解密数据"""
        return self._fernet.decrypt(encrypted_data).decode("utf-8")
    
    def encrypt_record(self, record: dict, sensitive_fields: list) -> dict:
        """
        选择性加密记录中的敏感字段
        
        只加密真正敏感的字段，其余字段保持明文（便于查询）
        这是一种常用的"部分加密"模式
        """
        encrypted_record = record.copy()
        for field in sensitive_fields:
            if field in encrypted_record:
                value = str(encrypted_record[field])
                encrypted_record[field] = {
                    "_encrypted": True,
                    "_data": self.encrypt(value).decode("latin-1")  # bytes转str存储
                }
        return encrypted_record
    
    def decrypt_record(self, encrypted_record: dict) -> dict:
        """解密记录中的加密字段"""
        decrypted_record = encrypted_record.copy()
        for key, value in decrypted_record.items():
            if isinstance(value, dict) and value.get("_encrypted"):
                encrypted_bytes = value["_data"].encode("latin-1")
                decrypted_record[key] = self.decrypt(encrypted_bytes)
        return decrypted_record


# 使用示例
# 实际生产中密码应从环境变量读取：os.environ["ENCRYPTION_KEY"]
store = SecureDataStore(password="your_secure_password_from_env")

user_record = {
    "user_id": "U001",
    "username": "zhang_wei",          # 不敏感，保持明文
    "phone": "13812345678",           # 敏感，需加密
    "id_card": "110101199001011234",  # 敏感，需加密
    "credit_score": 720,              # 不敏感（用于计算），保持明文
    "created_at": "2024-01-15"
}

sensitive_fields = ["phone", "id_card"]
encrypted = store.encrypt_record(user_record, sensitive_fields)
decrypted = store.decrypt_record(encrypted)

print("原始记录:", {k: v for k, v in user_record.items()})
print("\n加密后存储:", {k: str(v)[:50] + "..." if isinstance(v, dict) else v 
                      for k, v in encrypted.items()})
print("\n解密后:", {k: v for k, v in decrypted.items()})

import json
import hashlib
from datetime import datetime, timezone
from typing import Optional, Dict, Any
from dataclasses import dataclass, field, asdict
from enum import Enum

class AccessType(Enum):
    READ = "read"
    WRITE = "write"
    DELETE = "delete"
    EXPORT = "export"
    SHARE = "share"

@dataclass
class AuditLog:
    """审计日志条目"""
    timestamp: str
    user_id: str
    action: str
    resource_type: str
    resource_id: str
    access_type: str
    ip_address: str
    result: str  # "success" or "denied"
    details: Dict[str, Any] = field(default_factory=dict)
    log_hash: str = ""  # 防篡改哈希
    
    def __post_init__(self):
        # 计算哈希（用于防篡改验证）
        content = f"{self.timestamp}{self.user_id}{self.action}{self.resource_id}"
        self.log_hash = hashlib.sha256(content.encode()).hexdigest()[:16]


class AuditLogger:
    """
    合规审计日志系统
    
    满足要求：
    - PIPL第56条：数据处理记录至少保留3年
    - GDPR第30条：维护处理活动记录
    - 等保2.0：安全审计要求
    """
    
    def __init__(self, log_file_path: str = "/tmp/audit_log.jsonl"):
        self.log_file = log_file_path
        self._logs = []  # 内存缓存
    
    def log(
        self,
        user_id: str,
        action: str,
        resource_type: str,
        resource_id: str,
        access_type: AccessType,
        ip_address: str,
        result: str = "success",
        details: Optional[Dict] = None
    ) -> AuditLog:
        """记录一条审计日志"""
        log_entry = AuditLog(
            timestamp=datetime.now(timezone.utc).isoformat(),
            user_id=user_id,
            action=action,
            resource_type=resource_type,
            resource_id=resource_id,
            access_type=access_type.value,
            ip_address=ip_address,
            result=result,
            details=details or {}
        )
        
        self._logs.append(log_entry)
        
        # 写入文件（生产环境应使用结构化日志系统如ELK）
        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(asdict(log_entry), ensure_ascii=False) + "\n")
        
        return log_entry
    
    def query_user_access(self, user_id: str) -> list:
        """查询特定用户的所有访问记录（用于用户数据访问申请）"""
        return [log for log in self._logs if log.user_id == user_id]
    
    def query_resource_access(self, resource_id: str) -> list:
        """查询特定资源的所有访问记录（用于数据泄露调查）"""
        return [log for log in self._logs if log.resource_id == resource_id]
    
    def generate_compliance_report(self, start_date: str, end_date: str) -> Dict:
        """生成合规报告（定期向监管机构提交）"""
        filtered = [
            log for log in self._logs
            if start_date <= log.timestamp <= end_date
        ]
        
        return {
            "period": {"start": start_date, "end": end_date},
            "total_access_events": len(filtered),
            "by_access_type": {
                t.value: len([l for l in filtered if l.access_type == t.value])
                for t in AccessType
            },
            "denied_requests": len([l for l in filtered if l.result == "denied"]),
            "unique_users": len(set(l.user_id for l in filtered)),
            "delete_operations": len([l for l in filtered if l.access_type == "delete"]),
        }


# 使用示例
logger = AuditLogger()

# 记录各种操作
logger.log("admin_001", "查询用户信用报告", "user_credit", "U123456",
           AccessType.READ, "192.168.1.100",
           details={"query_purpose": "贷款审批", "loan_id": "L789"})

logger.log("system", "删除过期用户数据", "user_profile", "U000099",
           AccessType.DELETE, "internal",
           details={"deletion_reason": "用户申请被遗忘权", "request_id": "GDPR-2024-001"})

logger.log("agent_002", "尝试访问超权限数据", "financial_record", "U555555",
           AccessType.READ, "10.0.0.50", result="denied",
           details={"reason": "权限不足"})

# 生成报告
report = logger.generate_compliance_report("2024-01-01", "2024-12-31")
print("\n合规报告:", json.dumps(report, ensure_ascii=False, indent=2))

from typing import List, Dict, Optional
import json
import os

class RightToBeForgottenHandler:
    """
    被遗忘权处理器
    
    GDPR第17条 / PIPL第47条：用户有权要求删除其个人数据
    企业须在合理期限内（通常30天）完成删除并告知用户
    """
    
    def __init__(self):
        # 注册所有存储用户数据的系统
        self.data_stores: Dict[str, callable] = {}
        self.deletion_log: List[Dict] = []
    
    def register_data_store(self, store_name: str, delete_func: callable):
        """注册数据存储系统，每个系统提供一个删除函数"""
        self.data_stores[store_name] = delete_func
        print(f"已注册数据存储: {store_name}")
    
    def process_deletion_request(
        self, 
        user_id: str, 
        request_id: str,
        retention_exceptions: List[str] = None
    ) -> Dict:
        """
        处理用户的数据删除请求
        
        参数：
            user_id: 要删除数据的用户ID
            request_id: 申请单编号（用于追踪）
            retention_exceptions: 因法律要求需要保留的数据类型
                例如：["financial_records"]（财务记录需保留5年）
        
        返回：
            删除结果报告
        """
        retention_exceptions = retention_exceptions or []
        results = {}
        
        print(f"\n处理被遗忘权申请: {request_id} (用户: {user_id})")
        print("=" * 50)
        
        for store_name, delete_func in self.data_stores.items():
            # 检查是否有法律保留要求
            if any(exc in store_name for exc in retention_exceptions):
                results[store_name] = {
                    "status": "retained",
                    "reason": "法律合规要求保留，将在法定期限后删除",
                    "retention_end_date": "2029-12-31"
                }
                print(f"  {store_name}: ⏸ 保留（法律要求）")
                continue
            
            # 执行删除
            try:
                success = delete_func(user_id)
                results[store_name] = {
                    "status": "deleted" if success else "not_found",
                    "deleted_at": datetime.now(timezone.utc).isoformat()
                }
                status_icon = "✓" if success else "○"
                print(f"  {store_name}: {status_icon} {'已删除' if success else '无记录'}")
            except Exception as e:
                results[store_name] = {
                    "status": "error",
                    "error": str(e)
                }
                print(f"  {store_name}: ✗ 删除失败 - {e}")
        
        # 记录删除操作（合规审计）
        deletion_record = {
            "request_id": request_id,
            "user_id": user_id,
            "requested_at": datetime.now(timezone.utc).isoformat(),
            "results": results,
            "completed": all(
                r["status"] in ("deleted", "not_found", "retained")
                for r in results.values()
            )
        }
        self.deletion_log.append(deletion_record)
        
        # 生成用户通知
        deleted_count = sum(1 for r in results.values() if r["status"] == "deleted")
        retained_count = sum(1 for r in results.values() if r["status"] == "retained")
        
        user_notification = f"""
亲爱的用户：

您的数据删除申请（编号：{request_id}）已处理完成。

处理结果：
- 已删除数据系统：{deleted_count} 个
- 因法律合规保留：{retained_count} 个（将在法定期限后自动删除）

如有疑问，请联系我们的数据保护官（DPO）。
"""
        print(user_notification)
        return deletion_record


# 使用示例
handler = RightToBeForgottenHandler()

# 注册各个数据存储（模拟）
mock_data = {
    "user_profile_db": {"U001": {"name": "张伟", "phone": "138****5678"}},
    "chat_history_db": {"U001": ["对话记录1", "对话记录2"]},
    "financial_records_db": {"U001": {"loan_id": "L001", "amount": 50000}},
}

for db_name, db_data in mock_data.items():
    # 闭包捕获db_data
    def make_delete_func(data):
        def delete_func(uid):
            if uid in data:
                del data[uid]
                return True
            return False
        return delete_func
    handler.register_data_store(db_name, make_delete_func(db_data))

# 处理删除申请（财务记录因法律要求保留）
result = handler.process_deletion_request(
    user_id="U001",
    request_id="GDPR-2024-1025",
    retention_exceptions=["financial_records"]
)

from dataclasses import dataclass
from enum import Enum
from typing import Dict, List, Tuple
import re

class ContentCategory(Enum):
    """内容安全类别（对应《生成式AI管理办法》要求）"""
    SAFE = "safe"
    POLITICAL_SENSITIVE = "political_sensitive"
    VIOLENCE = "violence"
    ADULT_CONTENT = "adult_content"
    DISCRIMINATION = "discrimination"
    MISINFORMATION = "misinformation"
    PRIVACY_VIOLATION = "privacy_violation"

@dataclass
class ContentFilterResult:
    is_safe: bool
    category: ContentCategory
    confidence: float
    matched_rules: List[str]
    original_text: str
    filtered_text: str  # 过滤后的文本（如适用）

class ContentSafetyFilter:
    """
    内容安全过滤器
    
    满足：
    - 《生成式AI管理办法》第4条：不得生成违法内容
    - 《网络安全法》第12条：不得利用网络危害国家安全
    - 未成年人保护相关法规
    """
    
    # 各类别的关键词规则（生产环境需要更完整的规则库）
    FILTER_RULES = {
        ContentCategory.VIOLENCE: [
            r'(杀|打|伤|袭击).{0,5}(方法|步骤|教程)',
            r'暴力.{0,10}(如何|怎样)',
        ],
        ContentCategory.ADULT_CONTENT: [
            r'(色情|成人|pornography)',
            r'性.{0,5}(描写|内容|图片)',
        ],
        ContentCategory.DISCRIMINATION: [
            r'(种族|民族|性别|宗教).{0,10}(低劣|歧视|仇恨)',
        ],
        ContentCategory.PRIVACY_VIOLATION: [
            r'(公开|泄露|出售).{0,10}(个人信息|隐私)',
            r'(跟踪|监视|监控).{0,10}(他人|用户)',
        ],
    }
    
    def __init__(self, strict_mode: bool = False):
        """
        参数：
            strict_mode: 严格模式，true则任何匹配都拦截；
                         false则允许合理的教育/研究内容
        """
        self.strict_mode = strict_mode
        self._compiled_rules: Dict[ContentCategory, List] = {}
        
        for category, patterns in self.FILTER_RULES.items():
            self._compiled_rules[category] = [
                re.compile(p, re.IGNORECASE | re.UNICODE)
                for p in patterns
            ]
    
    def filter(self, text: str) -> ContentFilterResult:
        """对文本进行内容安全过滤"""
        matched_rules = []
        triggered_category = ContentCategory.SAFE
        
        for category, patterns in self._compiled_rules.items():
            for i, pattern in enumerate(patterns):
                if pattern.search(text):
                    matched_rules.append(f"{category.value}:{self.FILTER_RULES[category][i]}")
                    triggered_category = category
                    break
            if matched_rules:
                break
        
        is_safe = len(matched_rules) == 0
        confidence = 0.95 if matched_rules else 0.85
        
        # 内容替换（对轻微违规内容进行替换而非完全拦截）
        filtered_text = text
        if not is_safe and not self.strict_mode:
            for patterns in self._compiled_rules.values():
                for pattern in patterns:
                    filtered_text = pattern.sub("[内容已过滤]", filtered_text)
        
        return ContentFilterResult(
            is_safe=is_safe,
            category=triggered_category,
            confidence=confidence,
            matched_rules=matched_rules,
            original_text=text,
            filtered_text=filtered_text if not is_safe else text
        )
    
    def batch_filter(self, texts: List[str]) -> List[ContentFilterResult]:
        """批量过滤（用于训练数据清洗）"""
        results = []
        blocked_count = 0
        
        for text in texts:
            result = self.filter(text)
            results.append(result)
            if not result.is_safe:
                blocked_count += 1
        
        print(f"批量过滤完成: {len(texts)} 条，拦截 {blocked_count} 条 "
              f"({blocked_count/len(texts)*100:.1f}%)")
        return results
    
    def add_ai_watermark(self, text: str, model_id: str = "model-v1") -> str:
        """
        为AI生成内容添加水印标识
        满足《生成式AI管理办法》第9条要求
        
        注：真正的不可见水印需要专门的水印算法（如 AIGC水印技术）
        这里演示元数据标注方式
        """
        watermark = f"\n\n---\n[本内容由AI生成 | 模型: {model_id} | 时间: {datetime.now().strftime('%Y-%m-%d')}]"
        return text + watermark


# 使用示例
safety_filter = ContentSafetyFilter(strict_mode=False)

test_contents = [
    "今天天气真好，适合出去散步",
    "请问如何学习Python编程？",
    "如何泄露他人个人信息并出售",
]

print("内容安全过滤结果：")
print("=" * 60)
for content in test_contents:
    result = safety_filter.filter(content)
    status = "✓ 安全" if result.is_safe else f"✗ 拦截 [{result.category.value}]"
    print(f"\n内容: {content[:40]}")
    print(f"状态: {status}")
    if not result.is_safe:
        print(f"过滤后: {result.filtered_text[:60]}")

from dataclasses import dataclass
from typing import Optional, Set
from enum import Enum

class ConsentType(Enum):
    """用户同意类型"""
    NECESSARY = "necessary"        # 必要功能（不需要同意）
    ANALYTICS = "analytics"        # 数据分析
    MARKETING = "marketing"        # 营销
    AI_TRAINING = "ai_training"    # AI模型训练
    THIRD_PARTY = "third_party"    # 第三方共享

@dataclass
class UserPrivacySettings:
    """用户隐私设置（隐私设计原则：默认最大保护）"""
    user_id: str
    # 默认全部关闭（原则2：默认隐私）
    analytics_enabled: bool = False
    marketing_enabled: bool = False
    ai_training_enabled: bool = False
    third_party_sharing_enabled: bool = False
    
    def get_allowed_uses(self) -> Set[ConsentType]:
        """获取用户允许的数据使用方式"""
        allowed = {ConsentType.NECESSARY}
        if self.analytics_enabled:
            allowed.add(ConsentType.ANALYTICS)
        if self.marketing_enabled:
            allowed.add(ConsentType.MARKETING)
        if self.ai_training_enabled:
            allowed.add(ConsentType.AI_TRAINING)
        if self.third_party_sharing_enabled:
            allowed.add(ConsentType.THIRD_PARTY)
        return allowed
    
    def can_use_for(self, purpose: ConsentType) -> bool:
        """检查是否可以用于特定目的"""
        return purpose in self.get_allowed_uses()


class PrivacyAwareDataProcessor:
    """
    隐私感知数据处理器
    在每次数据处理前检查用户是否授权
    """
    
    def __init__(self):
        self._user_settings: Dict[str, UserPrivacySettings] = {}
    
    def update_user_consent(self, user_id: str, settings: UserPrivacySettings):
        """更新用户的隐私设置（响应用户同意变更）"""
        self._user_settings[user_id] = settings
        print(f"用户 {user_id} 同意设置已更新")
    
    def get_user_settings(self, user_id: str) -> UserPrivacySettings:
        """获取用户设置（未设置则返回最严格的默认值）"""
        return self._user_settings.get(user_id, UserPrivacySettings(user_id=user_id))
    
    def process_with_consent_check(
        self, 
        user_id: str, 
        data: dict, 
        purpose: ConsentType,
        process_func
    ):
        """
        带同意检查的数据处理
        
        在处理前验证用户授权，确保数据使用目的合法
        """
        settings = self.get_user_settings(user_id)
        
        if not settings.can_use_for(purpose):
            raise PermissionError(
                f"用户 {user_id} 未授权将数据用于 {purpose.value}。"
                f"请先获取用户同意。"
            )
        
        return process_func(data)
    
    def collect_data_with_minimization(self, full_data: dict, purpose: ConsentType) -> dict:
        """
        数据最小化采集：根据处理目的，只保留必要字段
        """
        # 不同目的所需的最少字段
        required_fields = {
            ConsentType.NECESSARY: ["user_id", "session_id"],
            ConsentType.ANALYTICS: ["user_id", "page_path", "duration", "device_type"],
            ConsentType.AI_TRAINING: ["interaction_text", "feedback_score"],  # 不含用户ID
            ConsentType.MARKETING: ["user_id", "preferred_categories", "last_active"],
        }
        
        allowed_fields = required_fields.get(purpose, [])
        minimized = {k: v for k, v in full_data.items() if k in allowed_fields}
        
        collected = set(full_data.keys())
        kept = set(minimized.keys())
        dropped = collected - kept
        
        if dropped:
            print(f"数据最小化：丢弃了 {len(dropped)} 个非必要字段: {dropped}")
        
        return minimized


# 演示隐私设计原则
processor = PrivacyAwareDataProcessor()

# 用户明确同意分析用途
user_settings = UserPrivacySettings(
    user_id="U001",
    analytics_enabled=True,    # 同意分析
    ai_training_enabled=False, # 不同意用于AI训练
)
processor.update_user_consent("U001", user_settings)

# 尝试用于分析（有授权）
try:
    raw_data = {
        "user_id": "U001", 
        "page_path": "/products",
        "duration": 45,
        "device_type": "mobile",
        "phone": "138****5678",    # 非必要字段
        "name": "张伟"              # 非必要字段
    }
    
    minimized = processor.collect_data_with_minimization(raw_data, ConsentType.ANALYTICS)
    processor.process_with_consent_check(
        "U001", minimized, ConsentType.ANALYTICS,
        lambda d: print(f"\n分析处理成功: {d}")
    )
except PermissionError as e:
    print(f"权限拒绝: {e}")

# 尝试用于AI训练（无授权）
try:
    processor.process_with_consent_check(
        "U001", raw_data, ConsentType.AI_TRAINING,
        lambda d: print("AI训练处理")
    )
except PermissionError as e:
    print(f"\n权限拒绝: {e}")