AI助力合规：GDPR与CCPA智能检测

# 基于NLP的隐私法规合规检测示例
# 运行环境：Python 3.8+, pip install spacy transformers pandas scikit-learn

import spacy
import pandas as pd
import numpy as np
from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, accuracy_score
import re
import json

# 加载NLP模型
print("加载NLP模型...")
# 加载spaCy模型用于命名实体识别和文本处理
nlp = spacy.load("en_core_web_lg")
# 加载预训练的文本分类模型用于合规风险评估
compliance_classifier = pipeline("text-classification", model="distilbert-base-uncased-finetuned-sst-2-english")

# 模拟合规检测数据集生成
def generate_compliance_dataset(n_samples=1000):
    """
    生成模拟的合规检测数据集
    包含数据处理描述和对应的合规状态（符合/不符合GDPR/CCPA）
    """
    # 数据处理活动类型
    processing_activities = [
        "collect user personal data", 
        "process health information", 
        "share customer data with third parties",
        "store payment information", 
        "analyze user behavior", 
        "transfer data to third countries",
        "delete user accounts", 
        "update privacy policies", 
        "conduct security audits",
        "notify data breaches"
    ]
    
    # 合规状态标签
    compliance_labels = ["GDPR compliant", "GDPR non-compliant", "CCPA compliant", "CCPA non-compliant"]
    
    # 生成数据集
    data = []
    for i in range(n_samples):
        # 随机选择数据处理活动
        activity = random.choice(processing_activities)
        
        # 添加一些细节，使描述更真实
        details = [
            "without user consent", 
            "with explicit user consent", 
            "for marketing purposes",
            "for fraud detection", 
            "without proper security measures", 
            "with encryption",
            "without data retention policy", 
            "with clear data retention policy", 
            "without data subject access rights",
            "with data subject access rights"
        ]
        
        # 随机选择1-3个细节
        n_details = random.randint(1, 3)
        selected_details = random.sample(details, n_details)
        
        # 构建完整的描述
        description = f"{activity} {', '.join(selected_details)}"
        
        # 根据描述内容确定合规状态（简化的规则）
        # 实际应用中需要更复杂的规则或机器学习模型
        if "without user consent" in description or "without proper security measures" in description:
            compliance_status = random.choice(["GDPR non-compliant", "CCPA non-compliant"])
        elif "with explicit user consent" in description and "with encryption" in description:
            compliance_status = random.choice(["GDPR compliant", "CCPA compliant"])
        else:
            # 随机选择合规状态，但偏向于合规
            compliance_status = random.choices(
                compliance_labels, 
                weights=[0.4, 0.1, 0.4, 0.1], 
                k=1
            )[0]
        
        data.append({
            "id": i+1,
            "description": description,
            "compliance_status": compliance_status
        })
    
    # 创建DataFrame
    df = pd.DataFrame(data)
    return df

# 敏感数据识别函数
def identify_sensitive_data(text):
    """
    识别文本中的敏感数据
    """
    doc = nlp(text)
    
    # 定义敏感实体类型（根据GDPR和CCPA）
    sensitive_entity_types = {
        "PERSON": "个人身份信息",
        "ORG": "组织信息",
        "GPE": "地理位置信息",
        "DATE": "日期信息",
        "TIME": "时间信息",
        "MONEY": "财务信息",
        "CARDINAL": "数量信息",
        "EMAIL": "电子邮箱",
        "PHONE": "电话号码",
        "ID": "身份证号"
    }
    
    # 自定义正则表达式用于识别特定的敏感数据
    patterns = {
        "EMAIL": r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+",
        "PHONE": r"\+?[1-9]\d{1,14}",  # E.164国际电话号码格式
        "ID": r"\b\d{8,18}\b"  # 身份证号或其他ID
    }
    
    sensitive_data = []
    
    # 使用spaCy的命名实体识别
    for ent in doc.ents:
        if ent.label_ in sensitive_entity_types:
            sensitive_data.append({
                "text": ent.text,
                "type": sensitive_entity_types[ent.label_],
                "category": "PII"
            })
    
    # 使用正则表达式识别特定的敏感数据
    for pattern_type, pattern in patterns.items():
        matches = re.finditer(pattern, text)
        for match in matches:
            # 避免重复识别
            is_duplicate = False
            for item in sensitive_data:
                if item["text"] == match.group():
                    is_duplicate = True
                    break
            
            if not is_duplicate:
                sensitive_data.append({
                    "text": match.group(),
                    "type": sensitive_entity_types[pattern_type],
                    "category": "PII"
                })
    
    return sensitive_data

