Data Anonymization skill for Claude Code

#!/usr/bin/env python3
"""Scan CSV files for likely PII, column by column (stdlib only).

Usage:
  python3 pii_scan.py data.csv [more.csv ...] [--rows 5000] [--threshold 0.05] [--fail-on-pii] [--json]

For each column it samples up to --rows values and reports:
  - value detectors that fired (email, phone, card with Luhn, IBAN with mod-97,
    US SSN, IPv4/IPv6, date-like, postcode-like) and the share of non-empty values matching
  - header hints (name, address, dob, ...) from the column name
  - a suggested treatment

Exit codes: 0 = scan finished (or no PII with --fail-on-pii), 2 = PII found with
--fail-on-pii (use on anonymized output in CI), 1 = usage or read error.
"""
import argparse
import csv
import ipaddress
import json
import re
import sys

csv.field_size_limit(10_000_000)

EMAIL = re.compile(r"^[A-Za-z0-9._%+'-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$")
PHONE = re.compile(r"^\+?[\d\s().-]{7,20}$")
SSN = re.compile(r"^(?!000|666|9\d\d)\d{3}-(?!00)\d{2}-(?!0000)\d{4}$")
CARD = re.compile(r"^(?:\d[ -]?){13,19}$")
IBAN = re.compile(r"^[A-Z]{2}\d{2}[A-Z0-9]{11,30}$")
DATE = re.compile(r"^(\d{4}-\d{2}-\d{2}|\d{1,2}[/.]\d{1,2}[/.]\d{2,4})([ T]\d{2}:\d{2}(:\d{2})?)?$")
POSTCODE = re.compile(r"^(\d{5}(-\d{4})?|[A-Z]{1,2}\d[A-Z\d]? ?\d[A-Z]{2}|[A-Z]\d[A-Z] ?\d[A-Z]\d)$", re.I)

HEADER_HINTS = {
    "name": ["name", "first", "last", "surname", "fullname", "given"],
    "contact": ["email", "e_mail", "phone", "mobile", "tel", "fax"],
    "address": ["address", "street", "city", "zip", "postcode", "postal", "addr"],
    "birth": ["dob", "birth", "birthday", "age"],
    "government_id": ["ssn", "nin", "passport", "license", "licence", "tax_id", "taxid", "national"],
    "financial": ["iban", "card", "account", "acct", "bic", "swift", "salary", "income"],
    "network": ["ip", "ip_address", "device_id", "mac", "user_agent", "cookie"],
    "location": ["lat", "lon", "lng", "latitude", "longitude", "geo"],
    "health": ["diagnosis", "medical", "health", "condition", "medication"],
    "free_text": ["note", "notes", "comment", "comments", "description", "message", "body"],
}

SUGGEST = {
    "email": "pseudonymize (HMAC) or replace with synthetic unique emails",
    "phone": "mask or replace with synthetic numbers",
    "card": "drop; never copy card numbers out of the PCI scope",
    "iban": "drop or tokenize",
    "ssn": "drop",
    "ip": "truncate (IPv4 /24, IPv6 /48) or drop",
    "date": "generalize (year or month) if it is a birth or event date tied to a person",
    "postcode": "generalize (first 3 chars / 3-digit ZIP) and check k-anonymity",
}


def luhn_ok(s: str) -> bool:
    digits = [int(c) for c in s if c.isdigit()]
    if not 13 <= len(digits) <= 19:
        return False
    total = 0
    for i, d in enumerate(reversed(digits)):
        if i % 2 == 1:
            d *= 2
            if d > 9:
                d -= 9
        total += d
    return total % 10 == 0


def iban_ok(s: str) -> bool:
    s = s.replace(" ", "").upper()
    if not IBAN.match(s):
        return False
    moved = s[4:] + s[:4]
    num = "".join(str(int(c, 36)) for c in moved)
    return int(num) % 97 == 1


def is_ip(s: str) -> bool:
    try:
        ipaddress.ip_address(s)
        return True
    except ValueError:
        return False


