pseudo_framework.py

"""
파일/문서 가명화·익명화 PoC 프레임워크.

표준 기반:
  - ISO/IEC 20889:2018  Privacy enhancing data de-identification terminology
                        and classification of techniques
  - ISO/IEC 27559:2022  Privacy enhancing data de-identification framework
용어 기반:
  - W3C DPV 2.0 (Data Privacy Vocabulary) — dpv-pd:* compact IRI

규제 매트릭스:
  - KR  개인정보보호법 + 개인정보보호위원회 가명정보 처리 가이드라인
  - JP  個人情報保護法 (APPI) + 仮名加工情報・匿名加工情報の作成基準
  - US  HIPAA Safe Harbor (45 CFR §164.514(b)(2)) + CCPA/CPRA + NIST SP 800-188
  - EU  GDPR Art. 4(5), Recital 26 + EDPB Guidelines 01/2025 + WP29 Op. 05/2014

본 모듈은 PII Scanner 의 검출 결과(findings)를 입력으로 받아,
ISO 20889 의 기법 카탈로그에 따라 변환을 적용하고
관할별 준수 여부를 판정합니다 (PoC — 실 운영 시 법무 검토 필수).
"""
from __future__ import annotations

import hashlib
import hmac
import re
import secrets
from dataclasses import asdict, dataclass, field
from typing import Dict, List, Optional, Tuple


# =========================================================================
# DPV 매핑 — PII Scanner entity_type → DPV 2.0 personal data category
# =========================================================================
# 형식: entity_type → (dpv_concept, 해설)
DPV_CATEGORY: Dict[str, Tuple[str, str]] = {
    "KR_RRN":            ("dpv-pd:NationalIdentificationNumber", "직접식별자: 한국 주민등록번호"),
    "KR_PASSPORT":       ("dpv-pd:PassportNumber",                "직접식별자: 한국 여권번호"),
    "KR_PHONE":          ("dpv-pd:TelephoneNumber",               "직접식별자: 한국 휴대폰"),
    "KR_BIZ_NO":         ("dpv-pd:Identifier",                    "법인 식별자 (사업자등록번호)"),
    "KR_ADDRESS":        ("dpv-pd:HomeAddress",                   "준식별자: 한국 주소"),
    "EMAIL_ADDRESS":     ("dpv-pd:EmailAddress",                  "직접식별자: 이메일"),
    "PHONE_NUMBER":      ("dpv-pd:TelephoneNumber",               "직접식별자: 일반 전화번호"),
    "CREDIT_CARD":       ("dpv-pd:CreditCardNumber",              "민감/금융: 신용카드"),
    "US_SSN":            ("dpv-pd:NationalIdentificationNumber",  "직접식별자: 미국 SSN"),
    "URL":               ("dpv-pd:URL",                           "준식별자(가능): URL"),
    "IP_ADDRESS":        ("dpv-pd:IPAddress",                     "준식별자/Tracking"),
    "IBAN_CODE":         ("dpv-pd:BankAccount",                   "민감/금융: IBAN"),
    "VIP_PERSON":        ("dpv-pd:Name",                           "직접식별자: 자연인 이름"),
    "INTERNAL_PROJECT":  ("dpv:NonPersonalData",                  "내부 코드명 (PII 아님)"),
    "AWS_ACCESS_KEY":    ("dpv:NonPersonalData",                  "비밀: 자격증명"),
    "GENERIC_API_KEY":   ("dpv:NonPersonalData",                  "비밀: 자격증명 후보"),
    # 일본 PII — APPI / マイナンバー法
    "JP_MY_NUMBER":         ("dpv-pd:NationalIdentificationNumber", "個人識別符号: マイナンバー (12자리). 별도법(マイナンバー法) 적용"),
    "JP_PASSPORT":          ("dpv-pd:PassportNumber",                "個人識別符号: 일본 여권번호"),
    "JP_DRIVERS_LICENSE":   ("dpv-pd:Identifier",                    "個人識別符号: 운전면허번호 (12자리)"),
    "JP_PHONE":             ("dpv-pd:TelephoneNumber",               "직접식별자: 일본 전화번호"),
    "JP_POSTAL_CODE":       ("dpv-pd:HomeAddress",                   "준식별자: 일본 우편번호"),
    "JP_ADDRESS":           ("dpv-pd:HomeAddress",                   "준식별자: 일본 주소"),
    "JP_CORPORATE_NUMBER":  ("dpv:NonPersonalData",                  "법인번호 (13자리) — 個人情報 아님"),
    "JP_BANK_ACCOUNT":      ("dpv-pd:BankAccount",                   "민감/금융: 일본 은행구좌"),
}


