trend_engine/place_extractor.py

"""
PlaceNameExtractor — 블로그/유튜브 텍스트에서 장소명을 추출하는 공통 모듈

추출 우선순위:
  1. 장소명 사전 매칭 (trend_spots + story_spots 이름)
  2. 유튜브 전용 패턴 (📍, @, 번호 리스트)
  3. 접미사 패턴 매칭 (카페, 식당, 해변 등)
"""

import re
import logging

from supabase import Client

from db import get_supabase

from trend_engine.spot_matcher import is_system_name

logger = logging.getLogger(__name__)

# 장소명 접미사 패턴 (사전에 없는 새 장소 발견용)
SUFFIX_PATTERNS = [
    re.compile(r"[\w가-힣]{2,}(?:카페|커피)"),
    re.compile(r"[\w가-힣]{2,}(?:식당|밥집|국수)"),
    re.compile(r"[\w가-힣]{2,}(?:해변|해안|포구)"),
    re.compile(r"[\w가-힣]{2,}(?:오름|공원|숲)"),
    re.compile(r"[\w가-힣]{2,}(?:전망대|뷰)"),
    re.compile(r"[\w가-힣]{2,}(?:베이커리|브런치|디저트)"),
]

# 유튜브 영상 설명에서 자주 사용되는 장소 소개 패턴
YOUTUBE_PLACE_PATTERNS = [
    re.compile(r"📍\s*([\w가-힣\s]{2,20})"),
    re.compile(r"(?:방문|다녀온|들른)\s*(?:곳|장소)[:\s]*([\w가-힣\s]{2,20})"),
    re.compile(r"(?:\d+\.\s*)([\w가-힣]{2,15})"),
    re.compile(r"@\s*([\w가-힣]{2,15})"),
]

# 너무 일반적인 단어 — 장소명이 아닌 경우 필터링
STOPWORDS = frozenset({
    # 지시대명사 + 카테고리
    "이카페", "그카페", "저카페", "이식당", "그식당",
    "이해변", "그해변", "이오름", "그오름",
    # 형용사 + 카테고리
    "좋은카페", "예쁜카페", "맛있는식당", "좋은식당",
    # 위치/수식어 + 카테고리
    "근처카페", "주변카페", "동네카페", "유명카페",
    "감성카페", "뷰카페", "오션뷰카페",
    "애월카페", "한림카페", "협재카페",
    "제주카페", "제주식당", "제주해변", "제주오름",
    "예쁜식당", "제주맛집", "애월맛집",
    # 카테고리 일반명사 (단독 사용 시 장소명 아님)
    "카페", "커피", "식당", "맛집", "밥집", "국수",
    "해변", "해안", "포구", "오름", "공원", "숲",
    "전망대", "뷰", "베이커리", "브런치", "디저트",
    "산책", "산책로", "드라이브",
    # 지역 + 일반 표현
    "제주여행", "애월여행", "제주도", "애월", "애월리",
})

# 주소 패턴 — 장소명 추출 전 제거
ADDRESS_PATTERNS = [
    re.compile(r"제주\s*(?:특별자치)?도?\s*제주시\s*애월읍\s*[\w가-힣\d\-]+"),
    re.compile(r"애월읍\s*[\w가-힣]+리\s*[\d\-]+"),
    re.compile(r"제주시\s*[\w가-힣]+(?:읍|면)\s*[\w가-힣]+(?:리|로|길)\s*[\d\-]*"),
]


def preprocess_text(text: str) -> str:
    """장소명 추출 전 주소 패턴 및 HTML 태그 제거."""
    text = re.sub(r"<[^>]+>", "", text)
    for pattern in ADDRESS_PATTERNS:
        text = pattern.sub("", text)
    return text


class PlaceNameExtractor:
    """텍스트에서 장소명을 추출하는 파이프라인."""

    def __init__(self, supabase: Client | None = None):
        self.supabase = supabase or get_supabase()
        # {장소명: spot_id} — 정확 매칭용
        self.known_places: dict[str, str] = {}
        self._load_place_dictionary()

    # ------------------------------------------------------------------
    # 사전 구축
    # ------------------------------------------------------------------

    def _load_place_dictionary(self) -> None:
        """trend_spots + story_spots에서 장소명 사전을 구축한다."""

        # 1) trend_spots
        try:
            resp = self.supabase.table("trend_spots").select("id, name").execute()
            for row in resp.data or []:
                name = row["name"]
                if not is_system_name(name):
                    self._register_name(name, row["id"])
        except Exception as e:
            logger.warning("trend_spots 로드 실패 (테이블 미존재 가능): %s", e)

        # 2) story_spots
        try:
            resp = (
                self.supabase.table("story_spots")
                .select("id, name, aliases")
                .execute()
            )
            for row in resp.data or []:
                name = row["name"]
                if name and not is_system_name(name):
                    self._register_name(name, row["id"])
                    for alias in row.get("aliases") or []:
                        if alias and not is_system_name(alias):
                            self._register_name(alias, row["id"])
        except Exception as e:
            logger.warning("story_spots 로드 실패: %s", e)

        logger.info("장소명 사전 구축 완료: %d건", len(self.known_places))

    def _register_name(self, name: str, spot_id: str) -> None:
        """이름과 공백 제거 변형을 사전에 등록한다."""
        name = name.strip()
        if not name or len(name) < 2:
            return
        # STOPWORDS 체크
        if name in STOPWORDS:
            return
        self.known_places[name] = spot_id
        no_space = name.replace(" ", "")
        if no_space != name and no_space not in STOPWORDS:
            self.known_places[no_space] = spot_id

    # ------------------------------------------------------------------
    # 추출
    # ------------------------------------------------------------------

    def extract(self, text: str) -> list[dict]:
        """
        텍스트에서 장소명을 추출한다.

        Returns:
            [{"name": str, "spot_id": str|None, "method": "dictionary"|"pattern"|"youtube_pattern"}, ...]
        """
        if not text:
            return []

        # 전처리: 주소 패턴 + HTML 태그 제거
        text = preprocess_text(text)

        found: list[dict] = []
        found_names: set[str] = set()

        # 1순위: 사전 매칭 — 긴 이름부터 매칭 (부분 매칭 방지)
        for name in sorted(self.known_places, key=len, reverse=True):
            if name in STOPWORDS:
                continue
            if name in text and name not in found_names:
                found.append({
                    "name": name,
                    "spot_id": self.known_places[name],
                    "method": "dictionary",
                })
                found_names.add(name)

        # 2순위: 유튜브 전용 패턴
        for pattern in YOUTUBE_PLACE_PATTERNS:
            for match in pattern.findall(text):
                match = match.strip()
                if match and match not in found_names and match not in STOPWORDS:
                    found.append({
                        "name": match,
                        "spot_id": None,
                        "method": "youtube_pattern",
                    })
                    found_names.add(match)

        # 3순위: 접미사 패턴
        for pattern in SUFFIX_PATTERNS:
            for match in pattern.findall(text):
                if match not in found_names and match not in STOPWORDS:
                    found.append({
                        "name": match,
                        "spot_id": None,
                        "method": "pattern",
                    })
                    found_names.add(match)

        return found