170 lines
6.1 KiB
Python
170 lines
6.1 KiB
Python
"""콘텐츠 구성요소 추출.
|
|
|
|
두 가지 구현:
|
|
- RuleExtractor: 형태소분석기 없이 동작하는 폴백 (의존성/비용 0)
|
|
- OpenAIExtractor: gpt-4o-mini 기반. JSON mode로 인물/장르/모티프/플롯 요약 추출
|
|
→ 추후 1단계 사내 sLLM(KLUE NER 파인튜닝)으로 교체할 자리.
|
|
|
|
두 구현 모두 ExtractedElements를 반환하므로 detector.py 입장에서는 swap-in 가능.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from collections import Counter
|
|
from typing import Protocol
|
|
|
|
from app.api.schemas import ExtractedElements
|
|
from app.core.config import Settings, get_settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Extractor(Protocol):
|
|
def extract(self, text: str) -> ExtractedElements: ...
|
|
|
|
|
|
# ---------- 룰 기반 폴백 ----------
|
|
|
|
_NAME_PATTERN = re.compile(r"[가-힣]{2,4}(?=(?:[은는이가을를과와의]|\s|$))")
|
|
_QUOTED_PATTERN = re.compile(r"[\"“”]([^\"“”]{1,20})[\"“”]")
|
|
_KEYWORD_PATTERN = re.compile(r"[가-힣A-Za-z]{2,}")
|
|
|
|
_GENRE_HINTS = {
|
|
"판타지": ["마법", "마왕", "용사", "엘프", "드래곤", "차원"],
|
|
"SF": ["우주", "외계", "로봇", "AI", "안드로이드", "행성"],
|
|
"로맨스": ["사랑", "연인", "키스", "고백", "결혼"],
|
|
"추리": ["살인", "탐정", "용의자", "범인", "수사", "단서"],
|
|
"에세이": ["나는", "내가", "생각", "느낀", "삶"],
|
|
}
|
|
_MOTIF_HINTS = {
|
|
"여행": ["여행", "떠나", "기차", "비행기"],
|
|
"성장": ["자라", "성장", "어른", "배우"],
|
|
"복수": ["복수", "원수", "되갚"],
|
|
"우정": ["친구", "우정", "동료"],
|
|
"이별": ["이별", "헤어", "떠나보"],
|
|
}
|
|
_STOPWORDS = {
|
|
"그리고", "그러나", "그래서", "하지만", "그런데", "이것", "그것", "저것",
|
|
"이는", "그는", "저는", "있다", "없다", "이다", "있는", "하는", "되는",
|
|
}
|
|
|
|
|
|
class RuleExtractor:
|
|
def extract(self, text: str) -> ExtractedElements:
|
|
characters = self._characters(text)
|
|
return ExtractedElements(
|
|
characters=characters[:10],
|
|
motifs=self._motifs(text)[:5],
|
|
genre=self._genre(text),
|
|
keywords=self._keywords(text)[:10],
|
|
)
|
|
|
|
@staticmethod
|
|
def _characters(text: str) -> list[str]:
|
|
raw = _NAME_PATTERN.findall(text) + [m.group(1) for m in _QUOTED_PATTERN.finditer(text)]
|
|
cnt = Counter(w for w in raw if w not in _STOPWORDS and len(w) >= 2)
|
|
return [w for w, _ in cnt.most_common(10)]
|
|
|
|
@staticmethod
|
|
def _genre(text: str) -> str | None:
|
|
scores = {g: sum(text.count(h) for h in hints) for g, hints in _GENRE_HINTS.items()}
|
|
best = max(scores.items(), key=lambda kv: kv[1])
|
|
return best[0] if best[1] > 0 else None
|
|
|
|
@staticmethod
|
|
def _motifs(text: str) -> list[str]:
|
|
found: list[tuple[str, int]] = []
|
|
for motif, hints in _MOTIF_HINTS.items():
|
|
score = sum(text.count(h) for h in hints)
|
|
if score > 0:
|
|
found.append((motif, score))
|
|
found.sort(key=lambda kv: kv[1], reverse=True)
|
|
return [m for m, _ in found]
|
|
|
|
@staticmethod
|
|
def _keywords(text: str) -> list[str]:
|
|
tokens = [t for t in _KEYWORD_PATTERN.findall(text) if len(t) >= 2 and t not in _STOPWORDS]
|
|
cnt = Counter(tokens)
|
|
return [w for w, _ in cnt.most_common(20)]
|
|
|
|
|
|
# ---------- OpenAI 기반 ----------
|
|
|
|
_EXTRACTION_PROMPT = """다음 출판 콘텐츠 텍스트에서 저작권 침해 판정에 사용할 구성요소를 추출하라.
|
|
|
|
[규칙]
|
|
- 결과는 반드시 JSON으로만 출력.
|
|
- 인물은 텍스트에 명시적으로 등장한 고유명사 인물만. 일반명사(소년/사람 등) 제외.
|
|
- 모티프는 추상적 주제(여행, 성장, 복수, 우정, 이별, 정의, 희생 등).
|
|
- 장르는 단일 값 (소설/에세이/판타지/SF/로맨스/추리/역사/사회 중 하나, 또는 null).
|
|
- 키워드는 작품 특유의 명사구 위주.
|
|
|
|
[출력 스키마]
|
|
{
|
|
"characters": ["..."],
|
|
"motifs": ["..."],
|
|
"genre": "..." | null,
|
|
"keywords": ["..."]
|
|
}
|
|
|
|
[텍스트]
|
|
"""
|
|
|
|
|
|
class OpenAIExtractor:
|
|
def __init__(self, settings: Settings):
|
|
from openai import OpenAI # 지연 import — 미설치 환경에서도 RuleExtractor 사용 가능
|
|
|
|
self._client = OpenAI(api_key=settings.openai_api_key)
|
|
self._model = settings.openai_extraction_model
|
|
self._fallback = RuleExtractor()
|
|
|
|
def extract(self, text: str) -> ExtractedElements:
|
|
truncated = text[:8000]
|
|
try:
|
|
resp = self._client.chat.completions.create(
|
|
model=self._model,
|
|
temperature=0.0,
|
|
response_format={"type": "json_object"},
|
|
messages=[
|
|
{"role": "system", "content": "You are a precise literary metadata extractor."},
|
|
{"role": "user", "content": _EXTRACTION_PROMPT + truncated},
|
|
],
|
|
)
|
|
raw = resp.choices[0].message.content or "{}"
|
|
data = json.loads(raw)
|
|
return ExtractedElements(
|
|
characters=_as_str_list(data.get("characters")),
|
|
motifs=_as_str_list(data.get("motifs")),
|
|
genre=_as_optional_str(data.get("genre")),
|
|
keywords=_as_str_list(data.get("keywords")),
|
|
)
|
|
except Exception as exc:
|
|
logger.warning("OpenAI extraction failed, falling back to rule-based: %s", exc)
|
|
return self._fallback.extract(text)
|
|
|
|
|
|
def _as_str_list(value, limit: int = 10) -> list[str]:
|
|
if not isinstance(value, list):
|
|
return []
|
|
return [str(v).strip() for v in value if isinstance(v, (str, int)) and str(v).strip()][:limit]
|
|
|
|
|
|
def _as_optional_str(value) -> str | None:
|
|
if value is None:
|
|
return None
|
|
s = str(value).strip()
|
|
return s or None
|
|
|
|
|
|
def get_extractor(settings: Settings | None = None) -> Extractor:
|
|
s = settings or get_settings()
|
|
if s.use_llm_extractor and s.has_openai:
|
|
logger.info("Using OpenAIExtractor (model=%s)", s.openai_extraction_model)
|
|
return OpenAIExtractor(s)
|
|
logger.info("Using RuleExtractor (fallback)")
|
|
return RuleExtractor()
|