"""문서 익명화 + 등급강등 (SPEC §15) ========================================== 설계 의도 --------- - PII findings + entity 별 정책(JSON) → 익명화된 텍스트. - 6가지 처리 방법: mask / remove / replace / generalize / shift / round. - **일관성(consistent)** — 같은 원본은 같은 placeholder 로 (문서 내 cross-reference 보존). - **익명화 → 재분석 → 다시 익명화** 사이클로 target 등급까지 자동 강등. - 매핑 테이블은 평문으로 영속 가능 (옵션 — node-seal HE 암호화는 차후). 사용 흐름 --------- 1) load_policy() / save_policy(p) — anonymization_policy.json 2) apply_policy(text, findings, policy) → {anonymized_text, replacements, mapping} 3) downgrade_to_target(text, findings, policy, classify_fn, analyze_fn) — target 등급 도달 시까지 반복. """ from __future__ import annotations import json import re from datetime import date, timedelta from pathlib import Path POLICY_FILE = Path(__file__).resolve().parent / "anonymization_policy.json" DEFAULT_POLICY: dict = { "version": "anon-policy-v1", "target_grade": "S", # 익명화 후 도달 목표 등급 "max_iterations": 3, # 강등 사이클 최대 반복 "entities": { # 직접식별자 — 강한 마스킹/제거 "KR_RRN": {"method": "mask", "pattern": "******-*******"}, "KR_PASSPORT": {"method": "mask", "pattern": "********"}, "KR_BIZ_NO": {"method": "mask", "pattern": "***-**-*****"}, "CREDIT_CARD": {"method": "mask", "preserve_last": 4}, "AWS_ACCESS_KEY": {"method": "remove"}, "GENERIC_API_KEY": {"method": "remove"}, "US_SSN": {"method": "mask", "pattern": "***-**-****"}, "IBAN_CODE": {"method": "mask", "preserve_last": 4}, # 약한 식별자 — 이름 유지로 가독성 보존 "KR_PHONE": {"method": "mask", "preserve_last": 4}, "PHONE_NUMBER": {"method": "mask", "preserve_last": 4}, "EMAIL_ADDRESS": {"method": "replace", "format": "[EMAIL_%d]", "consistent": True}, # NER 식별자 — 일관성 있는 placeholder (문서 내부 같은 이름은 같은 번호) "PERSON": {"method": "replace", "format": "[PERSON_%d]", "consistent": True}, "LOCATION": {"method": "replace", "format": "[LOCATION_%d]", "consistent": True}, "ORGANIZATION": {"method": "replace", "format": "[ORG_%d]", "consistent": True}, "VIP_NAMES": {"method": "replace", "format": "[VIP_%d]", "consistent": True}, "INTERNAL_PROJECTS": {"method": "replace", "format": "[PROJECT_%d]", "consistent": True}, # 주소/시간 — 일반화 / 시프트 "KR_ADDRESS": {"method": "generalize", "level": 1}, "DATE_TIME": {"method": "shift", "max_days": 30}, # 기타 "URL": {"method": "replace", "format": "[URL]"}, "IP_ADDRESS": {"method": "mask", "pattern": "***.***.***.***"}, # 일본 PII — APPI / マイナンバー法 대응 # マイナンバー: 마이넘버법 §19 利用目的外保管禁止 → 마스킹 불가, 완전 제거 "JP_MY_NUMBER": {"method": "remove", "_law": "マイナンバー法 §19"}, "JP_PASSPORT": {"method": "mask", "pattern": "*********"}, "JP_DRIVERS_LICENSE": {"method": "mask", "pattern": "************"}, "JP_PHONE": {"method": "mask", "preserve_last": 4}, "JP_POSTAL_CODE": {"method": "generalize", "level": 1, "_note": "앞 3자리만 유지"}, "JP_ADDRESS": {"method": "generalize", "level": 1}, "JP_CORPORATE_NUMBER": {"method": "mask", "pattern": "*************"}, "JP_BANK_ACCOUNT": {"method": "mask", "preserve_last": 4}, }, } # --------------------------------------------------------------------------- # 정책 영속 # --------------------------------------------------------------------------- def load_policy() -> dict: if not POLICY_FILE.exists(): save_policy(DEFAULT_POLICY) return dict(DEFAULT_POLICY) try: cur = json.loads(POLICY_FILE.read_text(encoding="utf-8")) # 새 entity 추가됐을 때 기본값 병합 merged_entities = {**DEFAULT_POLICY["entities"], **cur.get("entities", {})} return {**DEFAULT_POLICY, **cur, "entities": merged_entities} except Exception: return dict(DEFAULT_POLICY) def save_policy(p: dict) -> dict: POLICY_FILE.write_text( json.dumps(p, ensure_ascii=False, indent=2), encoding="utf-8", ) return p # --------------------------------------------------------------------------- # 익명화 적용 # --------------------------------------------------------------------------- def apply_policy(text: str, findings: list[dict], policy: dict | None = None) -> dict: """findings + policy → 익명화된 텍스트 + 변경 내역. Returns: anonymized_text: str replacements: [{start_orig, end_orig, original, replacement, entity_type, method}] mapping: {"PERSON|김철수": "[PERSON_1]", ...} # 재현가능성/감사용 stats: {entity_type: count, ...} """ pol = policy or load_policy() ent_pol = pol.get("entities", {}) # 같은 위치/엔티티 중복 제거 + 시작순 오름차순 dedup: dict[tuple, dict] = {} for f in findings or []: key = (f.get("start"), f.get("end"), f.get("entity_type")) if not all(x is not None for x in key[:2]): continue prev = dedup.get(key) if prev is None or (f.get("score", 0) > prev.get("score", 0)): dedup[key] = f sorted_f = sorted(dedup.values(), key=lambda f: f["start"]) counters: dict[str, int] = {} consistent_map: dict[tuple[str, str], str] = {} replacements: list[dict] = [] stats: dict[str, int] = {} # 좌→우 순회하며 결과 문자열을 새로 만듭니다 (offset 추적 단순화) out: list[str] = [] cursor = 0 for f in sorted_f: et = f.get("entity_type") if et not in ent_pol: continue start = f["start"]; end = f["end"] if start < cursor: # 겹치는 매치는 skip continue original = f.get("text") or text[start:end] cfg = ent_pol[et] ph = _compute_placeholder(cfg, et, original, counters, consistent_map) out.append(text[cursor:start]) out.append(ph) cursor = end replacements.append({ "start_orig": start, "end_orig": end, "original": original, "replacement": ph, "entity_type": et, "method": cfg.get("method", "mask"), }) stats[et] = stats.get(et, 0) + 1 out.append(text[cursor:]) anonymized = "".join(out) return { "anonymized_text": anonymized, "replacements": replacements, "mapping": {f"{k[0]}|{k[1]}": v for k, v in consistent_map.items()}, "stats": stats, "policy_version": pol.get("version", "anon-policy-v1"), "target_grade": pol.get("target_grade"), } def _compute_placeholder(cfg: dict, entity_type: str, original: str, counters: dict, consistent_map: dict) -> str: method = cfg.get("method", "mask") if method == "mask": return _mask(original, cfg) if method == "remove": return "" if method == "replace": fmt = cfg.get("format", f"[{entity_type}_%d]") consistent = bool(cfg.get("consistent", False)) if consistent: key = (entity_type, original) if key in consistent_map: return consistent_map[key] counters[entity_type] = counters.get(entity_type, 0) + 1 ph = (fmt % counters[entity_type]) if "%d" in fmt else fmt consistent_map[key] = ph return ph counters[entity_type] = counters.get(entity_type, 0) + 1 return (fmt % counters[entity_type]) if "%d" in fmt else fmt if method == "generalize": return _generalize(original, int(cfg.get("level", 1))) if method == "shift": return _shift_date(original, int(cfg.get("max_days", 30))) if method == "round": return _round_money(original, str(cfg.get("precision", "1만"))) return "[REDACTED]" def _mask(original: str, cfg: dict) -> str: if cfg.get("pattern"): return cfg["pattern"] preserve = int(cfg.get("preserve_last", 0) or 0) if preserve > 0 and len(original) > preserve: return ("*" * (len(original) - preserve)) + original[-preserve:] return "*" * max(1, len(original)) _JP_POSTAL_RE = re.compile(r"〒?\s*(\d{3})-?\d{4}") # 일본 주소 행정구역 접미사 — 매치 직후 자르기 _JP_ADDR_BOUNDARIES = ("丁目", "番地", "番", "号", "町", "村", "区", "市", "郡", "県", "府", "都", "道") def _generalize(text: str, level: int) -> str: """level 토큰만큼 끝에서 잘라낸다. 다음 순서로 시도: 1) 일본 우편번호 패턴 → 앞 3자리만 (〒XXX-****) 2) 공백 분해 (한국 주소 등) 3) 일본 행정구역 접미사 (都/府/県/市/区/町/丁目/番地) 까지만 유지""" if not text: return text # 우편번호 — 앞 3자리만 m = _JP_POSTAL_RE.search(text) if m and len(text) <= 16: return f"〒{m.group(1)}-****" # 공백 분해 우선 parts = text.split() if len(parts) > 1: keep = max(1, len(parts) - max(0, level)) return " ".join(parts[:keep]) # 일본 주소 — 행정구역 접미사 기준 끝에서 level 단계 제거 # 예: 東京都千代田区千代田1-1 + level=2 → 東京都千代田区 boundaries: list[int] = [] for suf in _JP_ADDR_BOUNDARIES: i = 0 while True: i = text.find(suf, i) if i < 0: break boundaries.append(i + len(suf)) i += len(suf) boundaries = sorted(set(boundaries)) if not boundaries: return text # level 만큼 끝에서 자른 경계 target_idx = max(0, len(boundaries) - 1 - max(0, level)) return text[: boundaries[target_idx]] _DATE_RE = re.compile(r"(\d{4})[-/.](\d{1,2})[-/.](\d{1,2})") def _shift_date(text: str, max_days: int) -> str: """텍스트 안의 첫 YYYY-MM-DD 만 시프트 (deterministic — text hash 기반).""" m = _DATE_RE.search(text) if not m: return text days = (hash(text) % (2 * max_days + 1)) - max_days try: d = date(int(m.group(1)), int(m.group(2)), int(m.group(3))) d2 = d + timedelta(days=days) return text[: m.start()] + d2.isoformat() + text[m.end():] except ValueError: return text _NUM_RE = re.compile(r"[\d,]+") def _round_money(text: str, precision: str) -> str: m = _NUM_RE.search(text) if not m: return text try: val = int(m.group(0).replace(",", "")) except ValueError: return text if precision == "1억": rounded = round(val / 1e8) * 1e8 return text[: m.start()] + f"{int(rounded/1e8)}억" + text[m.end():] if precision == "1천만": rounded = round(val / 1e7) * 1e7 return text[: m.start()] + f"{int(rounded/1e7)*10}백만" + text[m.end():] if precision == "1만": rounded = round(val / 1e4) * 1e4 return text[: m.start()] + f"{int(rounded/1e4):,}만" + text[m.end():] return text # --------------------------------------------------------------------------- # 강등 사이클 — anonymize → re-analyze → if not target, again # --------------------------------------------------------------------------- GRADE_RANK = {"O": 0, "S": 1, "C": 2} def downgrade_to_target( text: str, findings: list[dict], classify_fn, # callable(text) -> classification dict (grade/score/...) analyze_fn=None, # callable(text) -> findings — 없으면 1회만 적용 policy: dict | None = None, ) -> dict: pol = policy or load_policy() target = pol.get("target_grade", "S") max_iter = int(pol.get("max_iterations", 3)) target_rank = GRADE_RANK.get(target, 1) cur_text = text cur_findings = list(findings or []) history: list[dict] = [] last_result: dict | None = None initial = classify_fn(cur_text) history.append({ "iter": 0, "kind": "initial", "grade": initial.get("grade"), "score": initial.get("score"), "n_findings": len(cur_findings), "n_chars": len(cur_text), }) if GRADE_RANK.get(initial.get("grade"), 0) <= target_rank: return { "ok": True, "achieved": True, "final_text": cur_text, "final_grade": initial.get("grade"), "final_score": initial.get("score"), "iterations": history, "applied_replacements": [], "mapping": {}, "stats": {}, "n_iterations": 0, "policy_version": pol.get("version"), "target_grade": target, "note": "이미 target 등급 — 익명화 불필요", } for i in range(1, max_iter + 1): last_result = apply_policy(cur_text, cur_findings, pol) cur_text = last_result["anonymized_text"] if analyze_fn: cur_findings = analyze_fn(cur_text) else: cur_findings = [] # 다시 분석 안 하면 두 번째 이터레이션은 사실상 no-op re_class = classify_fn(cur_text) history.append({ "iter": i, "kind": "after_anonymize", "grade": re_class.get("grade"), "score": re_class.get("score"), "n_findings": len(cur_findings), "n_chars": len(cur_text), "n_replacements": len(last_result["replacements"]), }) if GRADE_RANK.get(re_class.get("grade"), 0) <= target_rank: return { "ok": True, "achieved": True, "final_text": cur_text, "final_grade": re_class.get("grade"), "final_score": re_class.get("score"), "iterations": history, "applied_replacements": last_result["replacements"], "mapping": last_result["mapping"], "stats": last_result["stats"], "n_iterations": i, "policy_version": pol.get("version"), "target_grade": target, } final_class = classify_fn(cur_text) return { "ok": True, "achieved": GRADE_RANK.get(final_class.get("grade"), 0) <= target_rank, "final_text": cur_text, "final_grade": final_class.get("grade"), "final_score": final_class.get("score"), "iterations": history, "applied_replacements": (last_result or {}).get("replacements", []), "mapping": (last_result or {}).get("mapping", {}), "stats": (last_result or {}).get("stats", {}), "n_iterations": max_iter, "policy_version": pol.get("version"), "target_grade": target, "note": "max_iterations 도달 — 정책 강화 또는 추가 항목 검토 필요" if not (last_result or {}) else None, }