Merge: 2단계 고도화 선행 작업 (군집화·요약·평가환경·데이터 스펙)

app/api/routes.py

@@ -15,11 +15,14 @@ from app.api.schemas import (
     DetectRequest,
     DetectResponse,
     HealthResponse,
+    SummaryRequest,
+    SummaryResponse,
     TaxonomyResponse,
 )
 from app.core.config import get_settings
 from app.engine.corpus import add_document, delete_document, list_documents
 from app.engine.detector import PlagiarismDetector
+from app.engine.summarizer import get_summarizer
 from app.jobs.store import JobStore
 
 router = APIRouter(prefix="/v1")
@@ -83,6 +86,35 @@ async def detect(req: DetectRequest, request: Request) -> DetectResponse:
     return _detector(request).detect_request(req)
 
 
+@router.post(
+    "/summary",
+    response_model=SummaryResponse,
+    tags=["summary"],
+)
+async def summarize(req: SummaryRequest) -> SummaryResponse:
+    """스토리 요약/분석 (계획서 과제2 ②). 추출적 요약은 외부 의존 0으로 동작.
+
+    use_abstractive=True 이고 LLM 키가 있으면 통합(hybrid) 요약, 아니면 추출적 요약 반환.
+    """
+    settings = get_settings()
+    result = get_summarizer(settings).summarize(
+        req.text,
+        ratio=req.ratio,
+        max_sentences=req.max_sentences,
+        use_abstractive=req.use_abstractive,
+    )
+    return SummaryResponse(
+        extractive=result.extractive,
+        abstractive=result.abstractive,
+        final=result.final,
+        mode=result.mode,
+        selected_indices=result.selected_indices,
+        num_sentences_in=result.num_sentences_in,
+        num_sentences_out=result.num_sentences_out,
+        engine_version=settings.engine_version,
+    )
+
+
 @router.post(
     "/plagiarism/batch",
     response_model=BatchCreatedResponse,

app/api/schemas.py

@@ -83,6 +83,19 @@ class ScoreBreakdown(BaseModel):
     lsh_jaccard: float | None = Field(default=None, ge=0.0, le=1.0)
 
 
+class PartialPlagiarismSignal(BaseModel):
+    """군집화 기반 요소별 부분 표절 분해 (계획서 2단계 고도화).
+
+    '본문은 그대로 두고 인물만 바꾼 표절' 같은 부분 표절을 요소 단위로 수치화.
+    """
+    cluster_id: int
+    verdict: Literal["near_duplicate", "element_swap_plagiarism", "weak", "none"]
+    signature_score: float = Field(..., ge=0.0, le=1.0)
+    per_element: dict[str, float] = Field(default_factory=dict)
+    retained_elements: list[str] = Field(default_factory=list)  # 그대로 유지된 요소
+    changed_elements: list[str] = Field(default_factory=list)    # 바꿔치기한 요소
+
+
 class MatchResult(BaseModel):
     source_doc: str
     source_title: str | None = None
@@ -93,6 +106,7 @@ class MatchResult(BaseModel):
     infringement_type: InfringementType = "unknown"
     evidence_spans: list[EvidenceSpan] = Field(default_factory=list)
     score_breakdown: ScoreBreakdown | None = None
+    partial_signal: PartialPlagiarismSignal | None = None
 
 
 class ExtractedElements(BaseModel):
@@ -144,6 +158,24 @@ class BatchStatusResponse(BaseModel):
     error: str | None = None
 
 
+class SummaryRequest(BaseModel):
+    text: str = Field(..., min_length=1)
+    ratio: float = Field(default=0.3, gt=0.0, le=1.0, description="요약 길이 비율 (입력 문장 대비)")
+    max_sentences: int | None = Field(default=None, ge=1, description="최대 문장 수 (옵션)")
+    use_abstractive: bool = Field(default=True, description="추상적(LLM) 단계 사용 — 키 없으면 추출적 폴백")
+
+
+class SummaryResponse(BaseModel):
+    extractive: str
+    abstractive: str | None = None
+    final: str
+    mode: Literal["extractive", "hybrid"]
+    selected_indices: list[int] = Field(default_factory=list)
+    num_sentences_in: int
+    num_sentences_out: int
+    engine_version: str
+
+
 class HealthResponse(BaseModel):
     status: Literal["ok"]
     engine_version: str

app/core/config.py

@@ -45,6 +45,10 @@ class Settings(BaseSettings):
     lsh_threshold: float = 0.3      # 1차 필터는 느슨하게 (재현율 우선)
     lsh_top_k: int = 50
 
+    # 2단계 고도화: 군집화 기반 부분 표절(요소 교체) 신호 (계획서 p.21)
+    use_clustering: bool = True
+    cluster_link_threshold: float = 0.35
+
     # PDF VII-4 자서전 모드
     autobiography_mode: bool = True
     enable_entity_masking: bool = True

app/engine/case_coverage.py

@@ -0,0 +1,64 @@
+"""39개 침해 케이스 커버리지 갭 분석 (계획서 2단계 정밀도 97% 근거).
+
+목적: 현재 표절 탐지 엔진이 자동 판별 가능한 케이스(detectable_internal=True) 중,
+평가/학습 샘플이 확보된 케이스와 그렇지 않은 케이스를 구분해 컴북스에 요청할
+데이터의 '케이스 단위 갭'을 산출한다. (정밀도 97% 달성은 케이스 커버리지가 전제)
+
+상태 분류:
+  - covered            : 탐지 대상 + 평가 샘플 보유
+  - detectable_no_data : 탐지 대상이나 샘플 없음  → 데이터 요청 대상
+  - out_of_engine_scope: 탐지 모듈 범위 밖(운영/약관/유족 등) → 데이터 불필요
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class CaseCoverage:
+    case_id: str
+    subgroup: str
+    actor: str
+    detectable: bool
+    sample_count: int
+    status: str
+
+
+def analyze_coverage(cases: list, case_sample_counts: dict[str, int] | None = None) -> dict:
+    """taxonomy 케이스 + (옵션) 케이스별 보유 샘플 수 → 커버리지 리포트.
+
+    cases: Taxonomy.cases (case_id/subgroup/actor/detectable_internal 속성 보유)
+    case_sample_counts: {case_id: 보유 샘플 수}. None 이면 모두 0(=데이터 미보유)으로 간주.
+    """
+    counts = case_sample_counts or {}
+    rows: list[CaseCoverage] = []
+    for c in cases:
+        n = counts.get(c.case_id, 0)
+        if not c.detectable_internal:
+            status = "out_of_engine_scope"
+        elif n > 0:
+            status = "covered"
+        else:
+            status = "detectable_no_data"
+        rows.append(CaseCoverage(
+            case_id=c.case_id, subgroup=c.subgroup, actor=c.actor,
+            detectable=c.detectable_internal, sample_count=n, status=status,
+        ))
+
+    detectable = [r for r in rows if r.detectable]
+    covered = [r for r in detectable if r.status == "covered"]
+    need_data = [r for r in detectable if r.status == "detectable_no_data"]
+
+    return {
+        "total_cases": len(rows),
+        "detectable_cases": len(detectable),
+        "covered_cases": len(covered),
+        "need_data_cases": len(need_data),
+        "coverage_ratio": round(len(covered) / len(detectable), 4) if detectable else 0.0,
+        "rows": rows,
+        "data_request_list": [
+            {"case_id": r.case_id, "subgroup": r.subgroup, "actor": r.actor}
+            for r in need_data
+        ],
+    }

app/engine/clustering.py

@@ -0,0 +1,225 @@
+"""군집화 기반 표절 판별 (계획서 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)
+        ]

app/engine/detector.py

@@ -25,10 +25,12 @@ from app.api.schemas import (
     InfringementTag,
     InfringementType,
     MatchResult,
+    PartialPlagiarismSignal,
     ScoreBreakdown,
 )
 from app.core.config import Settings, get_settings
 from app.engine.autobiography_filter import preprocess_for_autobiography
+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
@@ -87,6 +89,16 @@ class PlagiarismDetector:
         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}
 
@@ -137,8 +149,14 @@ class PlagiarismDetector:
         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
+
         matches = [
-            self._to_match(h, opts.return_evidence, lsh_jaccards.get(h.doc_id))
+            self._to_match(
+                h, opts.return_evidence, lsh_jaccards.get(h.doc_id),
+                self._partial_signal(h.doc_id, elements, query_lemmas),
+            )
             for h in hits if h.score >= threshold
         ]
         confidence = matches[0].similarity if matches else (hits[0].score if hits else 0.0)
@@ -161,7 +179,29 @@ class PlagiarismDetector:
     def detect_request(self, req: DetectRequest) -> DetectResponse:
         return self.detect(req.doc_id, req.text, req.metadata, req.options)
 
-    def _to_match(self, hit: SimilarityHit, return_evidence: bool, lsh_j: float | None) -> MatchResult:
+    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
@@ -182,6 +222,7 @@ class PlagiarismDetector:
                 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]:

app/engine/metadata_eval.py

