o2o-plagiarism-ai/app/engine/summarizer.py

"""스토리 요약/분석 모듈 (계획서 과제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