# =========================================================================
# 관할별 규제 매트릭스
# =========================================================================
# 각 관할에서 entity_type 을 어떻게 분류하는지(direct/quasi/sensitive/secret)
JURISDICTION: Dict[str, Dict] = {
    "KR": {
        "name": "대한민국",
        "law": "개인정보보호법 §2·§28-2~7 · 가명정보 처리 가이드라인 (PIPC)",
        "url": "https://www.pipc.go.kr/",
        "notes": (
            "직접식별자는 가명화 시 비가역적 변환 필요. 추가정보(매핑 키 등)는 "
            "별도 분리 보관. 준식별자 조합으로 재식별 가능성이 낮아야 함 "
            "(k-익명성·l-다양성 등 적정성 검토). 익명정보는 어떤 추가정보로도 "
            "복원 불가능해야 함 (§2 제1호의2)."
        ),
        "direct":   ["KR_RRN", "KR_PASSPORT", "EMAIL_ADDRESS", "PHONE_NUMBER", "KR_PHONE", "VIP_PERSON", "US_SSN"],
        "quasi":    ["KR_ADDRESS", "KR_BIZ_NO", "IP_ADDRESS", "URL"],
        "sensitive":["CREDIT_CARD", "IBAN_CODE"],
        "secret":   ["AWS_ACCESS_KEY", "GENERIC_API_KEY"],
    },
    "JP": {
        "name": "日本",
        "law": "個人情報保護法 (APPI) §2·§16-2·§35-2·§43 / マイナンバー法 §3·§19 / PPC『仮名加工情報・匿名加工情報の作成基準』",
        "url": "https://www.ppc.go.jp/",
        "notes": (
            "個人識別符号(マイナンバー·旅券·運転免許·指紋 등)는 仮名加工 시 "
            "削除 또는 復元不可能化. 要配慮個人情報(병력·범죄경력 등)는 사전 "
            "동의 필수. マイナンバーは特定個人情報として別途厳格規制("
            "利用目的限定·暗号化必須). 仮名加工情報는 추가정보 별도 관리 시 "
            "식별 가능. 匿名加工情報는 復元不可能 + 加工方法 정보 보존 필요."
        ),
        # 個人識別符号 (政令열거) — 단독으로 개인정보, 직접식별자
        "direct":         ["JP_MY_NUMBER", "JP_PASSPORT", "JP_DRIVERS_LICENSE",
                           "EMAIL_ADDRESS", "JP_PHONE", "PHONE_NUMBER", "VIP_PERSON"],
        "quasi":          ["JP_ADDRESS", "JP_POSTAL_CODE", "IP_ADDRESS", "URL"],
        # APPI 의 要配慮個人情報 — 별도 옵트인 동의 필수 (별도 라벨)
        "sensitive_appi": [],  # 의료/병력 검출기 추가 시 여기 확장
        "sensitive":      ["CREDIT_CARD", "JP_BANK_ACCOUNT", "IBAN_CODE"],
        "secret":         ["AWS_ACCESS_KEY", "GENERIC_API_KEY"],
        # マイナンバー法 § 적용 — 무조건 suppress 강제 + 미처리 시 verdict insufficient
        "my_number_act":  ["JP_MY_NUMBER"],
    },
    "US": {
        "name": "United States",
        "law": "HIPAA Safe Harbor (45 CFR §164.514(b)(2)) · CCPA/CPRA · NIST SP 800-188",
        "url": "https://www.hhs.gov/hipaa/",
        "notes": (
            "HIPAA Safe Harbor 18가지 식별자 모두 제거/일반화 — 이름·주소 "
            "(앞 3자리 ZIP 만 가능, 인구 ≥20,000), 모든 날짜(연도만), 전화·팩스·"
            "이메일·SSN·계정·인증서·차량·기기·URL·IP·생체정보·사진. CCPA "
            "deidentified data 는 'cannot reasonably identify' + 기술·계약 통제."
        ),
        "direct":   ["KR_RRN", "KR_PASSPORT", "EMAIL_ADDRESS", "PHONE_NUMBER", "KR_PHONE", "VIP_PERSON", "US_SSN", "URL", "IP_ADDRESS", "CREDIT_CARD", "IBAN_CODE"],
        "quasi":    ["KR_ADDRESS", "KR_BIZ_NO"],
        "sensitive":[],
        "secret":   ["AWS_ACCESS_KEY", "GENERIC_API_KEY"],
    },
    "EU": {
        "name": "European Union",
        "law": "GDPR Art. 4(5), Recital 26 · EDPB Guidelines 01/2025 on Pseudonymisation · WP29 Op. 05/2014",
        "url": "https://edpb.europa.eu/",
        "notes": (
            "Pseudonymisation = 추가정보를 별도로 보관하고 기술·조직적 조치로 "
            "재식별을 차단(Art 4(5)). Anonymisation = 어떤 합리적 수단으로도 "
            "재식별 불가능 (Recital 26). Special categories(Art 9: 건강·생체·"
            "민족 등) 은 추가 보호. Singling-out, linkability, inference 3개 "
            "리스크가 모두 제거되어야 익명."
        ),
        "direct":   ["KR_RRN", "KR_PASSPORT", "EMAIL_ADDRESS", "PHONE_NUMBER", "KR_PHONE", "VIP_PERSON", "US_SSN", "IP_ADDRESS"],
        "quasi":    ["KR_ADDRESS", "KR_BIZ_NO", "URL"],
        "sensitive":["CREDIT_CARD", "IBAN_CODE"],
        "secret":   ["AWS_ACCESS_KEY", "GENERIC_API_KEY"],
    },
}


# =========================================================================
# 권장 기법 (ISO/IEC 20889 분류 인용)
# =========================================================================
TECHNIQUE: Dict[str, Tuple[str, str, str]] = {
    # entity → (technique_id, ISO 20889 인용, DPV 표현)
    "KR_RRN":           ("tokenize_random",        "ISO 20889 §8.4 Tokenization (random) — 직접식별자, 매핑 키는 분리 보관",          "dpv:Pseudonymisation+dpv:Tokenisation"),
    "KR_PASSPORT":      ("tokenize_random",        "ISO 20889 §8.4 Tokenization",                                                      "dpv:Pseudonymisation+dpv:Tokenisation"),
    "KR_PHONE":         ("mask_partial",           "ISO 20889 §7.5 Masking — 앞 3 / 뒤 4 유지",                                         "dpv:DataMasking"),
    "PHONE_NUMBER":     ("mask_partial",           "ISO 20889 §7.5 Masking",                                                            "dpv:DataMasking"),
    "EMAIL_ADDRESS":    ("hash_local_keep_domain", "ISO 20889 §8.4 Cryptographic — local 부분 HMAC-BLAKE2b, 도메인 유지",               "dpv:Pseudonymisation+dpv:Encryption"),
    "KR_ADDRESS":       ("generalize_to_city",     "ISO 20889 §7.2 Generalization — 시·도 단위까지 일반화 (HIPAA SH §164.514(b)(2)(i)(B) 와 정합)", "dpv:Generalisation"),
    "CREDIT_CARD":      ("mask_pan",               "PCI-DSS Req 3.4 — 앞 6 / 뒤 4 유지, 중간 마스킹",                                   "dpv:DataMasking"),
    "US_SSN":           ("tokenize_random",        "직접식별자 → 비가역 토큰",                                                          "dpv:Pseudonymisation+dpv:Tokenisation"),
    "IBAN_CODE":        ("mask_partial",           "ISO 20889 §7.5 Masking — 국가코드+체크디지트 유지, 계좌부 중간 마스킹",            "dpv:DataMasking"),
    "VIP_PERSON":       ("pseudonym_consistent",   "ISO 20889 §8.4 Pseudonymisation — 동일인 일관 매핑(HMAC)",                          "dpv:Pseudonymisation"),
    "AWS_ACCESS_KEY":   ("suppress",               "비밀 자격증명 — 즉시 회전(rotate) + 완전 제거",                                     "dpv:Erasure"),
    "GENERIC_API_KEY":  ("suppress",               "비밀 후보 — 보수적 완전 제거",                                                       "dpv:Erasure"),
    "INTERNAL_PROJECT": ("tokenize_random",        "내부 코드명 — 외부 노출 시 무작위 토큰",                                             "dpv:Pseudonymisation"),
    "KR_BIZ_NO":        ("mask_partial",           "법인 ID — 부분 마스킹 (전체 비식별 시에는 토큰화)",                                  "dpv:DataMasking"),
    "URL":              ("generalize_url",         "ISO 20889 §7.2 — 호스트만 유지, 경로/쿼리 제거",                                    "dpv:Generalisation"),
    "IP_ADDRESS":       ("ip_truncate",            "ISO 20889 §7.2 — IPv4 마지막 옥텟 절단 (/24)",                                      "dpv:Generalisation"),
    # 일본 PII 권장 기법
    "JP_MY_NUMBER":         ("suppress",            "マイナンバー法 §19 利用目的外保管禁止 — 仮名加工·匿名加工 모두 완전 제거",         "dpv:Erasure"),
    "JP_PASSPORT":          ("tokenize_random",     "ISO 20889 §8.4 — 個人識別符号 (旅券番号) 비가역 토큰화",                            "dpv:Pseudonymisation+dpv:Tokenisation"),
    "JP_DRIVERS_LICENSE":   ("tokenize_random",     "ISO 20889 §8.4 — 個人識別符号 (公安委員会発行番号) 비가역 토큰화",                  "dpv:Pseudonymisation+dpv:Tokenisation"),
    "JP_PHONE":             ("mask_partial",        "ISO 20889 §7.5 — 앞 3 / 뒤 4 유지 (한국 KR_PHONE 과 동일 정책)",                    "dpv:DataMasking"),
    "JP_POSTAL_CODE":       ("generalize_postal",   "앞 3자리 (행정구역 단위) 만 유지 — HIPAA SH 의 ZIP3 정합",                         "dpv:Generalisation"),
    "JP_ADDRESS":           ("generalize_to_city",  "ISO 20889 §7.2 — 都道府県·市区町村 단위까지 일반화",                                 "dpv:Generalisation"),
    "JP_CORPORATE_NUMBER":  ("identity",            "법인번호는 個人情報 아님 (法人税法 §10-3 公開) — 기본 보존",                          "dpv:NonPersonalData"),
    "JP_BANK_ACCOUNT":      ("mask_partial",        "ISO 20889 §7.5 — 末尾4桁 유지, 본문 마스킹",                                        "dpv:DataMasking"),
}


# =========================================================================
# 데이터 클래스
# =========================================================================
@dataclass
class EntityRecord:
    index: int
    entity_type: str
    original: str
    start: int
    end: int
    score: float
    dpv_concept: str
    dpv_note: str
    technique: str
    technique_note: str
    technique_dpv: str
    transformed: str
    classifications: Dict[str, str]   # {jurisdiction: 'direct'|'quasi'|'sensitive'|'secret'|'unmapped'}


@dataclass
class ComplianceVerdict:
    jurisdiction: str
    name: str
    law: str
    url: str
    notes: str
    treatment_level: str          # 'pseudonymization' | 'anonymization'
    counts: Dict[str, int]        # direct/quasi/sensitive/secret
    untreated: List[str]
    verdict: str                  # 'compliant' | 'partial' | 'insufficient'
    rationale: str
    requirements_met: List[str]
    requirements_pending: List[str]


# =========================================================================
# 변환기 (ISO 20889 기법별)
# =========================================================================
class Pseudonymizer:
    """변환 기법 모음. salt 와 매핑 테이블은 인스턴스 내에 보관 — 실 운영 시
    이 매핑이 'additional information' (GDPR Art 4(5)) 에 해당하므로 별도 KMS·HSM
    보관 권장. PoC 에서는 메모리 내."""

    def __init__(self, salt: Optional[bytes] = None, anonymize: bool = False):
        self.salt = salt or secrets.token_bytes(16)
        self.anonymize = anonymize  # True 면 일관성 매핑도 끔
        self.consistent: Dict[Tuple[str, str], str] = {}
        self._counter: Dict[str, int] = {}

    def transform(self, entity_type: str, value: str) -> Tuple[str, str, str, str]:
        """반환: (변환문, technique_id, ISO 20889 노트, DPV 표현)"""
        tech_id, note, dpv = TECHNIQUE.get(
            entity_type,
            ("suppress", "기본 정책 — 미지정 엔티티는 보수적으로 제거", "dpv:Erasure"),
        )
        method = getattr(self, f"_{tech_id}", self._suppress)
        return method(entity_type, value), tech_id, note, dpv

    # ---- 기법 구현 ----
    def _suppress(self, et, v):
        return "[REDACTED]"

    def _tokenize_random(self, et, v):
        if not self.anonymize:
            key = (et, v)
            if key in self.consistent:
                return self.consistent[key]
        n = self._counter.get(et, 0) + 1
        self._counter[et] = n
        token = f"<{et}_{n:04d}>"
        if not self.anonymize:
            self.consistent[(et, v)] = token
        return token

    def _pseudonym_consistent(self, et, v):
        if self.anonymize:
            # 익명화: 동일성도 보존하지 않음 → 매번 임의 값
            return f"<PERSON_{secrets.token_hex(3).upper()}>"
        key = (et, v)
        if key in self.consistent:
            return self.consistent[key]
        h = hmac.new(self.salt, v.encode("utf-8"), hashlib.blake2b).hexdigest()[:8]
        token = f"<PERSON_{h.upper()}>"
        self.consistent[key] = token
        return token

    def _mask_partial(self, et, v):
        digit_pos = [i for i, c in enumerate(v) if c.isdigit()]
        if len(digit_pos) < 7:
            return self._suppress(et, v)
        keep_front, keep_back = 3, 4
        masked = set(digit_pos[keep_front : len(digit_pos) - keep_back])
        return "".join("*" if i in masked else c for i, c in enumerate(v))

    def _mask_pan(self, et, v):
        digit_pos = [i for i, c in enumerate(v) if c.isdigit()]
        if len(digit_pos) < 13:
            return self._suppress(et, v)
        masked = set(digit_pos[6 : len(digit_pos) - 4])
        return "".join("*" if i in masked else c for i, c in enumerate(v))

    def _hash_local_keep_domain(self, et, v):
        if "@" not in v:
            return self._suppress(et, v)
        local, domain = v.split("@", 1)
        h = hmac.new(self.salt, local.encode("utf-8"), hashlib.blake2b).hexdigest()[:8]
        return f"user-{h}@{domain}"

    _KR_PROVINCES = (
        "서울", "부산", "대구", "인천", "광주", "대전", "울산", "세종",
        "경기", "강원", "충북", "충남", "전북", "전남", "경북", "경남", "제주",
    )
    _JP_PROVINCES = (
        "東京都", "京都府", "大阪府", "北海道",
        "神奈川県", "埼玉県", "千葉県", "茨城県", "栃木県", "群馬県",
        "愛知県", "岐阜県", "三重県", "静岡県",
        "兵庫県", "奈良県", "和歌山県", "滋賀県",
        "福岡県", "佐賀県", "長崎県", "熊本県", "大分県", "宮崎県", "鹿児島県", "沖縄県",
        "宮城県", "福島県", "山形県", "秋田県", "岩手県", "青森県",
        "新潟県", "富山県", "石川県", "福井県", "山梨県", "長野県",
        "鳥取県", "島根県", "岡山県", "広島県", "山口県",
        "徳島県", "香川県", "愛媛県", "高知県",
    )

    def _generalize_to_city(self, et, v):
        # JP 우선 (긴 매치)
        for p in self._JP_PROVINCES:
            if v.startswith(p):
                return f"{p} (以下一般化)"
        for p in self._KR_PROVINCES:
            if v.startswith(p):
                return f"{p} (이하 일반화)"
        return self._suppress(et, v)

    def _generalize_postal(self, et, v):
        """일본 우편번호 — 앞 3자리만 유지 (지역 분류 단위)."""
        m = re.search(r"(\d{3})-?\d{4}", v)
        if m:
            return f"〒{m.group(1)}-****"
        return self._suppress(et, v)

    def _identity(self, et, v):
        """변환 없음 — 個人情報 가 아닌 항목 (법인번호 등)."""
        return v

    def _ip_truncate(self, et, v):
        m = re.match(r"(\d+)\.(\d+)\.(\d+)\.\d+", v)
        if m:
            return f"{m.group(1)}.{m.group(2)}.{m.group(3)}.0/24"
        return self._suppress(et, v)

    def _generalize_url(self, et, v):
        m = re.match(r"(https?://[^/]+)", v, flags=re.IGNORECASE)
        if m:
            return f"{m.group(1)}/[…]"
        return v


