PII Detection

This section demonstrates how to detect potential PII (characteristics) with Metadata in ydata-sdk, inspect the result, and adjust it before generating synthetic data. See PII identification for how the detection works and what to expect from it.

Example Code

"""Detect potential PII (characteristics) in a dataset and review the result before synthesis."""
import random

import pandas as pd
from faker import Faker

from ydata.characteristics import ColumnCharacteristic
from ydata.dataset import Dataset
from ydata.metadata import Metadata


def create_customers(size: int = 500) -> pd.DataFrame:
    fake = Faker("en_US")
    fake.seed_instance(42)
    random.seed(42)
    return pd.DataFrame({
        "customer_ref": [fake.bothify("CUS-#######") for _ in range(size)],
        "full_name": [fake.name() for _ in range(size)],
        "email": [fake.email() if random.random() > 0.1 else "unknown" for _ in range(size)],
        "phone": [fake.phone_number() for _ in range(size)],
        "city": [fake.city() for _ in range(size)],
        "website": [fake.url() for _ in range(size)],
        "segment": [random.choice(["Retail", "SME", "Corporate"]) for _ in range(size)],
        "balance": [round(random.uniform(0, 10_000), 2) for _ in range(size)],
    })


if __name__ == "__main__":
    dataset = Dataset(create_customers())

    # 1. Detect characteristics while computing the metadata.
    #    Detection uses the values, the column names (en, pt, es, fr, de, it, nl)
    #    and the metadata itself (unique columns are tagged as `id`).
    metadata = Metadata(dataset, infer_characteristics=True)

    for column, details in metadata.columns.items():
        if details.characteristics:
            print(column, sorted(c.value for c in details.characteristics))
    # customer_ref ['id']
    # full_name ['id', 'person']
    # email ['email', 'id']
    # phone ['id', 'phone']
    # city ['id', 'location']
    # website ['id', 'url']

    # 2. Inspect how each characteristic was found.
    #    `value` is the share of the column's values that matched (placeholders such as "unknown" are ignored),
    #    `source` is "values", "column_name", "metadata" (id) or "manual".
    for detail in metadata.summary["characteristics"]["phone"]:
        print(detail)
    # {'characteristic': <ColumnCharacteristic.PHONE: 'phone'>, 'value': 1, 'upper_bound': 1, 'lower_bound': 1,
    #  'source': 'column_name'}
    # {'characteristic': <ColumnCharacteristic.ID: 'id'>, 'value': 1, ..., 'source': 'metadata'}

    # 3. Fix the result: detected characteristics are suggestions to review.
    metadata.add_characteristics({"segment": ColumnCharacteristic.PII})
    metadata.remove_characteristics({"website": "url"})

    # 4. Declare known characteristics upfront. They are kept and merged with the detected ones.
    metadata = Metadata(
        dataset,
        infer_characteristics=True,
        characteristics={"segment": "pii", "customer_ref": ["id", "pii"]},
    )

    # 5. Tune the detection.
    #    - characteristics_language: spaCy model used for names and places ("en" by default; also "pt", "es",
    #      "fr", "de", "it", "nl" and "xx"). Non-English models are downloaded on first use (~500 MB).
    #    - characteristics_threshold: share of a column's values that must match (default 0.8).
    #    - characteristics_score_thresholds: minimum detection score per characteristic (default 0.8, URL 0.6).
    fake = Faker()
    mixed = Dataset(pd.DataFrame({
        "col_a": [fake.domain_name() for _ in range(300)],                              # domains, no https://
        "col_b": [fake.email() if i % 10 < 6 else fake.word() for i in range(300)],     # 60% emails
    }))
    strict = Metadata(mixed, infer_characteristics=True)
    lenient = Metadata(mixed, infer_characteristics=True,
                       characteristics_threshold=0.5,
                       characteristics_score_thresholds={"url": 0.5})
    print({c: sorted(x.value for x in m.characteristics) for c, m in strict.columns.items()})
    # {'col_a': ['id'], 'col_b': ['id']}
    print({c: sorted(x.value for x in m.characteristics) for c, m in lenient.columns.items()})
    # {'col_a': ['id', 'url'], 'col_b': ['email', 'id']}

    # 6. Run the detection later on an existing Metadata, without recomputing it.
    metadata = Metadata(dataset)
    metadata.compute_characteristics(dataset)