web-page-backend/insta-lab/app/trend_collector.py

"""외부 트렌드 수집 — NAVER 인기 + YouTube 인기 영상 + LLM 카테고리 분류.

NAVER: 카테고리별 시드 키워드로 인기 검색 → 빈도 상위 추출.
YouTube: Google Trends 비공식 endpoint(RSS / dailytrends JSON)가 모두 404 폐기되어
대체로 YouTube Data API v3 (`videos.list?chart=mostPopular&regionCode=KR`) 사용.
무료 일일 quota 10000, 한국 region 지원, 인기 영상 50개 제목에서 트렌드 추출.
LLM 분류 결과는 24h in-memory 캐시.
"""

import json
import logging
import re
import time
from typing import Any, Dict, List, Optional

import requests
from anthropic import Anthropic

from .config import (
    NAVER_CLIENT_ID, NAVER_CLIENT_SECRET, DEFAULT_CATEGORY_SEEDS,
    ANTHROPIC_API_KEY, ANTHROPIC_MODEL_HAIKU, YOUTUBE_DATA_API_KEY,
)
from . import db
from .news_collector import _clean
from .keyword_extractor import _count_nouns, _top_candidates

logger = logging.getLogger(__name__)

NEWS_URL = "https://openapi.naver.com/v1/search/news.json"
_NAVER_HEADERS = {
    "X-Naver-Client-Id": NAVER_CLIENT_ID,
    "X-Naver-Client-Secret": NAVER_CLIENT_SECRET,
}

YOUTUBE_TRENDING_URL = "https://www.googleapis.com/youtube/v3/videos"
# YouTube 제목 정제: 대괄호·이모지·과도한 길이 제거 후 카드 주제로 적합한 키워드 형태
_TITLE_BRACKET_RE = re.compile(r"[\[【「『\(][^\]】」』\)]{0,30}[\]】」』\)]")
_EMOJI_RE = re.compile(
    r"["
    r"\U0001F300-\U0001FAFF"   # symbols & pictographs, etc.
    r"\U00002600-\U000027BF"   # misc symbols, dingbats
    r"\U0001F1E6-\U0001F1FF"   # regional indicator
    r"]"
)
_TITLE_MAX_LEN = 60

_PLACEHOLDER_SEEDS = {"...", "…", "tbd", "todo", "placeholder", "example"}


def _is_valid_seed(s: str) -> bool:
    """프롬프트 템플릿에 placeholder/빈 값이 들어가 NAVER에 400을 유발하는 일을 막는 가드."""
    if not s:
        return False
    s = s.strip()
    if len(s) < 2:
        return False
    if s.lower() in _PLACEHOLDER_SEEDS:
        return False
    return True


def _seeds_for(category: str) -> List[str]:
    """category_seeds 프롬프트 템플릿이 있으면 사용, 없거나 모두 invalid면 config DEFAULT 폴백."""
    pt = db.get_prompt_template("category_seeds")
    if pt and pt.get("template"):
        try:
            data = json.loads(pt["template"])
            if category in data:
                filtered = [s for s in (data[category] or []) if _is_valid_seed(s)]
                if filtered:
                    return filtered
                logger.warning("category_seeds[%s]에 유효한 시드 없음 → DEFAULT 폴백", category)
        except Exception as e:
            logger.warning("category_seeds JSON 파싱 실패 → DEFAULT 폴백: %s", e)
    return list(DEFAULT_CATEGORY_SEEDS.get(category, []))


def fetch_naver_popular(category: str, per_seed: int = 30, top_n: int = 10) -> List[Dict[str, Any]]:
    """카테고리 시드 키워드들로 NAVER news.json `sort=sim` 호출,
    응답 기사 묶음에서 빈도어 추출 후 상위 N개 반환."""
    seeds = _seeds_for(category)
    if not seeds:
        return []
    blob_parts: List[str] = []
    for seed in seeds:
        try:
            resp = requests.get(
                NEWS_URL,
                headers=_NAVER_HEADERS,
                params={"query": seed, "display": per_seed, "sort": "sim"},
                timeout=10,
            )
            resp.raise_for_status()
            for item in resp.json().get("items", []):
                blob_parts.append(_clean(item.get("title", "")))
                blob_parts.append(_clean(item.get("description", "")))
        except Exception as e:
            logger.warning("fetch_naver_popular seed=%s err=%s", seed, e)
            continue
    text = "\n".join(blob_parts)
    counts = _count_nouns(text)
    candidates = _top_candidates(counts, n=top_n)
    if not candidates:
        return []
    max_count = candidates[0][1] or 1
    return [
        {
            "keyword": k,
            "category": category,
            "source": "naver_popular",
            "score": round(min(1.0, c / max_count), 4),
            "articles_count": c,
        }
        for k, c in candidates
    ]


