خصوصية البيانات والأمان: دليل المطور للتعامل مع البيانات الحساسة باستخدام DuckDB

import duckdb

import pandas as pd

​

def demonstrate_data_redaction():

    # Create a connection

    conn = duckdb.connect(':memory:')

    # Create sample data with various sensitive information

    conn.execute("""

        CREATE TABLE sensitive_info AS SELECT * FROM (

            VALUES

                ('John Doe', '[email protected]', 'CC: 4532-1234-5678-9012', 'Normal text'),

                ('Jane Smith', '[email protected]', 'SSN: 123-45-6789', 'Some notes'),

                ('Bob Wilson', '[email protected]', 'Password: SecretPass123!', 'Regular info'),

                ('Alice Brown', '[email protected]', 'API_KEY=abc123xyz', 'Basic text')

            ) AS t(name, email, sensitive_field, normal_text);

    """)

    # Define redaction rules

    redaction_rules = {

        'email': r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}',  # Email pattern

        'sensitive_field': r'(CC:\s*\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}|SSN:\s*\d{3}-\d{2}-\d{4}|Password:\s*\S+|API_KEY=\S+)',  # Various sensitive patterns

        'name': r'[A-Z][a-z]+ [A-Z][a-z]+'  # Full name pattern

    }

    # Show original data

    print("Original Data:")

    print(conn.execute("SELECT * FROM sensitive_info").fetchdf())

    # Apply redaction

    redact_sensitive_data(conn, 'sensitive_info', redaction_rules)

    # Show redacted data

    print("\nRedacted Data:")

    print(conn.execute("SELECT * FROM redacted_data").fetchdf())

    return conn

​

def redact_sensitive_data(conn, table_name, rules):

    """

    Redact sensitive data based on specified patterns.

    Parameters:

    - conn: DuckDB connection

    - table_name: Name of the table containing sensitive data

    - rules: Dictionary of column names and their corresponding regex patterns to match sensitive data

    """

    redaction_cases = []

    # This creates a CASE statement for each column

    # If the pattern matches, the value is redacted

    # If not, the original value is kept

    for column, pattern in rules.items():

        redaction_cases.append(f"""

            CASE 

                WHEN regexp_matches({column}, '{pattern}') 

                THEN '(REDACTED)' 

                ELSE {column} 

            END as {column}

        """)

    query = f"""

    CREATE TABLE redacted_data AS

    SELECT 

        {', '.join(redaction_cases)}

    FROM {table_name};

    """

    conn.execute(query)

​

# Example with custom redaction patterns

def demonstrate_custom_redaction():

    conn = duckdb.connect(':memory:')

    # Create sample data

    conn.execute("""

        CREATE TABLE customer_data AS SELECT * FROM (

            VALUES

                ('John Doe', '123-45-6789', 'ACC#12345', '$5000'),

                ('Jane Smith', '987-65-4321', 'ACC#67890', '$3000'),

                ('Bob Wilson', '456-78-9012', 'ACC#11111', '$7500')

            ) AS t(name, ssn, account, balance);

    """)

    # Define custom redaction rules with different patterns

    custom_rules = {

        'name': {

            'pattern': r'[A-Z][a-z]+ [A-Z][a-z]+',

            'replacement': lambda match: f"{match[0][0]}*** {match[0].split()[1][0]}***"

        },

        'ssn': {

            'pattern': r'\d{3}-\d{2}-\d{4}',

            'replacement': 'XXX-XX-XXXX'

        },

        'account': {

            'pattern': r'ACC#\d{5}',

            'replacement': 'ACC#*****'

        }

    }

    def apply_custom_redaction(conn, table_name, rules):

        redaction_cases = []

        for column, rule in rules.items():

            redaction_cases.append(f"""

                CASE 

                    WHEN regexp_matches({column}, '{rule['pattern']}') 

                    THEN '{rule['replacement']}'

                    ELSE {column} 

                END as {column}

            """)

        query = f"""

        CREATE TABLE custom_redacted AS

        SELECT 

            {', '.join(redaction_cases)},

            balance  -- Keep this column unchanged

        FROM {table_name};

        """

        conn.execute(query)

    # Show original data

    print("\nOriginal Customer Data:")

    print(conn.execute("SELECT * FROM customer_data").fetchdf())

    # Apply custom redaction

    apply_custom_redaction(conn, 'customer_data', custom_rules)

    # Show results

    print("\nCustom Redacted Data:")

    print(conn.execute("SELECT * FROM custom_redacted").fetchdf())

​

# Run demonstrations

print("=== Basic Redaction Demo ===")

demonstrate_data_redaction()

print("\n=== Custom Redaction Demo ===")

demonstrate_custom_redaction()