K-Anonymity System: Data Anonymization with Sensitivity Attribute Generalization

import pandas as pd
from tkinter import *
from tkinter import filedialog

# 定义K-AnonymitySystem类
class KAnonymitySystem:

    def validate_sensitivity_attributes(self):
        sensitivity_attributes = self.sensitivity_entry.get().split(",")
        sensitivity_attributes = [attr.strip() for attr in sensitivity_attributes]  # 去除空格
        self.sensitive_attributes = [attr for attr in sensitivity_attributes if attr in self.df.columns]  # 筛选出有效属性
        self.sensitivity_entry.delete(0, END)
        self.sensitivity_entry.insert(0, ", ".join(self.sensitive_attributes))

    def __init__(self, root):
        # 初始化窗口
        self.root = root
        self.root.title("K-Anonymity System")
        self.root.geometry("500x500")

        # 初始化变量
        self.file_path = ""
        self.df = None
        self.sensitive_attributes = []
        self.k = 0
        self.anonymized_df = None

        # 创建控件
        self.file_label = Label(self.root, text="未查找到文件.")
        self.file_label.pack()

        self.select_file_button = Button(self.root, text="插入文件", command=self.select_file)
        self.select_file_button.pack()

        self.preprocess_button = Button(self.root, text="运行程序", command=self.preprocess_data, state=DISABLED)
        self.preprocess_button.pack()

        self.sensitivity_label = Label(self.root, text="输入敏感数据 (逗号分割):", state=DISABLED)
        self.sensitivity_label.pack()

        self.sensitivity_entry = Entry(self.root, state=DISABLED)
        self.sensitivity_entry.pack()

        self.k_label = Label(self.root, text="输入k值：", state=DISABLED)
        self.k_label.pack()

        self.k_entry = Entry(self.root, state=DISABLED)
        self.k_entry.pack()

        self.anonymize_button = Button(self.root, text="匿名数据集", command=self.anonymize_data, state=DISABLED)
        self.anonymize_button.pack()

        self.save_button = Button(self.root, text="保存匿名数据集", command=self.save_anonymized_data, state=DISABLED)
        self.save_button.pack()

        self.result_text = Text(self.root, state=DISABLED, height=20)
        self.result_text.pack()

    # 选择文件函数
    def select_file(self):
        self.file_path = filedialog.askopenfilename(filetypes=[("CSV Files", "*.csv")])
        self.file_label.config(text=self.file_path)
        self.preprocess_button.config(state=NORMAL)

    # 预处理数据函数
    def preprocess_data(self):
        # 读取数据集
        self.df = pd.read_csv(self.file_path)
        # 去重
        self.df.drop_duplicates(inplace=True)
        # 删除缺失值
        self.df.dropna(inplace=True)
        # 启用控件
        self.sensitivity_label.config(state=NORMAL)
        self.sensitivity_entry.config(state=NORMAL)
        self.k_label.config(state=NORMAL)
        self.k_entry.config(state=NORMAL)
        self.anonymize_button.config(state=NORMAL)

    # 匿名化数据函数
    def anonymize_data(self):
        # 获取敏感属性和k值
        self.sensitive_attributes = self.sensitivity_entry.get().split(",")
        self.k = int(self.k_entry.get())

        # 初始化匿名数据集
        self.anonymized_df = pd.DataFrame(columns=self.df.columns)

        # 对每个敏感属性组进行处理
        for group_name, group_data in self.df.groupby(self.sensitive_attributes):
            # 如果组大小小于k值，直接添加到匿名数据集中
            if len(group_data) < self.k:
                self.anonymized_df = pd.concat([self.anonymized_df, group_data])
            # 如果组大小大于等于k值，进行匿名化处理
            else:
                # 复制一份组的数据，用于匿名化处理
                group_data_copy = group_data.copy()
                # 对所有属性进行泛化处理
                group_data_copy = self.generalize_data(group_data_copy)
                # 删除重复行
                group_data_copy.drop_duplicates(inplace=True)
                # 将数据集划分为k个簇
                group_data_copy["cluster"] = pd.cut(group_data_copy.index, bins=self.k, labels=False)
                group_data_copy["cluster"] += min(group_data_copy["cluster"])
                # 将划分后的数据集添加到组的数据中
                group_data_copy.drop(self.sensitive_attributes, axis=1, inplace=True)
                # 对非敏感属性进行泛化处理
                group_data_copy = self.generalize_data(group_data_copy)
                # 将处理后的数据添加到匿名数据集中，并重新添加敏感属性列
                group_data_copy = pd.concat([group_data[self.sensitive_attributes], group_data_copy], axis=1)
                # 将处理后的数据添加到匿名数据集中
                self.anonymized_df = pd.concat([self.anonymized_df, group_data_copy])

        # 计算匿名化的准确率
        accuracy = self.calculate_accuracy()
        # 在文本框中显示匿名数据集和准确率
        self.result_text.config(state=NORMAL)
        self.result_text.delete(1.0, END)
        self.result_text.insert(END, "Anonymized Data:

")
        self.result_text.insert(END, self.anonymized_df.to_string(index=False))
        self.result_text.insert(END, f"\n\nAccuracy: {accuracy}")
        self.result_text.config(state=DISABLED)
        # 启用保存按钮
        self.save_button.config(state=NORMAL)

    # 保存匿名数据集函数
    def save_anonymized_data(self):
        save_path = filedialog.asksaveasfilename(defaultextension=".csv", filetypes=[("CSV Files", "*.csv")])
        self.anonymized_df.to_csv(save_path, index=False)

    # 泛化数据函数
    def generalize_data(self, group_data):
        for column in group_data.columns:
            column_data = group_data[column]
            # 对于object类型的列，将最后一个字符替换为*
            if column_data.dtype == "object":
                group_data[column] = column_data.str[:-1] + "*"
            # 对于数值类型的列，将其分成10个区间进行泛化
            else:
                min_value = column_data.min()
                max_value = column_data.max()
                step = (max_value - min_value) / 10
                group_data[column] = pd.cut(column_data, bins=[min_value + step*i for i in range(11)], labels=[f"{min_value + step*i:.2f}-{min_value + step*(i+1):.2f}" for i in range(10)])
        return group_data

    # 计算匿名化准确率函数
    def calculate_accuracy(self):
        # 计算原始数据集中每个敏感属性组的大小
        accuracy_df = self.df.groupby(self.sensitive_attributes).size().reset_index(name="actual_count")
        # 计算匿名数据集中每个敏感属性组的大小
        accuracy_df = pd.merge(accuracy_df, self.anonymized_df.groupby(self.sensitive_attributes).size().reset_index(name="anonymized_count"), on=self.sensitive_attributes, how="left")
        # 将缺失值填充为0
        accuracy_df.fillna(0, inplace=True)
        # 计算每个敏感属性组的匿名化准确率
        accuracy_df["accuracy"] = accuracy_df["anonymized_count"] / accuracy_df["actual_count"]
        # 计算平均准确率
        accuracy = accuracy_df["accuracy"].mean()
        return accuracy

# 创建主窗口
root = Tk()
# 创建K-AnonymitySystem对象
k_anonymity_system = KAnonymitySystem(root)
# 进入主循环
root.mainloop()

代码说明：

1. 用户界面: 使用 tkinter 创建一个简单的用户界面，允许用户选择 CSV 文件、输入敏感属性和 k 值。
2. 数据预处理: 读取 CSV 文件，进行去重和删除缺失值操作，以确保数据质量。
3. 匿名化:
   • 分组: 根据敏感属性将数据分成不同的组。
   • 泛化: 对所有属性进行泛化处理，将敏感属性进行模糊化，并根据 k 值将数据划分成簇。
   • 合并: 将处理后的分组数据合并成一个匿名化数据集。
4. 准确率计算: 通过比较原始数据集和匿名化数据集中的敏感属性组的大小，计算匿名化准确率。
5. 输出: 在界面上显示匿名化数据集和准确率。

使用方法：

1. 运行 Python 代码。
2. 点击 "插入文件" 按钮选择一个 CSV 文件。
3. 在 "输入敏感数据" 文本框中输入敏感属性名称，以逗号分隔。
4. 在 "输入 k 值" 文本框中输入一个 k 值。
5. 点击 "匿名数据集" 按钮进行匿名化操作。
6. 匿名化后的数据集和准确率将显示在文本框中。
7. 点击 "保存匿名数据集" 按钮保存匿名化后的数据集。

特点：

• 敏感属性泛化: 对所有敏感属性进行泛化处理，以保护敏感信息。
• 可自定义 k 值: 用户可以根据需要调整 k 值，以控制匿名化程度。
• 准确率计算: 提供匿名化准确率的计算，以评估匿名化效果。
• 易于使用: 友好的用户界面，方便用户操作。

改进建议：

• 添加更多泛化方法，例如，对数值型属性使用基于范围的泛化，对字符串类型属性使用基于长度的泛化等。
• 提供数据可视化功能，例如，使用直方图或散点图展示匿名化前后的数据分布。
• 添加更多指标来评估匿名化效果，例如，信息损失度和数据可用性等。
• 考虑使用更高级的匿名化方法，例如，l-diversity 和 t-closeness，以提供更强的隐私保护。