Spaces:
Sleeping
Sleeping
| """문서 익명화 + 등급강등 (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, | |
| } | |