scripts/anonymize_sample.py

#!/usr/bin/env python3
"""obcon-hf — sample DB → anonymized seed SQL.

Reads from a live MariaDB (defaults to local 127.0.0.1:3306/obcondb) and writes
INSERT statements with PHI/PII scrubbed:

  · sites.name       → "Demo Site N"
  · users.authenticateId / emailWork / emailHome → "userN@demo.local"
  · users.name       → "User N"
  · users.hash       → MD5('Admin@2026')  (모든 사용자 동일 비밀번호로 통일)
  · devices.name / location → "Device N" / "loc-N"
  · GPS 좌표         → 한국 중심 근처에 ±0.5° 랜덤 지터
  · devicedata*      → 최근 7일치 중 사이트당 최대 N행으로 샘플링
  · 운영 로그 / 세션 / loginHistories / audit 등은 통째로 제외

요구사항: pip install pymysql  (이 스크립트만 추가 의존성).
"""

import argparse
import hashlib
import random
from datetime import datetime, timedelta

try:
    import pymysql
except ImportError:
    raise SystemExit("pymysql required: pip install pymysql")


# ---- PHI/PII 가 들어 있을 수 있어 통째로 제외하는 테이블 ----------------------
EXCLUDED_TABLES = {
    "loginHistories", "sessions", "auditLogs", "audits",
    "notifications", "tasks", "events", "messages",
    "smsHistories", "emailHistories",
    "thingplugs",
    # devicedata* 는 따로 샘플링
}

# ---- 사이트당 보존할 devicedata 행 개수 -----------------------------------
DEVICEDATA_PER_SITE_LIMIT = 200

# ---- 비결정성 막기 위해 시드 고정 -----------------------------------------
random.seed(20260101)

KR_CENTER_LAT = 36.5
KR_CENTER_LNG = 127.8
JITTER = 0.5  # 약 50km


def jittered_coord():
    return (
        round(KR_CENTER_LAT + random.uniform(-JITTER, JITTER), 6),
        round(KR_CENTER_LNG + random.uniform(-JITTER, JITTER), 6),
    )


def sql_literal(v):
    if v is None:
        return "NULL"
    if isinstance(v, bool):
        return "1" if v else "0"
    if isinstance(v, (int, float)):
        return str(v)
    if isinstance(v, (datetime,)):
        return "'" + v.strftime("%Y-%m-%d %H:%M:%S") + "'"
    if isinstance(v, bytes):
        return "0x" + v.hex()
    s = str(v).replace("\\", "\\\\").replace("'", "''")
    return "'" + s + "'"


def fetch_columns(cur, table):
    cur.execute(f"DESCRIBE `{table}`")
    return [row[0] for row in cur.fetchall()]


def list_tables(cur, dbname):
    cur.execute(
        "SELECT table_name FROM information_schema.tables "
        "WHERE table_schema=%s AND table_type='BASE TABLE' ORDER BY table_name",
        (dbname,),
    )
    return [r[0] for r in cur.fetchall()]


def anon_users(cur, out):
    cols = fetch_columns(cur, "users")
    admin_hash = hashlib.md5(b"Admin@2026").hexdigest()
    cur.execute(
        f"SELECT {','.join(f'`{c}`' for c in cols)} FROM users "
        "WHERE deleted=0 AND status='Active' "
        "ORDER BY isGlobalAdmin DESC, isAdmin DESC, id LIMIT 30"
    )
    rows = cur.fetchall()
    out.write(f"\n-- users (anonymized, {len(rows)} rows; password = Admin@2026)\n")
    for i, row in enumerate(rows, 1):
        d = dict(zip(cols, row))
        d["authenticateId"] = f"user{i}@demo.local" if i > 1 else "admin@demo"
        d["name"] = f"Demo User {i}" if i > 1 else "Demo Admin"
        d["hash"] = admin_hash
        for f in ("emailWork", "emailHome", "phoneMobile", "phoneWork",
                  "phoneFax", "address", "addressDetail", "postalCode",
                  "department", "title", "description"):
            if f in d and d[f] is not None:
                d[f] = "" if isinstance(d[f], str) else None
        values = ", ".join(sql_literal(d[c]) for c in cols)
        out.write(
            f"INSERT IGNORE INTO `users` (`{'`,`'.join(cols)}`) VALUES ({values});\n"
        )