def collect_naver_popular_for(categories: List[str]) -> int:
    total = 0
    for cat in categories:
        trends = fetch_naver_popular(cat)
        for t in trends:
            db.add_external_trend(t)
            total += 1
    return total


# ── LLM 분류 캐시 ────────────────────────────────────────────────────────────

_CACHE_TTL_SEC = 24 * 3600
_category_cache: Dict[str, tuple] = {}  # keyword -> (category, expires_ts)


def _llm_classify_one(keyword: str) -> str:
    """Claude Haiku 1회 호출로 단일 키워드 분류."""
    if not ANTHROPIC_API_KEY:
        return "uncategorized"
    seeds_template = db.get_prompt_template("category_seeds")
    if seeds_template and seeds_template.get("template"):
        try:
            allowed = sorted(json.loads(seeds_template["template"]).keys())
        except Exception:
            allowed = sorted(DEFAULT_CATEGORY_SEEDS.keys())
    else:
        allowed = sorted(DEFAULT_CATEGORY_SEEDS.keys())
    allowed.append("uncategorized")

    client = Anthropic(api_key=ANTHROPIC_API_KEY)
    msg = client.messages.create(
        model=ANTHROPIC_MODEL_HAIKU,
        max_tokens=20,
        messages=[{
            "role": "user",
            "content": (
                f"다음 한국어 트렌딩 키워드를 카테고리 중 하나로 분류해라. "
                f"카테고리: {allowed}. 키워드: '{keyword}'. "
                f"카테고리명 한 단어만 출력. 다른 텍스트 금지."
            ),
        }],
    )
    raw = msg.content[0].text.strip().lower()
    for cat in allowed:
        if cat.lower() in raw:
            return cat
    return "uncategorized"


def classify_keyword(keyword: str) -> str:
    now = time.time()
    cached = _category_cache.get(keyword)
    if cached and cached[1] > now:
        return cached[0]
    cat = _llm_classify_one(keyword)
    _category_cache[keyword] = (cat, now + _CACHE_TTL_SEC)
    return cat


# ── YouTube Trending ──────────────────────────────────────────────────────────
# YouTube Data API v3 videos.list?chart=mostPopular&regionCode=KR
# 한국 인기 영상 50개 제목에서 카드 주제로 적합한 키워드 추출.

def _clean_yt_title(title: str) -> str:
    """[공식]·【속보】·🔥 등 제거 후 60자 이내로 자른다."""
    if not title:
        return ""
    cleaned = _TITLE_BRACKET_RE.sub("", title)
    cleaned = _EMOJI_RE.sub("", cleaned)
    cleaned = re.sub(r"\s+", " ", cleaned).strip()
    return cleaned[:_TITLE_MAX_LEN]


def fetch_youtube_trending() -> List[Dict[str, Any]]:
    """YouTube Data API v3 mostPopular (한국, 50개). API 키 없거나 호출 실패 시 빈 리스트."""
    if not YOUTUBE_DATA_API_KEY:
        logger.info("YOUTUBE_DATA_API_KEY 미설정 — youtube_trending skip")
        return []
    try:
        resp = requests.get(
            YOUTUBE_TRENDING_URL,
            params={
                "part": "snippet",
                "chart": "mostPopular",
                "regionCode": "KR",
                "maxResults": 50,
                "key": YOUTUBE_DATA_API_KEY,
            },
            timeout=15,
        )
        resp.raise_for_status()
        videos = resp.json().get("items", []) or []
    except Exception as e:
        logger.warning("YouTube trending fetch failed: %s", e)
        return []

    items: List[Dict[str, Any]] = []
    seen = set()
    total = max(1, len(videos))
    for idx, v in enumerate(videos):
        title = (v.get("snippet") or {}).get("title", "")
        kw = _clean_yt_title(title)
        if not kw or kw in seen:
            continue
        seen.add(kw)
        try:
            cat = classify_keyword(kw)
        except Exception as e:
            logger.warning("classify_keyword(%s) 실패: %s", kw, e)
            cat = "uncategorized"
        rank_score = round(max(0.0, 1.0 - (idx / total)), 4)
        items.append({
            "keyword": kw,
            "category": cat,
            "source": "youtube_trending",
            "score": rank_score,
            "articles_count": 0,
        })
    return items


def collect_youtube_trending() -> int:
    items = fetch_youtube_trending()
    for it in items:
        db.add_external_trend(it)
    return len(items)


def collect_all(categories: List[str]) -> Dict[str, int]:
    naver_n = collect_naver_popular_for(categories)
    yt_n = collect_youtube_trending()
    return {"naver_popular": naver_n, "youtube_trending": yt_n}