Upload folder using huggingface_hub

requirements-trend.txt

@@ -0,0 +1,8 @@
+httpx>=0.27.0
+beautifulsoup4>=4.12.0
+lxml>=5.0.0
+requests>=2.31.0
+google-api-python-client>=2.100.0
+apify-client>=1.6.0
+supabase>=2.0.0
+python-dotenv>=1.0.0

scripts/run_trend_engine.py

@@ -0,0 +1,248 @@
+"""
+RE:Play Trend Engine v3 — 주간 배치 오케스트레이터
+
+순차 실행 파이프라인:
+  1. 카카오맵 그리드 스캔 + 리뷰 파싱 (trend_spots 마스터 생성)
+  2. SpotMatcher 초기화 (trend_spots + story_spots 사전 로드)
+  3. 네이버 블로그 수집 (URL 확보 + 크롤링 + DB 저장)
+  4. 블로그 본문 → 장소명 추출 + mention_count 집계
+  5. 유튜브 API (SpotMatcher 연동)
+  6. 인스타그램 Apify (SpotMatcher 연동)
+  7. 종합 스코어 계산 + 랭킹 생성
+
+Usage:
+  python backend/scripts/run_trend_engine.py
+"""
+
+import asyncio
+import json
+import logging
+import os
+import re
+import sys
+import time
+from datetime import date, timedelta
+
+# backend/ 디렉토리를 import path에 추가
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+# 로컬 실행 시 .env 파일 로드
+try:
+    from dotenv import load_dotenv
+    # 프로젝트 루트의 .env 파일 로드
+    env_path = os.path.join(os.path.dirname(__file__), "..", "..", ".env")
+    load_dotenv(env_path)
+except ImportError:
+    pass  # GitHub Actions 등 dotenv 없는 환경
+
+from supabase import create_client
+
+from trend_engine.collectors.naver_blog import NaverBlogCollector
+from trend_engine.collectors.kakaomap import KakaoMapCollector
+from trend_engine.collectors.youtube import YouTubeCollector
+from trend_engine.collectors.instagram import InstagramCollector
+from trend_engine.spot_matcher import SpotMatcher
+from trend_engine.trend_scorer import generate_weekly_ranking
+from trend_engine.place_extractor import PlaceNameExtractor
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger("trend_engine.orchestrator")
+
+
+def _get_supabase_client():
+    url = os.environ.get("SUPABASE_URL") or os.environ.get("VITE_SUPABASE_URL")
+    key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY") or os.environ.get("SUPABASE_SERVICE_KEY")
+    if not url or not key:
+        raise ValueError("SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY must be set")
+    return create_client(url, key)
+
+
+def run_step(name: str, func, results: dict):
+    """단일 파이프라인 단계를 실행하고 결과를 기록한다. 반환값을 돌려준다."""
+    logger.info("━━━ [START] %s ━━━", name)
+    start = time.time()
+    try:
+        result = func()
+        elapsed = time.time() - start
+        results[name] = {"status": "ok", "result": _summarize(result), "elapsed_sec": round(elapsed, 1)}
+        logger.info("✓ [DONE] %s — %.1f초", name, elapsed)
+        return result
+    except Exception as e:
+        elapsed = time.time() - start
+        results[name] = {"status": "error", "error": str(e), "elapsed_sec": round(elapsed, 1)}
+        logger.error("✗ [FAIL] %s — %s (%.1f초)", name, e, elapsed)
+        return None
+
+
+def run_async_step(name: str, coro, results: dict):
+    """asyncio 코루틴을 실행하는 run_step 변형."""
+    def wrapper():
+        return asyncio.run(coro)
+    return run_step(name, wrapper, results)
+
+
+def _summarize(result) -> str:
+    """결과를 로그용 요약 문자열로 변환."""
+    if isinstance(result, dict):
+        return json.dumps(result, ensure_ascii=False, default=str)[:200]
+    return str(result)[:200]
+
+
+def main() -> None:
+    total_start = time.time()
+    results: dict = {}
+
+    sb = _get_supabase_client()
+
+    # ── 1. 카카오맵 그리드 스캔 (먼저 실행 → trend_spots 마스터 생성) ──
+    kakao = KakaoMapCollector(sb)
+    run_async_step("1_kakaomap", kakao.run(), results)
+
+    # ── 2. SpotMatcher 초기화 (trend_spots + story_spots 사전 로드) ──
+    matcher = SpotMatcher(sb)
+    extractor = PlaceNameExtractor(sb)
+    logger.info(
+        "SpotMatcher 준비 완료 — trend_spots %d건, story_spots %d건",
+        len(matcher.trend_spots),
+        len(matcher.story_spots),
+    )
+
+    # ── 3. 유튜브 API (SpotMatcher 연동) ──
+    youtube = YouTubeCollector(sb, spot_matcher=matcher)
+    run_step("3_youtube", youtube.run, results)
+
+    # ── 4. 인스타그램 Apify (SpotMatcher 연동) ──
+    instagram = InstagramCollector(sb, spot_matcher=matcher)
+    run_step("4_instagram", instagram.run, results)
+
+    # ── 5. 네이버 플레이스 — 비활성 (Place ID 매칭 불가) ──
+    logger.info("네이버 플레이스: 비활성 (Place ID 매칭 불가, 2026-02)")
+    results["5_naver_place"] = {
+        "status": "skipped",
+        "reason": "Place ID matching unavailable",
+        "elapsed_sec": 0,
+    }
+
+    # ── 6. 네이버 블로그 수집 (URL + 병렬 크롤링 + 저장) ──
+    blog = NaverBlogCollector(sb)
+    run_step("6_naver_blog", blog.run, results)
+
+    # ── 7. 블로그 본문 → 장소명 추출 + mention_count 집계 ──
+    def extract_blog_places():
+        """블로그 포스트에서 장소명 추출 → mention_count 집계 → spot_trends 저장."""
+        today = date.today()
+        period_start = today - timedelta(days=7)
+
+        # spot_trends에서 naver_blog + __pending__ 레코드 조회 (페이지네이션)
+        records = []
+        page_size = 1000
+        offset = 0
+        try:
+            while True:
+                batch = (
+                    sb.table("spot_trends")
+                    .select("id, raw_data")
+                    .eq("source", "naver_blog")
+                    .eq("spot_id", "__pending__")
+                    .range(offset, offset + page_size - 1)
+                    .execute()
+                )
+                rows = batch.data or []
+                records.extend(rows)
+                if len(rows) < page_size:
+                    break
+                offset += page_size
+        except Exception as e:
+            logger.warning("블로그 pending 레코드 조회 실패: %s", e)
+            return {"error": str(e)}
+
+        logger.info("블로그 pending 레코드: %d건 조회", len(records))
+        if not records:
+            return {"pending_records": 0, "places_found": 0}
+
+        # 장소별 언급 횟수 집계
+        place_mentions: dict[str, int] = {}
+
+        for record in records:
+            raw = record.get("raw_data", {})
+            content = raw.get("content_preview", "")
+            title = raw.get("title", "")
+            text = f"{title} {content}"
+            text = re.sub(r"<[^>]+>", "", text)  # HTML 태그 제거
+
+            places = extractor.extract(text)
+            for place in places:
+                matched_id = matcher.match(place["name"])
+                if matched_id:
+                    place_mentions[matched_id] = place_mentions.get(matched_id, 0) + 1
+
+        # 집계 결과를 spot_trends에 저장 (장소별 mention_count)
+        saved = 0
+        for spot_id, count in place_mentions.items():
+            try:
+                sb.table("spot_trends").insert({
+                    "spot_id": spot_id,
+                    "source": "naver_blog",
+                    "metric_type": "mention_count",
+                    "metric_value": count,
+                    "period_start": period_start.isoformat(),
+                    "period_end": today.isoformat(),
+                    "raw_data": {"aggregated_from": "blog_post_extraction"},
+                }).execute()
+                saved += 1
+            except Exception as e:
+                logger.warning("mention_count 저장 실패 (%s): %s", spot_id, e)
+
+        return {
+            "pending_records": len(records),
+            "places_found": len(place_mentions),
+            "mention_records_saved": saved,
+        }
+
+    run_step("7_blog_place_extraction", extract_blog_places, results)
+
+    # ── 8. 종합 스코어 계산 + 랭킹 생성 ──
+    def calc_scores():
+        return generate_weekly_ranking(sb)
+
+    run_step("8_score_calculation", calc_scores, results)
+
+    # ── 결과 요약 ──
+    total_elapsed = time.time() - total_start
+    ok_count = sum(1 for r in results.values() if r.get("status") == "ok")
+    err_count = sum(1 for r in results.values() if r.get("status") == "error")
+    skip_count = sum(1 for r in results.values() if r.get("status") == "skipped")
+
+    summary = {
+        "total_steps": len(results),
+        "succeeded": ok_count,
+        "failed": err_count,
+        "skipped": skip_count,
+        "total_elapsed_sec": round(total_elapsed, 1),
+        "steps": {
+            k: {"status": v.get("status"), "elapsed_sec": v.get("elapsed_sec", 0)}
+            for k, v in results.items()
+        },
+    }
+
+    logger.info("━━━ TREND ENGINE COMPLETE ━━━")
+    logger.info(
+        "성공: %d / 실패: %d / 스킵: %d / 총 소요: %.1f초",
+        ok_count, err_count, skip_count, total_elapsed,
+    )
+
+    # JSON 요약 출력 (GitHub Actions 로그용)
+    print(json.dumps(summary, ensure_ascii=False, indent=2))
+
+    # 전체 실패 시에만 비정상 종료
+    if ok_count == 0:
+        logger.error("모든 단계가 실패했습니다.")
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

trend_engine/__init__.py

@@ -0,0 +1 @@
+"""RE:Play Trend Engine v3 — 트렌드 데이터 수집 파이프라인"""

trend_engine/collectors/__init__.py

@@ -0,0 +1,15 @@
+"""Trend Engine data collectors — 채널별 수집기"""
+
+from .naver_blog import NaverBlogCollector
+from .naver_place import NaverPlaceCollector
+from .kakaomap import KakaoMapCollector
+from .youtube import YouTubeCollector
+from .instagram import InstagramCollector
+
+__all__ = [
+    "NaverBlogCollector",
+    "NaverPlaceCollector",
+    "KakaoMapCollector",
+    "YouTubeCollector",
+    "InstagramCollector",
+]

trend_engine/collectors/instagram.py

