"""등록된 판례만 인용하는 저작권 위험도 보조 엔진. 유사도 점수를 법적 결론으로 바꾸지 않는다. 보호되는 표현, 의거관계, 권리 귀속 등 입력되지 않은 사실을 ``missing_factors`` 로 남기고 사람 검토를 요구한다. """ from __future__ import annotations import json import logging from dataclasses import dataclass, field from pathlib import Path from typing import Iterable, Protocol logger = logging.getLogger(__name__) @dataclass(frozen=True) class Precedent: case_id: str title: str source_url: str work_types: tuple[str, ...] legal_tags: tuple[str, ...] criteria: tuple[str, ...] holding_summary: str outcome: str | None = None @dataclass(frozen=True) class LegalRiskAssessment: status: str risk_level: str | None similarity_evidence: str protected_expression: str access_evidence: str missing_factors: tuple[str, ...] precedent_ids: tuple[str, ...] = field(default_factory=tuple) judgment_method: str = "rule_based" llm_verdict: str | None = None llm_confidence: float | None = None llm_review_required: bool | None = None llm_matched_precedent_ids: tuple[str, ...] = field(default_factory=tuple) supporting_reasons: tuple[str, ...] = field(default_factory=tuple) counter_reasons: tuple[str, ...] = field(default_factory=tuple) judge_model: str | None = None judge_prompt_version: str | None = None judge_note: str | None = None disclaimer: str = ( "이 결과는 등록 코퍼스와 판례에 기반한 검토 우선순위이며 법률상 침해 확정이 아닙니다." ) @dataclass(frozen=True) class LegalJudgeInput: max_similarity: float coverage: float longest_span: int legal_tags: tuple[str, ...] work_type: str access_evidence: bool | None protected_expression_reviewed: bool rights_verified: bool evidence: tuple[dict, ...] precedents: tuple[Precedent, ...] deterministic_missing_factors: tuple[str, ...] @dataclass(frozen=True) class LegalJudgeDecision: verdict: str confidence: float matched_precedent_ids: tuple[str, ...] supporting_reasons: tuple[str, ...] counter_reasons: tuple[str, ...] missing_factors: tuple[str, ...] review_required: bool model: str prompt_version: str class LegalJudge(Protocol): def judge(self, request: LegalJudgeInput) -> LegalJudgeDecision: ... def load_precedents(path: str | Path) -> list[Precedent]: file = Path(path) if not file.exists(): return [] rows: list[Precedent] = [] seen: set[str] = set() for lineno, line in enumerate(file.read_text(encoding="utf-8").splitlines(), 1): if not line.strip(): continue raw = json.loads(line) required = ("case_id", "title", "source_url", "holding_summary") missing = [key for key in required if not str(raw.get(key, "")).strip()] if missing: raise ValueError(f"{file}:{lineno} 필수 필드 없음: {missing}") case_id = str(raw["case_id"]) if case_id in seen: raise ValueError(f"중복 사건번호: {case_id}") if not str(raw["source_url"]).startswith("https://"): raise ValueError(f"검증 가능한 HTTPS 출처가 필요합니다: {case_id}") seen.add(case_id) rows.append(Precedent( case_id=case_id, title=str(raw["title"]), source_url=str(raw["source_url"]), work_types=tuple(raw.get("work_types", [])), legal_tags=tuple(raw.get("legal_tags", [])), criteria=tuple(raw.get("criteria", [])), holding_summary=str(raw["holding_summary"]), outcome=raw.get("outcome"), )) return rows class LegalRiskEngine: def __init__(self, precedents: Iterable[Precedent], judge: LegalJudge | None = None): self.precedents = list(precedents) self.judge = judge def assess( self, *, max_similarity: float, coverage: float, longest_span: int, legal_tags: Iterable[str], work_type: str = "literary", access_evidence: bool | None = None, protected_expression_reviewed: bool = False, rights_verified: bool = False, evidence: Iterable[dict] = (), ) -> LegalRiskAssessment: tags = set(legal_tags) related = sorted([ p for p in self.precedents if (not p.work_types or work_type in p.work_types) and (not p.legal_tags or tags.intersection(p.legal_tags)) ], key=lambda p: ( -len(tags.intersection(p.legal_tags)), 0 if work_type in p.work_types else 1, p.case_id, ))[:5] missing: list[str] = [] if not protected_expression_reviewed: missing.append("보호되는 창작적 표현인지에 대한 사람 검토") if access_evidence is None: missing.append("원저작물 접근·의거 가능성") if not rights_verified: missing.append("저작권 귀속·이용허락·인용 요건") if not self.precedents: status, level = "insufficient_precedent_data", None elif max_similarity <= 0 and coverage <= 0: status, level = "no_registered_corpus_match", "low" else: status = "review_required" strong_copy = coverage >= 0.30 or longest_span >= 100 level = "high" if strong_copy and max_similarity >= 0.75 else "medium" if not strong_copy and max_similarity < 0.60: level = "low" base = LegalRiskAssessment( status=status, risk_level=level, similarity_evidence=( f"검색 유사도 {max_similarity:.3f}, 질의 커버리지 {coverage:.3f}, " f"최장 연속 일치 {longest_span}자" ), protected_expression=("reviewed" if protected_expression_reviewed else "not_reviewed"), access_evidence=( "provided" if access_evidence is True else "not_found" if access_evidence is False else "not_provided" ), missing_factors=tuple(missing), precedent_ids=tuple(p.case_id for p in related), ) if self.judge is None or not related or (max_similarity <= 0 and coverage <= 0): return base request = LegalJudgeInput( max_similarity=max_similarity, coverage=coverage, longest_span=longest_span, legal_tags=tuple(sorted(tags)), work_type=work_type, access_evidence=access_evidence, protected_expression_reviewed=protected_expression_reviewed, rights_verified=rights_verified, evidence=tuple(evidence), precedents=tuple(related), deterministic_missing_factors=tuple(missing), ) try: decision = self.judge.judge(request) allowed = {p.case_id for p in related} unknown = set(decision.matched_precedent_ids) - allowed if unknown: raise ValueError(f"LLM이 제공되지 않은 판례 ID를 반환함: {sorted(unknown)}") if not decision.review_required: raise ValueError("LLM legal judge는 사람 검토를 해제할 수 없습니다") merged_missing = tuple(dict.fromkeys([ *missing, *decision.missing_factors, ])) return LegalRiskAssessment( **{ **base.__dict__, "missing_factors": merged_missing, "judgment_method": "llm", "llm_verdict": decision.verdict, "llm_confidence": decision.confidence, "llm_review_required": True, "llm_matched_precedent_ids": decision.matched_precedent_ids, "supporting_reasons": decision.supporting_reasons, "counter_reasons": decision.counter_reasons, "judge_model": decision.model, "judge_prompt_version": decision.prompt_version, "judge_note": ( "GPT 판정은 판례·탐지 증거에 대한 검토 보조 의견이며 법률상 확정이 아닙니다." ), } ) except Exception as exc: logger.warning("LLM legal judge failed; using deterministic fallback: %s", exc) return LegalRiskAssessment( **{ **base.__dict__, "judgment_method": "rule_fallback", "judge_note": "GPT 판단을 완료하지 못해 규칙 기반 결과로 대체했습니다.", } )