101 lines
3.0 KiB
Python
101 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
anonymize_buyers.py - Anonymisiert PII-Felder in buyers_raw.jsonl.
|
|
|
|
Ersetzt folgende Felder durch realistische Faker-Dummydaten:
|
|
name_company, prospective_buyer, company, phone, cell, email, address
|
|
Leere Felder (null / "" / []) bleiben leer.
|
|
|
|
Konsistenz: Gleicher Originalwert -> gleicher Dummy (ueber alle Zeilen).
|
|
So bleiben Personen ueber mehrere Sheets hinweg zusammenfuehrbar, ohne
|
|
echte Namen preiszugeben.
|
|
|
|
Alle uebrigen Felder (types_of_business, dates, checkbox, background etc.)
|
|
bleiben UNVERAENDERT - die sind fuer die Analyse relevant und nicht sensibel.
|
|
|
|
Aufruf:
|
|
python3 anonymize_buyers.py ./poc_out/buyers_raw.jsonl ./poc_out/buyers_raw_anonym.jsonl
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
|
|
try:
|
|
from faker import Faker
|
|
except ImportError:
|
|
print("Faker fehlt. Installieren mit: pip3 install faker --break-system-packages")
|
|
sys.exit(1)
|
|
|
|
fake = Faker("en_US")
|
|
Faker.seed(1234) # reproduzierbar
|
|
|
|
# Zu anonymisierende Felder
|
|
PII_FIELDS = ["name_company", "prospective_buyer", "company",
|
|
"phone", "cell", "email", "address"]
|
|
|
|
|
|
def is_empty(v):
|
|
return v in (None, "", [], {})
|
|
|
|
|
|
# Konsistenz-Caches: gleicher Originalwert -> gleicher Dummy
|
|
_caches = {f: {} for f in PII_FIELDS}
|
|
|
|
|
|
def fake_value(field):
|
|
"""Erzeugt einen feldgerechten Dummy."""
|
|
if field in ("name_company", "prospective_buyer"):
|
|
return fake.name()
|
|
if field == "company":
|
|
return fake.company()
|
|
if field in ("phone", "cell"):
|
|
return fake.phone_number()
|
|
if field == "email":
|
|
return fake.email()
|
|
if field == "address":
|
|
# einzeilige Adresse
|
|
return fake.address().replace("\n", ", ")
|
|
return fake.word()
|
|
|
|
|
|
def anonymize_field(field, value):
|
|
if is_empty(value):
|
|
return value # leer bleibt leer
|
|
key = str(value).strip().lower()
|
|
cache = _caches[field]
|
|
if key not in cache:
|
|
cache[key] = fake_value(field)
|
|
return cache[key]
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 3:
|
|
print("Aufruf: python3 anonymize_buyers.py <input.jsonl> <output.jsonl>")
|
|
sys.exit(1)
|
|
inp, outp = sys.argv[1], sys.argv[2]
|
|
|
|
n_lines = 0
|
|
n_fields_anonymized = 0
|
|
|
|
with open(inp, encoding="utf-8") as fin, open(outp, "w", encoding="utf-8") as fout:
|
|
for line in fin:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
rec = json.loads(line)
|
|
for field in PII_FIELDS:
|
|
if field in rec and not is_empty(rec[field]):
|
|
rec[field] = anonymize_field(field, rec[field])
|
|
n_fields_anonymized += 1
|
|
fout.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
|
n_lines += 1
|
|
|
|
print(f"Verarbeitet: {n_lines} Zeilen")
|
|
print(f"Anonymisierte Felder: {n_fields_anonymized}")
|
|
print(f"Distinkte Ersetzungen: " +
|
|
", ".join(f"{f}={len(_caches[f])}" for f in PII_FIELDS))
|
|
print(f"Ausgabe: {outp}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |