POSA_Copyrighter/scripts/score_distribution.py

"""위험도 스코어러의 밴드 분포를 신호 조합 전수로 측정한다.

**이 스크립트가 재는 것과 재지 않는 것**

재는 것: 스코어러의 구조적 거동. 어떤 신호 조합이 어느 밴드로 떨어지는지,
100점 포화가 얼마나 되는지, "웹 신호만으로 high 에 도달하는가"(v2 이후의 설계
원칙)가 지켜지는지.

재지 않는 것: **임계가 맞는지.** 이 격자는 신호 축을 균등 샘플링하므로 실제 큐의
분포가 아니다. 임계 검증은 실이미지 + 운영자 판정으로만 가능하고 그것은
scripts/sweep_thresholds.py 가 담당한다.

**왜 만들었나**: SCORER_VERSION 이 v1→v2→v3→v4 로 갔는데 v3(제공자별 풀링)와
v4(신뢰 출처)는 분포를 한 번도 측정하지 않고 나갔다. 2026-08-08 평가가 쓴
224조합 하네스는 저장소에 남아 있지 않고, 그 문서의 두 수치(174/224=77.7% 와
82.6%)가 서로 맞지 않아 격자가 달랐음을 시사한다 — 즉 과거 숫자와 비교 가능한
재현은 불가능하다. 그래서 격자를 여기에 **명시적으로** 고정해 새 기준선으로 삼는다.
앞으로 스코어러를 바꾸면 이 스크립트를 돌려 전후를 비교한다.

사용법:
    python scripts/score_distribution.py
    python scripts/score_distribution.py --json
"""

from __future__ import annotations

import argparse
import itertools
import json
import sys
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

from rights_filter.analysis.risk_scoring import SCORER_VERSION, RiskScorer  # noqa: E402
from rights_filter.domain.records import Evidence, EvidenceSource  # noqa: E402

TRUSTED_HOST = "trusted.example"
OUTSIDE_HOST = "outside.example"

# --- 신호 축 -----------------------------------------------------------------
#
# 각 축은 스코어러가 실제로 구분하는 입력만 담는다. provider 를 명시하는 이유는
# v3 부터 풀과 dedup 버킷이 제공자별로 갈리기 때문이다 - 같은 match 타입이라도
# 제공자가 다르면 점수가 달라진다.

WEB_PRESENCE = {
    "none": [],
    "full": [("full", 0.9)],
    "full+partial": [("full", 0.9), ("partial", 0.8)],
    "all4": [("full", 0.9), ("partial", 0.8), ("page", 0.8), ("visual", 0.6)],
}
ENTITY_COUNT = {"0": 0, "1": 1, "3": 3}
FACE = {"none": None, "background": "background", "identifiable": "identifiable"}
FINGERPRINT = {"none": None, "weak": 0.5, "strong": 0.95}
NAVER = {"none": False, "promoted": True}
SECOND_PROVIDER = {"none": False, "cse_page": True}


def build_evidence(presence, entities, face, fingerprint, naver, second, host):
    evidence = []
    for match, confidence in WEB_PRESENCE[presence]:
        evidence.append(
            Evidence(
                source=EvidenceSource.WEB_DETECTION,
                reason=f"vision {match}",
                confidence=confidence,
                data={"match": match, "provider": "google", "host": host},
            )
        )
    for index in range(ENTITY_COUNT[entities]):
        evidence.append(
            Evidence(
                source=EvidenceSource.WEB_DETECTION,
                reason=f"entity {index}",
                confidence=0.9,
                data={"match": "entity", "entity": f"E{index}", "provider": "google"},
            )
        )
    if SECOND_PROVIDER[second]:
        # 독립 제공자가 따로 찾은 게시본. v3 의 제공자별 풀이 이것을 Vision 과
        # 합산하지 않고 따로 세는지가 여기서 드러난다.
        evidence.append(
            Evidence(
                source=EvidenceSource.WEB_DETECTION,
                reason="cse page",
                confidence=0.35,
                data={"match": "page", "provider": "google_custom_search", "host": host},
            )
        )
    if FACE[face]:
        evidence.append(
            Evidence(
                source=EvidenceSource.FACE_PERSON,
                reason="face",
                confidence=1.0,
                data={"tier": FACE[face]},
            )
        )
    if FINGERPRINT[fingerprint] is not None:
        evidence.append(
            Evidence(
                source=EvidenceSource.FINGERPRINT,
                reason="knowledge match",
                confidence=1.0,
                data={"similarity": FINGERPRINT[fingerprint]},
            )
        )
    if NAVER[naver]:
        evidence.append(
            Evidence(
                source=EvidenceSource.NAVER_SEARCH,
                reason="naver promoted",
                confidence=0.8,
                data={"promoted": True},
            )
        )
    return evidence


