"""등록된 판례만 인용하는 저작권 위험도 보조 엔진. 유사도 점수를 법적 결론으로 바꾸지 않는다. 보호되는 표현, 의거관계, 권리 귀속 등 입력되지 않은 사실을 ``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 #: 사람이 판시 본문을 확인하고 매긴 등급. A=직접 적용, B=조건부, C=참고. #: None 은 "아직 검토되지 않음"이며 C(검토 후 부적합)와 다르다. #: 출처: docs/PRECEDENT_LISTUP.md / data/precedents/precedent_listup.csv grade: 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) #: 인용한 판례의 등급 (사건번호, 등급) 쌍. 등급 없는 건은 넣지 않는다. #: 검토자가 "이 근거가 사람 검토를 거친 것인지"를 구분할 수 있어야 한다. precedent_grades: tuple[tuple[str, 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 judgment_summary: str = "" 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: ... #: 인용 우선순위. A/B/C는 모두 사람이 판시 본문을 확인해 최종 리스트업한 #: 판례다. 등급이 없는 적재본은 리스트업 제외 대상이므로, 등급 데이터가 하나라도 #: 있는 운영 코퍼스에서는 인용 후보로 사용하지 않는다. _GRADE_ORDER = {"A": 0, "B": 1, "C": 2} def _grade_rank(grade: str | None) -> int: return _GRADE_ORDER.get(grade or "", len(_GRADE_ORDER)) 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"), grade=(str(raw["grade"]).strip().upper() or None) if raw.get("grade") else None, )) return rows class LegalRiskEngine: def __init__(self, precedents: Iterable[Precedent], judge: LegalJudge | None = None): self.precedents = list(precedents) self.judge = judge # 등급 반영 전의 외부/테스트 코퍼스는 계속 동작시키되, 등급을 반영한 운영 # 코퍼스는 최종 리스트업(A/B/C)을 명시적인 allowlist로 취급한다. self._has_reviewed_precedents = any(p.grade in _GRADE_ORDER for p in self.precedents) 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 self._has_reviewed_precedents or p.grade in _GRADE_ORDER) and (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: ( _grade_rank(p.grade), -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), precedent_grades=tuple((p.case_id, p.grade) for p in related if p.grade), judgment_summary=_rule_based_summary(status, 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 decision.verdict == "likely" and not decision.matched_precedent_ids: raise ValueError("침해 의심 의견에는 근거 판례 ID가 최소 1개 필요합니다") 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, "judgment_summary": _judgment_summary( decision, tuple(p.case_id for p in related), ), "judge_note": ( "판례·탐지 증거 비교 결과는 검토 보조 의견이며 법률상 확정이 아닙니다." ), } ) 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": "자동 판례 비교를 완료하지 못해 규칙 기반 결과로 대체했습니다.", } ) def _rule_based_summary(status: str, related: list[Precedent]) -> str: if status == "insufficient_precedent_data": return "등록된 판례가 없어 판례에 따른 검토 의견을 제시할 수 없습니다." if status == "no_registered_corpus_match": return ( "등록 코퍼스에서 일치 증거가 확인되지 않아 판례 기반 침해 의심을 제시하지 않습니다. " "다만 미매칭은 비침해 확정이 아닙니다." ) ids = ", ".join(p.case_id for p in related[:3]) suffix = f" 등 {len(related)}건" if len(related) > 3 else "" return f"{ids}{suffix} 판례가 관련 판례로 검색되어 구체적인 사실관계 검토가 필요합니다." def _judgment_summary( decision: LegalJudgeDecision, fallback_precedent_ids: tuple[str, ...] = (), ) -> str: cited_ids = decision.matched_precedent_ids or fallback_precedent_ids ids = ", ".join(cited_ids[:3]) suffix = ( f" 등 {len(cited_ids)}건" if len(cited_ids) > 3 else "" ) precedent_label = f"{ids}{suffix} 판례" def first(items: tuple[str, ...], fallback: str) -> str: value = items[0] if items else fallback return " ".join(value.split()).rstrip(".。")[:180] if decision.verdict == "likely": reason = first(decision.supporting_reasons, "표현의 유사성과 일치 범위가 확인됨") return ( f"{precedent_label}의 판단 기준과 탐지 증거를 비교한 결과, {reason} 사유로 " "저작권 침해가 의심되어 추가 검토가 필요합니다." ) if decision.verdict == "unlikely": reason = first(decision.counter_reasons, "침해를 뒷받침하는 표현상 증거가 충분하지 않음") return ( f"{precedent_label}의 판단 기준과 탐지 증거를 비교한 결과, {reason} 등을 고려할 때 " "현재 증거만으로 저작권 침해 의심은 낮습니다." ) missing = first(decision.missing_factors, "법적 판단에 필요한 사실관계") return ( f"{precedent_label}와 관련성은 확인되지만, {missing} 항목이 확인되지 않아 " "저작권 침해 여부는 추가 검토가 필요합니다." )