581 lines
24 KiB
Python
581 lines
24 KiB
Python
"""저작권 침해 탐지 파이프라인 (PDF VII장 권장 아키텍처).
|
|
|
|
3단 캐스케이딩:
|
|
1차) MinHash + LSH 1차 필터 — 대규모 코퍼스에서 후보 N건 빠르게 추출
|
|
2차) 자서전 모드 전처리 (옵션) — 공통 표현 제거 + NER 마스킹
|
|
3차) 삼중 유사도 정밀 비교 — text(임베딩) + lemma(형태소) + element(자카드)
|
|
4차) 분류 — 10종 법령 메타 태그 (주/보조) + 케이스 매핑
|
|
|
|
후방 호환:
|
|
- infringement_type (5종 enum): 기존 UI/통합 코드용으로 유지
|
|
- tags + case_id: PDF 분류체계 신규 필드
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from app.api.schemas import (
|
|
TAG_LABEL_KO,
|
|
AiGenerationSignal,
|
|
DetectOptions,
|
|
DetectRequest,
|
|
DetectResponse,
|
|
DocumentMetadata,
|
|
InfringementTag,
|
|
InfringementType,
|
|
LegalRiskSignal,
|
|
MatchResult,
|
|
PartialPlagiarismSignal,
|
|
ReviewSummary,
|
|
ScoreBreakdown,
|
|
)
|
|
from app.core.config import Settings, get_settings
|
|
from app.engine.autobiography_filter import preprocess_for_autobiography
|
|
from app.engine.ai_detector import get_ai_detector
|
|
from app.engine.clustering import ClusterIndex
|
|
from app.engine.corpus import load_corpus
|
|
from app.engine.extractor import Extractor, get_extractor
|
|
from app.engine.lsh_filter import LshIndex
|
|
from app.engine.legal_risk import LegalRiskEngine, load_precedents
|
|
from app.engine.persistent_index import PersistentCorpusIndex, PersistentHit
|
|
from app.engine.provenance import DocumentRecord, SegmentRecord, stable_id
|
|
from app.engine.source_extraction import chunk_with_offsets
|
|
from app.engine.similarity import (
|
|
DualSimilarityIndex,
|
|
SimilarityHit,
|
|
build_text_backend,
|
|
)
|
|
from app.engine.structural import extract_lemmas
|
|
from app.engine.taxonomy import Taxonomy, load_taxonomy
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PlagiarismDetector:
|
|
def __init__(self, settings: Settings | None = None, extractor: Extractor | None = None):
|
|
self.settings = settings or get_settings()
|
|
self._extractor: Extractor = extractor or get_extractor(self.settings)
|
|
self.taxonomy: Taxonomy | None = load_taxonomy(self.settings.taxonomy_path)
|
|
self._legal_engine = LegalRiskEngine(load_precedents(self.settings.precedent_path))
|
|
self._ai_detector = get_ai_detector(
|
|
self.settings.ai_detector_model_path,
|
|
allow_heuristic=self.settings.ai_detector_allow_heuristic,
|
|
use_pos=self.settings.ai_detector_use_pos,
|
|
)
|
|
self._persistent: PersistentCorpusIndex | None = None
|
|
|
|
if self.settings.use_persistent_index:
|
|
candidate = PersistentCorpusIndex(
|
|
self.settings.corpus_db, self.settings.persistent_index_path,
|
|
)
|
|
if candidate.ready:
|
|
self._persistent = candidate.load()
|
|
self._corpus = []
|
|
self._corpus_preprocessed_texts = []
|
|
self._corpus_elements = []
|
|
self._corpus_lemmas = []
|
|
self._index = None
|
|
self._lsh = None
|
|
self._cluster = None
|
|
self._docs_by_id = {}
|
|
logger.info("Loaded persistent index: %d segments", self._persistent.size)
|
|
return
|
|
logger.warning(
|
|
"USE_PERSISTENT_INDEX=true but index is absent; falling back to text corpus: %s",
|
|
self.settings.persistent_index_path,
|
|
)
|
|
|
|
self._corpus = load_corpus(self.settings.corpus_path)
|
|
|
|
# 자서전 모드면 코퍼스도 동일 전처리 적용 후 인덱싱
|
|
logger.info("Building corpus indexes (autobiography_mode=%s)", self.settings.autobiography_mode)
|
|
if self.settings.autobiography_mode:
|
|
self._corpus_preprocessed_texts = [
|
|
preprocess_for_autobiography(
|
|
d.text,
|
|
self.settings.autobiography_patterns_path,
|
|
self.settings.enable_entity_masking,
|
|
)
|
|
for d in self._corpus
|
|
]
|
|
else:
|
|
self._corpus_preprocessed_texts = [d.text for d in self._corpus]
|
|
|
|
# 정밀 비교용 인덱스 (전처리된 텍스트 사용)
|
|
from app.engine.corpus import ReferenceDoc
|
|
preprocessed_docs = [
|
|
ReferenceDoc(doc_id=d.doc_id, title=d.title, text=pt)
|
|
for d, pt in zip(self._corpus, self._corpus_preprocessed_texts)
|
|
]
|
|
self._corpus_elements = [self._extractor.extract(t) for t in self._corpus_preprocessed_texts]
|
|
self._corpus_lemmas = [extract_lemmas(t) for t in self._corpus_preprocessed_texts]
|
|
text_backend = build_text_backend(preprocessed_docs, self.settings)
|
|
self._index = DualSimilarityIndex(
|
|
docs=preprocessed_docs,
|
|
doc_elements=self._corpus_elements,
|
|
doc_lemmas=self._corpus_lemmas,
|
|
settings=self.settings,
|
|
text_backend=text_backend,
|
|
)
|
|
|
|
# 1차 LSH 필터 (PDF VII-3)
|
|
self._lsh: LshIndex | None = None
|
|
if self.settings.use_lsh_filter:
|
|
self._lsh = LshIndex(preprocessed_docs, threshold=self.settings.lsh_threshold)
|
|
|
|
# 2단계 고도화: 요소 군집화 인덱스 (요소 교체 부분 표절 신호)
|
|
self._cluster: ClusterIndex | None = None
|
|
if self.settings.use_clustering:
|
|
self._cluster = ClusterIndex(
|
|
doc_ids=[d.doc_id for d in self._corpus],
|
|
doc_elements=self._corpus_elements,
|
|
doc_lemmas=self._corpus_lemmas,
|
|
link_threshold=self.settings.cluster_link_threshold,
|
|
)
|
|
|
|
# source_doc → ReferenceDoc 매핑
|
|
self._docs_by_id = {d.doc_id: d for d in self._corpus}
|
|
|
|
@property
|
|
def corpus_size(self) -> int:
|
|
return self._persistent.size if self._persistent else len(self._corpus)
|
|
|
|
@property
|
|
def corpus_document_count(self) -> int:
|
|
if self._persistent:
|
|
return self._persistent.document_count
|
|
return len(self._corpus)
|
|
|
|
@property
|
|
def index_backend(self) -> str:
|
|
return "persistent-hashing-char-3-4" if self._persistent else "legacy-in-memory"
|
|
|
|
@property
|
|
def precedent_count(self) -> int:
|
|
return len(self._legal_engine.precedents)
|
|
|
|
@property
|
|
def ai_model_ready(self) -> bool:
|
|
return self._ai_detector.mode == "trained"
|
|
|
|
@property
|
|
def uses_persistent_index(self) -> bool:
|
|
return self._persistent is not None
|
|
|
|
def list_persistent_documents(self) -> list[dict]:
|
|
return self._persistent.store.list_documents() if self._persistent else []
|
|
|
|
def detect(
|
|
self,
|
|
doc_id: str,
|
|
text: str,
|
|
metadata: DocumentMetadata | None = None,
|
|
options: DetectOptions | None = None,
|
|
include_ai_segments: bool = True,
|
|
) -> DetectResponse:
|
|
opts = options or DetectOptions()
|
|
default_threshold = (
|
|
self.settings.persistent_similarity_threshold
|
|
if self._persistent else self.settings.similarity_threshold
|
|
)
|
|
threshold = opts.threshold if opts.threshold is not None else default_threshold
|
|
|
|
# 요청 단위 자서전 모드 override
|
|
autobio_mode = (
|
|
self.settings.autobiography_mode if opts.autobiography_mode is None
|
|
else opts.autobiography_mode
|
|
)
|
|
|
|
# 자서전 모드 전처리
|
|
query_text = (
|
|
preprocess_for_autobiography(
|
|
text, self.settings.autobiography_patterns_path,
|
|
self.settings.enable_entity_masking,
|
|
)
|
|
if autobio_mode else text
|
|
)
|
|
|
|
# 요소 추출 (원본 텍스트 기준 — 사용자 검토용)
|
|
elements = self._extractor.extract(text)
|
|
|
|
# 영속 CPU 인덱스: 전량 행렬곱으로 후보를 구하고 상위 후보만 증거 비교한다.
|
|
persistent_hits: list[PersistentHit] = []
|
|
if self._persistent:
|
|
persistent_query_lemmas = extract_lemmas(text)
|
|
persistent_hits = self._persistent.query(
|
|
text,
|
|
top_k=max(opts.top_k, self.settings.persistent_rerank_top_k),
|
|
)
|
|
hits = [
|
|
self._persistent_to_similarity_hit(h, persistent_query_lemmas, elements)
|
|
for h in persistent_hits
|
|
]
|
|
hits.sort(key=lambda h: h.score, reverse=True)
|
|
candidates_count = len(hits)
|
|
lsh_jaccards: dict[str, float] = {}
|
|
else:
|
|
hits = []
|
|
candidates_count = None
|
|
lsh_jaccards = {}
|
|
|
|
# 1차 LSH 필터 (레거시 옵션)
|
|
candidate_ids: set[str] | None = None
|
|
if not self._persistent and self._lsh:
|
|
cands = self._lsh.query(query_text, top_k=self.settings.lsh_top_k)
|
|
candidate_ids = {c.doc_id for c in cands}
|
|
candidates_count = len(cands)
|
|
lsh_jaccards = {c.doc_id: c.jaccard for c in cands}
|
|
|
|
# 정밀 비교 (LSH 후보가 있으면 그것만, 없으면 풀스캔)
|
|
if not self._persistent:
|
|
hits = self._index.query(query_text, elements, top_k=opts.top_k)
|
|
if candidate_ids is not None:
|
|
hits = [h for h in hits if h.doc_id in candidate_ids]
|
|
|
|
# 군집화 부분 표절 신호용 query lemma (전처리 텍스트 기준)
|
|
query_lemmas = extract_lemmas(query_text) if self._cluster else None
|
|
|
|
persistent_by_id = {h.segment_id: h for h in persistent_hits}
|
|
matches = []
|
|
for h in hits:
|
|
provenance_hit = persistent_by_id.get(h.doc_id)
|
|
exact_partial_match = bool(
|
|
provenance_hit and (
|
|
provenance_hit.longest_span >= self.settings.persistent_min_exact_span
|
|
or provenance_hit.coverage >= self.settings.persistent_min_coverage
|
|
)
|
|
)
|
|
if (h.score < threshold and not exact_partial_match) or len(matches) >= opts.top_k:
|
|
continue
|
|
match = self._to_match(
|
|
h, opts.return_evidence, lsh_jaccards.get(h.doc_id),
|
|
self._partial_signal(h.doc_id, elements, query_lemmas),
|
|
)
|
|
if provenance_hit:
|
|
match = self._add_provenance(match, provenance_hit)
|
|
matches.append(match)
|
|
confidence = matches[0].similarity if matches else (hits[0].score if hits else 0.0)
|
|
is_infringement = bool(matches) # 후방호환 필드. 법적 확정이 아니라 임계 초과 매칭.
|
|
ccl_basis = self._build_ccl_basis(matches) if is_infringement else None
|
|
|
|
top_coverage = max((m.matched_coverage for m in matches), default=0.0)
|
|
top_longest = max((m.longest_span for m in matches), default=0)
|
|
legal_tags = [t.tag for m in matches for t in m.tags]
|
|
legal = self._legal_engine.assess(
|
|
max_similarity=matches[0].similarity if matches else 0.0,
|
|
coverage=top_coverage,
|
|
longest_span=top_longest,
|
|
legal_tags=legal_tags,
|
|
work_type="literary",
|
|
)
|
|
|
|
# AI 탐지는 전처리 전 raw text에서만 실행한다.
|
|
ai_result = self._ai_detector.detect(text, with_segments=include_ai_segments)
|
|
ai_signal = AiGenerationSignal(
|
|
suspicion_level=ai_result.suspicion_level or "unknown",
|
|
score=ai_result.score,
|
|
available=ai_result.available,
|
|
provenance=ai_result.provenance,
|
|
is_stub=ai_result.is_stub,
|
|
model_version=ai_result.model_version,
|
|
feature_set_version=ai_result.feature_set_version,
|
|
pos_available=ai_result.pos_available,
|
|
warnings=ai_result.warnings,
|
|
segments=[{
|
|
"index": s.index, "start": s.start, "end": s.end,
|
|
"char_count": s.char_count, "score": s.score,
|
|
"suspicion_level": s.suspicion_level, "scored": s.scored,
|
|
"note": s.note,
|
|
} for s in ai_result.segments],
|
|
top_contributions=[
|
|
{"feature": name, "contribution": round(value, 4)}
|
|
for name, value in ai_result.top_contributions
|
|
],
|
|
note=ai_result.note,
|
|
)
|
|
sim_pct = _calibrate_similarity(confidence, threshold)
|
|
review = ReviewSummary(
|
|
originality_percent=100 - sim_pct,
|
|
similarity_percent=sim_pct,
|
|
similar_sentence_count=len(matches),
|
|
compared_count=self.corpus_size,
|
|
has_suspicion=is_infringement,
|
|
ai_suspicion_level=ai_signal.suspicion_level,
|
|
)
|
|
|
|
return DetectResponse(
|
|
doc_id=doc_id,
|
|
is_infringement=is_infringement,
|
|
confidence=round(confidence, 4),
|
|
extracted_elements=elements,
|
|
matches=matches,
|
|
ccl_basis=ccl_basis,
|
|
review_summary=review,
|
|
ai_generation=ai_signal,
|
|
has_similarity_match=bool(matches),
|
|
corpus_scope_note=(
|
|
f"현재 등록된 {self.corpus_document_count}개 원천 문서의 "
|
|
f"{self.corpus_size}개 검색 세그먼트만 대조했습니다. 미매칭은 비침해 확정이 아닙니다."
|
|
),
|
|
legal_risk=LegalRiskSignal(
|
|
status=legal.status,
|
|
risk_level=legal.risk_level,
|
|
similarity_evidence=legal.similarity_evidence,
|
|
protected_expression=legal.protected_expression,
|
|
access_evidence=legal.access_evidence,
|
|
missing_factors=list(legal.missing_factors),
|
|
precedent_ids=list(legal.precedent_ids),
|
|
disclaimer=legal.disclaimer,
|
|
),
|
|
autobiography_mode=autobio_mode,
|
|
candidates_before_filter=candidates_count,
|
|
engine_version=self.settings.engine_version,
|
|
analyzed_at=datetime.now(timezone.utc),
|
|
)
|
|
|
|
def detect_request(self, req: DetectRequest) -> DetectResponse:
|
|
return self.detect(req.doc_id, req.text, req.metadata, req.options)
|
|
|
|
def _persistent_to_similarity_hit(
|
|
self, hit: PersistentHit, query_lemmas: list[str], query_elements,
|
|
) -> SimilarityHit:
|
|
from app.api.schemas import EvidenceSpan
|
|
from app.engine.similarity import _element_similarities
|
|
from app.engine.structural import lemma_overlap_ratio
|
|
|
|
reference_elements = self._extractor.extract(hit.reference_text)
|
|
element_sim = _element_similarities(query_elements, reference_elements)
|
|
lemma_sim = lemma_overlap_ratio(query_lemmas, extract_lemmas(hit.reference_text))
|
|
s = self.settings
|
|
combined = (
|
|
s.weight_text_sim * hit.score
|
|
+ s.weight_lemma_sim * lemma_sim
|
|
+ s.weight_char_sim * element_sim["characters"]
|
|
+ s.weight_motif_sim * element_sim["motifs"]
|
|
)
|
|
|
|
return SimilarityHit(
|
|
doc_id=hit.segment_id,
|
|
title=hit.title,
|
|
score=combined,
|
|
text_sim=hit.score,
|
|
lemma_sim=lemma_sim,
|
|
element_sim=element_sim,
|
|
evidence=[EvidenceSpan(**span) for span in hit.evidence],
|
|
)
|
|
|
|
@staticmethod
|
|
def _add_provenance(match: MatchResult, hit: PersistentHit) -> MatchResult:
|
|
return match.model_copy(update={
|
|
"source_document_id": hit.document_id,
|
|
"source_segment_id": hit.segment_id,
|
|
"source_locator": hit.source_locator,
|
|
"coordinate_scope": hit.coordinate_scope,
|
|
"page_number": hit.page_number,
|
|
"paragraph_number": hit.paragraph_number,
|
|
"source_char_start": hit.source_char_start,
|
|
"source_char_end": hit.source_char_end,
|
|
"matched_coverage": round(hit.coverage, 4),
|
|
"longest_span": hit.longest_span,
|
|
})
|
|
|
|
def add_persistent_document(self, doc_id: str | None, title: str, text: str) -> str:
|
|
if not self._persistent:
|
|
raise RuntimeError("persistent index is not enabled")
|
|
document_id = (doc_id or stable_id("doc", title)).strip()
|
|
self._persistent.store.upsert_document(DocumentRecord(document_id=document_id, title=title))
|
|
clean_text = text.strip()
|
|
segments = [SegmentRecord(
|
|
segment_id=stable_id("seg", document_id, str(start), chunk),
|
|
document_id=document_id, text=chunk, ordinal=str(i),
|
|
coordinate_scope="document", char_start=start, char_end=end,
|
|
source_locator=f"api://corpus/{document_id}#chars={start}-{end}",
|
|
metadata={"provenance_quality": "api_document"},
|
|
) for i, (start, end, chunk) in enumerate(
|
|
chunk_with_offsets(clean_text, size=1000, stride=500), 1
|
|
)]
|
|
inserted, duplicates = self._persistent.store.add_segments(segments)
|
|
if duplicates and not inserted:
|
|
raise FileExistsError(f"doc_id '{document_id}' 또는 동일 본문이 이미 존재합니다")
|
|
self._replace_persistent_index()
|
|
return document_id
|
|
|
|
def delete_persistent_document(self, doc_id: str) -> bool:
|
|
if not self._persistent:
|
|
raise RuntimeError("persistent index is not enabled")
|
|
deleted = self._persistent.store.delete_document(doc_id)
|
|
if deleted:
|
|
self._replace_persistent_index()
|
|
return deleted
|
|
|
|
def _replace_persistent_index(self) -> None:
|
|
"""새 객체를 완성한 뒤 포인터를 한 번에 교체해 동시 질의 정합성을 보장."""
|
|
replacement = PersistentCorpusIndex(
|
|
self.settings.corpus_db, self.settings.persistent_index_path,
|
|
)
|
|
replacement.sync()
|
|
replacement.load()
|
|
self._persistent = replacement
|
|
|
|
def _partial_signal(self, doc_id, query_elements, query_lemmas) -> PartialPlagiarismSignal | None:
|
|
"""군집화 기반 요소별 부분 표절 분해 (옵션)."""
|
|
if not self._cluster:
|
|
return None
|
|
sig = self._cluster.partial_signal(doc_id, query_elements, query_lemmas)
|
|
if sig is None:
|
|
return None
|
|
return PartialPlagiarismSignal(
|
|
cluster_id=sig.cluster_id,
|
|
verdict=sig.verdict,
|
|
signature_score=sig.signature_score,
|
|
per_element=sig.per_element,
|
|
retained_elements=sig.retained_elements,
|
|
changed_elements=sig.changed_elements,
|
|
)
|
|
|
|
def _to_match(
|
|
self,
|
|
hit: SimilarityHit,
|
|
return_evidence: bool,
|
|
lsh_j: float | None,
|
|
partial: PartialPlagiarismSignal | None = None,
|
|
) -> MatchResult:
|
|
legacy_type = _classify_legacy(hit)
|
|
tags = self._assign_tags(hit, legacy_type)
|
|
case = self.taxonomy.find_case([t.tag for t in tags if t.role == "primary"]) if self.taxonomy else None
|
|
|
|
return MatchResult(
|
|
source_doc=hit.doc_id,
|
|
source_title=hit.title,
|
|
similarity=round(hit.score, 4),
|
|
tags=tags,
|
|
case_id=case.case_id if case else None,
|
|
case_title=case.title if case else None,
|
|
infringement_type=legacy_type,
|
|
evidence_spans=hit.evidence if return_evidence else [],
|
|
score_breakdown=ScoreBreakdown(
|
|
text_sim=round(hit.text_sim, 4),
|
|
lemma_sim=round(hit.lemma_sim, 4),
|
|
character_sim=round(hit.element_sim.get("characters", 0.0), 4),
|
|
motif_sim=round(hit.element_sim.get("motifs", 0.0), 4),
|
|
lsh_jaccard=round(lsh_j, 4) if lsh_j is not None else None,
|
|
),
|
|
partial_signal=partial,
|
|
)
|
|
|
|
def _assign_tags(self, hit: SimilarityHit, legacy: InfringementType) -> list[InfringementTag]:
|
|
"""삼중 유사도 분포 → 10종 법령 태그(주/보조) 매핑.
|
|
|
|
규칙(PDF IX장 매핑표 기반):
|
|
- copy/패러프레이즈 수준 표절 (lemma↑ or text↑) → 복제권(주) + 공중송신권(보조)
|
|
- lemma만 매우 높음 → 인용 표시 누락(주 보조)
|
|
- 인물 일치도 매우 높음(서사·구조 차용) → 2차적저작물작성권(주) + 자기창작인양표시(보조)
|
|
- 구조 미달 가공 신호 (text 낮음 + lemma 중간) → 2차적저작물 미달 가공
|
|
"""
|
|
text_sim = hit.text_sim
|
|
lemma_sim = hit.lemma_sim
|
|
char_sim = hit.element_sim.get("characters", 0.0)
|
|
motif_sim = hit.element_sim.get("motifs", 0.0)
|
|
|
|
primary: list[str] = []
|
|
secondary: list[str] = []
|
|
|
|
# 복제권: 표면 또는 lemma가 강하게 일치
|
|
if lemma_sim >= 0.70 or text_sim >= 0.70:
|
|
primary.append("reproduction")
|
|
secondary.append("public_transmission") # 전자책 게재 가정
|
|
# 표절 실무 - 인용 누락
|
|
primary.append("citation_missing")
|
|
|
|
# 2차적저작물작성권: 구조·서사 차용 (인물/모티프 일치 + 표면은 낮음)
|
|
elif (char_sim >= 0.40 or motif_sim >= 0.50) and text_sim < 0.40:
|
|
primary.append("derivative_work")
|
|
secondary.append("attribution")
|
|
secondary.append("citation_missing")
|
|
# 미달 가공 가능성
|
|
if text_sim < 0.30 and lemma_sim < 0.50:
|
|
secondary.append("substandard_derivative")
|
|
|
|
# 부분 변형 (text 중간 + 인물 일치)
|
|
elif text_sim >= 0.40 and char_sim >= 0.30:
|
|
primary.append("reproduction")
|
|
secondary.append("derivative_work")
|
|
secondary.append("citation_missing")
|
|
|
|
# 낮은 매칭이지만 임계 통과한 경우
|
|
else:
|
|
secondary.append("reproduction")
|
|
|
|
# 중복 제거 + 태그 객체화
|
|
primary = list(dict.fromkeys(primary))
|
|
secondary = [s for s in dict.fromkeys(secondary) if s not in primary]
|
|
|
|
out: list[InfringementTag] = []
|
|
for t in primary:
|
|
out.append(InfringementTag(tag=t, role="primary", label_ko=TAG_LABEL_KO[t]))
|
|
for t in secondary:
|
|
out.append(InfringementTag(tag=t, role="secondary", label_ko=TAG_LABEL_KO[t]))
|
|
return out
|
|
|
|
def _build_ccl_basis(self, matches: list[MatchResult]) -> str:
|
|
top = matches[0]
|
|
sb = top.score_breakdown
|
|
breakdown = ""
|
|
if sb:
|
|
breakdown = (
|
|
f" [text={sb.text_sim:.2f} / lemma={sb.lemma_sim:.2f} "
|
|
f"/ char={sb.character_sim:.2f} / motif={sb.motif_sim:.2f}]"
|
|
)
|
|
primary_labels = [t.label_ko for t in top.tags if t.role == "primary"]
|
|
tag_summary = ", ".join(primary_labels) if primary_labels else "확인 필요"
|
|
case_part = f" 추정 케이스 {top.case_id} ({top.case_title})." if top.case_id else ""
|
|
return (
|
|
f"'{top.source_title}'와 검색 유사도 {top.similarity:.2%}로 후보 매칭. "
|
|
f"검토 가설 태그: {tag_summary}.{case_part}{breakdown} "
|
|
"법적 침해 여부는 보호되는 표현·의거관계·권리관계를 별도 검토해야 합니다."
|
|
)
|
|
|
|
|
|
def _calibrate_similarity(raw: float, threshold: float) -> int:
|
|
"""결합 유사도(0~1) → 사용자 표시용 '유사도 %' 캘리브레이션.
|
|
|
|
임베딩 성분 때문에 무관한 글도 raw 0.5 전후가 나오므로, 그대로 %로 쓰면
|
|
깨끗한 글이 '유사도 50%'로 보인다. 아래 구간 변환으로 무관한 글은 0~5%,
|
|
임계 초과(실제 표절)만 40% 이상으로 눌러준다.
|
|
"""
|
|
floor = 0.55 # 무관한 글의 전형적 결합 유사도 상한
|
|
if raw <= floor:
|
|
disp = (raw / floor) * 5.0 if floor else 0.0
|
|
elif raw <= threshold:
|
|
disp = 5.0 + (raw - floor) / max(1e-6, threshold - floor) * 35.0
|
|
else:
|
|
disp = 40.0 + (raw - threshold) / max(1e-6, 1.0 - threshold) * 60.0
|
|
return max(0, min(100, round(disp)))
|
|
|
|
|
|
def _classify_legacy(hit: SimilarityHit) -> InfringementType:
|
|
"""후방 호환 - 단일 enum 분류 (UI/기존 통합 코드용)."""
|
|
elem = hit.element_sim
|
|
char_sim = elem.get("characters", 0.0)
|
|
motif_sim = elem.get("motifs", 0.0)
|
|
|
|
if hit.lemma_sim >= 0.70:
|
|
return "copy"
|
|
if hit.text_sim >= 0.70:
|
|
return "copy"
|
|
if hit.text_sim >= 0.40 and char_sim >= 0.30:
|
|
return "transform"
|
|
if hit.lemma_sim >= 0.40 and char_sim < 0.20:
|
|
return "plot"
|
|
if motif_sim >= 0.50 and char_sim < 0.20:
|
|
return "plot"
|
|
if char_sim >= 0.40:
|
|
return "character"
|
|
return "unknown"
|
|
|
|
|
|
# 후방 호환 alias (테스트가 _classify import)
|
|
_classify = _classify_legacy
|