378 lines
15 KiB
Python
378 lines
15 KiB
Python
"""등록된 판례만 인용하는 저작권 위험도 보조 엔진.
|
|
|
|
유사도 점수를 법적 결론으로 바꾸지 않는다. 보호되는 표현, 의거관계, 권리 귀속
|
|
등 입력되지 않은 사실을 ``missing_factors`` 로 남기고 사람 검토를 요구한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import math
|
|
import re
|
|
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: ...
|
|
|
|
|
|
_TOKEN_RE = re.compile(r"[가-힣A-Za-z0-9]{2,}")
|
|
_TOKEN_STOPWORDS = {
|
|
"그리고", "그러나", "대한", "관한", "있는", "없는", "으로", "에서",
|
|
"판결", "사건", "원고", "피고", "저작권", "저작물", "경우", "해당",
|
|
}
|
|
_KOREAN_SUFFIXES = ("에서는", "으로", "에서", "에게", "까지", "부터", "처럼", "보다", "은", "는", "이", "가", "을", "를", "과", "와", "의", "에", "로")
|
|
_GRADE_BOOST = {"A": 0.05, "B": 0.035, "C": 0.015}
|
|
_SOFTWARE_MARKERS = ("컴퓨터프로그램", "폰트파일", "글꼴파일", "글꼴 파일", "서체프로그램")
|
|
|
|
|
|
def _tokens(text: str) -> set[str]:
|
|
tokens: set[str] = set()
|
|
for raw in _TOKEN_RE.findall(text):
|
|
token = raw.lower()
|
|
if token in _TOKEN_STOPWORDS:
|
|
continue
|
|
tokens.add(token)
|
|
for suffix in _KOREAN_SUFFIXES:
|
|
if token.endswith(suffix) and len(token) >= len(suffix) + 2:
|
|
tokens.add(token[:-len(suffix)])
|
|
break
|
|
return tokens
|
|
|
|
|
|
def _normalized_work_types(raw: dict) -> tuple[str, ...]:
|
|
"""수집 라벨의 명백한 오분류를 판시 제목 기준으로 보정한다.
|
|
|
|
폰트 프로그램 사건 3건은 literary/visual로 수집돼 문학 판례 검색에 섞였다.
|
|
원천 레코드는 보존하되 엔진과 조회 API에서는 software로 취급한다.
|
|
"""
|
|
title = str(raw.get("title", "")).replace(" ", "")
|
|
if any(marker.replace(" ", "") in title for marker in _SOFTWARE_MARKERS):
|
|
return ("software",)
|
|
return tuple(raw.get("work_types", []))
|
|
|
|
|
|
def _precedent_score(
|
|
precedent: Precedent,
|
|
*,
|
|
tags: set[str],
|
|
work_type: str,
|
|
query_terms: set[str],
|
|
) -> float:
|
|
"""유형·법적 쟁점·판시 내용 관련성을 중심으로 한 결정적 검색 점수."""
|
|
tag_overlap = len(tags.intersection(precedent.legal_tags))
|
|
tag_score = tag_overlap / max(len(tags), 1)
|
|
precedent_terms = _tokens(" ".join([
|
|
precedent.title, *precedent.criteria, precedent.holding_summary,
|
|
]))
|
|
# 긴 판시문 때문에 관련 어휘가 희석되지 않도록 질의 쪽 포함률을 사용한다.
|
|
text_score = len(query_terms.intersection(precedent_terms)) / max(len(query_terms), 1)
|
|
criteria_terms = _tokens(" ".join(precedent.criteria))
|
|
criteria_score = len(query_terms.intersection(criteria_terms)) / max(len(query_terms), 1)
|
|
type_score = 1.0 if work_type in precedent.work_types else 0.5
|
|
quality_score = min(math.log1p(len(precedent.holding_summary)) / 10.0, 1.0)
|
|
return (
|
|
0.55 * tag_score
|
|
+ 0.20 * text_score
|
|
+ 0.10 * criteria_score
|
|
+ 0.08 * type_score
|
|
+ 0.02 * quality_score
|
|
+ _GRADE_BOOST.get(precedent.grade or "", 0.0)
|
|
)
|
|
|
|
|
|
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=_normalized_work_types(raw),
|
|
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
|
|
|
|
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] = (),
|
|
query_text: str = "",
|
|
) -> LegalRiskAssessment:
|
|
tags = set(legal_tags)
|
|
evidence_rows = tuple(evidence)
|
|
search_text = " ".join([
|
|
query_text,
|
|
*(str(excerpt) for row in evidence_rows for excerpt in row.get("excerpts", [])),
|
|
*(str(row.get("infringement_type", "")) for row in evidence_rows),
|
|
])
|
|
query_terms = _tokens(search_text)
|
|
candidates = [
|
|
p for p in self.precedents
|
|
if (not p.work_types or work_type in p.work_types)
|
|
]
|
|
related = sorted(
|
|
candidates,
|
|
key=lambda p: (
|
|
-_precedent_score(
|
|
p, tags=tags, work_type=work_type, query_terms=query_terms,
|
|
),
|
|
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=evidence_rows,
|
|
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} 항목이 확인되지 않아 "
|
|
"저작권 침해 여부는 추가 검토가 필요합니다."
|
|
)
|