@@ -0,0 +1,101 @@
+"""콘텐츠 요소(메타데이터) 추출 F1 평가 (계획서 성능지표 No.3, p.24).
+
+계획서 평가방식:
+    KLUE 데이터셋의 NER(개체명 인식) 데이터를 활용하여 학습/테스트,
+    KLUE Leaderboard 등재 baseline 대비 상대 평가. 목표 F1 83 이상(top5).
+
+본 모듈은 정답 라벨(gold) 세트가 주어졌을 때 추출기 출력과의 집합 단위
+precision/recall/F1 을 계산한다. 요소 유형(characters/motifs/keywords/genre)
+별로 분리 측정 + 마이크로 평균.
+
+gold 라벨은 다음 두 경로 중 하나로 공급:
+  ① 컴북스 1단계 '콘텐츠 구성요소 정의' 라벨 (article 단위)
+  ② KLUE NER 공개 데이터의 PS(인물) 태그 → characters 평가 (scripts 에서 변환)
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+def _norm(s: str) -> str:
+    return s.strip().lower()
+
+
+@dataclass
+class PRF:
+    precision: float
+    recall: float
+    f1: float
+    tp: int
+    fp: int
+    fn: int
+
+    def as_dict(self) -> dict:
+        return {
+            "precision": round(self.precision, 4),
+            "recall": round(self.recall, 4),
+            "f1": round(self.f1, 4),
+            "tp": self.tp, "fp": self.fp, "fn": self.fn,
+        }
+
+
+def _prf_from_counts(tp: int, fp: int, fn: int) -> PRF:
+    precision = tp / (tp + fp) if (tp + fp) else 0.0
+    recall = tp / (tp + fn) if (tp + fn) else 0.0
+    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
+    return PRF(precision, recall, f1, tp, fp, fn)
+
+
+def set_prf(predicted: list[str], gold: list[str]) -> PRF:
+    """단일 문서 한 요소 유형의 집합 단위 PRF."""
+    p = {_norm(x) for x in predicted if x and x.strip()}
+    g = {_norm(x) for x in gold if x and x.strip()}
+    tp = len(p & g)
+    fp = len(p - g)
+    fn = len(g - p)
+    return _prf_from_counts(tp, fp, fn)
+
+
+# 평가 대상 요소 유형 (genre 는 단일값이라 별도 처리)
+_LIST_FIELDS = ("characters", "motifs", "keywords")
+
+
+def evaluate_extraction(predictions: list[dict], golds: list[dict]) -> dict[str, dict]:
+    """예측/정답 메타데이터 리스트 → 요소 유형별 + 마이크로 평균 F1.
+
+    predictions / golds 각 원소는 {"characters": [...], "motifs": [...],
+    "keywords": [...], "genre": "..."} 형식.
+    """
+    if len(predictions) != len(golds):
+        raise ValueError("predictions/golds length mismatch")
+
+    per_field_counts = {f: [0, 0, 0] for f in _LIST_FIELDS}  # tp, fp, fn
+    genre_tp = genre_total = 0
+
+    for pred, gold in zip(predictions, golds):
+        for f in _LIST_FIELDS:
+            r = set_prf(pred.get(f, []) or [], gold.get(f, []) or [])
+            per_field_counts[f][0] += r.tp
+            per_field_counts[f][1] += r.fp
+            per_field_counts[f][2] += r.fn
+        if gold.get("genre"):
+            genre_total += 1
+            if _norm(str(pred.get("genre") or "")) == _norm(str(gold["genre"])):
+                genre_tp += 1
+
+    result: dict[str, dict] = {}
+    micro = [0, 0, 0]
+    for f in _LIST_FIELDS:
+        tp, fp, fn = per_field_counts[f]
+        result[f] = _prf_from_counts(tp, fp, fn).as_dict()
+        micro[0] += tp
+        micro[1] += fp
+        micro[2] += fn
+
+    result["micro_avg"] = _prf_from_counts(*micro).as_dict()
+    result["genre_accuracy"] = {
+        "accuracy": round(genre_tp / genre_total, 4) if genre_total else None,
+        "correct": genre_tp, "total": genre_total,
+    }
+    return result

app/engine/preference.py

@@ -0,0 +1,106 @@
+"""Human Feedback 기반 Preference Optimization 데이터 파이프라인 (계획서 p.22 고도화).
+
+계획서 2단계 표절 검출 고도화:
+    > Human Feedback 형태의 Preference Optimization 을 통해서 표절 문서를
+    > 비선호하게끔 학습하는 형태로 sLLM 을 고도화.
+
+선호 학습(DPO/ORPO)은 (prompt, chosen, rejected) 삼중쌍을 입력으로 한다.
+표절 도메인에서:
+    chosen   = 정당한 글(원본 또는 정상 2차 창작)  ← 선호
+    rejected = 표절 글(무단 복제/요소 교체)         ← 비선호
+
+본 모듈은 데이터(사람 선호 라벨) 도착 전까지:
+  1) 라벨링 '후보쌍 템플릿' 생성 — 사람이 chosen/rejected 를 확정할 수 있는 골격
+  2) 라벨 완료 파일 → DPO 학습 포맷(JSONL) 변환 + 검증/통계
+까지를 제공한다. 라벨이 들어오면 즉시 sLLM 선호학습에 투입 가능.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field, asdict
+
+# 선호 학습 기본 지시문 (표절 비선호 방향 고정)
+DEFAULT_PROMPT = (
+    "다음 원문을 참고한 두 글 중, 저작권을 침해하지 않은 정당한 글을 선호하라."
+)
+
+
+@dataclass
+class PreferenceCandidate:
+    """라벨링 전 후보쌍. label_status 가 'pending' 이면 사람 확정 대기."""
+    pair_id: str
+    prompt: str
+    candidate_a: str
+    candidate_b: str
+    source_doc: str | None = None         # 비교 기준 원본 doc_id
+    suggested_rejected: str | None = None  # 엔진이 표절로 추정한 쪽 ("a"/"b") — 약한 신호
+    label_status: str = "pending"          # pending | labeled
+    chosen: str | None = None              # 라벨 결과: candidate_a/b 중 선호 텍스트
+    rejected: str | None = None            # 라벨 결과: 비선호 텍스트
+    meta: dict = field(default_factory=dict)
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+def build_candidate(
+    pair_id: str,
+    original: str,
+    text_a: str,
+    text_b: str,
+    suggested_rejected: str | None = None,
+    source_doc: str | None = None,
+    prompt: str = DEFAULT_PROMPT,
+) -> PreferenceCandidate:
+    """원본 + 두 후보 글 → 라벨링 대기 후보쌍.
+
+    suggested_rejected: 표절 탐지 결과로 추정한 비선호 쪽('a'/'b'). 사람이 검토·확정.
+    """
+    return PreferenceCandidate(
+        pair_id=pair_id,
+        prompt=f"{prompt}\n\n[원문]\n{original}",
+        candidate_a=text_a,
+        candidate_b=text_b,
+        source_doc=source_doc,
+        suggested_rejected=suggested_rejected,
+        meta={"original_len": len(original)},
+    )
+
+
+def to_dpo_record(c: PreferenceCandidate) -> dict | None:
+    """라벨 완료 후보쌍 → DPO/ORPO 학습 레코드. 미라벨이면 None."""
+    if c.label_status != "labeled" or not c.chosen or not c.rejected:
+        return None
+    return {"prompt": c.prompt, "chosen": c.chosen, "rejected": c.rejected, "pair_id": c.pair_id}
+
+
+def to_dpo_dataset(candidates: list[PreferenceCandidate]) -> list[dict]:
+    return [r for c in candidates if (r := to_dpo_record(c)) is not None]
+
+
+def validate_labeled(c: PreferenceCandidate) -> list[str]:
+    """라벨 완료 후보쌍 검증 — 학습 투입 전 무결성 체크."""
+    errors: list[str] = []
+    if c.label_status == "labeled":
+        if not c.chosen:
+            errors.append(f"{c.pair_id}: chosen 누락")
+        if not c.rejected:
+            errors.append(f"{c.pair_id}: rejected 누락")
+        if c.chosen and c.rejected and c.chosen.strip() == c.rejected.strip():
+            errors.append(f"{c.pair_id}: chosen == rejected (구분 불가)")
+    return errors
+
+
+def dataset_stats(candidates: list[PreferenceCandidate]) -> dict:
+    labeled = [c for c in candidates if c.label_status == "labeled"]
+    pending = [c for c in candidates if c.label_status != "labeled"]
+    errors: list[str] = []
+    for c in labeled:
+        errors.extend(validate_labeled(c))
+    return {
+        "total": len(candidates),
+        "labeled": len(labeled),
+        "pending": len(pending),
+        "trainable": len(to_dpo_dataset(candidates)),
+        "errors": errors,
+    }

app/engine/rouge.py

@@ -0,0 +1,109 @@
+"""N-gram ROUGE 점수 (계획서 성능지표 No.7, p.24 수식).
+
+계획서 평가수식:
+    ROUGE-N = Σ_S∈ref Σ_gram_n Count_match(gram_n) / Σ_S∈ref Σ_gram_n Count(gram_n)
+
+즉 레퍼런스 n-gram 기준 재현율(recall) 형태. 본 모듈은 No.7 평가를 위해
+ROUGE-1 / ROUGE-2 / ROUGE-L 을 자체 구현한다 (외부 라이브러리 의존 0).
+
+토큰화는 두 가지 지원:
+  - "lemma": kiwi 형태소 기본형 (한국어 어미 변화에 강건, 권장)
+  - "char" : 공백/문자 기준 단순 토큰 (의존성 없이 동작)
+"""
+
+from __future__ import annotations
+
+import re
+from collections import Counter
+from dataclasses import dataclass
+
+
+def _char_tokens(text: str) -> list[str]:
+    return re.findall(r"[가-힣A-Za-z0-9]+", text.lower())
+
+
+def tokenize(text: str, mode: str = "lemma") -> list[str]:
+    if mode == "lemma":
+        try:
+            from app.engine.structural import extract_lemmas
+            toks = extract_lemmas(text)
+            if toks:
+                return toks
+        except Exception:
+            pass
+    return _char_tokens(text)
+
+
+def _ngrams(tokens: list[str], n: int) -> Counter:
+    if len(tokens) < n:
+        return Counter()
+    return Counter(tuple(tokens[i : i + n]) for i in range(len(tokens) - n + 1))
+
+
+@dataclass
+class RougeScore:
+    precision: float
+    recall: float
+    f1: float
+
+    def as_dict(self) -> dict[str, float]:
+        return {"precision": round(self.precision, 4), "recall": round(self.recall, 4), "f1": round(self.f1, 4)}
+
+
+def _prf(match: int, sys_total: int, ref_total: int) -> RougeScore:
+    precision = match / sys_total if sys_total else 0.0
+    recall = match / ref_total if ref_total else 0.0
+    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
+    return RougeScore(precision, recall, f1)
+
+
+def rouge_n(system: str, reference: str, n: int = 1, mode: str = "lemma") -> RougeScore:
+    sys_g = _ngrams(tokenize(system, mode), n)
+    ref_g = _ngrams(tokenize(reference, mode), n)
+    match = sum((sys_g & ref_g).values())
+    return _prf(match, sum(sys_g.values()), sum(ref_g.values()))
+
+
+def _lcs_length(a: list[str], b: list[str]) -> int:
+    if not a or not b:
+        return 0
+    prev = [0] * (len(b) + 1)
+    for x in a:
+        cur = [0] * (len(b) + 1)
+        for j, y in enumerate(b, 1):
+            cur[j] = prev[j - 1] + 1 if x == y else max(prev[j], cur[j - 1])
+        prev = cur
+    return prev[-1]
+
+
+def rouge_l(system: str, reference: str, mode: str = "lemma") -> RougeScore:
+    s, r = tokenize(system, mode), tokenize(reference, mode)
+    lcs = _lcs_length(s, r)
+    return _prf(lcs, len(s), len(r))
+
+
+def evaluate_pairs(
+    pairs: list[tuple[str, str]],
+    mode: str = "lemma",
+) -> dict[str, dict[str, float]]:
+    """(system, reference) 페어 리스트 → 코퍼스 평균 ROUGE-1/2/L.
+
+    계획서 No.2-1년차 목표: N-gram ROUGE 65점 (gpt-4o 줄글 요약 64 대비).
+    """
+    if not pairs:
+        return {}
+    acc = {"rouge1": [], "rouge2": [], "rougeL": []}
+    for system, reference in pairs:
+        acc["rouge1"].append(rouge_n(system, reference, 1, mode))
+        acc["rouge2"].append(rouge_n(system, reference, 2, mode))
+        acc["rougeL"].append(rouge_l(system, reference, mode))
+
+    def avg(scores: list[RougeScore]) -> dict[str, float]:
+        k = len(scores)
+        return {
+            "precision": round(sum(s.precision for s in scores) / k, 4),
+            "recall": round(sum(s.recall for s in scores) / k, 4),
+            "f1": round(sum(s.f1 for s in scores) / k, 4),
+        }
+
+    return {metric: avg(scores) for metric, scores in acc.items()}

app/engine/summarizer.py