# 合规风险评估函数
def assess_compliance_risk(text):
    """
    评估文本描述的数据处理活动的合规风险
    """
    # 使用预训练模型进行情感分析（作为风险评估的一部分）
    result = compliance_classifier(text)[0]
    
    # 定义风险关键词
    high_risk_keywords = [
        "without user consent", "no consent", 
        "without proper security", "insecure",
        "data breach", "breach",
        "non-compliant", "violation",
        "unauthorized access", "unauthorized sharing"
    ]
    
    medium_risk_keywords = [
        "lack of policy", "no policy",
        "insufficient documentation", "poor documentation",
        "limited access controls", "weak controls",
        "incomplete consent", "partial consent"
    ]
    
    # 计算风险得分
    risk_score = 0
    
    # 基础得分基于情感分析结果
    if result["label"] == "NEGATIVE":
        risk_score = 50  # 负面情感基础分为50
    else:
        risk_score = 20  # 正面情感基础分为20
    
    # 根据关键词调整风险得分
    for keyword in high_risk_keywords:
        if keyword in text.lower():
            risk_score += 20
    
    for keyword in medium_risk_keywords:
        if keyword in text.lower():
            risk_score += 10
    
    # 限制风险得分在0-100之间
    risk_score = min(100, max(0, risk_score))
    
    # 确定风险等级
    if risk_score >= 70:
        risk_level = "高风险"
    elif risk_score >= 40:
        risk_level = "中风险"
    else:
        risk_level = "低风险"
    
    return {
        "risk_score": risk_score,
        "risk_level": risk_level,
        "sentiment": result["label"],
        "sentiment_score": result["score"]
    }

# 训练合规分类模型
def train_compliance_model(df):
    """
    训练机器学习模型用于合规状态分类
    """
    # 特征提取：使用TF-IDF
    vectorizer = TfidfVectorizer(max_features=5000, stop_words="english")
    X = vectorizer.fit_transform(df["description"])
    
    # 标签编码
    y = df["compliance_status"]
    
    # 划分训练集和测试集
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    # 训练随机森林分类器
    model = RandomForestClassifier(n_estimators=100, random_state=42)
    model.fit(X_train, y_train)
    
    # 评估模型
    y_pred = model.predict(X_test)
    accuracy = accuracy_score(y_test, y_pred)
    report = classification_report(y_test, y_pred)
    
    print(f"模型准确率: {accuracy:.4f}")
    print("分类报告:")
    print(report)
    
    return model, vectorizer

# 主函数
def main():
    # 1. 生成模拟合规检测数据
    print("生成模拟合规检测数据...")
    df = generate_compliance_dataset(n_samples=1000)
    print(f"数据集大小: {df.shape}")
    print("数据样例:")
    print(df.head())
    
    # 2. 敏感数据识别演示
    print("\n敏感数据识别演示...")
    # 模拟包含敏感数据的文本
    sample_text = "John Doe from New York shared his email john.doe@example.com and phone number +1234567890 with our service on 2024-01-15."
    print(f"示例文本: {sample_text}")
    
    # 识别敏感数据
    sensitive_data = identify_sensitive_data(sample_text)
    print("识别到的敏感数据:")
    for item in sensitive_data:
        print(f"- {item['text']} (类型: {item['type']}, 类别: {item['category']})")
    
    # 3. 合规风险评估演示
    print("\n合规风险评估演示...")
    # 模拟数据处理活动描述
    sample_activities = [
        "Collect user personal data without explicit consent and store it without encryption",
        "Process health information with explicit user consent and proper security measures",
        "Share customer data with third parties for marketing purposes without user authorization"
    ]
    
    for i, activity in enumerate(sample_activities):
        print(f"\n活动 {i+1}: {activity}")
        risk_assessment = assess_compliance_risk(activity)
        print(f"风险评估: {risk_assessment['risk_level']} (得分: {risk_assessment['risk_score']})")
        print(f"情感分析: {risk_assessment['sentiment']} (得分: {risk_assessment['sentiment_score']:.4f})")
    
    # 4. 训练合规分类模型
    print("\n训练合规分类模型...")
    model, vectorizer = train_compliance_model(df)
    
    # 5. 使用模型进行预测演示
    print("\n使用模型进行预测演示...")
    # 模拟新的合规检测文本
    new_activities = [
        "Collect user data with explicit consent and implement encryption for storage",
        "Share payment information with third parties without proper security measures",
        "Delete user accounts upon request and maintain clear audit logs"
    ]
    
    # 特征提取
    X_new = vectorizer.transform(new_activities)
    # 预测
    predictions = model.predict(X_new)
    
    for i, (activity, prediction) in enumerate(zip(new_activities, predictions)):
        print(f"\n活动 {i+1}: {activity}")
        print(f"预测合规状态: {prediction}")
    
    # 6. 生成合规报告示例
    print("\n生成合规报告示例...")
    # 模拟合规检测结果
    compliance_results = {
        "organization": "示例金融科技公司",
        "assessment_date": "2024-09-10",
        "total_processing_activities": 100,
        "compliant_activities": 75,
        "non_compliant_activities": 25,
        "high_risk_activities": 10,
        "medium_risk_activities": 15,
        "low_risk_activities": 75,
        "top_findings": [
            "缺少用户明确同意的情况下收集个人数据 (15起)",
            "未采取适当的安全措施存储敏感数据 (10起)",
            "未向数据主体提供访问和更正其个人数据的权利 (8起)"
        ],
        "recommendations": [
            "实施自动化的用户同意管理系统",
            "加强数据加密和访问控制措施",
            "建立数据主体权利请求处理流程"
        ]
    }
    
    # 生成JSON格式的报告
    report_json = json.dumps(compliance_results, indent=2)
    print("合规报告 (JSON格式):")
    print(report_json)
    
    print("\n合规检测演示完成！")

if __name__ == "__main__":
    import random
    main()