def combinations():
    return itertools.product(
        WEB_PRESENCE, ENTITY_COUNT, FACE, FINGERPRINT, NAVER, SECOND_PROVIDER
    )


def measure(trusted: bool):
    """trusted=True 면 모든 웹 행이 등록된 신뢰 출처에 게시된 것으로 본다."""
    host = TRUSTED_HOST if trusted else OUTSIDE_HOST
    scorer = RiskScorer(frozenset({TRUSTED_HOST}) if trusted else None)

    bands = Counter()
    saturated = 0
    web_only_high = 0
    web_only_total = 0
    rows = []

    for presence, entities, face, fingerprint, naver, second in combinations():
        evidence = build_evidence(presence, entities, face, fingerprint, naver, second, host)
        result = scorer.score(evidence)
        bands[result.band] += 1
        if result.score == 100:
            saturated += 1

        # "웹 신호만" = 인물·지식DB·네이버가 전부 없는 조합. v2 가 세운 설계
        # 원칙은 이 부류가 high 에 도달하지 않는다는 것이다.
        if face == "none" and fingerprint == "none" and naver == "none":
            web_only_total += 1
            if result.band == "high":
                web_only_high += 1

        rows.append(
            {
                "presence": presence,
                "entities": entities,
                "face": face,
                "fingerprint": fingerprint,
                "naver": naver,
                "second": second,
                "score": result.score,
                "band": result.band,
            }
        )

    total = sum(bands.values())
    return {
        "total": total,
        "bands": {band: bands.get(band, 0) for band in ("high", "medium", "low")},
        "bandPct": {
            band: round(100 * bands.get(band, 0) / total, 1) for band in ("high", "medium", "low")
        },
        "saturated": saturated,
        "saturatedPct": round(100 * saturated / total, 1),
        "webOnlyHigh": web_only_high,
        "webOnlyTotal": web_only_total,
        "rows": rows,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--json", action="store_true", help="전체 행을 JSON 으로 출력")
    args = parser.parse_args()

    plain = measure(trusted=False)
    trusted = measure(trusted=True)

    if args.json:
        print(json.dumps({"scorer": SCORER_VERSION, "plain": plain, "trusted": trusted}, ensure_ascii=False))
        return 0

    # 비ASCII 는 stdout 에 넣지 않는다 (콘솔이 cp949 라 죽은 이력).
    print("scorer %s | grid %d combinations" % (SCORER_VERSION, plain["total"]))
    print("")
    print("%-22s %8s %8s" % ("", "default", "trusted"))
    for band in ("high", "medium", "low"):
        print(
            "%-22s %7d%% %7d%%"
            % (band, plain["bandPct"][band], trusted["bandPct"][band])
        )
    print(
        "%-22s %7d%% %7d%%"
        % ("saturated (==100)", plain["saturatedPct"], trusted["saturatedPct"])
    )
    print("")
    print(
        "web-signal-only high : %d/%d (default) | %d/%d (trusted)"
        % (
            plain["webOnlyHigh"],
            plain["webOnlyTotal"],
            trusted["webOnlyHigh"],
            trusted["webOnlyTotal"],
        )
    )
    print("")
    print("counts  default: %s" % plain["bands"])
    print("counts  trusted: %s" % trusted["bands"])
    return 0


if __name__ == "__main__":
    sys.exit(main())