@@ -0,0 +1,277 @@
+"""
+Instagram Collector — Apify SaaS (Instagram Hashtag Scraper)
+
+수집 방식:
+  1. Apify의 instagram-hashtag-scraper Actor로 해시태그별 게시물 수집
+  2. 위치별 평균 참여도(좋아요+댓글) 집계
+  3. spot_trends 테이블에 저장
+"""
+
+import os
+import logging
+from datetime import date
+
+from apify_client import ApifyClient
+
+from trend_engine.place_extractor import PlaceNameExtractor
+
+logger = logging.getLogger(__name__)
+
+# ──────────────────────────────────────────────────
+# 설정
+# ──────────────────────────────────────────────────
+
+INSTAGRAM_HASHTAGS = [
+    "애월카페",
+    "애월맛집",
+    "애월가볼만한곳",
+    "애월해안",
+    "애월여행",
+    "제주애월",
+    "애월감성",
+    "한담해안",
+    "곽지해변",
+    "애월핫플",
+    "애월디저트",
+]
+
+RESULTS_LIMIT_PER_HASHTAG = 50
+
+ACTOR_ID = "apify/instagram-hashtag-scraper"
+
+
+class InstagramCollector:
+    """Apify Instagram Hashtag Scraper 기반 수집기."""
+
+    def __init__(self, supabase_client, spot_matcher=None):
+        self.supabase = supabase_client
+        self.apify = ApifyClient(os.environ["APIFY_API_TOKEN"])
+        self.spot_matcher = spot_matcher
+        self.extractor = PlaceNameExtractor(supabase_client)
+
+    # ------------------------------------------------------------------
+    # 해시태그 게시물 수집
+    # ------------------------------------------------------------------
+
+    def collect_hashtag_posts(self) -> list[dict]:
+        """
+        Apify Actor를 실행하여 해시태그별 게시물을 수집한다.
+
+        Returns:
+            [{hashtag, location_name, likes_count, comments_count, caption, timestamp}, ...]
+        """
+        run_input = {
+            "hashtags": INSTAGRAM_HASHTAGS,
+            "resultsLimit": RESULTS_LIMIT_PER_HASHTAG,
+        }
+
+        logger.info(
+            "Apify Actor 실행 시작: %d개 해시태그, 해시태그당 %d건",
+            len(INSTAGRAM_HASHTAGS), RESULTS_LIMIT_PER_HASHTAG,
+        )
+
+        try:
+            run = self.apify.actor(ACTOR_ID).call(run_input=run_input)
+        except Exception as e:
+            logger.error("Apify Actor 실행 실패: %s", e)
+            return []
+
+        all_posts: list[dict] = []
+        dataset_id = run["defaultDatasetId"]
+
+        for item in self.apify.dataset(dataset_id).iterate_items():
+            all_posts.append({
+                "hashtag": item.get("hashtag", ""),
+                "location_name": item.get("locationName", ""),
+                "likes_count": item.get("likesCount", 0),
+                "comments_count": item.get("commentsCount", 0),
+                "caption": item.get("caption", ""),
+                "timestamp": item.get("timestamp", ""),
+                "url": item.get("url", ""),
+            })
+
+        logger.info("게시물 수집 완료: %d건", len(all_posts))
+        return all_posts
+
+    # ------------------------------------------------------------------
+    # 위치별 메트릭 집계
+    # ------------------------------------------------------------------
+
+    def aggregate_location_metrics(self, posts: list[dict]) -> dict[str, dict]:
+        """
+        수집된 게시물에서 위치별 인기도 메트릭을 집계한다.
+
+        1차: 위치 태그 기반 (location_name)
+        2차: 캡션 기반 장소명 추출 (위치 태그 없는 게시물)
+
+        Returns:
+            {location_name: {post_count, total_likes, total_comments, avg_engagement}}
+        """
+        location_metrics: dict[str, dict] = {}
+        no_location_posts: list[dict] = []
+
+        for post in posts:
+            loc = post["location_name"]
+            if not loc:
+                no_location_posts.append(post)
+                continue
+
+            if loc not in location_metrics:
+                location_metrics[loc] = {
+                    "post_count": 0,
+                    "total_likes": 0,
+                    "total_comments": 0,
+                    "hashtags": set(),
+                }
+
+            location_metrics[loc]["post_count"] += 1
+            location_metrics[loc]["total_likes"] += post["likes_count"]
+            location_metrics[loc]["total_comments"] += post["comments_count"]
+            if post["hashtag"]:
+                location_metrics[loc]["hashtags"].add(post["hashtag"])
+
+        # 2차: 캡션 기반 장소명 추출 (위치 태그 없는 게시물)
+        caption_extracted = 0
+        for post in no_location_posts:
+            caption = post.get("caption", "")
+            if not caption or len(caption) < 5:
+                continue
+
+            places = self.extractor.extract(caption)
+            for place in places:
+                loc = place["name"]
+                if loc not in location_metrics:
+                    location_metrics[loc] = {
+                        "post_count": 0,
+                        "total_likes": 0,
+                        "total_comments": 0,
+                        "hashtags": set(),
+                    }
+                location_metrics[loc]["post_count"] += 1
+                location_metrics[loc]["total_likes"] += post["likes_count"]
+                location_metrics[loc]["total_comments"] += post["comments_count"]
+                if post["hashtag"]:
+                    location_metrics[loc]["hashtags"].add(post["hashtag"])
+                caption_extracted += 1
+                break  # 게시물당 1개 장소만 카운트
+
+        # 평균 참여도 계산 + set → list 변환
+        for loc, metrics in location_metrics.items():
+            count = max(metrics["post_count"], 1)
+            metrics["avg_engagement"] = (
+                metrics["total_likes"] + metrics["total_comments"]
+            ) / count
+            metrics["hashtags"] = sorted(metrics["hashtags"])
+
+        logger.info(
+            "위치별 집계 완료: %d개 위치 (위치태그 %d건, 캡션추출 %d건, 미식별 %d건)",
+            len(location_metrics),
+            sum(1 for p in posts if p["location_name"]),
+            caption_extracted,
+            len(no_location_posts) - caption_extracted,
+        )
+        return location_metrics
+
+    # ------------------------------------------------------------------
+    # DB 저장
+    # ------------------------------------------------------------------
+
+    def _save_to_db(
+        self,
+        location_metrics: dict[str, dict],
+        total_posts: int,
+    ) -> int:
+        """집계된 메트릭을 spot_trends 테이블에 저장한다.
+
+        SpotMatcher가 설정된 경우 trend_spots.id로 정규화.
+        매칭 실패한 위치는 스킵.
+        """
+        today = date.today()
+        saved = 0
+        skipped = 0
+
+        for loc_name, metrics in location_metrics.items():
+            # SpotMatcher로 spot_id 정규화
+            if self.spot_matcher:
+                spot_id = self.spot_matcher.match(loc_name)
+                if not spot_id:
+                    skipped += 1
+                    continue
+            else:
+                spot_id = f"ig_{loc_name}"
+
+            # post_count 메트릭
+            try:
+                self.supabase.table("spot_trends").insert({
+                    "spot_id": spot_id,
+                    "source": "instagram",
+                    "metric_type": "post_count",
+                    "metric_value": metrics["post_count"],
+                    "period_start": today.isoformat(),
+                    "period_end": today.isoformat(),
+                    "raw_data": {
+                        "location_name": loc_name,
+                        "total_likes": metrics["total_likes"],
+                        "total_comments": metrics["total_comments"],
+                        "avg_engagement": metrics["avg_engagement"],
+                        "hashtags": metrics["hashtags"],
+                    },
+                }).execute()
+                saved += 1
+            except Exception as e:
+                logger.warning("spot_trends insert 실패 (instagram post_count, %s): %s", spot_id, e)
+
+            # avg_engagement 메트릭 (채널 스코어링에 필요)
+            avg_eng = metrics["avg_engagement"]
+            if avg_eng > 0:  # -1 값 필터링 (Apify 미수집 케이스)
+                try:
+                    self.supabase.table("spot_trends").insert({
+                        "spot_id": spot_id,
+                        "source": "instagram",
+                        "metric_type": "avg_engagement",
+                        "metric_value": int(round(avg_eng)),
+                        "period_start": today.isoformat(),
+                        "period_end": today.isoformat(),
+                        "raw_data": {"location_name": loc_name},
+                    }).execute()
+                except Exception as e:
+                    logger.warning("spot_trends insert 실패 (instagram avg_engagement, %s): %s", spot_id, e)
+
+        if skipped:
+            logger.info("Instagram 장소 매칭 실패로 %d건 스킵", skipped)
+        logger.info("Instagram DB 저장 완료: %d건", saved)
+        return saved
+
+    # ------------------------------------------------------------------
+    # 메인 실행
+    # ------------------------------------------------------------------
+
+    def run(self) -> dict:
+        """
+        Instagram 수집 파이프라인 전체 실행.
+
+        Returns:
+            {"total_posts": int, "locations_found": int, "saved": int}
+        """
+        logger.info("=== Instagram 수집 시작 ===")
+
+        # 1단계: 해시태그 ���시물 수집
+        posts = self.collect_hashtag_posts()
+
+        if not posts:
+            logger.warning("수집된 게시물 없음 — 종료")
+            return {"total_posts": 0, "locations_found": 0, "saved": 0}
+
+        # 2단계: 위치별 메트릭 집계
+        location_metrics = self.aggregate_location_metrics(posts)
+
+        # 3단계: DB 저장
+        saved = self._save_to_db(location_metrics, len(posts))
+
+        result = {
+            "total_posts": len(posts),
+            "locations_found": len(location_metrics),
+            "saved": saved,
+        }
+        logger.info("=== Instagram 수집 완료: %s ===", result)
+        return result

trend_engine/collectors/kakaomap.py

