"""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 publication_verdict: 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"), publication_verdict=c.get("publication_verdict"), ) 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