def detect(v: str) -> set:
    hits = set()
    if EMAIL.match(v):
        hits.add("email")
    if SSN.match(v):
        hits.add("ssn")
    if CARD.match(v) and luhn_ok(v):
        hits.add("card")
    if iban_ok(v):
        hits.add("iban")
    if is_ip(v):
        hits.add("ip")
    if DATE.match(v):
        hits.add("date")
    if POSTCODE.match(v) and not v.isdigit() or re.fullmatch(r"\d{5}-\d{4}", v):
        hits.add("postcode")
    digits = sum(c.isdigit() for c in v)
    if not hits & {"card", "date", "ip"} and PHONE.match(v) and 7 <= digits <= 15 and not v.isdigit():
        hits.add("phone")
    return hits


def header_hints(col: str) -> list:
    tokens = re.split(r"[^a-z0-9]+", col.lower())
    found = []
    for cat, words in HEADER_HINTS.items():
        if any(t in words for t in tokens) or any(w in col.lower() for w in words if len(w) > 4):
            found.append(cat)
    return found


def scan(path: str, rows: int, threshold: float) -> list:
    with open(path, newline="", encoding="utf-8-sig", errors="replace") as fh:
        sample = fh.read(65536)
        fh.seek(0)
        try:
            dialect = csv.Sniffer().sniff(sample, delimiters=",;\t|")
        except csv.Error:
            dialect = csv.excel
        reader = csv.DictReader(fh, dialect=dialect)
        cols = reader.fieldnames or []
        counts = {c: {} for c in cols}
        nonempty = {c: 0 for c in cols}
        distinct = {c: set() for c in cols}
        n = 0
        for row in reader:
            n += 1
            for c in cols:
                v = (row.get(c) or "").strip()
                if not v:
                    continue
                nonempty[c] += 1
                if len(distinct[c]) < 10001:
                    distinct[c].add(v)
                for h in detect(v):
                    counts[c][h] = counts[c].get(h, 0) + 1
            if n >= rows:
                break
    report = []
    for c in cols:
        ne = nonempty[c] or 1
        detectors = {k: round(v / ne, 3) for k, v in sorted(counts[c].items(), key=lambda kv: -kv[1]) if v / ne >= threshold}
        hints = header_hints(c)
        uniq = len(distinct[c]) / ne if nonempty[c] else 0
        risk = "high" if detectors.keys() & {"email", "ssn", "card", "iban", "phone", "ip"} else (
            "medium" if hints or detectors else "low")
        if risk == "low" and uniq > 0.9 and nonempty[c] > 50:
            risk = "check"  # near-unique column: possible identifier
        suggestion = "; ".join(SUGGEST[d] for d in detectors if d in SUGGEST)
        if not suggestion and hints:
            suggestion = "review: header suggests " + ", ".join(hints)
        report.append({"file": path, "column": c, "rows_sampled": n, "non_empty": nonempty[c],
                       "distinct_ratio": round(uniq, 3), "detectors": detectors,
                       "header_hints": hints, "risk": risk, "suggestion": suggestion})
    return report


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("files", nargs="+")
    ap.add_argument("--rows", type=int, default=5000)
    ap.add_argument("--threshold", type=float, default=0.05, help="min share of values matching a detector")
    ap.add_argument("--fail-on-pii", action="store_true")
    ap.add_argument("--json", action="store_true")
    a = ap.parse_args()
    results = []
    for f in a.files:
        try:
            results.extend(scan(f, a.rows, a.threshold))
        except OSError as e:
            print(f"error: {f}: {e}", file=sys.stderr)
            return 1
    if a.json:
        print(json.dumps(results, indent=2))
    else:
        order = {"high": 0, "medium": 1, "check": 2, "low": 3}
        print(f"{'risk':7} {'column':28} {'detectors':34} hints / suggestion")
        for r in sorted(results, key=lambda r: (order[r["risk"]], r["file"], r["column"])):
            det = ",".join(f"{k}:{v:.0%}" for k, v in r["detectors"].items()) or "-"
            extra = r["suggestion"] or ("near-unique values; possible identifier" if r["risk"] == "check" else "")
            print(f"{r['risk']:7} {r['column'][:28]:28} {det[:34]:34} {extra}")
        flagged = sum(r["risk"] in ("high", "medium") for r in results)
        print(f"\n{flagged} of {len(results)} columns flagged high/medium.")
    if a.fail_on_pii and any(r["risk"] == "high" for r in results):
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main())