@@ -0,0 +1,309 @@
+"""
+KakaoMap Collector — 그리드 분할 카테고리 스캔 + 리뷰 수 파싱
+
+수집 방식:
+  1. 애월 지역을 2km×2km 그리드로 분할
+  2. 각 셀 × 4개 카테고리(CE7/FD6/AT4/CT1) × 3페이지 조회
+  3. 상위 100개 장소의 리뷰 수를 비공식 API로 파싱 (graceful degradation)
+"""
+
+import os
+import logging
+from datetime import date, datetime
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# ──────────────────────────────────────────────────
+# 애월 그리드 설정
+# ──────────────────────────────────────────────────
+
+# 애월리 중심 그리드 (애월항~한담해안~곽지해변 커버)
+AEWOL_GRID = {
+    "west": 126.30,
+    "east": 126.36,
+    "south": 33.44,
+    "north": 33.47,
+    "cell_lng": 0.023,  # ~2km 경도
+    "cell_lat": 0.018,  # ~2km 위도
+}
+
+CATEGORY_CODES = {
+    "CE7": "카페",
+    "FD6": "음식점",
+    "AT4": "관광명소",
+    "CT1": "문화시설",
+}
+
+KAKAO_CATEGORY_URL = "https://dapi.kakao.com/v2/local/search/category"
+KAKAO_PLACE_DETAIL_URL = "https://place.map.kakao.com/main/v/{place_id}"
+
+REVIEW_PARSE_LIMIT = 100  # 리뷰 수 파싱 대상 상위 N개
+
+
+class KakaoMapCollector:
+    """카카오맵 그리드 분할 스캔 + 리뷰 수 파싱 수집기."""
+
+    def __init__(self, supabase_client):
+        self.supabase = supabase_client
+        self.api_key = os.environ["KAKAO_REST_API_KEY"]
+
+    # ------------------------------------------------------------------
+    # 그리드 생성
+    # ------------------------------------------------------------------
+
+    def generate_grid_cells(self) -> list[dict]:
+        """애월 지역을 2km×2km 셀로 분할하여 rect 파라미터 목록을 생성한다."""
+        cells = []
+        lng = AEWOL_GRID["west"]
+        while lng < AEWOL_GRID["east"]:
+            lat = AEWOL_GRID["south"]
+            while lat < AEWOL_GRID["north"]:
+                cells.append({
+                    "x1": lng,
+                    "y1": lat,
+                    "x2": min(lng + AEWOL_GRID["cell_lng"], AEWOL_GRID["east"]),
+                    "y2": min(lat + AEWOL_GRID["cell_lat"], AEWOL_GRID["north"]),
+                })
+                lat += AEWOL_GRID["cell_lat"]
+            lng += AEWOL_GRID["cell_lng"]
+
+        logger.info("그리드 셀 %d개 생성 완료", len(cells))
+        return cells
+
+    # ------------------------------------------------------------------
+    # 카테고리 스캔
+    # ------------------------------------------------------------------
+
+    def collect_spots(self) -> dict[str, dict]:
+        """
+        카카오맵 카테고리 검색 API로 애월 전역 장소를 수집한다.
+
+        Returns:
+            {place_id: {kakao_id, name, category, lat, lng, address, place_url, phone, search_rank}}
+        """
+        all_spots: dict[str, dict] = {}
+        grid_cells = self.generate_grid_cells()
+        headers = {"Authorization": f"KakaoAK {self.api_key}"}
+        api_calls = 0
+
+        for cell in grid_cells:
+            rect_str = f"{cell['x1']},{cell['y1']},{cell['x2']},{cell['y2']}"
+
+            for code, category_name in CATEGORY_CODES.items():
+                for page in range(1, 4):  # 최대 3페이지
+                    params = {
+                        "category_group_code": code,
+                        "rect": rect_str,
+                        "page": page,
+                        "size": 15,
+                        "sort": "accuracy",
+                    }
+
+                    try:
+                        resp = httpx.get(
+                            KAKAO_CATEGORY_URL,
+                            headers=headers,
+                            params=params,
+                            timeout=10,
+                        )
+                        resp.raise_for_status()
+                        data = resp.json()
+                        api_calls += 1
+                    except Exception as e:
+                        logger.warning(
+                            "카카오맵 API 호출 실패 (cell=%s, code=%s, page=%d): %s",
+                            rect_str, code, page, e,
+                        )
+                        break
+
+                    for place in data.get("documents", []):
+                        place_id = place["id"]
+                        if place_id not in all_spots:
+                            all_spots[place_id] = {
+                                "kakao_id": place_id,
+                                "name": place["place_name"],
+                                "category": category_name,
+                                "lat": float(place["y"]),
+                                "lng": float(place["x"]),
+                                "address": place["address_name"],
+                                "place_url": place.get("place_url", ""),
+                                "phone": place.get("phone", ""),
+                                "search_rank": len(all_spots) + 1,
+                            }
+
+                    # 마지막 페이지면 다음 카테고리로
+                    if data.get("meta", {}).get("is_end", True):
+                        break
+
+        logger.info(
+            "카카오맵 스캔 완료: %d개 장소 수집 (API 호출 %d회)",
+            len(all_spots), api_calls,
+        )
+        return all_spots
+
+    # ------------------------------------------------------------------
+    # 리뷰 수 파싱 (비공식 API)
+    # ------------------------------------------------------------------
+
+    async def fetch_review_counts(
+        self, spots: dict[str, dict], limit: int = REVIEW_PARSE_LIMIT
+    ) -> dict[str, dict]:
+        """
+        상위 limit개 장소의 리뷰 수를 카카오맵 비공식 API로 파싱한다.
+
+        실패 시 review_count=0 처리 (graceful degradation).
+        """
+        # search_rank 기준 상위 N개만 파싱
+        sorted_spots = sorted(spots.values(), key=lambda s: s["search_rank"])[:limit]
+        results: dict[str, dict] = {}
+        success_count = 0
+        fail_count = 0
+
+        headers = {
+            "User-Agent": (
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/120.0.0.0 Safari/537.36"
+            ),
+            "Referer": "https://map.kakao.com/",
+        }
+
+        async with httpx.AsyncClient(headers=headers, timeout=10) as client:
+            for spot in sorted_spots:
+                place_id = spot["kakao_id"]
+                url = KAKAO_PLACE_DETAIL_URL.format(place_id=place_id)
+
+                try:
+                    resp = await client.get(url)
+                    data = resp.json()
+
+                    results[place_id] = {
+                        "review_count": data.get("comment", {}).get("scorecnt", 0),
+                        "avg_rating": data.get("comment", {}).get("scoretotalavgstar", 0),
+                        "blog_review_count": data.get("blogReview", {}).get("blogrvwcnt", 0),
+                        "success": True,
+                    }
+                    success_count += 1
+                except Exception as e:
+                    results[place_id] = {
+                        "review_count": 0,
+                        "avg_rating": 0,
+                        "blog_review_count": 0,
+                        "success": False,
+                        "error": str(e),
+                    }
+                    fail_count += 1
+
+        total = success_count + fail_count
+        fail_rate = fail_count / max(total, 1)
+        logger.info(
+            "리뷰 파싱 완료: 성공 %d, 실패 %d (실패율 %.1f%%)",
+            success_count, fail_count, fail_rate * 100,
+        )
+
+        if fail_rate > 0.5:
+            logger.warning(
+                "리뷰 파싱 실패율 50%% 초과 — 이전 데이터 유지 권장"
+            )
+
+        return results
+
+    # ------------------------------------------------------------------
+    # DB 저장
+    # ------------------------------------------------------------------
+
+    def _save_to_db(self, spots: dict[str, dict], reviews: dict[str, dict]) -> int:
+        """수집된 장소+리뷰를 trend_spots / spot_trends 테이블에 저장한다."""
+        today = date.today()
+        saved = 0
+
+        for place_id, spot in spots.items():
+            spot_id = f"trend_{spot['name']}_{place_id[:8]}"
+
+            # trend_spots upsert
+            try:
+                self.supabase.table("trend_spots").upsert(
+                    {
+                        "id": spot_id,
+                        "name": spot["name"],
+                        "category": spot["category"],
+                        "lat": spot["lat"],
+                        "lng": spot["lng"],
+                        "address": spot["address"],
+                        "source_ids": {"kakaomap": place_id},
+                    },
+                    on_conflict="id",
+                ).execute()
+            except Exception as e:
+                logger.warning("trend_spots upsert 실패 (%s): %s", spot_id, e)
+                continue
+
+            # spot_trends — search_rank 메트릭
+            try:
+                self.supabase.table("spot_trends").insert({
+                    "spot_id": spot_id,
+                    "source": "kakaomap",
+                    "metric_type": "search_rank",
+                    "metric_value": spot["search_rank"],
+                    "period_start": today.isoformat(),
+                    "period_end": today.isoformat(),
+                    "raw_data": {"kakao_id": place_id, "address": spot["address"]},
+                }).execute()
+            except Exception as e:
+                logger.warning("spot_trends insert 실패 (search_rank, %s): %s", spot_id, e)
+
+            # spot_trends — review_count 메트릭 (파싱 성공 시)
+            review = reviews.get(place_id, {})
+            if review.get("success"):
+                try:
+                    self.supabase.table("spot_trends").insert({
+                        "spot_id": spot_id,
+                        "source": "kakaomap",
+                        "metric_type": "review_count",
+                        "metric_value": review["review_count"],
+                        "period_start": today.isoformat(),
+                        "period_end": today.isoformat(),
+                        "raw_data": {
+                            "avg_rating": review["avg_rating"],
+                            "blog_review_count": review["blog_review_count"],
+                        },
+                    }).execute()
+                except Exception as e:
+                    logger.warning("spot_trends insert 실패 (review_count, %s): %s", spot_id, e)
+
+            saved += 1
+
+        logger.info("DB 저장 완료: %d건", saved)
+        return saved
+
+    # ------------------------------------------------------------------
+    # 메인 실행
+    # ------------------------------------------------------------------
+
+    async def run(self) -> dict:
+        """
+        카카오맵 수집 파이프라인 전체 실행.
+
+        Returns:
+            {"spots_count": int, "reviews_parsed": int, "saved": int}
+        """
+        logger.info("=== 카카오맵 수집 시작 ===")
+
+        # 1단계: 그리드 스캔으로 장소 수집
+        spots = self.collect_spots()
+
+        # 2단계: 상위 장소 리뷰 수 파싱
+        reviews = await self.fetch_review_counts(spots)
+
+        # 3단계: DB 저장
+        saved = self._save_to_db(spots, reviews)
+
+        result = {
+            "spots_count": len(spots),
+            "reviews_parsed": len(reviews),
+            "saved": saved,
+        }
+        logger.info("=== 카카오맵 수집 완료: %s ===", result)
+        return result

trend_engine/collectors/naver_blog.py

