291 lines
11 KiB
Python
291 lines
11 KiB
Python
"""등록된 판례만 인용하는 저작권 위험도 보조 엔진.
|
|
|
|
유사도 점수를 법적 결론으로 바꾸지 않는다. 보호되는 표현, 의거관계, 권리 귀속
|
|
등 입력되지 않은 사실을 ``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
|
|
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: ...
|
|
|
|
|
|
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),
|
|
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 not decision.review_required:
|
|
raise ValueError("LLM legal judge는 사람 검토를 해제할 수 없습니다")
|
|
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": (
|
|
"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 판단을 완료하지 못해 규칙 기반 결과로 대체했습니다.",
|
|
}
|
|
)
|
|
|
|
|
|
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} 항목이 확인되지 않아 "
|
|
"저작권 침해 여부는 추가 검토가 필요합니다."
|
|
)
|