# =========================================================================
# 분류 / 평가 함수
# =========================================================================
def classify_entity(entity_type: str, jurisdictions: List[str]) -> Dict[str, str]:
    out: Dict[str, str] = {}
    for j in jurisdictions:
        rules = JURISDICTION.get(j)
        if not rules:
            out[j] = "unmapped"
            continue
        # JP 전용 — マイナンバー法 적용 항목 우선 (direct 보다 더 엄격한 분류)
        if entity_type in rules.get("my_number_act", []):
            out[j] = "my_number_act"
        elif entity_type in rules.get("sensitive_appi", []):
            out[j] = "sensitive_appi"
        elif entity_type in rules.get("direct", []):
            out[j] = "direct"
        elif entity_type in rules.get("quasi", []):
            out[j] = "quasi"
        elif entity_type in rules.get("sensitive", []):
            out[j] = "sensitive"
        elif entity_type in rules.get("secret", []):
            out[j] = "secret"
        else:
            out[j] = "unmapped"
    return out


def _requirements_per_jurisdiction(j: str, level: str) -> List[str]:
    """관할/처리수준별 핵심 요구사항 체크리스트."""
    common = [
        "직접식별자 모두 변환/제거",
        "민감정보(금융·건강 등) 마스킹/제거",
        "비밀 자격증명 완전 제거",
    ]
    by_level = {
        "pseudonymization": [
            "추가정보(매핑·키) 별도 분리 보관",
            "재식별 시도 방지 기술·조직적 조치",
        ],
        "anonymization": [
            "추가정보를 포함한 어떤 합리적 수단으로도 재식별 불가",
            "준식별자 조합 재식별 위험 검증 (k-익명성 등)",
            "결과의 분포·일반화 수준 통계적 검증",
        ],
    }
    j_specific = {
        "KR": ["가명정보 적정성 검토(가명정보 처리 가이드라인)"] if level == "pseudonymization"
              else ["적정성 평가 + 추가정보 폐기"],
        "JP": ["仮名加工情報の安全管理措置 (APPI §35-2)"] if level == "pseudonymization"
              else ["匿名加工情報の作成方法等の公表 (APPI §43)"],
        "US": ["HIPAA Safe Harbor 18 식별자 항목 모두 처리",
               "Expert Determination 트랙 시 통계 전문가 검증"],
        "EU": ["EDPB 01/2025 — singling-out / linkability / inference 3개 위험 평가",
               "Art 32 적절한 보안조치"],
    }
    return common + by_level[level] + j_specific.get(j, [])