@@ -0,0 +1,197 @@
+"""스토리 요약/분석 모듈 (계획서 과제2 ②, p.13 / p.21 고도화).
+
+계획서 요약 시스템 흐름:
+    입력 ⇒ 추출적 요약(중요 정보 식별) ⇒ 추상적 요약(내용 재구성) ⇒ 최종 요약
+
+본 모듈은 데이터(요약 정답셋) 없이도 동작하는 부분까지 구현한다.
+
+  ① 추출적 요약 (extractive) — 정답셋 불필요. 비지도 TextRank 변형.
+       문장 분할 → 문장 간 lemma 코사인 유사도 그래프 → PageRank 중심성 →
+       상위 문장 선택. (계획서: "구성 요소 추출 모델 활용해 핵심 내용 추출")
+  ② 추상적 요약 (abstractive) — LLM 훅(옵션). 키 없으면 추출적 결과로 폴백.
+       자체 sLLM(고려대 2차저작 생성지원 모델) 으로 교체할 자리.
+  ③ 통합 요약 (hybrid) — ①의 핵심 문장을 ②의 입력으로 (계획서 통합 요약 시스템).
+
+사용자 맞춤형 옵션(요약 길이/비율)은 계획서 2단계 '사용자 맞춤형 요약' 반영.
+
+평가지표 No.7(N-gram ROUGE)는 scripts/eval_rouge.py 로 측정한다 (정답셋 들어오면).
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+import re
+from collections import Counter
+from dataclasses import dataclass
+
+from app.engine.structural import extract_lemmas
+
+logger = logging.getLogger(__name__)
+
+# 문장 분할 — 종결부호 기준 (한국어 '다./요./까?/!' + 줄바꿈)
+_SENT_SPLIT = re.compile(r"(?<=[.!?。…])\s+|\n+")
+
+
+def split_sentences(text: str) -> list[str]:
+    raw = _SENT_SPLIT.split(text.strip())
+    return [s.strip() for s in raw if s.strip()]
+
+
+def _lemma_vector(sentence: str) -> Counter:
+    return Counter(extract_lemmas(sentence))
+
+
+def _cosine(a: Counter, b: Counter) -> float:
+    if not a or not b:
+        return 0.0
+    common = set(a) & set(b)
+    if not common:
+        return 0.0
+    dot = sum(a[t] * b[t] for t in common)
+    na = math.sqrt(sum(v * v for v in a.values()))
+    nb = math.sqrt(sum(v * v for v in b.values()))
+    return dot / (na * nb) if na and nb else 0.0
+
+
+def _textrank_scores(vectors: list[Counter], damping: float = 0.85, iters: int = 30) -> list[float]:
+    """문장 그래프 PageRank. 정답셋 불필요한 비지도 중심성."""
+    n = len(vectors)
+    if n == 0:
+        return []
+    if n == 1:
+        return [1.0]
+
+    # 유사도 인접행렬 (자기 자신 제외)
+    sim = [[0.0] * n for _ in range(n)]
+    for i in range(n):
+        for j in range(i + 1, n):
+            s = _cosine(vectors[i], vectors[j])
+            sim[i][j] = sim[j][i] = s
+
+    # 행 정규화
+    row_sum = [sum(sim[i]) for i in range(n)]
+    scores = [1.0 / n] * n
+    for _ in range(iters):
+        new = [(1 - damping) / n] * n
+        for i in range(n):
+            for j in range(n):
+                if i == j or row_sum[j] == 0:
+                    continue
+                new[i] += damping * scores[j] * sim[j][i] / row_sum[j]
+        scores = new
+    return scores
+
+
+@dataclass
+class SummaryResult:
+    extractive: str               # 추출적 요약 (선택된 원문 문장)
+    abstractive: str | None       # 추상적 요약 (LLM, 없으면 None)
+    final: str                    # 최종 요약 (abstractive 우선, 없으면 extractive)
+    selected_indices: list[int]   # 선택된 문장 인덱스 (원문 순서)
+    mode: str                     # "extractive" | "hybrid"
+    num_sentences_in: int
+    num_sentences_out: int
+
+
+def extractive_summary(text: str, ratio: float = 0.3, max_sentences: int | None = None) -> SummaryResult:
+    """비지도 추출적 요약 — 정답셋/LLM/외부호출 불필요."""
+    sentences = split_sentences(text)
+    n = len(sentences)
+    if n == 0:
+        return SummaryResult("", None, "", [], "extractive", 0, 0)
+    if n <= 2:
+        joined = " ".join(sentences)
+        return SummaryResult(joined, None, joined, list(range(n)), "extractive", n, n)
+
+    k = max(1, math.ceil(n * ratio))
+    if max_sentences is not None:
+        k = min(k, max_sentences)
+
+    vectors = [_lemma_vector(s) for s in sentences]
+    scores = _textrank_scores(vectors)
+
+    # 상위 k개 문장 선택 → 원문 등장 순서로 재정렬 (가독성)
+    top = sorted(range(n), key=lambda i: scores[i], reverse=True)[:k]
+    top_sorted = sorted(top)
+    summary = " ".join(sentences[i] for i in top_sorted)
+    return SummaryResult(
+        extractive=summary,
+        abstractive=None,
+        final=summary,
+        selected_indices=top_sorted,
+        mode="extractive",
+        num_sentences_in=n,
+        num_sentences_out=len(top_sorted),
+    )
+
+
+_ABSTRACTIVE_PROMPT = """다음은 어떤 글에서 추출한 핵심 문장들이다. 이 내용을 바탕으로
+간결하고 자연스러운 한국어 요약문을 작성하라. 원문에 없는 사실을 지어내지 말 것.
+
+[핵심 문장]
+"""
+
+
+class Summarizer:
+    """통합 요약기 — 추출적(항상) + 추상적(LLM 옵션).
+
+    계획서 통합 요약 시스템: 추출적으로 중요 문장을 뽑은 뒤 추상적으로 재구성.
+    use_llm=False 이거나 키가 없으면 추출적 결과를 최종 요약으로 사용.
+    """
+
+    def __init__(self, settings=None):
+        from app.core.config import get_settings
+        self.settings = settings or get_settings()
+
+    def summarize(
+        self,
+        text: str,
+        ratio: float = 0.3,
+        max_sentences: int | None = None,
+        use_abstractive: bool = True,
+    ) -> SummaryResult:
+        base = extractive_summary(text, ratio=ratio, max_sentences=max_sentences)
+        if not base.extractive:
+            return base
+
+        if use_abstractive and self.settings.use_llm_extractor and self.settings.has_openai:
+            abstractive = self._abstractive(base.extractive)
+            if abstractive:
+                return SummaryResult(
+                    extractive=base.extractive,
+                    abstractive=abstractive,
+                    final=abstractive,
+                    selected_indices=base.selected_indices,
+                    mode="hybrid",
+                    num_sentences_in=base.num_sentences_in,
+                    num_sentences_out=base.num_sentences_out,
+                )
+        return base
+
+    def _abstractive(self, extractive_text: str) -> str | None:
+        try:
+            from openai import OpenAI
+            client = OpenAI(api_key=self.settings.openai_api_key)
+            resp = client.chat.completions.create(
+                model=self.settings.openai_extraction_model,
+                temperature=0.2,
+                messages=[
+                    {"role": "system", "content": "You are a concise Korean summarizer. Never hallucinate."},
+                    {"role": "user", "content": _ABSTRACTIVE_PROMPT + extractive_text},
+                ],
+            )
+            return (resp.choices[0].message.content or "").strip() or None
+        except Exception as exc:  # pragma: no cover - 네트워크/키 의존
+            logger.warning("Abstractive summary failed, using extractive: %s", exc)
+            return None
+
+
+_default_summarizer: Summarizer | None = None
+
+
+def get_summarizer(settings=None) -> Summarizer:
+    global _default_summarizer
+    if _default_summarizer is None:
+        _default_summarizer = Summarizer(settings)
+    return _default_summarizer

docs/CASE_COVERAGE.md

@@ -0,0 +1,49 @@
+# 39개 침해 케이스 커버리지 갭 분석
+
+- 전체 케이스: 39
+- 엔진 탐지 대상: 10
+- 평가 샘플 보유(covered): 0
+- 탐지 대상이나 데이터 없음(요청 대상): 10
+- 커버리지: 0.0%
+
+| case_id | subgroup | actor | 탐지대상 | 샘플수 | 상태 |
+|---|---|---|---|---|---|
+| A1 | A-1 외부 텍스트 인용·수록 | 저자(가해) | O | 0 | detectable_no_data |
+| A2 | A-1 외부 텍스트 인용·수록 | 저자(가해) | O | 0 | detectable_no_data |
+| A3 | A-1 외부 텍스트 인용·수록 | 저자(가해) | O | 0 | detectable_no_data |
+| A4 | A-1 외부 텍스트 인용·수록 | 저자(가해) | O | 0 | detectable_no_data |
+| A5 | A-1 외부 텍스트 인용·수록 | 저자(가해) | O | 0 | detectable_no_data |
+| A6 | A-2 타인 자서전·회고 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A7 | A-2 타인 자서전·회고 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A8 | A-2 타인 자서전·회고 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A9 | A-2 타인 자서전·회고 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A10 | A-2 타인 자서전·회고 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A11 | A-2 타인 자서전·회고 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A12 | A-2 타인 자서전·회고 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A13 | A-3 구술·강연·녹음 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A14 | A-4 학술·교육 자료 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A15 | A-4 학술·교육 자료 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A16 | A-5 번역물 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A17 | A-6 이미지·시각 자산 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A18 | A-6 이미지·시각 자산 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A19 | A-6 이미지·시각 자산 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A20 | A-6 이미지·시각 자산 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A21 | A-7 음원·영상 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A22 | A-8 디지털 사적 통신 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A23 | A-9 사후·고인 자료 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A24 | A-10 AI 도구 사용 | 저자(가해, 비의도) | O | 0 | detectable_no_data |
+| A25 | A-10 AI 도구 사용 | 저자(가해, 비의도) | O | 0 | detectable_no_data |
+| A26 | A-10 AI 도구 사용 | 저자(가해) | - | 0 | out_of_engine_scope |
+| A27 | A-11 대필 | 저자·플랫폼 | - | 0 | out_of_engine_scope |
+| B1 | B 저자(피해) | 저자(피해) | O | 0 | detectable_no_data |
+| B2 | B 저자(피해) | 저자(피해) | O | 0 | detectable_no_data |
+| B3 | B 저자(피해) | 저자(피해) | - | 0 | out_of_engine_scope |
+| B4 | B 저자(피해) | 저자(피해) | - | 0 | out_of_engine_scope |
+| C1 | C 플랫폼 | 플랫폼 | - | 0 | out_of_engine_scope |
+| C2 | C 플랫폼 | 저자·플랫폼 | - | 0 | out_of_engine_scope |
+| C3 | C 플랫폼 | 플랫폼 | - | 0 | out_of_engine_scope |
+| D1 | D 다른 사용자 | 다른 사용자 | O | 0 | detectable_no_data |
+| E1 | E 유족 | 유족 | - | 0 | out_of_engine_scope |
+| E2 | E 유족 | 유족 | - | 0 | out_of_engine_scope |
+| X1 | X 분류체계 외 | 저자(가해) | - | 0 | out_of_engine_scope |
+| X2 | X 분류체계 외 | 저자(가해) | - | 0 | out_of_engine_scope |

docs/DATA_REQUEST_SPEC.md

@@ -0,0 +1,103 @@
+# 컴북스 요청 데이터 스펙 & 요약 정답셋 작성 가이드
+
+> 오투오 과제2(콘텐츠 표절 탐지 / 요소 분석) 2단계 고도화에 필요한 데이터 스펙.
+> 데이터 수령 즉시 학습·검증에 투입할 수 있도록, 포맷을 본 문서로 사전 고정한다.
+> 관련 평가 스크립트: `scripts/eval_rouge.py`, `scripts/eval_metadata_f1.py`,
+> `scripts/analyze_case_coverage.py`
+
+## 0. 요약 — 무엇을, 왜, 어떤 포맷으로
+
+| # | 데이터 | 용도(성능지표) | 제공 주체 | 비고 |
+|---|---|---|---|---|
+| 1 | 표절/비표절 샘플 글 | 표절 정밀도 97% (No.4) | 컴북스 | 39 케이스·자서전 도메인 커버 |
+| 2 | article 본문 3만건 + 저작권 확보분 | 요소 추출·군집화 코퍼스 | 컴북스 | 1단계 계획 수량 |
+| 3 | 도메인별 텍스트 | 요약 모델 범용성 (No.7) | 컴북스+공개 | 장르 다양성 |
+| 4 | 콘텐츠 요소(메타) 정답 라벨 | 메타 추출 F1 83 (No.3) | 컴북스 1단계분 | KLUE NER 공개로 보완 |
+| 5 | **요약 정답셋(reference summary)** | 요약 ROUGE 65 (No.7) | **별도 구축** | 컴북스 미보유 → 역할분담 |
+| 6 | **Human Feedback 선호 라벨** | 표절검출 HF 고도화 | **별도 구축** | 라벨링 공수 필요 |
+
+→ #1~#4 는 컴북스 직접 제공, **#5·#6 은 컴북스가 줄 수 없는 데이터**로 제작 주체를 먼저 합의해야 한다.
+
+---
+
+## 1. 표절/비표절 샘플 글 (정밀도 97% 평가)
+
+- **포맷 (JSONL)**: 표절 페어 단위
+  ```json
+  {"pair_id": "A1-001", "source_text": "원본 ...", "suspect_text": "검사 대상 ...",
+   "is_plagiarism": true, "case_id": "A1", "note": "시·노래 가사 무단 인용"}
+  ```
+- **필수 커버리지**: 자동 탐지 대상 케이스(`detectable_internal=True`, 현재 10종)를
+  **모두** 포함. 케이스별 최소 30건 이상 권장(정밀도 0.97 신뢰구간 확보).
+  - 현재 요청 대상 케이스 목록은 `python -m scripts.analyze_case_coverage` 로 산출.
+- **도메인**: 출판 콘텐츠뿐 아니라 **자서전 도메인** 표절/비표절을 별도 분리 제공.
+  (현 평가셋 999쌍은 출판 콘텐츠 기준 → 자서전 도메인 정밀도 미검증)
+- **균형**: 표절:비표절 ≈ 1:1. 비표절에는 '합법적 인용·정상 2차 창작'을 포함해
+  과탐(FP)을 줄이는 hard-negative 로 활용.
+
+## 2. article 본문 데이터 (요소 추출·군집화 코퍼스)
+
+- **포맷**: `data/reference/` 와 동일한 `.txt` 또는 JSONL `{"doc_id","title","text"}`
+- **수량**: 계획서 기준 3만건. 우선 1.5천~3천건 표본 선제공 가능하면 군집화·요소
+  추출 튜닝을 조기 착수.
+- **저작권**: 학습/평가 사용 가능 범위(CCL 또는 계약)를 메타로 명시 → `license` 필드.
+
+## 3. 도메인별 텍스트 (요약 범용성)
+
+- 장르 다양성 확보용(소설/에세이/자서전/실용 등). 요약 모델의 도메인 편향 방지.
+- 포맷은 #2 와 동일. `genre` 필드 권장.
+
+## 4. 콘텐츠 요소(메타) 정답 라벨 (메타 F1)
+
+- **포맷 (JSONL)** — `scripts/eval_metadata_f1.py` 입력과 동일:
+  ```json
+  {"text": "원문 ...", "characters": ["홍길동"], "motifs": ["복수"],
+   "keywords": ["활빈당","탐관오리"], "genre": "역사"}
+  ```
+- 컴북스 1단계 '콘텐츠 구성요소 정의' 라벨을 article 단위로 제공.
+- 공개 보완: KLUE NER(`--klue`)로 인물(PS) 추출 F1 을 즉시 측정 가능.
+
+---
+
+## 5. 요약 정답셋(reference summary) 작성 가이드 — **별도 구축 필요**
+
+> 컴북스 데이터에는 '본문'만 있고 '요약 정답'이 없다. ROUGE(No.7) 평가는 정답
+> 요약이 전제이므로 아래 가이드에 따라 별도 구축한다. **제작 주체 합의 필요**:
+> (A) 컴북스가 작성 / (B) 오투오가 GPT 생성 후 컴북스 검수 / (C) 혼합.
+
+- **포맷 (JSONL)** — `scripts/eval_rouge.py` 입력과 동일:
+  ```json
+  {"text": "원문 전체 ...", "reference": "사람이 작성한 정답 요약 ..."}
+  ```
+- **작성 원칙**
+  1. 길이: 원문의 약 20~30% (또는 3~5문장). 일관된 비율 유지.
+  2. 내용: 원문에 **없는 사실 추가 금지**(환각 방지). 핵심 사건·인물·결말 포함.
+  3. 표현: 단순 문장 복사가 아니라 재구성(추상적 요약 평가 목적).
+  4. 1건당 1명이 작성하되, 신뢰도 위해 일부는 2명 작성 후 교차검수(IAA 확인).
+- **수량**: No.7 신뢰 평가 위해 최소 200~300건(도메인 분산).
+- **검수**: GPT 생성안 사용 시(경로 B), 컴북스 편집자가 사실관계·표현 검수 후 확정.
+
+## 6. Human Feedback 선호 라벨 — **별도 구축 필요**
+
+> 표절 검출 HF Preference Optimization(계획서 p.22)용. '표절 글을 비선호'로 학습.
+> 단순 표절/비표절 데이터가 아니라 **사람의 선호 판단 라벨**이 필요(라벨링 공수).
+
+- **파이프라인**: `scripts/build_preference_dataset.py`
+  1. `template` — 후보쌍(원본+글A+글B) → 라벨링 템플릿 생성
+  2. (사람) 각 행에 `chosen`/`rejected` 확정, `label_status="labeled"`
+  3. `convert` — 라벨 완료 파일 → DPO 학습셋(JSONL) + 검증/통계
+- **라벨링 형식 (JSONL)**:
+  ```json
+  {"pair_id":"p1","prompt":"...[원문]...","candidate_a":"정상 변형글",
+   "candidate_b":"표절글","label_status":"labeled",
+   "chosen":"정상 변형글","rejected":"표절글"}
+  ```
+- **수량**: 선호학습 최소 500쌍 이상 권장.
+
+---
+
+## 7. 공통 — 납기 명시 요청
+
+정밀도 97%·요약 ROUGE 65 는 위 데이터가 전제이므로, **각 항목 제공 시점**을
+함께 확정한다(연말 인수시험 역산). 부분 표본 선제공이 가능하면 군집화/요소추출
+튜닝을 데이터 도착 전 표본으로 조기 착수한다.

docs/INTEGRATION_INTERFACE.md

@@ -0,0 +1,71 @@
+# 2단계 통합 인터페이스 명세 (오투오 ↔ 바이칼/컴북스)
+
+> 계획서 마일스톤 2.4(요소 추출+표절 검출 고도화 통합), 3.x(침해요소 DB 공유),
+> 5.x(공유서비스 고도화) 연동을 위한 API 계약. 데이터 수령 전 사전 확정용.
+
+## 1. 오투오가 제공하는 API (현행)
+
+| Method | Path | 용도 | 비고 |
+|---|---|---|---|
+| POST | `/v1/plagiarism/detect` | 단건 표절 탐지 | 군집 부분표절 신호 포함 |
+| POST | `/v1/plagiarism/batch` | 배치(≤500) | 비동기 잡 |
+| POST | `/v1/summary` | 스토리 요약 | 신규(과제2 ②) |
+| GET | `/v1/taxonomy` | 10태그·39케이스 | 컴북스/바이칼 라벨 공유 |
+| GET | `/v1/health` | 엔진 상태 | |
+
+## 2. detect 응답 — 2단계 신규 필드 `partial_signal`
+
+군집화 기반 요소별 부분 표절 분해. 바이칼 침해요소 DB 에 '무엇을 바꿔치기했는지'를
+구조화해 적재할 수 있도록 제공.
+
+```json
+{
+  "matches": [{
+    "source_doc": "ref-0003",
+    "similarity": 0.88,
+    "tags": [{"tag": "reproduction", "role": "primary", "label_ko": "복제권"}],
+    "case_id": "A1",
+    "partial_signal": {
+      "cluster_id": 2,
+      "verdict": "element_swap_plagiarism",
+      "signature_score": 0.74,
+      "per_element": {"lemmas": 0.92, "keywords": 0.88, "characters": 0.0, "motifs": 0.1},
+      "retained_elements": ["lemmas", "keywords"],
+      "changed_elements": ["characters", "motifs"]
+    }
+  }]
+}
+```
+
+- `verdict`: `near_duplicate` | `element_swap_plagiarism` | `weak` | `none`
+- `element_swap_plagiarism` = 본문(lemma/키워드) 유지 + 인물/모티프만 교체한 표절.
+
+## 3. 바이칼 침해요소 DB 연동 (계획서 3.x)
+
+- 오투오 detect 결과 → 바이칼 침해요소 케이스 DB 적재 매핑:
+  - `case_id`, `tags[]`, `partial_signal.verdict`, `score_breakdown` → DB 컬럼.
+  - 분류체계 버전 동기화: `GET /v1/taxonomy` 의 `cases_version`/`meta_tags_version`.
+- 합의 필요: ① DB 스키마(필드/타입), ② 적재 방식(API push vs 배치 export),
+  ③ 침해요소 식별자 체계(컴북스 콘텐츠>제품>아티클 식별코드와 매핑).
+
+## 4. 공유서비스 / 저작권 자동 분석 경계
+
+- 저작권 침해 **자동 분석 모듈**(1차/2차 침해 자동 판단, 권한 기반 적용범위 계산)은
+  계획서상 **바이칼** 담당(p.22). 오투오는 표절 '유사도/태그/케이스' 신호까지 제공,
+  권한·계약 기반 침해 '확정' 판단은 바이칼 모듈로 위임.
+- 경계 인터페이스: 오투오 `MatchResult` → 바이칼 침해 판단 입력. 본 매핑 표를
+  통합 설계 회의에서 확정.
+
+## 5. 인증/배포
+
+- 현재 API Key 인증(`app/core/auth.py`). 공유서비스 통합 시 OAuth2(계획서 p.18)와의
+  연동 방식 협의.
+- 상용 데이터 구간 암호화(AES256-CBC)·키 배포는 바이칼 공유서비스 아키텍처 기준 적용.
+
+## 6. 통합 테스트 항목(연말 인수시험 역산)
+
+- [ ] taxonomy 버전 동기화 라운드트립
+- [ ] detect → 바이칼 DB 적재 E2E
+- [ ] partial_signal 필드 계약 테스트
+- [ ] summary 엔드포인트 통합
+- [ ] 배치 처리 성공률 95%(마일스톤 점검기준)

docs/PHASE2_PROGRESS.md

@@ -0,0 +1,50 @@
+# 오투오 2단계 진행 현황 (데이터 수령 전 선행 작업 완료분)
+
+> 데이터 수령 전까지 가능한 작업을 선행 구현하여, 컴북스 데이터가 들어오면
+> 즉시 학습·검증에 착수할 수 있도록 준비한 결과 요약.
+
+## 1. 한눈에 보기
+
+| 남은 작업(계획서 2단계) | 선행 구현 상태 | 데이터 도착 후 할 일 |
+|---|---|---|
+| 표절 검출 고도화 — 군집화 | ✅ `engine/clustering.py` 구현·테스트 | 실데이터로 임계값 튜닝 |
+| 표절 검출 고도화 — Human Feedback | ✅ 선호데이터 파이프라인 골격 | 사람 라벨 → DPO 학습 |
+| 스토리 요약 모듈 | ✅ 추출적 요약+통합 골격+API | 추상적(sLLM) 연결, 정답셋 평가 |
+| 메타 추출 F1(No.3) | ✅ 평가 하니스(KLUE NER 호환) | 정답 라벨로 정식 측정·향상 |
+| 표절 정밀도 97%(No.4) | ✅ 케이스 갭 분석으로 요청목록 산출 | 자서전 도메인 샘플로 달성 |
+| 요약 ROUGE 65(No.7) | ✅ ROUGE 평가 모듈·CLI | 요약 정답셋으로 정식 측정 |
+| SW 저작권 등록 | ✅ 등록 준비 문서 | 서류 제출 |
+| 2단계 통합 | ✅ 통합 인터페이스 명세 | 바이칼/컴북스 E2E |
+
+## 2. 구현 산출물
+
+### 코드 모듈 (테스트 포함)
+- `app/engine/clustering.py` — 군집화 기반 부분 표절(요소 교체) 판별. detect 응답에
+  `partial_signal` 필드로 노출. (`tests/test_clustering.py`)
+- `app/engine/summarizer.py` — 추출적(TextRank)+통합 요약. `/v1/summary` 엔드포인트.
+  외부 의존 0으로 동작, LLM 키 있으면 추상적 단계 결합. (`tests/test_summarizer.py`)
+- `app/engine/rouge.py` — ROUGE-1/2/L 자체 구현 (성능지표 No.7). (`tests/test_rouge.py`)
+- `app/engine/metadata_eval.py` — 요소 추출 F1(성능지표 No.3). (`tests/test_metadata_eval.py`)
+- `app/engine/case_coverage.py` — 39 케이스 커버리지 갭 분석. (`tests/test_case_coverage.py`)
+- `app/engine/preference.py` — HF 선호학습 데이터 골격. (`tests/test_preference.py`)
+
+### 평가/유틸 스크립트 (데이터 도착 시 즉시 사용)
+- `scripts/eval_rouge.py` — 요약 ROUGE 평가 (dry-run 내장)
+- `scripts/eval_metadata_f1.py` — 메타 추출 F1 (KLUE NER `--klue` 지원)
+- `scripts/analyze_case_coverage.py` — 케이스 갭 → 데이터 요청 목록
+- `scripts/build_preference_dataset.py` — 선호데이터 template/convert
+
+### 문서
+- `docs/DATA_REQUEST_SPEC.md` — 컴북스 요청 데이터 스펙 + 요약 정답셋 가이드
+- `docs/INTEGRATION_INTERFACE.md` — 2단계 통합 인터페이스
+- `docs/SW_COPYRIGHT_REGISTRATION.md` — SW 저작권 등록 준비
+
+## 3. 데이터에 묶여 남는 것 (수령 후 착수)
+- 표절 **정밀도 97% 최종 달성** — 자서전 도메인 표절 샘플 필요
+- 요약 **ROUGE 65 학습·최종 평가** — 요약 정답셋 필요
+- HF **실제 선호학습 수행** — 사람 선호 라벨 필요
+- sLLM **학습 실행** — A100급 GPU 인프라 필요
+
+## 4. 평가환경 (계획서 p.24 기준)
+- 공인인증 평가환경: A100 GPU / Python 3.9 / transformers 4.39.3
+- 본 저장소 평가 스크립트는 위 환경에서 정답셋만 연결하면 동작하도록 작성됨.

docs/SW_COPYRIGHT_REGISTRATION.md

@@ -0,0 +1,43 @@
+# SW 저작권 등록 준비 자료 (오투오 2단계 산출물 1건)
+
+> 계획서 2단계 오투오 산출물: **SW 저작권 등록 1건**. 데이터 무관하게 지금 진행 가능.
+> 등록 대상: 콘텐츠 표절 여부 AI 탐지 모듈(+요소 분석/요약).
+
+## 1. 등록 대상 소프트웨어
+
+- **명칭(안)**: O2O 콘텐츠 표절 탐지 및 요소 분석 엔진 (o2o-plagiarism-api)
+- **버전**: `app/core/config.py` 의 `engine_version` (예: `o2o-plagiarism-2.x`)
+- **언어/환경**: Python 3.13, FastAPI, scikit-learn, kiwipiepy, sentence-transformers
+- **구성**: REST API 서버 + 탐지 엔진(삼중 유사도+군집화) + 요약 + 분류체계
+
+## 2. 핵심 모듈(창작성 소명 포인트)
+
+| 파일 | 기능 | 창작적 요소 |
+|---|---|---|
+| `app/engine/similarity.py` | 삼중 유사도 | text·lemma·element 가중 결합 알고리즘 |
+| `app/engine/structural.py` | 형태소 lemma 교집합 | query 기준 다중집합 비율(어미변경 표절 탐지) |
+| `app/engine/clustering.py` | 군집화 부분표절 | 요소 시그니처 군집 + 요소교체 표절 분해 |
+| `app/engine/taxonomy.py` | 법령 10태그·39케이스 | 분류체계 매핑 규칙 |
+| `app/engine/summarizer.py` | 스토리 요약 | TextRank 추출+통합 요약 |
+| `app/engine/autobiography_filter.py` | 자서전 특화 전처리 | 공통표현 제거+NER 마스킹 |
+
+## 3. 등록 제출물 체크리스트 (한국저작권위원회 SW 등록)
+
+- [ ] 신청서(저작자/창작일/공표 여부)
+- [ ] 소스코드 명세서 — 전체 또는 처음·마지막 30면(영업비밀 보호 시 일부 마스킹)
+- [ ] 소프트웨어 설명서 — 본 저장소 `README.md` 기반 기능 명세
+- [ ] 창작 사실 소명 — 본 문서 §2 모듈별 창작 요소
+- [ ] 저작권 귀속 확인 — 오투오 사내 규정/과제 협약(지식재산 귀속 조항) 확인
+
+## 4. 사전 준비 액션 (지금 가능)
+
+1. `engine_version` 최종 고정 후 git 태그(`vX.Y.Z`)로 창작 시점 고정.
+2. 소스코드 명세서용 PDF 산출(주석 포함). 의존성 코드 제외, 자체 작성분만.
+3. 제3자 라이브러리 라이선스 정리(`requirements.txt` 각 패키지 라이선스 호환 확인).
+4. 과제 협약서의 IP 귀속 조항 확인 → 단독/공동 저작 여부 결정(컴북스 주관과 협의).
+
+## 5. 유의
+
+- 2단계 산출물은 '고도화' 1건. 1단계 등록분이 있으면 **변경/추가 등록**으로 진행
+  (군집화·요약·partial_signal 등 신규 창작분 반영).
+- 등록 전 외부 공개·배포 이력 정리(공표일 산정에 영향).

scripts/analyze_case_coverage.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""39개 침해 케이스 커버리지 갭 분석 — 컴북스 데이터 요청 근거 산출.
+
+데이터 없이 지금 실행 가능: taxonomy(분류체계)만으로 '탐지 대상이나 샘플이 없는
+케이스' 목록을 뽑아 정밀도 97% 달성을 위해 요청할 데이터를 케이스 단위로 정리한다.
+
+옵션으로 케이스별 보유 샘플 수 파일을 주면 실제 커버리지를 계산한다.
+  샘플 수 파일 형식 (JSON): {"A1": 12, "A2": 0, ...}
+
+사용:
+    python -m scripts.analyze_case_coverage
+    python -m scripts.analyze_case_coverage --samples data/eval/case_counts.json
+    python -m scripts.analyze_case_coverage --md > docs/CASE_COVERAGE.md
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from app.core.config import get_settings  # noqa: E402
+from app.engine.case_coverage import analyze_coverage  # noqa: E402
+from app.engine.taxonomy import load_taxonomy  # noqa: E402
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="39 케이스 커버리지 갭 분석")
+    ap.add_argument("--samples", default=None, help="케이스별 보유 샘플 수 JSON")
+    ap.add_argument("--md", action="store_true", help="Markdown 표로 출력")
+    args = ap.parse_args()
+
+    tax = load_taxonomy(get_settings().taxonomy_path)
+    if tax is None:
+        print("taxonomy 로드 실패", file=sys.stderr)
+        sys.exit(1)
+
+    counts = None
+    if args.samples and Path(args.samples).exists():
+        counts = json.loads(Path(args.samples).read_text(encoding="utf-8"))
+
+    rep = analyze_coverage(tax.cases, counts)
+
+    if args.md:
+        print("# 39개 침해 케이스 커버리지 갭 분석\n")
+        print(f"- 전체 케이스: {rep['total_cases']}")
+        print(f"- 엔진 탐지 대상: {rep['detectable_cases']}")
+        print(f"- 평가 샘플 보유(covered): {rep['covered_cases']}")
+        print(f"- 탐지 대상이나 데이터 없음(요청 대상): {rep['need_data_cases']}")
+        print(f"- 커버리지: {rep['coverage_ratio']:.1%}\n")
+        print("| case_id | subgroup | actor | 탐지대상 | 샘플수 | 상태 |")
+        print("|---|---|---|---|---|---|")
+        for r in rep["rows"]:
+            print(f"| {r.case_id} | {r.subgroup} | {r.actor} | "
+                  f"{'O' if r.detectable else '-'} | {r.sample_count} | {r.status} |")
+        return
+
+    print("\n=== 39개 침해 케이스 커버리지 ===")
+    print(f"전체 {rep['total_cases']} / 탐지대상 {rep['detectable_cases']} / "
+          f"covered {rep['covered_cases']} / 요청대상 {rep['need_data_cases']} / "
+          f"커버리지 {rep['coverage_ratio']:.1%}")
+
+    print("\n[데이터 요청 대상 — 탐지 가능하나 샘플 없는 케이스]")
+    if not rep["data_request_list"]:
+        print("  (없음 — 모든 탐지대상 케이스에 샘플 보유)")
+    for item in rep["data_request_list"]:
+        print(f"  - {item['case_id']:<4} {item['subgroup']}  [{item['actor']}]")
+
+    print("\n※ '탐지 대상이나 데이터 없음' 케이스가 정밀도 97% 달성을 위한 컴북스 요청 데이터 목록입니다.")
+    print("  out_of_engine_scope(유족/플랫폼/약관 등)는 탐지 모듈 범위 밖 → 데이터 불필요.")
+
+
+if __name__ == "__main__":
+    main()

scripts/build_preference_dataset.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+"""Human Feedback 선호학습(DPO) 데이터 파이프라인 — 계획서 2단계 표절검출 고도화.
+
+데이터(사람 선호 라벨) 도착 전까지 다음을 미리 가능하게 한다.
+  template : 후보쌍 입력(JSONL) → 라벨링 대기 템플릿 생성
+  convert  : 라벨 완료 파일 → DPO 학습 포맷(JSONL) 변환 + 검증/통계
+
+후보쌍 입력 형식 (JSONL):
+    {"pair_id": "p1", "original": "원문 ...", "text_a": "글A ...",
+     "text_b": "글B ...", "suggested_rejected": "b", "source_doc": "ref-0003"}
+
+라벨링 템플릿 출력에서 사람은 각 행에 chosen/rejected 와 label_status="labeled" 를 채운다.
+
+사용:
+    # 1) 후보쌍 → 라벨링 템플릿
+    python -m scripts.build_preference_dataset template candidates.jsonl -o label_me.jsonl
+    # 2) 라벨 완료 파일 → DPO 학습셋 + 통계
+    python -m scripts.build_preference_dataset convert labeled.jsonl -o dpo_train.jsonl
+    # 인자 없이 실행하면 내장 샘플로 파이프라인 검증
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from app.engine.preference import (  # noqa: E402
+    PreferenceCandidate,
+    build_candidate,
+    dataset_stats,
+    to_dpo_dataset,
+)
+
+_SAMPLE_CANDIDATES = [
+    {
+        "pair_id": "demo-1",
+        "original": "홍길동은 활빈당을 만들어 탐관오리의 재물을 빼앗아 백성에게 나눠주었다.",
+        "text_a": "임꺽정은 도적단을 조직해 부패한 양반의 곡식을 빼앗아 농민에게 베풀었다.",  # 정상 변형
+        "text_b": "홍길동은 활빈당을 만들어 탐관오리의 재물을 빼앗아 백성에게 나눠주었다.",  # 거의 복제
+        "suggested_rejected": "b",
+        "source_doc": "ref-0003",
+    },
+]
+
+# convert 데모용: 사람이 라벨을 완료했다고 가정한 샘플
+_SAMPLE_LABELED = [
+    {
+        "pair_id": "demo-1",
+        "prompt": "다음 원문을 참고한 두 글 중 ...\n\n[원문]\n홍길동은 ...",
+        "candidate_a": "임꺽정은 도적단을 조직해 ...",
+        "candidate_b": "홍길동은 활빈당을 만들어 ...",
+        "label_status": "labeled",
+        "chosen": "임꺽정은 도적단을 조직해 부패한 양반의 곡식을 빼앗아 농민에게 베풀었다.",
+        "rejected": "홍길동은 활빈당을 만들어 탐관오리의 재물을 빼앗아 백성에게 나눠주었다.",
+    },
+]
+
+
+def _read_jsonl(path: str, sample: list[dict]) -> list[dict]:
+    p = Path(path)
+    if not p.exists():
+        print(f"[warn] 파일 없음: {path} → 내장 샘플 사용", file=sys.stderr)
+        return sample
+    return [json.loads(l) for l in p.read_text(encoding="utf-8").splitlines() if l.strip()]
+
+
+def _write_jsonl(rows: list[dict], path: str | None) -> None:
+    out = "\n".join(json.dumps(r, ensure_ascii=False) for r in rows)
+    if path:
+        Path(path).write_text(out + "\n", encoding="utf-8")
+        print(f"[ok] {len(rows)}건 저장 → {path}")
+    else:
+        print(out)
+
+
+def cmd_template(args) -> None:
+    rows = _read_jsonl(args.input, _SAMPLE_CANDIDATES)
+    candidates = [
+        build_candidate(
+            pair_id=r.get("pair_id", f"p{i}"),
+            original=r["original"],
+            text_a=r["text_a"],
+            text_b=r["text_b"],
+            suggested_rejected=r.get("suggested_rejected"),
+            source_doc=r.get("source_doc"),
+        )
+        for i, r in enumerate(rows)
+    ]
+    _write_jsonl([c.to_dict() for c in candidates], args.output)
+    print(f"\n라벨링 안내: 각 행의 chosen/rejected 를 candidate_a/b 텍스트로 채우고 "
+          f"label_status 를 'labeled' 로 변경하세요.", file=sys.stderr)
+
+
+def cmd_convert(args) -> None:
+    rows = _read_jsonl(args.input, _SAMPLE_LABELED)
+    candidates = [PreferenceCandidate(**{k: v for k, v in r.items() if k in PreferenceCandidate.__dataclass_fields__})
+                  for r in rows]
+    stats = dataset_stats(candidates)
+    print(f"\n=== 선호 데이터 통계 ===", file=sys.stderr)
+    print(f"전체 {stats['total']} / 라벨완료 {stats['labeled']} / 대기 {stats['pending']} / "
+          f"학습가능 {stats['trainable']}", file=sys.stderr)
+    if stats["errors"]:
+        print("[검증 오류]", file=sys.stderr)
+        for e in stats["errors"]:
+            print(f"  - {e}", file=sys.stderr)
+    dpo = to_dpo_dataset(candidates)
+    _write_jsonl(dpo, args.output)
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="HF 선호학습 데이터 파이프라인")
+    sub = ap.add_subparsers(dest="cmd")
+
+    t = sub.add_parser("template", help="후보쌍 → 라벨링 템플릿")
+    t.add_argument("input", nargs="?", default="__sample__")
+    t.add_argument("-o", "--output", default=None)
+    t.set_defaults(func=cmd_template)
+
+    c = sub.add_parser("convert", help="라벨 완료 → DPO 학습셋")
+    c.add_argument("input", nargs="?", default="__sample__")
+    c.add_argument("-o", "--output", default=None)
+    c.set_defaults(func=cmd_convert)
+
+    args = ap.parse_args()
+    if not args.cmd:
+        print("[info] 서브커맨드 없음 → template 데모 실행\n", file=sys.stderr)
+        cmd_template(argparse.Namespace(input="__sample__", output=None))
+        return
+    args.func(args)
+
+
+if __name__ == "__main__":
+    main()

