뉴스데이터로 클러스터링1

naver-news-cluster/
├── app/
│   ├── services/
│   │   ├── news_fetcher.py       # 네이버 API 호출
│   │   ├── text_preprocessor.py  # HTML 제거, 특수문자 정리
│   │   └── cluster_service.py    # BERTopic (UMAP + HDBSCAN)
│   ├── infra/
│   │   └── embedding.py          # SentenceTransformer 래퍼
│   └── core/
│       └── settings.py           # 환경변수 중앙 관리
└── evaluations/
    └── metrics/
        └── clustering_metrics.py # noise_ratio 등 계산

# app/core/settings.py
"""
환경 설정 모듈
- pydantic-settings 기반 타입 안전 환경 변수 파싱
- .env만 바꾸면 코드 수정 없이 모델/파라미터 교체 가능
"""
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # ── 네이버 검색 API ──────────────────────────────
    naver_client_id: str
    naver_client_secret: str

    # ── 임베딩 모델 경로 (2편/3편에서 사용) ───────────
    embedding_model_path: str = "BAAI/bge-m3"

    # ── BERTopic 파라미터 ─────────────────────────────
    # min_topic_size: 몇 건 이상 모여야 하나의 토픽으로 인정할지
    bertopic_min_topic_size: int = 3

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
    )


# 앱 전체에서 이 인스턴스 하나만 import해서 재사용 (싱글턴)
settings = Settings()

# app/services/text_preprocessor.py
"""
텍스트 전처리 모듈
- 목적: 임베딩/클러스터링에 들어가기 전 노이즈 제거
- 순수 함수로 작성 (외부 IO 없음) -> 단위 테스트 쉬움
"""
import re


_HTML_TAG_PATTERN = re.compile(r"<[^>]+>")
_SPECIAL_CHAR_PATTERN = re.compile(r"[^\w\s가-힣.,!?]")


def clean_text(raw_text: str) -> str:
    """네이버 API 원문에서 HTML 태그, 특수문자를 제거한다.

    Args:
        raw_text: 네이버 뉴스 API가 반환한 title/description 원문

    Returns:
        정제된 텍스트. 빈 문자열이 나올 수도 있으니 호출부에서 검증 필요.
    """
    text_without_html: str = _HTML_TAG_PATTERN.sub("", raw_text)
    text_without_special: str = _SPECIAL_CHAR_PATTERN.sub("", text_without_html)
    return text_without_special.strip()

# app/services/cluster_service.py
"""
뉴스 기사 클러스터링 서비스
- BERTopic(UMAP + HDBSCAN)으로 유사 기사를 묶는다
- 목적: 중복/유사 기사를 그룹화해 검색 후보군의 다양성을 확보
"""
from bertopic import BERTopic
from hdbscan import HDBSCAN
from sentence_transformers import SentenceTransformer
from umap import UMAP

from app.core.settings import settings


class NewsClusterService:
    """기사 리스트를 입력받아 토픽별로 묶어주는 서비스."""

    def __init__(self, embedding_model: SentenceTransformer) -> None:
        # UMAP: 고차원 임베딩을 저차원으로 압축해 군집화 성능을 높인다
        umap_model = UMAP(
            n_neighbors=15,
            n_components=5,
            min_dist=0.0,
            metric="cosine",
            random_state=42,  # 재현 가능한 실험을 위해 시드 고정
        )
        # HDBSCAN: 군집 개수를 미리 정하지 않아도 되는 밀도 기반 클러스터링
        hdbscan_model = HDBSCAN(
            min_cluster_size=settings.bertopic_min_topic_size,
            metric="euclidean",
            cluster_selection_method="eom",
            prediction_data=True,
        )
        self._topic_model = BERTopic(
            embedding_model=embedding_model,
            umap_model=umap_model,
            hdbscan_model=hdbscan_model,
            language="multilingual",  # 한국어 지원을 위해 다국어 모드
            calculate_probabilities=False,
        )

    def cluster_articles(self, clean_texts: list[str]) -> dict[str, object]:
        """정제된 기사 텍스트 리스트를 클러스터링한다.

        Args:
            clean_texts: text_preprocessor를 거친 기사 본문 리스트

        Returns:
            topics: 각 기사가 속한 토픽 번호 (-1은 노이즈)
            topic_info: 토픽별 대표 키워드, 기사 수 등 메타데이터
        """
        topics, _ = self._topic_model.fit_transform(clean_texts)
        topic_info = self._topic_model.get_topic_info()

        # -1(노이즈) 비율이 너무 높으면 파라미터를 재검토해야 한다는 신호
        noise_count: int = topics.count(-1)
        noise_ratio: float = noise_count / len(topics) if topics else 0.0

        return {
            "topics": topics,
            "topic_info": topic_info.to_dict(orient="records"),
            "noise_ratio": round(noise_ratio, 4),
            "total_topics": int(topic_info.shape[0]) - 1,  # -1 행 제외
        }

# 1. 의존성 설치 (uv 기반)
uv add bertopic hdbscan umap-learn sentence-transformers pydantic-settings

# 2. .env 작성
echo 'NAVER_CLIENT_ID=발급받은_id' >> .env
echo 'NAVER_CLIENT_SECRET=발급받은_secret' >> .env
echo 'BERTOPIC_MIN_TOPIC_SIZE=3' >> .env

# 3. 뉴스 수집 -> 전처리 -> 클러스터링 순서로 실행
python -m app.services.news_fetcher