def evaluate_compliance(
    entities: List[EntityRecord],
    jurisdictions: List[str],
    treatment_level: str,
) -> List[ComplianceVerdict]:
    out: List[ComplianceVerdict] = []
    for j in jurisdictions:
        rules = JURISDICTION.get(j)
        if not rules:
            continue

        counts = {"direct": 0, "quasi": 0, "sensitive": 0, "secret": 0,
                  "sensitive_appi": 0, "my_number_act": 0, "unmapped": 0}
        untreated: List[str] = []
        # マイナンバー法 — 'suppress' 기법(=[REDACTED]) 이외는 모두 위반
        my_number_violations: List[str] = []
        # 要配慮個人情報 — 옵트인 동의 여부는 자동 판정 불가 → pending 으로 누적
        sensitive_appi_present: List[str] = []
        for e in entities:
            cls = e.classifications.get(j, "unmapped")
            counts[cls] = counts.get(cls, 0) + 1
            if cls in ("direct", "secret", "sensitive", "sensitive_appi", "my_number_act") \
                    and e.transformed == e.original:
                untreated.append(e.entity_type)
            if cls == "my_number_act" and e.technique != "suppress":
                my_number_violations.append(e.entity_type)
            if cls == "sensitive_appi":
                sensitive_appi_present.append(e.entity_type)

        reqs = _requirements_per_jurisdiction(j, treatment_level)

        # 단순 휴리스틱 판정 (PoC)
        # 우선 マイナンバー法 위반 검사 (다른 모든 판정 우선)
        if my_number_violations:
            verdict = "insufficient"
            rationale = (
                f"マイナンバー法 §19 위반: {', '.join(sorted(set(my_number_violations)))} — "
                "마이넘버는 利用目的外保管禁止 → 마스킹/토큰화 불가, 완전 제거(suppress) 필수."
            )
            met = ["검출/분류 완료"]
            pending = [
                "JP_MY_NUMBER 의 technique 을 'suppress' 로 변경 (완전 제거)",
                "マイナンバー法 §12 安全管理措置 — 暗号化·접근통제 증빙",
                "目的外利用·提供禁止 (法 §20)",
            ] + reqs
        elif untreated:
            verdict = "insufficient"
            rationale = (
                f"미처리 직접/민감/비밀 항목 존재: {', '.join(sorted(set(untreated)))} — "
                f"{rules['name']} 기준 {treatment_level} 미충족."
            )
            met = ["검출/분류 완료"]
            pending = ["미처리 항목 변환"] + reqs
        elif treatment_level == "anonymization":
            # 익명: 준식별자 다수 잔존 시 부분 (수동 검증 필요)
            quasi_n = counts.get("quasi", 0)
            if quasi_n >= 2:
                verdict = "partial"
                rationale = (
                    f"준식별자 {quasi_n}개 — 조합 재식별 위험. "
                    "k-익명성·l-다양성·t-근접성 등 정량 검증 필요."
                )
                met = [
                    "직접·민감·비밀 모두 변환",
                    "PII 검출·분류·DPV 매핑 완료",
                ]
                pending = [
                    "준식별자 조합에 대한 재식별 위험 정량 평가",
                    "추가정보(매핑) 폐기 절차",
                ]
            else:
                verdict = "compliant"
                rationale = (
                    "직접·민감·비밀 모두 변환되었고 준식별자 일반화 적용. "
                    "단 PoC 휴리스틱 — 실 운영 시 통계적 적정성 검토 필수."
                )
                met = [
                    "직접·민감·비밀 모두 변환",
                    "준식별자 일반화/제거",
                    "DPV 매핑 + 관할 분류 명세화",
                ]
                pending = [
                    "추가정보(매핑) 폐기 또는 분리 폐기 증빙",
                    "통계적 재식별 위험 평가 보고서",
                ]
        else:
            # 가명화 — 추가정보 분리 보관 전제로 일단 compliant
            verdict = "compliant"
            n_treated = counts["direct"] + counts["sensitive"] + counts["secret"]
            rationale = (
                f"직접·민감·비밀 {n_treated}건 모두 변환 완료. 매핑 테이블은 "
                "본 PoC 가 메모리에 보관 — 실 운영 시 KMS/HSM 분리 보관 필요."
            )
            met = [
                "직접식별자 모두 변환/제거",
                "민감정보 마스킹/제거",
                "비밀 자격증명 완전 제거",
                "DPV 매핑 + 처리 흐름 문서화",
            ]
            pending = [
                "매핑 테이블의 별도 보관 (KMS·HSM)",
                "재식별 시도 방지 기술·조직적 조치 (Art 32 / APPI 安全管理)",
                "(KR) 가명정보 적정성 검토",
            ]
            # KR 만의 가이드라인 적정성 검토 강조
            if j == "KR":
                pending.append("가명정보 처리 가이드라인 §III-3 적정성 검토위원회 의사록")
            # JP 要配慮個人情報 — 옵트인 동의 자동 검증 불가
            if j == "JP" and sensitive_appi_present:
                pending.append(
                    f"要配慮個人情報 {len(set(sensitive_appi_present))}건 — "
                    "APPI §20 사전 옵트인 동의 확보 증빙 필요"
                )

        out.append(ComplianceVerdict(
            jurisdiction=j,
            name=rules["name"],
            law=rules["law"],
            url=rules["url"],
            notes=rules["notes"],
            treatment_level=treatment_level,
            counts={k: v for k, v in counts.items() if k != "unmapped"},
            untreated=sorted(set(untreated)),
            verdict=verdict,
            rationale=rationale,
            requirements_met=met,
            requirements_pending=pending,
        ))
    return out


