226 lines
9.1 KiB
Python
226 lines
9.1 KiB
Python
"""군집화 기반 표절 판별 (계획서 2단계 표절 검출 기술 고도화, p.21).
|
|
|
|
목표: 기존 pairwise 삼중 유사도만으로는 "일부 등장인물만 바꾸거나 특정 요소만 바꾼
|
|
표절"의 표절률을 수치화하기 어렵다. 계획서는 고도화 전략으로 다음을 명시한다.
|
|
|
|
> 분류된 요소가 비슷한 텍스트들을 군집화하여 해당 군집 내의 요소 간 유사도를
|
|
> 비교하는 방식으로 표절 기술을 고도화. ... 일부 등장 인물만 바꾸거나 특정 요소만
|
|
> 바꾼 표절률도 표절여부의 수치화가 가능.
|
|
|
|
본 모듈은 데이터 없이(현 코퍼스만으로) 동작하는 군집화 1차 라우팅 계층을 제공한다.
|
|
|
|
설계:
|
|
1) 코퍼스 문서를 요소(인물/모티프/키워드) + lemma 시그니처로 묶어 군집 생성
|
|
(그래프 connected-components: 요소 자카드 ≥ link_threshold 이면 같은 군집).
|
|
2) query 를 가장 가까운 군집으로 라우팅 → 군집 내 문서끼리만 정밀 비교 (탐색량 감소).
|
|
3) 군집 내 "요소별 부분 표절 점수" 산출 — 어떤 요소(인물/모티프/키워드/lemma)가
|
|
얼마나 겹치는지를 분해해, 인물만 바꾼 표절을 별도 신호로 노출.
|
|
|
|
순수 함수 위주로 작성되어 단위테스트가 쉽다. detector 에 옵션으로 결합한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from app.api.schemas import ExtractedElements
|
|
|
|
|
|
def _jaccard(a: set[str], b: set[str]) -> float:
|
|
if not a and not b:
|
|
return 0.0
|
|
return len(a & b) / max(1, len(a | b))
|
|
|
|
|
|
def _element_signature(elem: ExtractedElements, lemmas: list[str] | None = None) -> dict[str, set[str]]:
|
|
"""문서를 요소별 집합 시그니처로 변환 (소문자 정규화)."""
|
|
sig = {
|
|
"characters": {c.lower() for c in elem.characters},
|
|
"motifs": {m.lower() for m in elem.motifs},
|
|
"keywords": {k.lower() for k in elem.keywords},
|
|
"genre": {elem.genre.lower()} if elem.genre else set(),
|
|
}
|
|
if lemmas is not None:
|
|
sig["lemmas"] = set(lemmas)
|
|
return sig
|
|
|
|
|
|
# 군집 링크/요소 표절 점수에 쓰는 요소 가중치 (lemma·키워드 = 본문 차용, 인물·모티프 = 구조 차용)
|
|
_SIGNATURE_WEIGHTS = {
|
|
"lemmas": 0.40,
|
|
"keywords": 0.25,
|
|
"characters": 0.15,
|
|
"motifs": 0.15,
|
|
"genre": 0.05,
|
|
}
|
|
|
|
|
|
def signature_similarity(a: dict[str, set[str]], b: dict[str, set[str]]) -> float:
|
|
"""두 시그니처의 가중 자카드 결합 (군집 링크 판정용)."""
|
|
total_w = 0.0
|
|
acc = 0.0
|
|
for key, w in _SIGNATURE_WEIGHTS.items():
|
|
if key not in a or key not in b:
|
|
continue
|
|
# 양쪽 모두 비어있는 요소는 정보가 없으므로 제외 (중립)
|
|
if not a[key] and not b[key]:
|
|
continue
|
|
total_w += w
|
|
acc += w * _jaccard(a[key], b[key])
|
|
return acc / total_w if total_w else 0.0
|
|
|
|
|
|
@dataclass
|
|
class Cluster:
|
|
cluster_id: int
|
|
members: list[str] = field(default_factory=list) # doc_id 리스트
|
|
centroid: dict[str, set[str]] = field(default_factory=dict) # 요소별 합집합 시그니처
|
|
|
|
|
|
@dataclass
|
|
class PartialPlagiarismSignal:
|
|
"""요소별 부분 표절 분해 — '무엇을 그대로 두고 무엇만 바꿨는지'."""
|
|
cluster_id: int
|
|
per_element: dict[str, float] # characters/motifs/keywords/lemmas/genre 별 자카드
|
|
signature_score: float # 가중 결합
|
|
changed_elements: list[str] # 거의 안 겹치는(=바꾼) 요소
|
|
retained_elements: list[str] # 강하게 겹치는(=유지한) 요소
|
|
verdict: str # "element_swap_plagiarism" / "near_duplicate" / "weak" / "none"
|
|
|
|
|
|
class ClusterIndex:
|
|
"""코퍼스 요소 시그니처를 군집화하고 query 를 라우팅한다."""
|
|
|
|
def __init__(
|
|
self,
|
|
doc_ids: list[str],
|
|
doc_elements: list[ExtractedElements],
|
|
doc_lemmas: list[list[str]] | None = None,
|
|
link_threshold: float = 0.35,
|
|
):
|
|
if doc_lemmas is not None and len(doc_lemmas) != len(doc_ids):
|
|
raise ValueError("doc_lemmas length mismatch")
|
|
self.link_threshold = link_threshold
|
|
self._doc_ids = doc_ids
|
|
self._sigs: dict[str, dict[str, set[str]]] = {
|
|
did: _element_signature(elem, doc_lemmas[i] if doc_lemmas else None)
|
|
for i, (did, elem) in enumerate(zip(doc_ids, doc_elements))
|
|
}
|
|
self.clusters: list[Cluster] = self._build_clusters()
|
|
self._cluster_of: dict[str, int] = {
|
|
did: c.cluster_id for c in self.clusters for did in c.members
|
|
}
|
|
|
|
# ---------- 군집 구성 (그래프 연결요소) ----------
|
|
|
|
def _build_clusters(self) -> list[Cluster]:
|
|
ids = self._doc_ids
|
|
parent = {d: d for d in ids}
|
|
|
|
def find(x: str) -> str:
|
|
while parent[x] != x:
|
|
parent[x] = parent[parent[x]]
|
|
x = parent[x]
|
|
return x
|
|
|
|
def union(x: str, y: str) -> None:
|
|
parent[find(x)] = find(y)
|
|
|
|
for i in range(len(ids)):
|
|
for j in range(i + 1, len(ids)):
|
|
if signature_similarity(self._sigs[ids[i]], self._sigs[ids[j]]) >= self.link_threshold:
|
|
union(ids[i], ids[j])
|
|
|
|
groups: dict[str, list[str]] = {}
|
|
for d in ids:
|
|
groups.setdefault(find(d), []).append(d)
|
|
|
|
clusters: list[Cluster] = []
|
|
for cid, members in enumerate(groups.values()):
|
|
centroid: dict[str, set[str]] = {}
|
|
for m in members:
|
|
for key, s in self._sigs[m].items():
|
|
centroid.setdefault(key, set()).update(s)
|
|
clusters.append(Cluster(cluster_id=cid, members=members, centroid=centroid))
|
|
return clusters
|
|
|
|
def cluster_of(self, doc_id: str) -> int | None:
|
|
return self._cluster_of.get(doc_id)
|
|
|
|
# ---------- query 라우팅 + 부분 표절 분해 ----------
|
|
|
|
def route(self, query_elem: ExtractedElements, query_lemmas: list[str] | None = None) -> Cluster | None:
|
|
"""query 와 가장 유사한 군집 반환 (centroid 가중 자카드 최대)."""
|
|
if not self.clusters:
|
|
return None
|
|
qsig = _element_signature(query_elem, query_lemmas)
|
|
return max(self.clusters, key=lambda c: signature_similarity(qsig, c.centroid))
|
|
|
|
def candidate_ids(self, query_elem: ExtractedElements, query_lemmas: list[str] | None = None) -> set[str]:
|
|
"""라우팅된 군집의 멤버 doc_id (정밀 비교 후보 축소용)."""
|
|
c = self.route(query_elem, query_lemmas)
|
|
return set(c.members) if c else set()
|
|
|
|
def partial_signal(
|
|
self,
|
|
doc_id: str,
|
|
query_elem: ExtractedElements,
|
|
query_lemmas: list[str] | None = None,
|
|
retain_threshold: float = 0.6,
|
|
change_threshold: float = 0.2,
|
|
) -> PartialPlagiarismSignal | None:
|
|
"""특정 코퍼스 문서 대비 요소별 부분 표절 신호 분해.
|
|
|
|
인물만 바꾸고 본문(lemma)·키워드는 유지한 표절을 element_swap_plagiarism 으로 식별.
|
|
"""
|
|
if doc_id not in self._sigs:
|
|
return None
|
|
qsig = _element_signature(query_elem, query_lemmas)
|
|
dsig = self._sigs[doc_id]
|
|
|
|
per_element: dict[str, float] = {}
|
|
for key in _SIGNATURE_WEIGHTS:
|
|
if key in qsig and key in dsig:
|
|
# 양쪽 모두 비어있으면 신호 없음 → 제외 (거짓 '변경' 방지)
|
|
if not qsig[key] and not dsig[key]:
|
|
continue
|
|
per_element[key] = round(_jaccard(qsig[key], dsig[key]), 4)
|
|
|
|
sig_score = signature_similarity(qsig, dsig)
|
|
retained = [k for k, v in per_element.items() if v >= retain_threshold]
|
|
changed = [k for k, v in per_element.items() if v <= change_threshold]
|
|
|
|
content_retained = any(k in retained for k in ("lemmas", "keywords"))
|
|
structure_changed = any(k in changed for k in ("characters", "motifs"))
|
|
|
|
high_content = per_element.get("lemmas", 0.0) >= 0.85 or per_element.get("keywords", 0.0) >= 0.85
|
|
if content_retained and structure_changed:
|
|
verdict = "element_swap_plagiarism" # 본문 유지 + 인물/모티프만 교체 (구조 차용)
|
|
elif high_content:
|
|
verdict = "near_duplicate" # 본문·구조 모두 유지 = 사실상 복제
|
|
elif sig_score >= change_threshold:
|
|
verdict = "weak"
|
|
else:
|
|
verdict = "none"
|
|
|
|
cid = self._cluster_of.get(doc_id, -1)
|
|
return PartialPlagiarismSignal(
|
|
cluster_id=cid,
|
|
per_element=per_element,
|
|
signature_score=round(sig_score, 4),
|
|
changed_elements=changed,
|
|
retained_elements=retained,
|
|
verdict=verdict,
|
|
)
|
|
|
|
@property
|
|
def num_clusters(self) -> int:
|
|
return len(self.clusters)
|
|
|
|
def summary(self) -> list[dict]:
|
|
"""군집 구성 요약 (디버그/리포트용)."""
|
|
return [
|
|
{"cluster_id": c.cluster_id, "size": len(c.members), "members": c.members}
|
|
for c in sorted(self.clusters, key=lambda x: len(x.members), reverse=True)
|
|
]
|