@@ -0,0 +1,330 @@
+"""
+네이버 블로그 2단계 수집기 (Naver Blog 2-Stage Collector)
+
+1단계: 네이버 검색 API로 블로그 URL 목록 확보
+2단계: 모바일 URL 크롤링으로 본문 전체 수집
+
+수집된 본문은 spot_trends 테이블에 저장되며,
+장소명 추출은 별도 PlaceNameExtractor가 담당한다.
+"""
+
+import logging
+import os
+import time
+import threading
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import date, datetime, timedelta
+
+import requests
+from bs4 import BeautifulSoup
+
+logger = logging.getLogger(__name__)
+
+# 검색 키워드 목록 (애월리 집중)
+NAVER_BLOG_KEYWORDS = [
+    "애월 카페", "애월 맛집", "애월 가볼만한곳", "애월 산책",
+    "애월 해안", "애월 뷰", "애월 디저트", "애월 브런치",
+    "애월 감성", "애월 숨은 명소", "애월 오션뷰",
+    "애월리 카페", "애월리 맛집",
+    "한담 해안 산책로", "곽지 해변",
+]
+
+# 모바일 User-Agent
+MOBILE_USER_AGENT = (
+    "Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) "
+    "AppleWebKit/605.1.15 (KHTML, like Gecko) "
+    "Version/16.0 Mobile/15E148 Safari/604.1"
+)
+
+
+class NaverBlogCollector:
+    """네이버 블로그 2단계 수집기."""
+
+    def __init__(self, supabase_client):
+        self.supabase = supabase_client
+        self.client_id = os.environ["NAVER_CLIENT_ID"]
+        self.client_secret = os.environ["NAVER_CLIENT_SECRET"]
+        self.keywords = NAVER_BLOG_KEYWORDS
+        self._stats = {
+            "api_calls": 0,
+            "urls_found": 0,
+            "urls_unique": 0,
+            "crawl_success": 0,
+            "crawl_fail": 0,
+            "saved": 0,
+        }
+
+    # ── 1단계: 네이버 검색 API로 URL 목록 확보 ────────────────────
+
+    def collect_blog_urls(self) -> list[dict]:
+        """네이버 검색 API로 블로그 포스트 URL 목록 확보.
+
+        키워드당 최대 1,100건(start 1~1000, display 100) 조회 가능하나,
+        최근 1주 포스팅 기준으로 실제 100건 미만이 대부분이다.
+
+        Returns:
+            URL 기준 중복 제거된 포스트 목록
+        """
+        all_posts: list[dict] = []
+        headers = {
+            "X-Naver-Client-Id": self.client_id,
+            "X-Naver-Client-Secret": self.client_secret,
+        }
+
+        for keyword in self.keywords:
+            keyword_count = 0
+            for start in range(1, 1001, 100):
+                params = {
+                    "query": keyword,
+                    "display": 100,
+                    "start": start,
+                    "sort": "date",
+                }
+                try:
+                    resp = requests.get(
+                        "https://openapi.naver.com/v1/search/blog",
+                        headers=headers,
+                        params=params,
+                        timeout=10,
+                    )
+                    resp.raise_for_status()
+                    data = resp.json()
+                    self._stats["api_calls"] += 1
+                except requests.RequestException as e:
+                    logger.warning("검색 API 실패 [%s start=%d]: %s", keyword, start, e)
+                    break
+
+                items = data.get("items", [])
+                for item in items:
+                    all_posts.append({
+                        "title": item["title"],
+                        "link": item["link"],
+                        "bloggername": item["bloggername"],
+                        "postdate": item["postdate"],  # YYYYMMDD
+                        "keyword": keyword,
+                    })
+                    keyword_count += 1
+
+                # 결과가 100건 미만이면 더 이상 데이터 없음
+                if len(items) < 100:
+                    break
+
+                time.sleep(0.1)  # API 부하 방지
+
+            logger.info("키워드 [%s]: %d건 수집", keyword, keyword_count)
+
+        self._stats["urls_found"] = len(all_posts)
+
+        # URL 기준 중복 제거
+        seen_urls: set[str] = set()
+        unique_posts: list[dict] = []
+        for post in all_posts:
+            if post["link"] not in seen_urls:
+                seen_urls.add(post["link"])
+                unique_posts.append(post)
+
+        self._stats["urls_unique"] = len(unique_posts)
+        logger.info(
+            "1단계 완료: 총 %d건 → 중복 제거 후 %d건 (API 호출 %d회)",
+            self._stats["urls_found"],
+            self._stats["urls_unique"],
+            self._stats["api_calls"],
+        )
+        return unique_posts
+
+    # ── 2단계: 모바일 크롤링으로 본문 전체 수집 ──────────────────
+
+    def _convert_to_mobile_url(self, url: str) -> str:
+        """PC 블로그 URL을 ��바일 URL로 변환."""
+        url = url.replace("https://blog.naver.com", "https://m.blog.naver.com")
+        url = url.replace("http://blog.naver.com", "https://m.blog.naver.com")
+        return url
+
+    def _crawl_single_blog(self, url: str) -> str | None:
+        """단일 블로그 모바일 페이지에서 본문 텍스트 추출."""
+        m_url = self._convert_to_mobile_url(url)
+        headers = {"User-Agent": MOBILE_USER_AGENT}
+
+        try:
+            resp = requests.get(m_url, headers=headers, timeout=10)
+            resp.raise_for_status()
+        except requests.RequestException as e:
+            logger.debug("크롤링 HTTP 실패: %s — %s", m_url, e)
+            return None
+
+        soup = BeautifulSoup(resp.text, "lxml")
+
+        # 셀렉터 우선순위: 스마트에디터3 > 구형 에디터 > 기타
+        content_div = (
+            soup.find("div", {"class": "se-main-container"})
+            or soup.find("div", {"id": "postViewArea"})
+            or soup.find("div", {"class": "post_ct"})
+        )
+
+        if content_div:
+            return content_div.get_text(separator=" ", strip=True)
+
+        return None
+
+    def crawl_blog_contents(
+        self, posts: list[dict], max_workers: int = 5
+    ) -> list[dict]:
+        """블로그 URL 목록을 받아 본문 병렬 크롤링.
+
+        ThreadPoolExecutor로 max_workers개 동시 요청.
+        네이버 서버 부하 방지를 위해 워커 간 0.2초 간격 유지.
+
+        Args:
+            posts: collect_blog_urls()의 반환값
+            max_workers: 동시 크롤링 워커 수 (기본 5)
+
+        Returns:
+            full_content 필드가 추가된 포스트 목록 (크롤링 성공분만)
+        """
+        results: list[dict] = []
+        total = len(posts)
+        lock = threading.Lock()
+
+        def crawl_one(post: dict) -> tuple[dict, str | None]:
+            content = self._crawl_single_blog(post["link"])
+            return post, content
+
+        logger.info(
+            "병렬 크롤링 시작: %d건 (워커 %d개)", total, max_workers
+        )
+
+        with ThreadPoolExecutor(max_workers=max_workers) as executor:
+            # 워커 간 0.2초 간격으로 submit (초기 burst 방지)
+            futures = []
+            for i, post in enumerate(posts):
+                futures.append(executor.submit(crawl_one, post))
+                if (i + 1) % max_workers == 0:
+                    time.sleep(0.2)
+
+            done_count = 0
+            for future in as_completed(futures):
+                try:
+                    post, content = future.result()
+                except Exception as e:
+                    logger.debug("크롤링 예외: %s", e)
+                    with lock:
+                        self._stats["crawl_fail"] += 1
+                        done_count += 1
+                    continue
+
+                with lock:
+                    if content:
+                        post["full_content"] = content
+                        results.append(post)
+                        self._stats["crawl_success"] += 1
+                    else:
+                        self._stats["crawl_fail"] += 1
+
+                    done_count += 1
+                    if done_count % 500 == 0 or done_count == total:
+                        logger.info(
+                            "크롤링 진행: %d/%d (성공: %d, 실패: %d)",
+                            done_count, total,
+                            self._stats["crawl_success"],
+                            self._stats["crawl_fail"],
+                        )
+
+        logger.info(
+            "2단계 완료: %d건 크롤링 → 성공 %d건, 실패 %d건",
+            total,
+            self._stats["crawl_success"],
+            self._stats["crawl_fail"],
+        )
+        return results
+
+    # ── DB 저장 ────────────────────────────────────────────────
+
+    def save_to_db(self, posts: list[dict]) -> int:
+        """크롤링 결과를 spot_trends 테이블에 저장.
+
+        source='naver_blog', metric_type='blog_post'로 원본 데이터 보관.
+        장소명 추출 및 mention_count 집계는 별도 파이프라인에서 처리한다.
+
+        Args:
+            posts: crawl_blog_contents()의 반환값 (full_content 포함)
+
+        Returns:
+            저장된 레코드 수
+        """
+        today = date.today()
+        period_start = today - timedelta(days=7)
+        saved_count = 0
+
+        # 배치 insert를 위한 rows 수집
+        rows: list[dict] = []
+        for post in posts:
+            rows.append({
+                "spot_id": "__pending__",  # 장소명 추출 전이므로 placeholder
+                "source": "naver_blog",
+                "metric_type": "blog_post",
+                "metric_value": 1,
+                "period_start": period_start.isoformat(),
+                "period_end": today.isoformat(),
+                "raw_data": {
+                    "title": post["title"],
+                    "link": post["link"],
+                    "bloggername": post["bloggername"],
+                    "postdate": post["postdate"],
+                    "keyword": post["keyword"],
+                    "content_length": len(post.get("full_content", "")),
+                    "content_preview": post.get("full_content", "")[:1500],
+                },
+            })
+
+        # Supabase bulk insert (1000건씩 배치)
+        batch_size = 1000
+        for i in range(0, len(rows), batch_size):
+            batch = rows[i:i + batch_size]
+            try:
+                result = self.supabase.table("spot_trends").insert(batch).execute()
+                saved_count += len(result.data) if result.data else 0
+            except Exception as e:
+                logger.error("spot_trends 저장 실패 (batch %d): %s", i // batch_size, e)
+
+        self._stats["saved"] = saved_count
+        logger.info("DB 저장 완료: %d건", saved_count)
+        return saved_count
+
+    # ── 전체 파이프라인 실행 ───────────────────────────────────
+
+    def run(self) -> dict:
+        """전체 수집 파이프라인 실행.
+
+        1단계: 검색 API로 URL 확보
+        2단계: 모바일 크롤링으로 본문 수집
+        3단계: DB 저장 (원본 보관)
+
+        Returns:
+            수집 통계 dict
+        """
+        logger.info("=== 네이버 블로그 수집 시작 ===")
+        start_time = datetime.now()
+
+        # 1단계
+        posts = self.collect_blog_urls()
+        if not posts:
+            logger.warning("검색 결과 없음 — 수집 종료")
+            return {**self._stats, "duration_seconds": 0}
+
+        # 2단계
+        crawled = self.crawl_blog_contents(posts)
+        if not crawled:
+            logger.warning("크롤링 결과 없음 — 수집 종료")
+            return {**self._stats, "duration_seconds": 0}
+
+        # 3단계
+        self.save_to_db(crawled)
+
+        duration = (datetime.now() - start_time).total_seconds()
+        self._stats["duration_seconds"] = duration
+        logger.info(
+            "=== 네이버 블로그 수집 완료 (%.1f초) === %s",
+            duration,
+            self._stats,
+        )
+        return self._stats

trend_engine/collectors/naver_place.py

@@ -0,0 +1,297 @@
+"""
+네이버 플레이스 보조 수집기 (Naver Place Auxiliary Collector)
+
+보조 채널 (가중치 5%): 카카오맵 스캔 결과 기반으로 네이버 플레이스 리뷰 수만 추가 수집.
+비공식 API 의존 — Graceful Degradation 설계.
+
+알려진 제약 (2026-02):
+- 네이버 지역 검색 API link 필드: 외부 URL만 반환 (Place ID 미포함)
+- 네이버 맵 내부 검색 API: 봇 트래픽 CAPTCHA 차단
+- 결과: Place ID 매칭 불가 → 가중치 재분배 (trend_scorer에서 None 처리)
+
+- 네이버 지역 검색 API: 장소명→플레이스 ID 매칭 (display 최대 5건)
+- 비공식 API: https://map.naver.com/p/api/place/summary/{place_id} → 리뷰 수
+- 실패율 50% 이상 시 해당 주기 건너뛰기
+"""
+
+import logging
+import os
+import re
+from datetime import date, datetime, timedelta
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+
+def _is_similar_address(naver_addr: str, kakao_addr: str) -> bool:
+    """네이버/카카오 주소 유사도 비교.
+
+    정확한 주소 일치가 아니라 핵심 구성 요소(읍/면/동, 번지/도로명)가
+    겹치는지 확인한다.
+    """
+    if not naver_addr or not kakao_addr:
+        return False
+
+    # HTML 태그 제거 (네이버 검색 결과에 <b> 태그가 포함될 수 있음)
+    naver_clean = re.sub(r"<[^>]+>", "", naver_addr).strip()
+    kakao_clean = kakao_addr.strip()
+
+    # 공백/특수문자 정규화
+    naver_tokens = set(re.findall(r"[\w가-힣]+", naver_clean))
+    kakao_tokens = set(re.findall(r"[\w가-힣]+", kakao_clean))
+
+    # 핵심 토큰(읍면동 이하) 겹침 비율 확인
+    overlap = naver_tokens & kakao_tokens
+    if not kakao_tokens:
+        return False
+
+    return len(overlap) / len(kakao_tokens) >= 0.4
+
+
+class NaverPlaceCollector:
+    """네이버 플레이스 보조 수집기."""
+
+    def __init__(self, supabase_client):
+        self.supabase = supabase_client
+        self.client_id = os.environ["NAVER_CLIENT_ID"]
+        self.client_secret = os.environ["NAVER_CLIENT_SECRET"]
+        self._stats = {
+            "total_spots": 0,
+            "matched": 0,
+            "match_failed": 0,
+            "review_success": 0,
+            "review_failed": 0,
+            "saved": 0,
+            "skipped_high_failure": False,
+        }
+
+    # ── 네이버 플레이스 ID 매칭 ──────────────────────────────
+
+    async def match_naver_place_id(self, name: str, address: str) -> str | None:
+        """카카오맵 장소명 → 네이버 플레이스 ID 매칭.
+
+        네이버 지역 검색 API로 장소명 검색 후 주소 비교로 동일 장소 판별.
+        display 최대 5건만 반환되므로 정확한 매칭이 중요하다.
+
+        Args:
+            name: 카카오맵 장소명
+            address: 카카오맵 주소
+
+        Returns:
+            네이버 플레이스 ID (매칭 실패 시 None)
+        """
+        headers = {
+            "X-Naver-Client-Id": self.client_id,
+            "X-Naver-Client-Secret": self.client_secret,
+        }
+        params = {
+            "query": f"애월 {name}",
+            "display": 5,
+            "start": 1,
+            "sort": "comment",
+        }
+
+        try:
+            async with httpx.AsyncClient() as client:
+                resp = await client.get(
+                    "https://openapi.naver.com/v1/search/local",
+                    headers=headers,
+                    params=params,
+                    timeout=10,
+                )
+                resp.raise_for_status()
+                data = resp.json()
+        except httpx.HTTPError as e:
+            logger.debug("지역 검색 API 실패 [%s]: %s", name, e)
+            return None
+
+        for item in data.get("items", []):
+            item_addr = item.get("address", "")
+            if _is_similar_address(item_addr, address):
+                # link에서 플레이스 ID 추출
+                link = item.get("link", "")
+                match = re.search(r"/(\d{5,})/?", link)
+                if match:
+                    return match.group(1)
+
+        return None
+
+    # ── 비공식 API로 리뷰 수 조회 ────────────────────────────
+
+    async def fetch_review_count(self, place_id: str) -> dict:
+        """네이버 플레이스 내부 API로 리뷰 수 조회.
+
+        비공식 API — 언제든 변경/차단될 수 있다.
+        실패 시 해당 장소만 None 처리, 다른 채널에 영향 없음.
+
+        Args:
+            place_id: 네이버 플레이스 ID
+
+        Returns:
+            리뷰 수 데이터 dict (success=False인 경우 0값)
+        """
+        api_url = f"https://map.naver.com/p/api/place/summary/{place_id}"
+        headers = {
+            "User-Agent": (
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/120.0.0.0 Safari/537.36"
+            ),
+            "Referer": "https://map.naver.com/",
+        }
+
+        try:
+            async with httpx.AsyncClient() as client:
+                resp = await client.get(api_url, headers=headers, timeout=10)
+                resp.raise_for_status()
+                data = resp.json()
+
+            return {
+                "visitor_review_count": data.get("visitorReviewCount", 0),
+                "blog_review_count": data.get("blogReviewCount", 0),
+                "avg_rating": data.get("visitorReviewScore", 0),
+                "success": True,
+            }
+        except Exception as e:
+            logger.debug("플레이스 리뷰 조회 실패 [%s]: %s", place_id, e)
+            return {
+                "visitor_review_count": 0,
+                "blog_review_count": 0,
+                "avg_rating": 0,
+                "success": False,
+                "error": str(e),
+            }
+
+    # ── DB 저장 ────────────────────────────────────────────────
+
+    def _save_results(self, results: list[dict]) -> int:
+        """수집 결과를 spot_trends 테이블에 저장."""
+        today = date.today()
+        period_start = today - timedelta(days=7)
+        saved_count = 0
+
+        rows: list[dict] = []
+        for r in results:
+            if not r.get("review_data") or not r["review_data"].get("success"):
+                continue
+            rd = r["review_data"]
+            rows.append({
+                "spot_id": r.get("spot_id", r.get("kakao_id", "__unknown__")),
+                "source": "naver_place",
+                "metric_type": "review_count",
+                "metric_value": rd["visitor_review_count"],
+                "period_start": period_start.isoformat(),
+                "period_end": today.isoformat(),
+                "raw_data": {
+                    "name": r.get("name"),
+                    "naver_place_id": r.get("naver_place_id"),
+                    "visitor_review_count": rd["visitor_review_count"],
+                    "blog_review_count": rd["blog_review_count"],
+                    "avg_rating": rd["avg_rating"],
+                },
+            })
+
+        if not rows:
+            return 0
+
+        batch_size = 500
+        for i in range(0, len(rows), batch_size):
+            batch = rows[i:i + batch_size]
+            try:
+                result = self.supabase.table("spot_trends").insert(batch).execute()
+                saved_count += len(result.data) if result.data else 0
+            except Exception as e:
+                logger.error("spot_trends 저장 실패 (naver_place batch %d): %s", i // batch_size, e)
+
+        self._stats["saved"] = saved_count
+        return saved_count
+
+    # ── 전체 파이프라인 실행 ───────────────────────────────────
+
+    async def run(self, kakaomap_spots: list[dict]) -> dict:
+        """네이버 플레이스 보조 수집 실행.
+
+        카카오맵 스캔 결과(상위 100개 이하) 기반으로:
+        1. 장소명 → 네이버 플레이스 ID 매칭
+        2. 비공식 API로 리뷰 수 조회
+        3. 실패율 50% 이상 시 해당 주기 건너뛰기
+
+        Args:
+            kakaomap_spots: 카카오맵 수집기에서 전달받은 장소 목록
+                [{"kakao_id": "...", "name": "...", "address": "...", ...}, ...]
+
+        Returns:
+            수집 통계 dict
+        """
+        logger.info("=== 네이버 플레이스 보조 수집 시작 ===")
+        start_time = datetime.now()
+
+        spots = kakaomap_spots[:100]  # 상위 100개만 대상
+        self._stats["total_spots"] = len(spots)
+
+        results: list[dict] = []
+
+        # 1단계: 네이버 플레이스 ID 매칭
+        for spot in spots:
+            name = spot.get("name", "")
+            address = spot.get("address", "")
+
+            place_id = await self.match_naver_place_id(name, address)
+            if place_id:
+                self._stats["matched"] += 1
+                spot["naver_place_id"] = place_id
+            else:
+                self._stats["match_failed"] += 1
+                spot["naver_place_id"] = None
+
+        # 매칭 성공한 장소만 리뷰 조회
+        matched_spots = [s for s in spots if s.get("naver_place_id")]
+        logger.info(
+            "ID 매칭: %d/%d 성공",
+            self._stats["matched"],
+            self._stats["total_spots"],
+        )
+
+        if not matched_spots:
+            logger.warning("매칭된 장소 없음 — 수집 종료")
+            duration = (datetime.now() - start_time).total_seconds()
+            self._stats["duration_seconds"] = duration
+            return self._stats
+
+        # 2단계: 리뷰 수 조회 (실패율 모니터링)
+        for spot in matched_spots:
+            review_data = await self.fetch_review_count(spot["naver_place_id"])
+            spot["review_data"] = review_data
+
+            if review_data["success"]:
+                self._stats["review_success"] += 1
+            else:
+                self._stats["review_failed"] += 1
+
+            results.append(spot)
+
+            # 실패율 50% 이상 시 조기 중단
+            total_attempts = self._stats["review_success"] + self._stats["review_failed"]
+            if total_attempts >= 10:
+                failure_rate = self._stats["review_failed"] / total_attempts
+                if failure_rate >= 0.5:
+                    logger.warning(
+                        "리뷰 API 실패율 %.0f%% — 해당 주기 건너뛰기",
+                        failure_rate * 100,
+                    )
+                    self._stats["skipped_high_failure"] = True
+                    break
+
+        # 3단계: DB 저장
+        if not self._stats["skipped_high_failure"]:
+            self._save_results(results)
+
+        duration = (datetime.now() - start_time).total_seconds()
+        self._stats["duration_seconds"] = duration
+        logger.info(
+            "=== 네이버 플레이스 보조 수집 완료 (%.1f초) === %s",
+            duration,
+            self._stats,
+        )
+        return self._stats

trend_engine/collectors/youtube.py

@@ -0,0 +1,317 @@
+"""
+YouTube Collector — YouTube Data API v3 키워드 검색 + 위치 기반 검색
+
+수집 방식:
+  1. 6개 키워드로 최근 1주 영상 검색 (search.list)
+  2. 영상 상세 정보 조회 (videos.list — 조회수, 좋아요)
+  3. 위치 기반 보조 검색 (애월 중심 10km)
+  4. 영상 제목+설명에서 장소명 추출 (PlaceNameExtractor 연동)
+"""
+
+import os
+import logging
+from datetime import date, datetime, timedelta, timezone
+
+from googleapiclient.discovery import build
+
+from trend_engine.place_extractor import PlaceNameExtractor
+
+logger = logging.getLogger(__name__)
+
+# ──────────────────────────────────────────────────
+# 설정
+# ──────────────────────────────────────────────────
+
+YOUTUBE_KEYWORDS = [
+    "애월 여행",
+    "제주 애월 카페",
+    "애월 브이로그",
+    "애월 해안 산책",
+    "애월 맛집 추천",
+    "한담 해안",
+]
+
+# 애월 중심 좌표
+AEWOL_CENTER = {"lat": "33.46", "lng": "126.31"}
+LOCATION_RADIUS = "10km"
+
+MAX_RESULTS_PER_KEYWORD = 20
+MAX_RESULTS_LOCATION = 30
+
+
+class YouTubeCollector:
+    """YouTube Data API v3 기반 트렌드 영상 수집기."""
+
+    def __init__(self, supabase_client, spot_matcher=None):
+        self.supabase = supabase_client
+        api_key = os.environ["AEWOL_AI_SYSTEM"]
+        self.youtube = build("youtube", "v3", developerKey=api_key)
+        self.extractor = PlaceNameExtractor(supabase_client)
+        self.spot_matcher = spot_matcher
+
+    # ------------------------------------------------------------------
+    # 키워드 검색
+    # ------------------------------------------------------------------
+
+    def collect_keyword_videos(self) -> list[dict]:
+        """6개 키워드로 최근 1주 영상을 검색한다."""
+        one_week_ago = (
+            datetime.now(timezone.utc) - timedelta(days=7)
+        ).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+        all_video_ids: list[str] = []
+        keyword_map: dict[str, list[str]] = {}  # video_id → keywords
+
+        for keyword in YOUTUBE_KEYWORDS:
+            try:
+                search_resp = (
+                    self.youtube.search()
+                    .list(
+                        q=keyword,
+                        type="video",
+                        part="id,snippet",
+                        order="date",
+                        publishedAfter=one_week_ago,
+                        maxResults=MAX_RESULTS_PER_KEYWORD,
+                    )
+                    .execute()
+                )
+            except Exception as e:
+                logger.warning("YouTube 검색 실패 (keyword=%s): %s", keyword, e)
+                continue
+
+            for item in search_resp.get("items", []):
+                vid = item["id"]["videoId"]
+                if vid not in keyword_map:
+                    keyword_map[vid] = []
+                    all_video_ids.append(vid)
+                keyword_map[vid].append(keyword)
+
+        logger.info(
+            "키워드 검색 완료: %d개 고유 영상 발견 (%d개 키워드)",
+            len(all_video_ids), len(YOUTUBE_KEYWORDS),
+        )
+
+        # 영상 상세 정보 조회 (50개씩 배치)
+        videos = self._fetch_video_details(all_video_ids, keyword_map)
+        return videos
+
+    # ------------------------------------------------------------------
+    # 위치 기반 검색
+    # ------------------------------------------------------------------
+
+    def collect_location_videos(self) -> list[dict]:
+        """애월 중심 10km 반경 내 최근 1주 영상을 검색한다."""
+        one_week_ago = (
+            datetime.now(timezone.utc) - timedelta(days=7)
+        ).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+        try:
+            search_resp = (
+                self.youtube.search()
+                .list(
+                    part="id,snippet",
+                    type="video",
+                    location=f"{AEWOL_CENTER['lat']},{AEWOL_CENTER['lng']}",
+                    locationRadius=LOCATION_RADIUS,
+                    order="date",
+                    publishedAfter=one_week_ago,
+                    maxResults=MAX_RESULTS_LOCATION,
+                )
+                .execute()
+            )
+        except Exception as e:
+            logger.warning("YouTube 위치 검색 실패: %s", e)
+            return []
+
+        video_ids = [item["id"]["videoId"] for item in search_resp.get("items", [])]
+        logger.info("위치 검색 완료: %d개 영상 발견", len(video_ids))
+
+        keyword_map = {vid: ["location_search"] for vid in video_ids}
+        return self._fetch_video_details(video_ids, keyword_map)
+
+    # ------------------------------------------------------------------
+    # 영상 상세 조회
+    # ------------------------------------------------------------------
+
+    def _fetch_video_details(
+        self,
+        video_ids: list[str],
+        keyword_map: dict[str, list[str]],
+    ) -> list[dict]:
+        """videos.list API로 조회수/좋아요 등 상세 정보를 가져온다."""
+        videos: list[dict] = []
+
+        # 50개씩 배치 처리
+        for i in range(0, len(video_ids), 50):
+            batch = video_ids[i : i + 50]
+            try:
+                resp = (
+                    self.youtube.videos()
+                    .list(
+                        part="statistics,snippet",
+                        id=",".join(batch),
+                    )
+                    .execute()
+                )
+            except Exception as e:
+                logger.warning("YouTube videos.list 실패: %s", e)
+                continue
+
+            for video in resp.get("items", []):
+                stats = video.get("statistics", {})
+                videos.append({
+                    "video_id": video["id"],
+                    "title": video["snippet"]["title"],
+                    "description": video["snippet"].get("description", ""),
+                    "channel_title": video["snippet"].get("channelTitle", ""),
+                    "view_count": int(stats.get("viewCount", 0)),
+                    "like_count": int(stats.get("likeCount", 0)),
+                    "comment_count": int(stats.get("commentCount", 0)),
+                    "published_at": video["snippet"]["publishedAt"],
+                    "keywords": keyword_map.get(video["id"], []),
+                })
+
+        return videos
+
+    # ------------------------------------------------------------------
+    # 장소명 추출 + 메트릭 집계
+    # ------------------------------------------------------------------
+
+    def _extract_place_mentions(self, videos: list[dict]) -> dict[str, dict]:
+        """
+        영상 제목+설명에서 장소명을 추출하고 메트릭을 집계한다.
+
+        Returns:
+            {spot_id_or_name: {name, spot_id, mention_video_count, total_views, total_likes}}
+        """
+        place_metrics: dict[str, dict] = {}
+
+        for video in videos:
+            text = video["title"] + " " + video["description"]
+            places = self.extractor.extract(text)
+
+            for place in places:
+                key = place["spot_id"] or place["name"]
+                if key not in place_metrics:
+                    place_metrics[key] = {
+                        "name": place["name"],
+                        "spot_id": place["spot_id"],
+                        "method": place["method"],
+                        "mention_video_count": 0,
+                        "total_views": 0,
+                        "total_likes": 0,
+                    }
+                place_metrics[key]["mention_video_count"] += 1
+                place_metrics[key]["total_views"] += video["view_count"]
+                place_metrics[key]["total_likes"] += video["like_count"]
+
+        logger.info("장소명 추출 완료: %d개 장소 식별", len(place_metrics))
+        return place_metrics
+
+    # ------------------------------------------------------------------
+    # DB 저장
+    # ------------------------------------------------------------------
+
+    def _save_to_db(self, place_metrics: dict[str, dict], videos: list[dict]) -> int:
+        """추출된 메트릭을 spot_trends 테이블에 저장한다.
+
+        SpotMatcher가 설정된 경우 trend_spots.id로 정규화.
+        매칭 실패한 장소는 스킵.
+        """
+        today = date.today()
+        saved = 0
+        skipped = 0
+
+        for key, metrics in place_metrics.items():
+            # SpotMatcher로 spot_id 정규화
+            if self.spot_matcher:
+                spot_id = self.spot_matcher.match(metrics["name"])
+                if not spot_id:
+                    skipped += 1
+                    continue
+            else:
+                spot_id = metrics["spot_id"] or key
+
+            # view_count 메트릭
+            try:
+                self.supabase.table("spot_trends").insert({
+                    "spot_id": spot_id,
+                    "source": "youtube",
+                    "metric_type": "view_count",
+                    "metric_value": metrics["total_views"],
+                    "period_start": today.isoformat(),
+                    "period_end": today.isoformat(),
+                    "raw_data": {
+                        "mention_video_count": metrics["mention_video_count"],
+                        "total_likes": metrics["total_likes"],
+                        "name": metrics["name"],
+                        "method": metrics["method"],
+                    },
+                }).execute()
+                saved += 1
+            except Exception as e:
+                logger.warning("spot_trends insert 실패 (youtube view_count, %s): %s", spot_id, e)
+
+            # video_count 메트릭 (채널 스코어링에 필요)
+            try:
+                self.supabase.table("spot_trends").insert({
+                    "spot_id": spot_id,
+                    "source": "youtube",
+                    "metric_type": "video_count",
+                    "metric_value": metrics["mention_video_count"],
+                    "period_start": today.isoformat(),
+                    "period_end": today.isoformat(),
+                    "raw_data": {"name": metrics["name"]},
+                }).execute()
+            except Exception as e:
+                logger.warning("spot_trends insert 실패 (youtube video_count, %s): %s", spot_id, e)
+
+        if skipped:
+            logger.info("YouTube 장소 매칭 실패로 %d건 스킵", skipped)
+        logger.info("YouTube DB 저장 완료: %d건", saved)
+        return saved
+
+    # ------------------------------------------------------------------
+    # 메인 실행
+    # ------------------------------------------------------------------
+
+    def run(self) -> dict:
+        """
+        YouTube 수집 파이프라인 전체 실행.
+
+        Returns:
+            {"keyword_videos": int, "location_videos": int, "places_found": int, "saved": int}
+        """
+        logger.info("=== YouTube 수집 시작 ===")
+
+        # 1단계: 키워드 검색
+        keyword_videos = self.collect_keyword_videos()
+
+        # 2단계: 위치 기반 보조 검색
+        location_videos = self.collect_location_videos()
+
+        # 중복 제거 (video_id 기준)
+        seen_ids = {v["video_id"] for v in keyword_videos}
+        for v in location_videos:
+            if v["video_id"] not in seen_ids:
+                keyword_videos.append(v)
+                seen_ids.add(v["video_id"])
+
+        all_videos = keyword_videos
+
+        # 3단계: 장소명 추출 + 메트릭 집계
+        place_metrics = self._extract_place_mentions(all_videos)
+
+        # 4단계: DB 저장
+        saved = self._save_to_db(place_metrics, all_videos)
+
+        result = {
+            "keyword_videos": len(keyword_videos),
+            "location_videos": len(location_videos),
+            "total_unique_videos": len(all_videos),
+            "places_found": len(place_metrics),
+            "saved": saved,
+        }
+        logger.info("=== YouTube 수집 완료: %s ===", result)
+        return result

trend_engine/place_extractor.py

@@ -0,0 +1,152 @@
+"""
+PlaceNameExtractor — 블로그/유튜브 텍스트에서 장소명을 추출하는 공통 모듈
+
+추출 우선순위:
+  1. 장소명 사전 매칭 (trend_spots + story_spots 이름)
+  2. 접미사 패턴 매칭 (카페, 식당, 해변 등)
+"""
+
+import os
+import re
+import logging
+
+from supabase import create_client, Client
+
+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,}(?:베이커리|브런치|디저트)"),
+]
+
+# 너무 일반적인 단어 — 장소명이 아닌 경우 필터링
+STOPWORDS = frozenset({
+    # 지시대명사 + 카테고리
+    "이카페", "그카페", "저카페", "이식당", "그식당",
+    "이해변", "그해변", "이오름", "그오름",
+    # 형용사 + 카테고리
+    "좋은카페", "예쁜카페", "맛있는식당", "좋은식당",
+    # 위치/수식어 + 카테고리
+    "근처카페", "주변카페", "동네카페", "유명카페",
+    "감성카페", "뷰카페", "오션뷰카페",
+    "애월카페", "한림카페", "협재카페",
+    "제주카페", "제주식당", "제주해변", "제주오름",
+    "예쁜식당", "제주맛집", "애월맛집",
+    # 카테고리 일반명사 (단독 사용 시 장소명 아님)
+    "카페", "커피", "식당", "맛집", "밥집", "국수",
+    "해변", "해안", "포구", "오름", "공원", "숲",
+    "전망대", "뷰", "베이커리", "브런치", "디저트",
+    "산책", "산책로", "드라이브",
+    # 지역 + 일반 표현
+    "제주여행", "애월여행", "제주도", "애월", "애월리",
+})
+
+
+def _get_supabase_client() -> Client:
+    url = os.environ.get("SUPABASE_URL") or os.environ.get("VITE_SUPABASE_URL")
+    key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY") or os.environ.get("SUPABASE_SERVICE_KEY")
+    if not url or not key:
+        raise ValueError("SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY must be set")
+    return create_client(url, key)
+
+
+class PlaceNameExtractor:
+    """텍스트에서 장소명을 추출하는 파이프라인."""
+
+    def __init__(self, supabase: Client | None = None):
+        self.supabase = supabase or _get_supabase_client()
+        # {장소명: 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 []:
+                self._register_name(row["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 []:
+                self._register_name(row["name"], row["id"])
+                # aliases 배열 등록
+                for alias in row.get("aliases") or []:
+                    if 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:
+            return
+        self.known_places[name] = spot_id
+        # 공백 제거 변형 ("봄날의 카페" → "봄날의카페")
+        no_space = name.replace(" ", "")
+        if no_space != name:
+            self.known_places[no_space] = spot_id
+
+    # ------------------------------------------------------------------
+    # 추출
+    # ------------------------------------------------------------------
+
+    def extract(self, text: str) -> list[dict]:
+        """
+        텍스트에서 장소명을 추출한다.
+
+        Returns:
+            [{"name": str, "spot_id": str|None, "method": "dictionary"|"pattern"}, ...]
+        """
+        if not text:
+            return []
+
+        found: list[dict] = []
+        found_names: set[str] = set()
+
+        # 1순위: 사전 매칭 — 긴 이름부터 매칭 (부분 매칭 방지)
+        for name in sorted(self.known_places, key=len, reverse=True):
+            # 2글자 이하 일반명사 필터 + STOPWORDS 체크
+            if len(name) <= 2 and 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 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

trend_engine/spot_matcher.py

@@ -0,0 +1,111 @@
+"""
+SpotMatcher — 채널별 수집 결과의 spot_id를 trend_spots 마스터 ID로 통합 매칭
+
+모든 채널의 수집 결과를 trend_spots(카카오맵 기반)과
+story_spots(향토지 기반)에서 정의된 장소 ID로 매칭한다.
+
+매칭 우선순위:
+  1. trend_spots 정확 매칭 (공백 제거 변형 포함)
+  2. story_spots 정확 매칭
+  3. trend_spots 부분 매칭 (최소 3글자 겹침)
+"""
+
+import logging
+import os
+
+from supabase import create_client, Client
+
+logger = logging.getLogger(__name__)
+
+
+def _get_supabase_client() -> Client:
+    url = os.environ.get("SUPABASE_URL") or os.environ.get("VITE_SUPABASE_URL")
+    key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY") or os.environ.get("SUPABASE_SERVICE_KEY")
+    if not url or not key:
+        raise ValueError("SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY must be set")
+    return create_client(url, key)
+
+
+class SpotMatcher:
+    """모든 채널의 수집 결과를 trend_spots/story_spots ID로 매칭.
+
+    카카오맵 스캔 결과(trend_spots)가 기준 사전 역할을 한다.
+    """
+
+    def __init__(self, supabase: Client | None = None):
+        self.supabase = supabase or _get_supabase_client()
+        self.trend_spots: dict[str, str] = {}
+        self.story_spots: dict[str, str] = {}
+        self._load_dictionaries()
+
+    def _load_dictionaries(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"].strip()
+                if not name:
+                    continue
+                self.trend_spots[name] = row["id"]
+                no_space = name.replace(" ", "")
+                if no_space != name:
+                    self.trend_spots[no_space] = 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")
+                .not_.is_("name", "null")
+                .execute()
+            )
+            for row in resp.data or []:
+                name = (row.get("name") or "").strip()
+                if name:
+                    self.story_spots[name] = row["id"]
+        except Exception as e:
+            logger.warning("story_spots 로드 실패: %s", e)
+
+        logger.info(
+            "SpotMatcher 사전 로드: trend_spots %d건, story_spots %d건",
+            len(self.trend_spots),
+            len(self.story_spots),
+        )
+
+    def match(self, name: str) -> str | None:
+        """장소명 → trend_spots.id 또는 story_spots.id 반환.
+
+        매칭 우선순위:
+          1. trend_spots 정확 매칭 (공백 제거 포함)
+          2. story_spots 정확 매칭
+          3. trend_spots 부분 매칭 (최소 3글자 겹침)
+
+        Returns:
+            매칭된 spot_id, 실패 시 None
+        """
+        if not name:
+            return None
+        name = name.strip()
+
+        # 1. trend_spots 정확 매칭
+        if name in self.trend_spots:
+            return self.trend_spots[name]
+        no_space = name.replace(" ", "")
+        if no_space in self.trend_spots:
+            return self.trend_spots[no_space]
+
+        # 2. story_spots 정확 매칭
+        if name in self.story_spots:
+            return self.story_spots[name]
+
+        # 3. 부분 매칭 (긴 이름부터 시도)
+        for known_name in sorted(self.trend_spots.keys(), key=len, reverse=True):
+            shorter = min(known_name, name, key=len)
+            if len(shorter) >= 3 and (known_name in name or name in known_name):
+                return self.trend_spots[known_name]
+
+        return None

trend_engine/trend_scorer.py

@@ -0,0 +1,344 @@
+"""
+Trend Scorer — 채널별 인기도 스코어 계산 + 종합 랭킹 생성
+
+v3 채널 가중치:
+  naver_blog  0.30
+  kakaomap    0.25
+  instagram   0.25
+  youtube     0.15
+  naver_place 0.05
+"""
+
+import os
+import logging
+from datetime import datetime, timedelta, date
+
+from supabase import create_client, Client
+
+logger = logging.getLogger(__name__)
+
+# 채널 가중치 (v3 — naver_place 비활성, 4채널 체제)
+CHANNEL_WEIGHTS: dict[str, float] = {
+    "naver_blog": 0.30,
+    "kakaomap": 0.25,
+    "instagram": 0.25,
+    "youtube": 0.15,
+    # "naver_place": 0.05,  # 비활성 — Place ID 매칭 불가 (2026-02)
+}
+
+
+def _get_supabase_client() -> Client:
+    url = os.environ.get("SUPABASE_URL") or os.environ.get("VITE_SUPABASE_URL")
+    key = os.environ.get("SUPABASE_SERVICE_ROLE_KEY") or os.environ.get("SUPABASE_SERVICE_KEY")
+    if not url or not key:
+        raise ValueError("SUPABASE_URL and SUPABASE_SERVICE_ROLE_KEY must be set")
+    return create_client(url, key)
+
+
+# ------------------------------------------------------------------
+# 정규화 유틸
+# ------------------------------------------------------------------
+
+
+def normalize_score(value: float, max_value: float) -> int:
+    """채널 내 최대값 기준 0~100 정규화."""
+    if max_value <= 0:
+        return 0
+    return min(100, int((value / max_value) * 100))
+
+
+# ------------------------------------------------------------------
+# 채널별 스코어 계산
+# ------------------------------------------------------------------
+
+
+def calc_naver_blog_score(
+    weekly_mentions: int,
+    max_weekly_mentions: int,
+    mention_growth: float = 0.0,
+) -> int:
+    """네이버 블로그: 주간 언급 수 + 급상승 가산점."""
+    base = normalize_score(weekly_mentions, max_weekly_mentions)
+    growth_bonus = min(20, int(mention_growth * 10)) if mention_growth > 0 else 0
+    return min(100, base + growth_bonus)
+
+
+def calc_kakaomap_score(
+    review_count: int,
+    max_review_count: int,
+    review_growth: int = 0,
+    max_review_growth: int = 1,
+    search_rank: int = 0,
+    max_rank: int = 256,
+) -> int:
+    """카카오맵: 리뷰 수(60%) + 전주 대비 리뷰 증감(40%).
+
+    리뷰 데이터 없으면 search_rank 기반 폴백 (1위=100점, max_rank위=0점).
+    """
+    if review_count > 0:
+        review_base = normalize_score(review_count, max_review_count) * 0.6
+        growth_base = normalize_score(review_growth, max_review_growth) * 0.4
+        return min(100, int(review_base + growth_base))
+
+    # search_rank 폴백 (비공식 리뷰 API 차단 시)
+    if search_rank > 0:
+        return max(0, int((1 - search_rank / max(max_rank, 1)) * 100))
+
+    return 0
+
+
+def calc_instagram_score(
+    hashtag_post_count: int,
+    max_post_count: int,
+    avg_engagement: float = 0.0,
+    max_engagement: float = 1.0,
+) -> int:
+    """인스타그램: 게시물 수(50%) + 평균 참여도(50%)."""
+    post_base = normalize_score(hashtag_post_count, max_post_count) * 0.5
+    engagement_base = normalize_score(avg_engagement, max_engagement) * 0.5
+    return min(100, int(post_base + engagement_base))
+
+
+def calc_youtube_score(
+    mention_video_count: int,
+    max_video_count: int,
+    total_views: int = 0,
+    max_total_views: int = 1,
+) -> int:
+    """유튜브: 언급 영상 수(40%) + 총 조회수(60%)."""
+    video_base = normalize_score(mention_video_count, max_video_count) * 0.4
+    views_base = normalize_score(total_views, max_total_views) * 0.6
+    return min(100, int(video_base + views_base))
+
+
+def calc_naver_place_score(
+    visitor_review_count: int,
+    max_visitor_reviews: int,
+    naver_review_growth: int = 0,
+    max_review_growth: int = 1,
+    data_available: bool = True,
+) -> int | None:
+    """네이버 플레이스 (보조): 데이터 없으면 None 반환."""
+    if not data_available:
+        return None
+    review_base = normalize_score(visitor_review_count, max_visitor_reviews) * 0.7
+    growth_base = normalize_score(naver_review_growth, max_review_growth) * 0.3
+    return min(100, int(review_base + growth_base))
+
+
+# ------------------------------------------------------------------
+# 종합 스코어
+# ------------------------------------------------------------------
+
+
+def calc_composite_score(channel_scores: dict[str, int | None]) -> int:
+    """
+    채널별 스코어를 가중 합산하여 종합 인기도 스코어 (0~100) 산출.
+
+    데이터가 없는 채널(None)의 가중치는 나머지 채널에 비례 재분배.
+
+    Args:
+        channel_scores: {"naver_blog": 80, "kakaomap": 60, ..., "naver_place": None}
+    """
+    available_weight = 0.0
+    weighted_parts: list[tuple[float, float]] = []  # (score, weight)
+
+    for channel, weight in CHANNEL_WEIGHTS.items():
+        score = channel_scores.get(channel)
+        if score is not None:
+            weighted_parts.append((float(score), weight))
+            available_weight += weight
+
+    if available_weight == 0:
+        return 0
+
+    composite = sum(score * (weight / available_weight) for score, weight in weighted_parts)
+    return min(100, int(composite))
+
+
+# ------------------------------------------------------------------
+# 주간 랭킹 생성
+# ------------------------------------------------------------------
+
+
+def generate_weekly_ranking(supabase: Client | None = None) -> dict:
+    """
+    주간 종합 인기 장소 랭킹 생성.
+
+    1. spot_trends에서 이번 주 + 지난주 메트릭 조회
+    2. 채널별 스코어 계산 → 종합 스코어
+    3. trend_spots.popularity_score 업데이트
+    4. 전체/채널별 랭킹 반환
+
+    Returns:
+        {"overall": [...], "naver_blog": [...], ..., "trending_up": [...]}
+    """
+    sb = supabase or _get_supabase_client()
+    now = datetime.utcnow()
+    this_week_start = (now - timedelta(days=now.weekday())).date()
+    last_week_start = this_week_start - timedelta(days=7)
+
+    # -- 이번 주 메트릭 조회 (유효한 spot_id만) --
+    this_week_resp = (
+        sb.table("spot_trends")
+        .select("spot_id, source, metric_type, metric_value")
+        .gte("period_end", this_week_start.isoformat())
+        .neq("spot_id", "__pending__")
+        .execute()
+    )
+    this_week_rows = this_week_resp.data or []
+
+    # -- 지난주 메트릭 조회 (유효한 spot_id만) --
+    last_week_resp = (
+        sb.table("spot_trends")
+        .select("spot_id, source, metric_type, metric_value")
+        .gte("period_end", last_week_start.isoformat())
+        .lt("period_end", this_week_start.isoformat())
+        .neq("spot_id", "__pending__")
+        .execute()
+    )
+    last_week_rows = last_week_resp.data or []
+
+    # -- 스팟별 메트릭 집계 --
+    spots_this: dict[str, dict[str, dict[str, int]]] = {}
+    for row in this_week_rows:
+        sid = row["spot_id"]
+        src = row["source"]
+        mt = row["metric_type"]
+        spots_this.setdefault(sid, {}).setdefault(src, {})[mt] = row["metric_value"]
+
+    spots_last: dict[str, dict[str, dict[str, int]]] = {}
+    for row in last_week_rows:
+        sid = row["spot_id"]
+        src = row["source"]
+        mt = row["metric_type"]
+        spots_last.setdefault(sid, {}).setdefault(src, {})[mt] = row["metric_value"]
+
+    # -- 전체 최대값 계산 (정규화용) --
+    maxes = _calc_maxes(spots_this)
+
+    # -- 스팟별 채널 스코어 + 종합 스코어 계산 --
+    spot_scores: list[dict] = []
+
+    for spot_id, src_data in spots_this.items():
+        last_src = spots_last.get(spot_id, {})
+
+        # 네이버 블로그
+        blog = src_data.get("naver_blog", {})
+        blog_last = last_src.get("naver_blog", {})
+        wm = blog.get("mention_count", 0)
+        wm_last = blog_last.get("mention_count", 0)
+        growth = (wm - wm_last) / max(wm_last, 1) if wm_last else 0.0
+        nb_score = calc_naver_blog_score(wm, maxes["naver_blog_mentions"], growth)
+
+        # 카카오맵 (search_rank 폴백 지원)
+        km = src_data.get("kakaomap", {})
+        km_last = last_src.get("kakaomap", {})
+        rc = km.get("review_count", 0)
+        rc_last = km_last.get("review_count", 0)
+        sr = km.get("search_rank", 0)
+        km_score = calc_kakaomap_score(
+            rc, maxes["kakaomap_reviews"],
+            rc - rc_last, maxes["kakaomap_review_growth"],
+            search_rank=sr, max_rank=maxes["kakaomap_max_rank"],
+        )
+
+        # 인스타그램
+        ig = src_data.get("instagram", {})
+        ig_score = calc_instagram_score(
+            ig.get("post_count", 0), maxes["instagram_posts"],
+            ig.get("avg_engagement", 0), maxes["instagram_engagement"],
+        )
+
+        # 유튜브
+        yt = src_data.get("youtube", {})
+        yt_score = calc_youtube_score(
+            yt.get("video_count", 0), maxes["youtube_videos"],
+            yt.get("view_count", 0), maxes["youtube_views"],
+        )
+
+        # 네이버 플레이스 — 비활성 (Place ID 매칭 불가, 2026-02)
+        # np_score는 항상 None → CHANNEL_WEIGHTS에서도 제거됨
+
+        channel_scores = {
+            "naver_blog": nb_score,
+            "kakaomap": km_score,
+            "instagram": ig_score,
+            "youtube": yt_score,
+        }
+        composite = calc_composite_score(channel_scores)
+
+        spot_scores.append({
+            "spot_id": spot_id,
+            "popularity_score": composite,
+            "naver_blog_score": nb_score,
+            "kakaomap_score": km_score,
+            "instagram_score": ig_score,
+            "youtube_score": yt_score,
+            "channel_scores": channel_scores,
+        })
+
+    # -- trend_spots 업데이트 --
+    for entry in spot_scores:
+        try:
+            sb.table("trend_spots").update({
+                "popularity_score": entry["popularity_score"],
+                "popularity_updated_at": now.isoformat(),
+            }).eq("id", entry["spot_id"]).execute()
+        except Exception as e:
+            logger.warning("trend_spots 업데이트 실패 (%s): %s", entry["spot_id"], e)
+
+    # -- 랭킹 생성 --
+    overall = sorted(spot_scores, key=lambda s: s["popularity_score"], reverse=True)
+
+    rankings = {
+        "overall": overall[:50],
+        "naver_blog": sorted(spot_scores, key=lambda s: s["naver_blog_score"], reverse=True)[:30],
+        "kakaomap": sorted(spot_scores, key=lambda s: s["kakaomap_score"], reverse=True)[:30],
+        "instagram": sorted(spot_scores, key=lambda s: s["instagram_score"], reverse=True)[:30],
+        "youtube": sorted(spot_scores, key=lambda s: s["youtube_score"], reverse=True)[:30],
+        "trending_up": overall[:10],  # 전주 대비 비교는 다음 주기부터 가능
+    }
+
+    logger.info("주간 랭킹 생성 완료: %d개 스팟", len(spot_scores))
+    return rankings
+
+
+# ------------------------------------------------------------------
+# 내부 헬퍼
+# ------------------------------------------------------------------
+
+
+def _calc_maxes(spots_data: dict[str, dict[str, dict[str, int]]]) -> dict[str, int]:
+    """전체 스팟에서 채널별 최대값 계산."""
+    maxes = {
+        "naver_blog_mentions": 1,
+        "kakaomap_reviews": 1,
+        "kakaomap_review_growth": 1,
+        "kakaomap_max_rank": 1,
+        "instagram_posts": 1,
+        "instagram_engagement": 1,
+        "youtube_videos": 1,
+        "youtube_views": 1,
+    }
+
+    for src_data in spots_data.values():
+        blog = src_data.get("naver_blog", {})
+        maxes["naver_blog_mentions"] = max(maxes["naver_blog_mentions"], blog.get("mention_count", 0))
+
+        km = src_data.get("kakaomap", {})
+        maxes["kakaomap_reviews"] = max(maxes["kakaomap_reviews"], km.get("review_count", 0))
+        maxes["kakaomap_max_rank"] = max(maxes["kakaomap_max_rank"], km.get("search_rank", 0))
+
+        ig = src_data.get("instagram", {})
+        maxes["instagram_posts"] = max(maxes["instagram_posts"], ig.get("post_count", 0))
+        # avg_engagement: -1 값 필터링 (Apify 미수집 케이스)
+        ig_eng = ig.get("avg_engagement", 0)
+        if ig_eng > 0:
+            maxes["instagram_engagement"] = max(maxes["instagram_engagement"], ig_eng)
+
+        yt = src_data.get("youtube", {})
+        maxes["youtube_videos"] = max(maxes["youtube_videos"], yt.get("video_count", 0))
+        maxes["youtube_views"] = max(maxes["youtube_views"], yt.get("view_count", 0))
+
+    return maxes