# =========================================================================
# 메인 진입점 — 텍스트 + findings → 변환 텍스트 + 평가
# =========================================================================
def run(
    text: str,
    findings: List[Dict],
    jurisdictions: List[str],
    treatment_level: str,
    salt: Optional[bytes] = None,
) -> Dict:
    """text 와 PII Scanner findings 를 받아 변환 + 평가 결과를 반환."""
    if treatment_level not in ("pseudonymization", "anonymization"):
        treatment_level = "pseudonymization"
    valid = [j for j in jurisdictions if j in JURISDICTION]
    if not valid:
        valid = list(JURISDICTION.keys())

    pz = Pseudonymizer(salt=salt, anonymize=(treatment_level == "anonymization"))

    # 위치 오름차순 정렬 후 비겹침 선택
    sorted_findings = sorted(findings, key=lambda f: (f["start"], -f.get("score", 0)))
    chosen = []
    last_end = -1
    for f in sorted_findings:
        if f["start"] >= last_end:
            chosen.append(f)
            last_end = f["end"]

    records: List[EntityRecord] = []
    for i, f in enumerate(chosen):
        et = f["entity_type"]
        original = f["text"]
        dpv_concept, dpv_note = DPV_CATEGORY.get(et, ("dpv:NonPersonalData", "DPV 매핑 없음"))
        transformed, tech_id, tech_note, tech_dpv = pz.transform(et, original)
        classifications = classify_entity(et, valid)
        records.append(EntityRecord(
            index=i + 1,
            entity_type=et,
            original=original,
            start=f["start"],
            end=f["end"],
            score=float(f.get("score", 0)),
            dpv_concept=dpv_concept,
            dpv_note=dpv_note,
            technique=tech_id,
            technique_note=tech_note,
            technique_dpv=tech_dpv,
            transformed=transformed,
            classifications=classifications,
        ))

    # 끝→앞 으로 치환 (인덱스 유지)
    out_text = text
    for r in sorted(records, key=lambda r: r.start, reverse=True):
        out_text = out_text[: r.start] + r.transformed + out_text[r.end :]

    verdicts = evaluate_compliance(records, valid, treatment_level)

    return {
        "treatment_level": treatment_level,
        "jurisdictions": valid,
        "original_text": text,
        "transformed_text": out_text,
        "entities": [asdict(r) for r in records],
        "verdicts": [asdict(v) for v in verdicts],
        # 매핑 테이블 (PoC 시연용 — 실 운영 시 절대 응답에 포함 금지)
        "mapping_demo": [
            {"entity_type": k[0], "original": k[1], "token": v}
            for k, v in pz.consistent.items()
        ],
    }