2026 技术观察：数据脱敏平台进入自动治理阶段，字段识别、访问分级和动态脱敏成为安全新重点

import json
import re
from datetime import datetime
from collections import defaultdict


DATA_TABLES = [
    {
        "table_name": "customer_profile",
        "fields": [
            "customer_id",
            "name",
            "phone",
            "id_card",
            "address",
            "city"
        ]
    },
    {
        "table_name": "employee_salary",
        "fields": [
            "employee_id",
            "employee_name",
            "department",
            "salary",
            "bank_account"
        ]
    },
    {
        "table_name": "order_detail",
        "fields": [
            "order_id",
            "customer_id",
            "amount",
            "receiver_phone",
            "receiver_address"
        ]
    }
]


SAMPLE_ROWS = {
    "customer_profile": {
        "name": "张三",
        "phone": "13812345678",
        "id_card": "310101199001011234",
        "address": "上海市浦东新区某路100号"
    },
    "employee_salary": {
        "employee_name": "李四",
        "salary": "25000",
        "bank_account": "6222020202020202020"
    },
    "order_detail": {
        "receiver_phone": "13987654321",
        "receiver_address": "杭州市西湖区某街道"
    }
}

SENSITIVE_RULES = {
    "phone": ["phone", "mobile", "tel"],
    "id_card": ["id_card", "identity"],
    "address": ["address"],
    "salary": ["salary", "income"],
    "bank_account": ["bank", "account"]
}


def detect_sensitive_field(field_name):
    lower_name = field_name.lower()

    for sensitive_type, keywords in SENSITIVE_RULES.items():
        for keyword in keywords:
            if keyword in lower_name:
                return {
                    "sensitive": 30654.t.kuaisou.com 
                    "sensitive_type": sensitive_type
                }

    return {
        "sensitive": False,
        "sensitive_type": None
    }

def mask_value(value, sensitive_type):
    text = str(value)

    if sensitive_type == "phone":
        return text[:3] + "****" + text[-4:]

    if sensitive_type == "id_card":
        return text[:6] + "********" + text[-4:]

    if sensitive_type == "address":
        return text[:3] + "****"

    if sensitive_type == "salary":
        return "***"

    if sensitive_type == "bank_account":
        return text[:4] + "********" + text[-4:]

    return "***"

ROLE_POLICIES = {
    "admin": {
        "view_raw": True
    },
    "data_analyst": {
        "view_raw": False
    },
    "developer": {
        "view_raw": False
    },
    "hr_manager": {
        "view_raw": True
    },
    "external_partner": {
        "view_raw": False
    }
}


def can_view_raw(user_role, sensitive_type):
    policy = ROLE_POLICIES.get(
        user_role,
        {
            "view_raw": False
        }
    )

    if user_role == "hr_manager" and sensitive_type in ["salary", "bank_account"]:
        return True

    return policy["view_raw"]

def apply_dynamic_masking(table_name, row, user_role):
    masked_row = {}

    for field, value in row.items():
        sensitive_result = detect_sensitive_field(field)

        if not sensitive_result["sensitive"]:
            masked_row[field] = value
            continue

        sensitive_type = sensitive_result["sensitive_type"]

        if can_view_raw(user_role, sensitive_type):
            masked_row[field] = value
        else:
            masked_row[field] = mask_value(
                value,
                sensitive_type
            )

    return {
        "table_name": table_name,
        "user_role": user_role,
        "data": masked_row
    }

ACCESS_LOGS = []


def write_access_log(user_id, user_role, table_name, fields):
    sensitive_fields = []

    for field in fields:
        result = detect_sensitive_field(field)

        if result["sensitive"]:
            sensitive_fields.append({
                "field": field,
                "sensitive_type": result["sensitive_type"]
            })

    risk_level = "normal"

    if user_role == "external_partner" and sensitive_fields:
        risk_level = "high"
    elif user_role == "developer" and len(sensitive_fields) >= 2:
        risk_level = "medium"

    log = {
        "user_id": user_id,
        "user_role": user_role,
        "table_name": table_name,
        "sensitive_fields": sensitive_fields,
        "risk_level":30549.t.kuaisou.com 
        "access_time": datetime.now().isoformat()
    }

    ACCESS_LOGS.append(log)

    return log

def run_data_masking_governance():
    user = {
        "user_id": "user_001",
        "role": "data_analyst"
    }

    table_name = "customer_profile"
    row = SAMPLE_ROWS[table_name]

    masked_result = apply_dynamic_masking(
        table_name,
        row,
        user["role"]
    )

    access_log = write_access_log(
        user["user_id"],
        user["role"],
        table_name,
        row.keys()
    )

    field_scan_results = []

    for table in DATA_TABLES:
        for field in table["fields"]:
            result = detect_sensitive_field(field)

            field_scan_results.append({
                "table_name": table["table_name"],
                "field": field,
                "sensitive": result["sensitive"],
                "sensitive_type": result["sensitive_type"]
            })

    report = {
        "report_name": "数据脱敏平台治理报告",
        "field_scan_results": field_scan_results,
        "masked_result": masked_result,
        "access_log": 30549.t.kuaisou.com 
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_data_masking_governance()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))