scripts/eval_metadata_f1.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""콘텐츠 요소(메타데이터) 추출 F1 평가 — 계획서 성능지표 No.3.
+
+데이터(요소 정답 라벨)가 들어오면 즉시 측정할 수 있도록 평가환경을 선구축한다.
+
+입력 형식 (JSONL, 한 줄당 한 건):
+    {"text": "원문 ...", "characters": ["홍길동"], "motifs": ["복수"],
+     "keywords": ["활빈당"], "genre": "역사"}
+
+동작:
+  - 현 추출기(RuleExtractor / OpenAIExtractor)로 text 에서 요소 예측
+  - 정답 라벨과 집합 단위 F1 (요소 유형별 + 마이크로 평균)
+  - 정답셋 없으면 내장 dry-run 샘플로 파이프라인 검증
+  - --klue 옵션: KLUE NER 공개 데이터(PS 태그)로 인물 추출 F1 평가 (datasets 필요)
+
+사용:
+    python -m scripts.eval_metadata_f1                       # dry-run
+    python -m scripts.eval_metadata_f1 data/eval/meta.jsonl  # 정답셋 평가
+    python -m scripts.eval_metadata_f1 --klue --limit 500    # KLUE NER 평가
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from app.engine.extractor import get_extractor  # noqa: E402
+from app.engine.metadata_eval import evaluate_extraction  # noqa: E402
+
+_DRY_RUN = [
+    {
+        "text": "홍길동은 활빈당을 만들어 탐관오리의 재물을 빼앗아 백성에게 나누어 주었다. 복수를 꿈꿨다.",
+        "characters": ["홍길동"], "motifs": ["복수"],
+        "keywords": ["활빈당", "탐관오리", "재물", "백성"], "genre": "역사",
+    },
+    {
+        "text": "어린왕자는 별을 떠나 여행하며 여우를 만나 우정을 배웠다.",
+        "characters": ["어린왕자", "여우"], "motifs": ["여행", "우정"],
+        "keywords": ["별", "여우"], "genre": "에세이",
+    },
+]
+
+
+def _load(path: str | None) -> tuple[list[dict], bool]:
+    if path is None:
+        return _DRY_RUN, True
+    p = Path(path)
+    if not p.exists():
+        print(f"[warn] 파일 없음: {path} → 내장 dry-run 샘플로 진행", file=sys.stderr)
+        return _DRY_RUN, True
+    rows = [json.loads(line) for line in p.read_text(encoding="utf-8").splitlines() if line.strip()]
+    return rows, False
+
+
+def _load_klue(limit: int) -> tuple[list[dict], bool]:
+    """KLUE NER 공개 데이터 → 인물(PS) 정답으로 변환. datasets 미설치 시 dry-run."""
+    try:
+        from datasets import load_dataset
+    except Exception:
+        print("[warn] `datasets` 미설치 → `pip install datasets` 후 --klue 사용 가능. dry-run 진행.",
+              file=sys.stderr)
+        return _DRY_RUN, True
+    ds = load_dataset("klue", "ner", split=f"validation[:{limit}]")
+    rows: list[dict] = []
+    for ex in ds:
+        tokens, tags = ex["tokens"], ex["ner_tags"]
+        names = ex["ner_tags"]  # placeholder; 실제 PS 추출은 라벨맵 기반
+        text = "".join(tokens)
+        # BIO → PS 엔티티 복원 (라벨 스킴은 데이터셋 버전에 맞춰 조정)
+        persons, cur = [], ""
+        label_names = ds.features["ner_tags"].feature.names
+        for tok, tag in zip(tokens, tags):
+            name = label_names[tag]
+            if name.endswith("PS") and name.startswith("B"):
+                if cur:
+                    persons.append(cur)
+                cur = tok
+            elif name.endswith("PS") and name.startswith("I"):
+                cur += tok
+            else:
+                if cur:
+                    persons.append(cur)
+                cur = ""
+        if cur:
+            persons.append(cur)
+        rows.append({"text": text, "characters": persons, "motifs": [], "keywords": [], "genre": None})
+    return rows, False
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="메타데이터 추출 F1 평가 (성능지표 No.3)")
+    ap.add_argument("dataset", nargs="?", default=None, help="JSONL 정답셋 (없으면 dry-run)")
+    ap.add_argument("--klue", action="store_true", help="KLUE NER 공개 데이터로 인물 F1 평가")
+    ap.add_argument("--limit", type=int, default=200, help="KLUE 평가 샘플 수")
+    args = ap.parse_args()
+
+    if args.klue:
+        rows, is_dry = _load_klue(args.limit)
+    else:
+        rows, is_dry = _load(args.dataset)
+
+    extractor = get_extractor()
+    predictions = []
+    for row in rows:
+        elem = extractor.extract(row.get("text", ""))
+        predictions.append({
+            "characters": elem.characters, "motifs": elem.motifs,
+            "keywords": elem.keywords, "genre": elem.genre,
+        })
+
+    scores = evaluate_extraction(predictions, rows)
+
+    src = "DRY-RUN (내장 샘플)" if is_dry else ("KLUE NER" if args.klue else f"{args.dataset}")
+    print(f"\n=== 메타데이터 추출 F1 — {src} ({len(rows)}건) ===")
+    print(f"{'field':<14}{'precision':>12}{'recall':>12}{'f1':>12}")
+    for field in ("characters", "motifs", "keywords", "micro_avg"):
+        s = scores[field]
+        print(f"{field:<14}{s['precision']:>12.4f}{s['recall']:>12.4f}{s['f1']:>12.4f}")
+    ga = scores["genre_accuracy"]
+    if ga["total"]:
+        print(f"\n장르 정확도: {ga['accuracy']:.4f} ({ga['correct']}/{ga['total']})")
+
+    micro_f1 = scores["micro_avg"]["f1"]
+    status = "달성" if micro_f1 >= 0.83 else "미달"
+    print(f"\n목표(No.3) F1 0.83 대비 마이크로 F1 = {micro_f1:.4f} → {status}")
+    if is_dry:
+        print("\n※ dry-run 수치. 컴북스 요소 정답 라벨 / KLUE NER 수령 후 정식 측정.")
+        print("   현 추출기는 룰 기반 폴백 → 1단계 sLLM(KLUE NER 파인튜닝) 교체 시 향상 예정.")
+
+
+if __name__ == "__main__":
+    main()

