第9章 大模型的伦理安全与隐私9.3 隐私保护技术9.3.1 数据匿名化与脱敏

    import random
    
    def data_anonymization(data, sensitive_column):
       # Step 1: Choose a sorting method
       sorted_data = sorted(data, key=lambda x: x[sensitive_column])
    
       # Step 2: Record the index before sorting
       index = [i for i in range(len(sorted_data))]
    
       # Step 3: Replace the sensitive attribute with new values
       anonymized_data = []
       for i in range(len(sorted_data)):
       temp_dict = dict(sorted_data[i])
       temp_dict[sensitive_column] = index[i]
       anonymized_data.append(temp_dict)
    
       return anonymized_data

    from sklearn.cluster import KMeans
    
    def data_anonymization(data, sensitive_column):
       # Step 1: Cluster analysis
       kmeans = KMeans(n_clusters=3)
       kmeans.fit(data[:, :-1])
       clusters = kmeans.predict(data[:, :-1])
    
       # Step 2: Replace the sensitive attribute with cluster centers or random values
       anonymized_data = []
       for i in range(len(clusters)):
       temp_dict = dict(data[i])
       if random.random() < 0.5:
           temp_dict[sensitive_column] = kmeans.cluster_centers_[clusters[i]]
       else:
           temp_dict[sensitive_column] = random.randint(1, 10)
       anonymized_data.append(temp_dict)
    
       return anonymized_data

    from sklearn.tree import DecisionTreeClassifier
    
    def data_anonymization(data, sensitive_column):
       # Step 1: Decision tree analysis
       clf = DecisionTreeClassifier()
       clf.fit(data[:, :-1], data[:, -1])
       decision_tree = clf.tree_
    
       # Step 2: Replace the sensitive attribute with decision tree nodes or random values
       anonymized_data = []
       for i in range(len(decision_tree.nodes)):
       node = decision_tree.nodes[i]
       if node.feature == sensitive_column:
           temp_dict = dict(data[node.index])
           if random.random() < 0.5:
               temp_dict[sensitive_column] = node.value
           else:
               temp_dict[sensitive_column] = random.randint(1, 10)
           anonymized_data.append(temp_dict)
    
       return anonymized_data

    import base64
    from Crypto.Cipher import AES
    
    def data_encryption(data, key):
       cipher = AES.new(key, AES.MODE_EAX)
       ciphertext, tag = cipher.encrypt_and_digest(data)
       nonce = cipher.nonce
       encrypted_data = base64.b64encode(nonce + ciphertext + tag).decode()
       return encrypted_data
    
    def data_decryption(encrypted_data, key):
       encrypted_data = base64.b64decode(encrypted_data.encode())
       nonce = encrypted_data[:16]
       ciphertext = encrypted_data[16:-16]
       tag = encrypted_data[-16:]
       cipher = AES.new(key, AES.MODE_EAX, nonce=nonce)
       plaintext = cipher.decrypt_and_verify(ciphertext, tag)
       return plaintext

    from Crypto.PublicKey import RSA
    
    def data_encryption(data, public_key):
       key = RSA.importKey(public_key)
       cipher = key.encrypt(data, 32)
       encrypted_data = ''.join(map(lambda x: chr(x), cipher))
       return encrypted_data
    
    def data_decryption(encrypted_data, private_key):
       private_key = RSA.importKey(private_key)
       cipher = [ord(char) for char in encrypted_data]
       plaintext = private_key.decrypt(cipher)
       return plaintext

    import random
    
    def data_disturbance(data, sensitive_column):
       anonymized_data = []
       for i in range(len(data)):
       temp_dict = dict(data[i])
       if temp_dict[sensitive_column] % 2 == 0:
           temp_dict[sensitive_column] += random.randint(-5, 5)
       else:
           temp_dict[sensitive_column] -= random.randint(-5, 5)
       anonymized_data.append(temp_dict)
    
       return anonymized_data

    from collections import Counter
    
    def data_summary(data, sensitive_column):
       anonymized_data = []
       counter = Counter([x[sensitive_column] for x in data])
       for value in counter:
       count = counter[value]
       temp_dict = {}
       temp_dict['value'] = value
       temp_dict['count'] = count
       anonymized_data.append(temp_dict)
    
       return anonymized_data

    import pandas as pd
    
    data = [{'name': 'Alice', 'age': 25, 'gender': 'F'},
       {'name': 'Bob', 'age': 30, 'gender': 'M'},
       {'name': 'Charlie', 'age': 35, 'gender': 'M'},
       {'name': 'David', 'age': 40, 'gender': 'M'},
       {'name': 'Eve', 'age': 45, 'gender': 'F'}]
    
    # Method 1: Data anonymization using permutation
    df = pd.DataFrame(data)
    sensitive_column = 'age'
    df[sensitive_column] = df[sensitive_column].apply(lambda x: df[sensitive_column].index[df[sensitive_column] == x].tolist()[0])
    print('Original Data:')
    print(df)
    df = df.sample(frac=1).reset_index(drop=True)
    df[sensitive_column] = df[sensitive_column].apply(lambda x: x + 10)
    print('\nAnonymized Data (Method 1):')
    print(df)
    
    # Method 2: Data anonymization using k-anonymity
    from pyclustering.cluster.kmedoids import kmedoids
    
    def k_anonymity(data, sensitive_column, k):
       clusters = kmedoids().process(data, k)
       centroids = clusters.get_centroids()
       anonymized_data = []
       for centroid in centroids:
       sub_data = data[np.where(np.array(list(map(tuple, data))) == tuple(centroid))[0]]
       sensitive_values = list(set([x[sensitive_column] for x in sub_data]))
       for sensitive_value in sensitive_values:
           new_sub_data = []
           for row in sub_data:
               if row[sensitive_column] == sensitive_value:
                  new_row = row.copy()
                  del new_row[sensitive_column]
                  new_sub_data.append(new_row)
           new_sub_data = pd.DataFrame(new_sub_data)
           random_row = new_sub_data.sample(1).iloc[0]
           anonymized_row = row.copy()
           anonymized_row[sensitive_column] = sensitive_value
           anonymized_data.append(anonymized_row)
       return pd.DataFrame(anonymized_data)
    
    df = pd.DataFrame(data)
    sensitive_column = 'age'
    k = 2
    print('\nAnonymized Data (Method 2):')
    print(k_anonymity(df.values.tolist(), sensitive_column, k))

    import pandas as pd
    from Crypto.Cipher import AES
    
    data = [{'name': 'Alice', 'age': 25, 'salary': 5000},
       {'name': 'Bob', 'age': 30, 'salary': 6000},
       {'name': 'Charlie', 'age': 35, 'salary': 7000},
       {'name': 'David', 'age': 40, 'salary': 8000},
       {'name': 'Eve', 'age': 45, 'salary': 9000}]
    
    # Method 1: Data encryption using AES
    key = b'This is a key123456'
    iv = b'This is an IV456'
    encryptor = AES.new(key, AES.MODE_CFB, iv)
    def data_encryption(data, sensitive_column):
       encrypted_data = []
       for i in range(len(data)):
       temp_dict = dict(data[i])
       encrypted_temp_dict = {}
       encrypted_temp_dict['name'] = temp_dict['name']
       encrypted_temp_dict['age'] = temp_dict['age']
       encrypted_temp_dict[sensitive_column] = base64.b64encode(encryptor.encrypt(str(temp_dict[sensitive_column]).encode())).decode()
       encrypted_data.append(encrypted_temp_dict)
       return encrypted_data
    
    df = pd.DataFrame(data)
    sensitive_column = 'salary'
    print('Original Data:')
    print(df)
    print('\nEncrypted Data:')
    print(pd.DataFrame(data_encryption(df.values.tolist(), sensitive_column)))
    
    # Method 2: Data obfuscation using noise addition
    def data_obfuscation(data, sensitive_column):
       anonymized_data = []
       for i in range(len(data)):
       temp_dict = dict(data[i])
       anonymized_temp_dict = {}
       anonymized_temp_dict['name'] = temp_dict['name']
       anonymized_temp_dict['age'] = temp_dict['age']
       anonymized_temp_dict[sensitive_column] = temp_dict[sensitive_column] * (1 + random.uniform(-0.1, 0.1))
       anonymized_data.append(anonymized_temp_dict)
       return anonymized_data
    
    df = pd.DataFrame(data)
    sensitive_column = 'salary'
    print('\nAnonymized Data:')
    print(pd.DataFrame(data_obfuscation(df.values.tolist(), sensitive_column)))