def anon_sites(cur, out):
    cols = fetch_columns(cur, "sites")
    cur.execute(
        f"SELECT {','.join(f'`{c}`' for c in cols)} FROM sites "
        "WHERE deleted=0 AND status='Active' ORDER BY id LIMIT 20"
    )
    rows = cur.fetchall()
    out.write(f"\n-- sites (anonymized, {len(rows)} rows)\n")
    for i, row in enumerate(rows, 1):
        d = dict(zip(cols, row))
        d["name"] = f"Demo Site {i}"
        d["siteKey"] = f"{i:02d}"
        for f in ("address", "addressDetail", "description"):
            if f in d and d[f] is not None:
                d[f] = "" if isinstance(d[f], str) else None
        values = ", ".join(sql_literal(d[c]) for c in cols)
        out.write(
            f"INSERT IGNORE INTO `sites` (`{'`,`'.join(cols)}`) VALUES ({values});\n"
        )


def anon_devices(cur, out, keep_site_ids):
    cols = fetch_columns(cur, "devices")
    cur.execute(
        f"SELECT {','.join(f'`{c}`' for c in cols)} FROM devices "
        f"WHERE deleted=0 AND statusUse='Active' AND site IN ({','.join(str(s) for s in keep_site_ids)}) "
        "ORDER BY site, type, deviceKey"
    )
    rows = cur.fetchall()
    out.write(f"\n-- devices (anonymized GPS + names, {len(rows)} rows)\n")
    for i, row in enumerate(rows, 1):
        d = dict(zip(cols, row))
        d["name"] = f"Device {i}"
        if "location" in d:
            d["location"] = f"loc-{i}"
        if "description" in d:
            d["description"] = ""
        if "latitude" in d and "longitude" in d:
            lat, lng = jittered_coord()
            d["latitude"] = lat
            d["longitude"] = lng
            if "latitudeDetail" in d:
                d["latitudeDetail"] = lat
            if "longitudeDetail" in d:
                d["longitudeDetail"] = lng
        values = ", ".join(sql_literal(d[c]) for c in cols)
        out.write(
            f"INSERT IGNORE INTO `devices` (`{'`,`'.join(cols)}`) VALUES ({values});\n"
        )


def anon_devicedata(cur, out, table, keep_site_keys):
    cols = fetch_columns(cur, table)
    if "statusDatetime" not in cols or "siteKey" not in cols:
        return 0
    cutoff = (datetime.now() - timedelta(days=7)).strftime("%Y%m%d%H%M%S")
    placeholders = ",".join(["%s"] * len(keep_site_keys))
    cur.execute(
        f"SELECT {','.join(f'`{c}`' for c in cols)} FROM `{table}` "
        f"WHERE statusDatetime > %s AND siteKey IN ({placeholders}) "
        f"ORDER BY statusDatetime DESC LIMIT {DEVICEDATA_PER_SITE_LIMIT * len(keep_site_keys)}",
        [cutoff] + list(keep_site_keys),
    )
    rows = cur.fetchall()
    if not rows:
        return 0
    out.write(f"\n-- {table} (recent sample, {len(rows)} rows)\n")
    for row in rows:
        d = dict(zip(cols, row))
        values = ", ".join(sql_literal(d[c]) for c in cols)
        out.write(
            f"INSERT IGNORE INTO `{table}` (`{'`,`'.join(cols)}`) VALUES ({values});\n"
        )
    return len(rows)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=3306)
    ap.add_argument("--user", default="scada")
    ap.add_argument("--password", default="scada")
    ap.add_argument("--db", default="obcondb")
    ap.add_argument("--out", required=True)
    args = ap.parse_args()

    conn = pymysql.connect(
        host=args.host, port=args.port, user=args.user,
        password=args.password, database=args.db, charset="utf8mb4"
    )
    cur = conn.cursor()

    with open(args.out, "w", encoding="utf-8") as out:
        out.write("-- obcon-hf anonymized sample data\n")
        out.write(f"-- generated {datetime.now().isoformat(timespec='seconds')}\n")
        out.write("SET FOREIGN_KEY_CHECKS=0;\n")

        anon_sites(cur, out)
        # 위 sites INSERT 한 사이트들만 device + devicedata 에 포함
        cur.execute(
            "SELECT id, siteKey FROM sites WHERE deleted=0 AND status='Active' "
            "ORDER BY id LIMIT 20"
        )
        site_pairs = list(cur.fetchall())
        keep_site_ids = [s[0] for s in site_pairs]
        keep_site_keys = [s[1] for s in site_pairs if s[1]]
        anon_users(cur, out)
        anon_devices(cur, out, keep_site_ids)

        total = 0
        for table in list_tables(cur, args.db):
            if table.startswith("devicedata") and table.endswith("s"):
                total += anon_devicedata(cur, out, table, keep_site_keys)
        out.write(f"\n-- devicedata total inserts: {total}\n")
        out.write("SET FOREIGN_KEY_CHECKS=1;\n")

    print(f"[anon] wrote {args.out}")


if __name__ == "__main__":
    main()