scripts/eval_rouge.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""요약 성능(ROUGE) 평가 하니스 — 계획서 성능지표 No.7.
+
+데이터(요약 정답셋)가 들어오면 즉시 측정할 수 있도록 평가환경을 선구축한다.
+
+입력 형식 (JSONL, 한 줄당 한 건):
+    {"text": "원문 ...", "reference": "사람 작성 요약 정답 ..."}
+  또는 이미 시스템 요약이 있는 경우:
+    {"system": "엔진 요약 ...", "reference": "정답 요약 ..."}
+
+동작:
+  - "text" 만 있으면 자체 Summarizer 로 system 요약을 생성한 뒤 reference 와 비교
+  - "system" 이 있으면 그대로 사용
+  - 정답셋 파일이 없으면 내장 dry-run 샘플로 파이프라인 동작만 검증
+
+사용:
+    python -m scripts.eval_rouge                          # dry-run (내장 샘플)
+    python -m scripts.eval_rouge data/eval/summary.jsonl  # 정답셋 평가
+    python -m scripts.eval_rouge data/eval/summary.jsonl --mode char --ratio 0.3
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from app.engine.rouge import evaluate_pairs  # noqa: E402
+from app.engine.summarizer import get_summarizer  # noqa: E402
+
+# 정답셋 도착 전 파이프라인 검증용 내장 샘플 (원문/정답 요약)
+_DRY_RUN = [
+    {
+        "text": (
+            "홍길동은 조선시대 의적이었다. 그는 활빈당을 만들어 탐관오리의 재물을 빼앗았다. "
+            "빼앗은 재물은 가난한 백성들에게 나누어 주었다. 조정에서는 그를 잡으려 했으나 실패했다. "
+            "결국 홍길동은 율도국으로 떠나 새 나라를 세웠다."
+        ),
+        "reference": "홍길동은 활빈당을 만들어 탐관오리의 재물을 빼앗아 백성에게 나눠주고 율도국을 세웠다.",
+    },
+    {
+        "text": (
+            "어린 왕자는 자신의 작은 별을 떠나 여러 행성을 여행했다. 여행 중 다양한 어른들을 만났다. "
+            "지구에서 여우를 만나 관계의 의미를 배웠다. 그는 자신의 장미가 소중함을 깨달았다."
+        ),
+        "reference": "어린 왕자는 여러 행성을 여행하며 여우를 통해 자신의 장미가 소중함을 깨닫는다.",
+    },
+]
+
+
+def _load(path: str | None) -> tuple[list[dict], bool]:
+    if path is None:
+        return _DRY_RUN, True
+    p = Path(path)
+    if not p.exists():
+        print(f"[warn] 파일 없음: {path} → 내장 dry-run 샘플로 진행", file=sys.stderr)
+        return _DRY_RUN, True
+    rows = [json.loads(line) for line in p.read_text(encoding="utf-8").splitlines() if line.strip()]
+    return rows, False
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description="요약 ROUGE 평가 (성능지표 No.7)")
+    ap.add_argument("dataset", nargs="?", default=None, help="JSONL 정답셋 (없으면 dry-run)")
+    ap.add_argument("--mode", choices=["lemma", "char"], default="lemma", help="토큰화 방식")
+    ap.add_argument("--ratio", type=float, default=0.3, help="자체 요약 길이 비율")
+    args = ap.parse_args()
+
+    rows, is_dry = _load(args.dataset)
+    summarizer = get_summarizer()
+
+    pairs: list[tuple[str, str]] = []
+    for row in rows:
+        reference = row.get("reference", "")
+        if not reference:
+            continue
+        if "system" in row and row["system"]:
+            system = row["system"]
+        else:
+            system = summarizer.summarize(row.get("text", ""), ratio=args.ratio).final
+        pairs.append((system, reference))
+
+    if not pairs:
+        print("평가할 (system, reference) 페어가 없습니다.", file=sys.stderr)
+        sys.exit(1)
+
+    scores = evaluate_pairs(pairs, mode=args.mode)
+
+    banner = "DRY-RUN (내장 샘플)" if is_dry else f"{args.dataset} ({len(pairs)}건)"
+    print(f"\n=== 요약 ROUGE 평가 — {banner} / 토큰={args.mode} ===")
+    print(f"{'metric':<10}{'precision':>12}{'recall':>12}{'f1':>12}")
+    for metric in ("rouge1", "rouge2", "rougeL"):
+        s = scores[metric]
+        print(f"{metric:<10}{s['precision']:>12.4f}{s['recall']:>12.4f}{s['f1']:>12.4f}")
+
+    target = 0.65
+    r1f1 = scores["rouge1"]["f1"]
+    status = "달성" if r1f1 >= target else "미달"
+    print(f"\n목표(No.7) ROUGE 0.65 대비 ROUGE-1 F1 = {r1f1:.4f} → {status}")
+    if is_dry:
+        print("\n※ 이는 파이프라인 검증용 dry-run 수치입니다. 컴북스 요약 정답셋 수령 후 본 평가로 정식 측정.")
+
+
+if __name__ == "__main__":
+    main()

