o2o-plagiarism-ai/app/engine/taxonomy.py

"""10종 메타 태그 분류체계 + 39개 침해 케이스 로더.

IV장(메타태그)과 IX장 「침해 케이스 ↔ 메타태그 매핑 총괄표」를 그대로 JSON으로
보관 (data/taxonomy/). 부팅 시 1회 로드.

케이스 데이터는 실무방안 v1.3 기준이며, 케이스별 **대표판례**와 처리 구분
(● 기술검출 근거 / ○ 약관·신고)을 함께 담는다.
"""

from __future__ import annotations

import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable

from app.api.schemas import LegalTag

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class MetaTagDef:
    id: str
    label_ko: str
    category: str
    law_ref: str
    scope: str
    description: str


@dataclass(frozen=True)
class RepresentativePrecedent:
    case_id: str            # 법원 사건번호 (케이스 ID 아님)
    in_runtime_corpus: bool  # precedents.jsonl 적재 여부. False 면 인용 시 출처 미확보.


@dataclass(frozen=True)
class CaseDef:
    case_id: str          # A1, A7, B1, ...
    old_no: int
    subgroup: str
    title: str
    actor: str
    primary_tags: tuple[str, ...]
    secondary_tags: tuple[str, ...] = field(default_factory=tuple)
    detectable_internal: bool = False
    high_risk: bool = False
    note: str | None = None
    #: v1.3 총괄표의 처리 구분. technical_detection(●) / terms_or_report(○)
    handling: str = "terms_or_report"
    representative_precedents: tuple[RepresentativePrecedent, ...] = field(default_factory=tuple)
    #: 대표판례가 지정되지 않은 이유("확립 판례 없음" 등). 빈 목록과 구분한다.
    representative_precedent_note: str | None = None


@dataclass(frozen=True)
class Taxonomy:
    meta_tags_version: str
    cases_version: str
    meta_tags: tuple[MetaTagDef, ...]
    cases: tuple[CaseDef, ...]

    def tag_label(self, tag_id: str) -> str:
        for t in self.meta_tags:
            if t.id == tag_id:
                return t.label_ko
        return tag_id

    def find_cases(self, primary_tags: Iterable[str]) -> tuple[CaseDef, ...]:
        """주 태그 조합에 가장 잘 맞는 케이스 **전부**를 반환한다.

        태그만으로는 케이스가 1:1로 좁혀지지 않는다. A1·A2·A3·A4·A5·A6·A15 는
        주/보조 태그가 완전히 같고, 실제 구별자는 베껴온 **원본의 종류**(가사·기사·
        위키·교재·타인 자서전)다. 원본 종류 라벨은 코퍼스에 아직 없으므로, 동점
        케이스 중 하나를 임의로 고르는 대신 동점군을 그대로 돌려준다. 호출자는
        이를 "이 중 하나"로 제시하고 사람이 원본 종류로 확정한다.

        # ponytail: 원본 종류 라벨이 코퍼스에 들어오면 이 함수에 source_type 인자를
        # 추가해 동점군을 좁힌다. 라벨 없는 상태에서 미리 배선하면 항상 None 이다.
        """
        target = set(primary_tags)
        if not target:
            return ()
        scored: list[tuple[float, CaseDef]] = []
        for c in self.cases:
            primary = set(c.primary_tags)
            if not primary:
                continue
            score = len(target & primary) / len(target | primary)
            # v1.3 에서 기술검출 근거(●)로 확인된 케이스에 작은 가중치.
            if c.detectable_internal:
                score += 0.1
            scored.append((score, c))
        if not scored:
            return ()
        best = max(score for score, _ in scored)
        if best <= 0.3:
            return ()
        return tuple(c for score, c in scored if abs(score - best) < 1e-9)


def load_taxonomy(taxonomy_dir: Path) -> Taxonomy | None:
    mt_path = taxonomy_dir / "meta_tags_v1.0.json"
    cs_path = taxonomy_dir / "cases_v1.3.json"
    if not mt_path.exists() or not cs_path.exists():
        logger.warning("Taxonomy files not found in %s", taxonomy_dir)
        return None

    mt_data = json.loads(mt_path.read_text(encoding="utf-8"))
    cs_data = json.loads(cs_path.read_text(encoding="utf-8"))

    meta_tags = tuple(
        MetaTagDef(
            id=t["id"], label_ko=t["label_ko"], category=t["category"],
            law_ref=t["law_ref"], scope=t["scope"], description=t["description"],
        )
        for t in mt_data["tags"]
    )
    cases = tuple(
        CaseDef(
            case_id=c["case_id"],
            old_no=c.get("old_no", 0),
            subgroup=c.get("subgroup", ""),
            title=c["title"],
            actor=c.get("actor", ""),
            primary_tags=tuple(c.get("primary_tags", [])),
            secondary_tags=tuple(c.get("secondary_tags", [])),
            detectable_internal=c.get("detectable_internal", False),
            high_risk=c.get("high_risk", False),
            note=c.get("note"),
            handling=c.get("handling", "terms_or_report"),
            representative_precedents=tuple(
                RepresentativePrecedent(case_id=p["case_id"], in_runtime_corpus=p["in_runtime_corpus"])
                for p in c.get("representative_precedents", [])
            ),
            representative_precedent_note=c.get("representative_precedent_note"),
        )
        for c in cs_data["cases"]
    )

    tax = Taxonomy(
        meta_tags_version=mt_data["version"],
        cases_version=cs_data["version"],
        meta_tags=meta_tags,
        cases=cases,
    )
    logger.info("Taxonomy loaded: tags=%s cases=%s (%d tags, %d cases)",
                tax.meta_tags_version, tax.cases_version, len(meta_tags), len(cases))
    return tax