tests/test_case_coverage.py

@@ -0,0 +1,43 @@
+"""39 케이스 커버리지 갭 분석 단위테스트."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.engine.case_coverage import analyze_coverage
+
+
+@dataclass
+class _Case:
+    case_id: str
+    subgroup: str
+    actor: str
+    detectable_internal: bool
+
+
+_CASES = [
+    _Case("A1", "A-1", "저자(가해)", True),
+    _Case("A2", "A-1", "저자(가해)", True),
+    _Case("C1", "C", "플랫폼", False),   # 탐지 범위 밖
+]
+
+
+def test_no_data_all_detectable_need_data():
+    rep = analyze_coverage(_CASES, None)
+    assert rep["detectable_cases"] == 2
+    assert rep["need_data_cases"] == 2
+    assert rep["covered_cases"] == 0
+    assert rep["coverage_ratio"] == 0.0
+
+
+def test_with_samples_marks_covered():
+    rep = analyze_coverage(_CASES, {"A1": 10})
+    assert rep["covered_cases"] == 1
+    assert rep["need_data_cases"] == 1
+    assert rep["coverage_ratio"] == 0.5
+
+
+def test_out_of_scope_excluded_from_request():
+    rep = analyze_coverage(_CASES, {"A1": 5, "A2": 5})
+    ids = {x["case_id"] for x in rep["data_request_list"]}
+    assert "C1" not in ids
+    assert rep["need_data_cases"] == 0

tests/test_clustering.py

@@ -0,0 +1,109 @@
+"""군집화 기반 표절 판별 단위테스트 (계획서 2단계 고도화).
+
+핵심 시나리오: "본문(lemma)·키워드는 그대로 두고 인물만 바꾼 표절" 을
+element_swap_plagiarism 으로 분해해서 잡는다.
+"""
+from __future__ import annotations
+
+from app.api.schemas import ExtractedElements
+from app.engine.clustering import (
+    ClusterIndex,
+    signature_similarity,
+    _element_signature,
+)
+
+
+def _elem(characters=None, motifs=None, keywords=None, genre=None) -> ExtractedElements:
+    return ExtractedElements(
+        characters=characters or [],
+        motifs=motifs or [],
+        keywords=keywords or [],
+        genre=genre,
+    )
+
+
+def test_signature_similarity_identical():
+    a = _element_signature(_elem(characters=["홍길동"], keywords=["활빈당", "탐관오리"], genre="역사"))
+    assert signature_similarity(a, a) == 1.0
+
+
+def test_clusters_group_similar_docs():
+    # 두 홍길동 변형은 한 군집, 어린왕자는 별도 군집
+    ids = ["hong-a", "hong-b", "prince"]
+    elems = [
+        _elem(characters=["홍길동"], keywords=["활빈당", "탐관오리", "재물"], genre="역사"),
+        _elem(characters=["홍길동"], keywords=["활빈당", "탐관오리", "재물"], genre="역사"),
+        _elem(characters=["어린왕자", "여우"], keywords=["사막", "장미", "별"], genre="동화"),
+    ]
+    idx = ClusterIndex(ids, elems, link_threshold=0.35)
+    assert idx.cluster_of("hong-a") == idx.cluster_of("hong-b")
+    assert idx.cluster_of("prince") != idx.cluster_of("hong-a")
+
+
+def test_route_to_nearest_cluster():
+    ids = ["hong", "prince"]
+    elems = [
+        _elem(characters=["홍길동"], keywords=["활빈당", "탐관오리", "재물"], genre="역사"),
+        _elem(characters=["어린왕자"], keywords=["사막", "장미", "별"], genre="동화"),
+    ]
+    idx = ClusterIndex(ids, elems, link_threshold=0.9)  # 분리 유지
+    q = _elem(characters=["홍길동"], keywords=["활빈당", "재물"], genre="역사")
+    routed = idx.route(q)
+    assert routed is not None
+    assert "hong" in routed.members
+
+
+def test_element_swap_plagiarism_detected():
+    """본문 lemma·키워드 유지 + 인물만 교체 → element_swap_plagiarism."""
+    ids = ["original"]
+    elems = [_elem(characters=["홍길동"], motifs=["복수"], keywords=["활빈당", "탐관오리", "재물", "의적"])]
+    lemmas = [["만들다", "빼앗다", "재물", "탐관오리", "활빈당", "나누다"]]
+    idx = ClusterIndex(ids, elems, doc_lemmas=lemmas, link_threshold=0.35)
+
+    # 인물(홍길동→김민수)·모티프만 바꾸고 본문 lemma/키워드는 그대로
+    q_elem = _elem(characters=["김민수"], motifs=["정의"], keywords=["활빈당", "탐관오리", "재물", "의적"])
+    q_lemmas = ["만들다", "빼앗다", "재물", "탐관오리", "활빈당", "나누다"]
+
+    sig = idx.partial_signal("original", q_elem, q_lemmas)
+    assert sig is not None
+    assert sig.verdict == "element_swap_plagiarism", sig.per_element
+    assert "characters" in sig.changed_elements
+    assert "lemmas" in sig.retained_elements
+
+
+def test_near_duplicate_verdict():
+    ids = ["original"]
+    elems = [_elem(characters=["홍길동"], keywords=["활빈당", "탐관오리", "재물"])]
+    lemmas = [["만들다", "빼앗다", "재물"]]
+    idx = ClusterIndex(ids, elems, doc_lemmas=lemmas)
+
+    sig = idx.partial_signal(
+        "original",
+        _elem(characters=["홍길동"], keywords=["활빈당", "탐관오리", "재물"]),
+        ["만들다", "빼앗다", "재물"],
+    )
+    assert sig is not None
+    assert sig.verdict == "near_duplicate"
+
+
+def test_unrelated_gets_none_or_weak():
+    ids = ["original"]
+    elems = [_elem(characters=["홍길동"], keywords=["활빈당", "탐관오리"])]
+    lemmas = [["만들다", "빼앗다"]]
+    idx = ClusterIndex(ids, elems, doc_lemmas=lemmas)
+
+    sig = idx.partial_signal(
+        "original",
+        _elem(characters=["우주비행사"], keywords=["로켓", "행성"]),
+        ["날다", "탐사하다"],
+    )
+    assert sig is not None
+    assert sig.verdict in ("none", "weak")
+    assert sig.verdict != "element_swap_plagiarism"
+
+
+def test_empty_index():
+    idx = ClusterIndex([], [])
+    assert idx.num_clusters == 0
+    assert idx.route(_elem(characters=["x"])) is None
+    assert idx.candidate_ids(_elem(characters=["x"])) == set()

tests/test_metadata_eval.py

@@ -0,0 +1,42 @@
+"""메타데이터 추출 F1 평가 단위테스트 (성능지표 No.3)."""
+from __future__ import annotations
+
+import pytest
+
+from app.engine.metadata_eval import set_prf, evaluate_extraction
+
+
+def test_set_prf_perfect():
+    r = set_prf(["홍길동", "활빈당"], ["홍길동", "활빈당"])
+    assert r.f1 == 1.0 and r.fp == 0 and r.fn == 0
+
+
+def test_set_prf_partial():
+    r = set_prf(["홍길동", "임꺽정"], ["홍길동", "활빈당"])
+    assert r.tp == 1 and r.fp == 1 and r.fn == 1
+    assert 0.0 < r.f1 < 1.0
+
+
+def test_set_prf_case_insensitive():
+    r = set_prf(["Hong"], ["hong"])
+    assert r.f1 == 1.0
+
+
+def test_evaluate_extraction_micro():
+    preds = [{"characters": ["홍길동"], "motifs": ["복수"], "keywords": ["활빈당"], "genre": "역사"}]
+    golds = [{"characters": ["홍길동"], "motifs": ["복수"], "keywords": ["활빈당"], "genre": "역사"}]
+    out = evaluate_extraction(preds, golds)
+    assert out["micro_avg"]["f1"] == 1.0
+    assert out["genre_accuracy"]["accuracy"] == 1.0
+
+
+def test_evaluate_extraction_genre_mismatch():
+    preds = [{"characters": [], "motifs": [], "keywords": [], "genre": "SF"}]
+    golds = [{"characters": [], "motifs": [], "keywords": [], "genre": "역사"}]
+    out = evaluate_extraction(preds, golds)
+    assert out["genre_accuracy"]["accuracy"] == 0.0
+
+
+def test_length_mismatch_raises():
+    with pytest.raises(ValueError):
+        evaluate_extraction([{}], [{}, {}])

tests/test_preference.py

@@ -0,0 +1,59 @@
+"""HF 선호학습 데이터 파이프라인 단위테스트 (계획서 2단계 고도화)."""
+from __future__ import annotations
+
+from app.engine.preference import (
+    PreferenceCandidate,
+    build_candidate,
+    to_dpo_record,
+    to_dpo_dataset,
+    validate_labeled,
+    dataset_stats,
+)
+
+
+def test_build_candidate_is_pending():
+    c = build_candidate("p1", "원문", "글A", "글B", suggested_rejected="b")
+    assert c.label_status == "pending"
+    assert "원문" in c.prompt
+    assert to_dpo_record(c) is None  # 미라벨은 학습 레코드 없음
+
+
+def test_labeled_becomes_dpo_record():
+    c = build_candidate("p1", "원문", "정상글", "표절글", suggested_rejected="b")
+    c.label_status = "labeled"
+    c.chosen = "정상글"
+    c.rejected = "표절글"
+    rec = to_dpo_record(c)
+    assert rec is not None
+    assert rec["chosen"] == "정상글" and rec["rejected"] == "표절글"
+
+
+def test_validate_catches_same_chosen_rejected():
+    c = PreferenceCandidate("p1", "p", "a", "b", label_status="labeled", chosen="x", rejected="x")
+    errors = validate_labeled(c)
+    assert any("chosen == rejected" in e for e in errors)
+
+
+def test_validate_catches_missing():
+    c = PreferenceCandidate("p1", "p", "a", "b", label_status="labeled", chosen="x", rejected=None)
+    assert any("rejected 누락" in e for e in validate_labeled(c))
+
+
+def test_dataset_stats():
+    pending = build_candidate("p1", "o", "a", "b")
+    labeled = build_candidate("p2", "o", "a", "b")
+    labeled.label_status = "labeled"
+    labeled.chosen, labeled.rejected = "a", "b"
+    stats = dataset_stats([pending, labeled])
+    assert stats["total"] == 2
+    assert stats["labeled"] == 1
+    assert stats["pending"] == 1
+    assert stats["trainable"] == 1
+    assert stats["errors"] == []
+
+
+def test_to_dpo_dataset_filters_pending():
+    cs = [build_candidate(f"p{i}", "o", "a", "b") for i in range(3)]
+    cs[0].label_status = "labeled"
+    cs[0].chosen, cs[0].rejected = "a", "b"
+    assert len(to_dpo_dataset(cs)) == 1

tests/test_rouge.py

@@ -0,0 +1,50 @@
+"""ROUGE 평가 모듈 단위테스트 (성능지표 No.7)."""
+from __future__ import annotations
+
+from app.engine.rouge import rouge_n, rouge_l, evaluate_pairs, tokenize
+
+
+def test_identical_is_perfect():
+    s = rouge_n("홍길동은 활빈당을 만들었다", "홍길동은 활빈당을 만들었다", n=1)
+    assert s.f1 == 1.0
+    assert s.recall == 1.0
+
+
+def test_no_overlap_is_zero():
+    s = rouge_n("우주선이 행성을 탐사한다", "사랑은 아름다운 감정이다", n=1, mode="char")
+    assert s.f1 == 0.0
+
+
+def test_partial_overlap_between_zero_and_one():
+    s = rouge_n("홍길동은 활빈당을 만들어 재물을 빼앗았다",
+                "홍길동은 활빈당을 조직했다", n=1)
+    assert 0.0 < s.f1 < 1.0
+
+
+def test_rouge2_stricter_than_rouge1():
+    sys = "홍길동은 활빈당을 만들어 재물을 빼앗았다"
+    ref = "홍길동은 재물을 활빈당으로 만들어 빼앗았다"  # 단어 순서 섞임
+    r1 = rouge_n(sys, ref, n=1).f1
+    r2 = rouge_n(sys, ref, n=2).f1
+    assert r2 <= r1  # bigram 은 순서에 민감 → 더 낮거나 같음
+
+
+def test_rouge_l_rewards_sequence():
+    s = rouge_l("홍길동은 활빈당을 만들어 재물을 빼앗았다",
+                "홍길동은 활빈당을 만들어 재물을 빼앗았다")
+    assert s.f1 == 1.0
+
+
+def test_evaluate_pairs_aggregates():
+    pairs = [
+        ("홍길동은 활빈당을 만들었다", "홍길동은 활빈당을 만들었다"),
+        ("어린왕자가 여우를 만났다", "어린왕자가 여우를 만났다"),
+    ]
+    out = evaluate_pairs(pairs)
+    assert set(out) == {"rouge1", "rouge2", "rougeL"}
+    assert out["rouge1"]["f1"] == 1.0
+
+
+def test_tokenize_char_fallback():
+    toks = tokenize("Hello 월드 123", mode="char")
+    assert "hello" in toks and "월드" in toks and "123" in toks

tests/test_summarizer.py

@@ -0,0 +1,67 @@
+"""스토리 요약 모듈 단위테스트 (계획서 과제2 ②).
+
+추출적 요약은 외부 의존(LLM/정답셋) 없이 동작해야 한다.
+"""
+from __future__ import annotations
+
+from app.core.config import Settings
+from app.engine.summarizer import (
+    Summarizer,
+    extractive_summary,
+    split_sentences,
+)
+
+_DOC = (
+    "홍길동은 조선시대 의적이었다. 그는 활빈당을 만들어 탐관오리의 재물을 빼앗았다. "
+    "빼앗은 재물은 가난한 백성들에게 나누어 주었다. 홍길동은 둔갑술과 도술에 능했다. "
+    "조정에서는 그를 잡으려 했으나 번번이 실패했다. 결국 홍길동은 율도국으로 떠나 새 나라를 세웠다. "
+    "오늘 날씨는 맑고 화창하다. 점심으로 김밥을 먹었다."
+)
+
+
+def test_split_sentences():
+    sents = split_sentences("첫 문장이다. 둘째 문장이다! 셋째 문장인가?")
+    assert len(sents) == 3
+
+
+def test_extractive_summary_shortens():
+    result = extractive_summary(_DOC, ratio=0.4)
+    assert result.num_sentences_in == 8
+    assert 0 < result.num_sentences_out < result.num_sentences_in
+    assert result.mode == "extractive"
+    assert result.final == result.extractive
+
+
+def test_extractive_preserves_source_order():
+    result = extractive_summary(_DOC, ratio=0.5)
+    # 선택된 인덱스는 원문 등장 순서여야 함 (가독성)
+    assert result.selected_indices == sorted(result.selected_indices)
+
+
+def test_extractive_picks_central_over_offtopic():
+    """본문 핵심(홍길동 서사)이 무관한 문장(날씨/점심)보다 우선 선택."""
+    result = extractive_summary(_DOC, ratio=0.4)
+    text = result.extractive
+    assert "홍길동" in text
+    # 마지막 두 무관 문장은 덜 선택되어야 함
+    assert not ("김밥" in text and "날씨" in text)
+
+
+def test_max_sentences_cap():
+    result = extractive_summary(_DOC, ratio=1.0, max_sentences=2)
+    assert result.num_sentences_out <= 2
+
+
+def test_empty_and_short():
+    assert extractive_summary("").final == ""
+    one = extractive_summary("한 문장만 있다.")
+    assert one.final == "한 문장만 있다."
+
+
+def test_summarizer_without_llm_is_extractive():
+    """LLM 비활성 시 통합 요청도 추출적 결과를 반환 (외부 의존 0 폴백)."""
+    settings = Settings(use_llm_extractor=False, openai_api_key="", use_kosimcse=False)
+    result = Summarizer(settings).summarize(_DOC, ratio=0.4, use_abstractive=True)
+    assert result.mode == "extractive"
+    assert result.abstractive is None
+    assert result.final