Merge branch 'main' of https://gitea.o2o.kr/castad/o2o-castad-backend into dashboard

.gitignore

@@ -49,4 +49,6 @@ logs/
 
 *.yml
 Dockerfile
-.dockerignore
+.dockerignore
+
+zzz/

app/core/common.py

@@ -24,6 +24,11 @@ async def lifespan(app: FastAPI):
 
             await create_db_tables()
             logger.info("Database tables created (DEBUG mode)")
+
+        # dashboard 테이블 초기화 및 기존 데이터 마이그레이션 (모든 환경)
+        from app.dashboard.migration import init_dashboard_table
+        await init_dashboard_table()
+
         await NvMapPwScraper.initiate_scraper()
     except asyncio.TimeoutError:
         logger.error("Database initialization timed out")

app/core/exceptions.py

@@ -309,6 +309,20 @@ def add_exception_handlers(app: FastAPI):
             content=content,
         )
 
+    # DashboardException 핸들러 추가
+    from app.dashboard.exceptions import DashboardException
+
+    @app.exception_handler(DashboardException)
+    def dashboard_exception_handler(request: Request, exc: DashboardException) -> Response:
+        logger.debug(f"Handled DashboardException: {exc.__class__.__name__} - {exc.message}")
+        return JSONResponse(
+            status_code=exc.status_code,
+            content={
+                "detail": exc.message,
+                "code": exc.code,
+            },
+        )
+
     @app.exception_handler(status.HTTP_500_INTERNAL_SERVER_ERROR)
     def internal_server_error_handler(request, exception):
         # 에러 메시지 로깅 (한글 포함 가능)

app/dashboard/__init__.py

@@ -0,0 +1,5 @@
+"""
+Dashboard Module
+
+YouTube Analytics API를 활용한 대시보드 기능을 제공합니다.
+"""

app/dashboard/api/__init__.py

@@ -0,0 +1,3 @@
+"""
+Dashboard API Module
+"""

app/dashboard/api/routers/__init__.py

@@ -0,0 +1,3 @@
+"""
+Dashboard Routers
+"""

app/dashboard/api/routers/v1/__init__.py

@@ -0,0 +1,7 @@
+"""
+Dashboard V1 Routers
+"""
+
+from app.dashboard.api.routers.v1.dashboard import router
+
+__all__ = ["router"]

app/dashboard/api/routers/v1/dashboard.py

@@ -0,0 +1,480 @@
+"""
+Dashboard API 라우터
+
+YouTube Analytics 기반 대시보드 통계를 제공합니다.
+"""
+
+import json
+import logging
+from datetime import date, datetime, timedelta
+from typing import Literal
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.dashboard.exceptions import (
+    NoVideosFoundError,
+    YouTubeAccountNotConnectedError,
+    YouTubeAccountNotFoundError,
+    YouTubeAccountSelectionRequiredError,
+    YouTubeTokenExpiredError,
+)
+from app.dashboard.schemas import (
+    CacheDeleteResponse,
+    ConnectedAccount,
+    ConnectedAccountsResponse,
+    DashboardResponse,
+    TopContent,
+)
+from app.dashboard.services import DataProcessor, YouTubeAnalyticsService
+from app.dashboard.redis_cache import (
+    delete_cache,
+    delete_cache_pattern,
+    get_cache,
+    set_cache,
+)
+from app.database.session import get_session
+from app.dashboard.models import Dashboard
+from app.social.exceptions import TokenExpiredError
+from app.social.services import SocialAccountService
+from app.user.dependencies.auth import get_current_user
+from app.user.models import SocialAccount, User
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/dashboard", tags=["Dashboard"])
+
+
+@router.get(
+    "/accounts",
+    response_model=ConnectedAccountsResponse,
+    summary="연결된 소셜 계정 목록 조회",
+    description="""
+연결된 소셜 계정 목록을 반환합니다.
+
+여러 계정이 연결된 경우, 반환된 `id` 값을 `/dashboard/stats?social_account_id=<id>`에 전달하여 계정을 선택합니다.
+""",
+)
+async def get_connected_accounts(
+    current_user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> ConnectedAccountsResponse:
+    result = await session.execute(
+        select(SocialAccount).where(
+            SocialAccount.user_uuid == current_user.user_uuid,
+            SocialAccount.platform == "youtube",
+            SocialAccount.is_active == True,  # noqa: E712
+        )
+    )
+    accounts_raw = result.scalars().all()
+
+    # platform_user_id 기준
+    seen_platform_ids: set[str] = set()
+    connected = []
+    for acc in sorted(
+        accounts_raw, key=lambda a: a.connected_at or datetime.min, reverse=True
+    ):
+        if acc.platform_user_id in seen_platform_ids:
+            continue
+        seen_platform_ids.add(acc.platform_user_id)
+        data = acc.platform_data if isinstance(acc.platform_data, dict) else {}
+        connected.append(
+            ConnectedAccount(
+                id=acc.id,
+                platform=acc.platform,
+                platform_username=acc.platform_username,
+                platform_user_id=acc.platform_user_id,
+                channel_title=data.get("channel_title"),
+                connected_at=acc.connected_at,
+                is_active=acc.is_active,
+            )
+        )
+
+    logger.info(
+        f"[ACCOUNTS] YouTube 계정 목록 조회 - "
+        f"user_uuid={current_user.user_uuid}, count={len(connected)}"
+    )
+    return ConnectedAccountsResponse(accounts=connected)
+
+
+@router.get(
+    "/stats",
+    response_model=DashboardResponse,
+    summary="대시보드 통계 조회",
+    description="""
+YouTube Analytics API를 활용한 대시보드 통계를 조회합니다.
+
+## 주요 기능
+- 최근 30개 업로드 영상 기준 통계 제공
+- KPI 지표: 업로드 영상 수, 총 조회수, 시청 시간, 평균 시청 시간, 신규 구독자, 좋아요, 댓글, 공유
+- 월별 추이: 최근 12개월 vs 이전 12개월 비교
+- 인기 영상 TOP 4
+- 시청자 분석: 연령/성별/지역 분포
+
+## 성능 최적화
+- 8개 YouTube Analytics API를 병렬로 호출
+- Redis 캐싱 적용 (TTL: 12시간)
+
+## 사전 조건
+- YouTube 계정이 연동되어 있어야 합니다
+- 최소 1개 이상의 영상이 업로드되어 있어야 합니다
+
+## 조회 모드
+- `day`: 최근 30일 통계 (현재 날짜 -2일 기준)
+- `month`: 최근 12개월 통계 (현재 날짜 -2일 기준, 기본값)
+
+## 데이터 특성
+- **지연 시간**: 48시간 (2일) - 2월 14일 요청 시 2월 12일 자까지 확정
+- **업데이트 주기**: 하루 1회 (PT 자정, 한국 시간 오후 5~8시)
+- **실시간 아님**: 전날 데이터가 다음날 확정됩니다
+""",
+)
+async def get_dashboard_stats(
+    mode: Literal["day", "month"] = Query(
+        default="month",
+        description="조회 모드: day(최근 30일), month(최근 12개월)",
+    ),
+    platform_user_id: str | None = Query(
+        default=None,
+        description="사용할 YouTube 채널 ID (platform_user_id)",
+    ),
+    current_user: User = Depends(get_current_user),
+    session: AsyncSession = Depends(get_session),
+) -> DashboardResponse:
+    """
+    대시보드 통계 조회
+
+    Args:
+        mode: 조회 모드 (day: 최근 30일, month: 최근 12개월)
+        platform_user_id: 사용할 YouTube 채널 ID (여러 계정 연결 시 필수, 재연동해도 불변)
+        current_user: 현재 인증된 사용자
+        session: 데이터베이스 세션
+
+    Returns:
+        DashboardResponse: 대시보드 통계 데이터
+
+    Raises:
+        YouTubeAccountNotConnectedError: YouTube 계정이 연동되어 있지 않음
+        YouTubeAccountSelectionRequiredError: 여러 계정이 연결되어 있으나 계정 미선택
+        YouTubeAccountNotFoundError: 지정한 계정을 찾을 수 없음
+        NoVideosFoundError: 업로드된 영상이 없음
+        YouTubeTokenExpiredError: YouTube 토큰 만료 (재연동 필요)
+        YouTubeAPIError: YouTube Analytics API 호출 실패
+    """
+    logger.info(
+        f"[DASHBOARD] 통계 조회 시작 - "
+        f"user_uuid={current_user.user_uuid}, mode={mode}, platform_user_id={platform_user_id}"
+    )
+
+    # 1. 모드별 날짜 자동 계산
+    today = date.today()
+
+    if mode == "day":
+        # 48시간 지연 적용: 오늘 기준 -2일을 end로 사용
+        # ex) 오늘 2/20 → end=2/18, start=1/20
+        end_dt = today - timedelta(days=2)
+        kpi_end_dt = end_dt
+        start_dt = end_dt - timedelta(days=29)
+        period_desc = "최근 30일"
+    else:  # mode == "month"
+        # 월별 차트: dimensions=month API는 YYYY-MM-01 형식 필요
+        # ex) 오늘 2/24 → end=2026-02-01, start=2025-03-01 → 2025-03 ~ 2026-02 (12개월)
+        end_dt = today.replace(day=1)
+        # KPI 등 집계형 API: 48시간 지연 적용하여 현재 월 전체 데이터 포함
+        kpi_end_dt = today - timedelta(days=2)
+
+        start_month = end_dt.month - 11
+        if start_month <= 0:
+            start_month += 12
+            start_year = end_dt.year - 1
+        else:
+            start_year = end_dt.year
+        start_dt = date(start_year, start_month, 1)
+        period_desc = "최근 12개월"
+
+    start_date = start_dt.strftime("%Y-%m-%d")
+    end_date = end_dt.strftime("%Y-%m-%d")
+    kpi_end_date = kpi_end_dt.strftime("%Y-%m-%d")
+
+    logger.debug(
+        f"[1] 날짜 계산 완료 - period={period_desc}, start={start_date}, end={end_date}"
+    )
+
+    # 2. YouTube 계정 연동 확인
+    result = await session.execute(
+        select(SocialAccount).where(
+            SocialAccount.user_uuid == current_user.user_uuid,
+            SocialAccount.platform == "youtube",
+            SocialAccount.is_active == True,  # noqa: E712
+        )
+    )
+    social_accounts_raw = result.scalars().all()
+
+    # platform_user_id 기준으로 중복 제거 (가장 최근 연동 계정 우선)
+    seen_platform_ids_stats: set[str] = set()
+    social_accounts = []
+    for acc in sorted(
+        social_accounts_raw, key=lambda a: a.connected_at or datetime.min, reverse=True
+    ):
+        if acc.platform_user_id not in seen_platform_ids_stats:
+            seen_platform_ids_stats.add(acc.platform_user_id)
+            social_accounts.append(acc)
+
+    if not social_accounts:
+        logger.warning(
+            f"[NO YOUTUBE ACCOUNT] YouTube 계정 미연동 - "
+            f"user_uuid={current_user.user_uuid}"
+        )
+        raise YouTubeAccountNotConnectedError()
+
+    if platform_user_id is not None:
+        matched = [a for a in social_accounts if a.platform_user_id == platform_user_id]
+        if not matched:
+            logger.warning(
+                f"[ACCOUNT NOT FOUND] 지정 계정 없음 - "
+                f"user_uuid={current_user.user_uuid}, platform_user_id={platform_user_id}"
+            )
+            raise YouTubeAccountNotFoundError()
+        social_account = matched[0]
+    elif len(social_accounts) == 1:
+        social_account = social_accounts[0]
+    else:
+        logger.warning(
+            f"[MULTI ACCOUNT] 계정 선택 필요 - "
+            f"user_uuid={current_user.user_uuid}, count={len(social_accounts)}"
+        )
+        raise YouTubeAccountSelectionRequiredError()
+
+    logger.debug(
+        f"[2] YouTube 계정 확인 완료 - platform_user_id={social_account.platform_user_id}"
+    )
+
+    # 3. 기간 내 업로드 영상 수 조회
+    count_result = await session.execute(
+        select(func.count())
+        .select_from(Dashboard)
+        .where(
+            Dashboard.user_uuid == current_user.user_uuid,
+            Dashboard.platform == "youtube",
+            Dashboard.platform_user_id == social_account.platform_user_id,
+            Dashboard.uploaded_at >= start_dt,
+            Dashboard.uploaded_at <= kpi_end_dt,
+        )
+    )
+    period_video_count = count_result.scalar() or 0
+    logger.debug(f"[3] 기간 내 업로드 영상 수 - count={period_video_count}")
+
+    # 4. Redis 캐시 조회
+    # platform_user_id 기준 캐시 키: 재연동해도 채널 ID는 불변 → 캐시 유지됨
+    cache_key = f"dashboard:{current_user.user_uuid}:{social_account.platform_user_id}:{mode}"
+    cached_raw = await get_cache(cache_key)
+
+    if cached_raw:
+        try:
+            payload = json.loads(cached_raw)
+            logger.info(f"[CACHE HIT] 캐시 반환 - user_uuid={current_user.user_uuid}")
+            response = DashboardResponse.model_validate(payload["response"])
+            for metric in response.content_metrics:
+                if metric.id == "uploaded-videos":
+                    metric.value = str(period_video_count)
+                    break
+            return response
+        except (json.JSONDecodeError, KeyError):
+            logger.warning(f"[CACHE PARSE ERROR] 포맷 오류, 무시 - key={cache_key}")
+
+    logger.debug("[4] 캐시 MISS - YouTube API 호출 필요")
+
+    # 5. 최근 30개 업로드 영상 조회 (Analytics API 전달용)
+    # YouTube Analytics API 제약사항:
+    # - 영상 개수: 20~30개 권장 (최대 50개, 그 이상은 응답 지연 발생)
+    # - URL 길이: 2000자 제한 (video ID 11자 × 30개 = 330자로 안전)
+    result = await session.execute(
+        select(
+            Dashboard.platform_video_id,
+            Dashboard.title,
+            Dashboard.uploaded_at,
+        )
+        .where(
+            Dashboard.user_uuid == current_user.user_uuid,
+            Dashboard.platform == "youtube",
+            Dashboard.platform_user_id == social_account.platform_user_id,
+        )
+        .order_by(Dashboard.uploaded_at.desc())
+        .limit(30)
+    )
+    rows = result.all()
+    # 테스트를 위해 주석처리
+    # if not rows:
+    #     logger.warning(
+    #         f"[NO VIDEOS] 업로드된 영상 없음 - " f"user_uuid={current_user.user_uuid}"
+    #     )
+    #     raise NoVideosFoundError()
+
+    # logger.debug(f"[5] 영상 조회 완료 - count={len(rows)}")
+
+    # 6. video_ids + 메타데이터 조회용 dict 구성
+    video_ids = []
+    video_lookup: dict[str, tuple[str, datetime]] = {}  # {video_id: (title, uploaded_at)}
+
+    for row in rows:
+        platform_video_id, title, uploaded_at = row
+        video_ids.append(platform_video_id)
+        video_lookup[platform_video_id] = (title, uploaded_at)
+
+    logger.debug(
+        f"[6] 영상 메타데이터 구성 완료 - count={len(video_ids)}, sample={video_ids[:3]}"
+    )
+
+    # 7. 토큰 유효성 확인 및 자동 갱신 (만료 10분 전 갱신)
+    try:
+        access_token = await SocialAccountService().ensure_valid_token(
+            social_account, session
+        )
+    except TokenExpiredError:
+        logger.warning(
+            f"[TOKEN EXPIRED] 재연동 필요 - user_uuid={current_user.user_uuid}"
+        )
+        raise YouTubeTokenExpiredError()
+
+    logger.debug("[7] 토큰 유효성 확인 완료")
+
+    # 8. YouTube Analytics API 호출 (8개 병렬)
+    youtube_service = YouTubeAnalyticsService()
+    raw_data = await youtube_service.fetch_all_metrics(
+        video_ids=video_ids,
+        start_date=start_date,
+        end_date=end_date,
+        kpi_end_date=kpi_end_date,
+        access_token=access_token,
+        mode=mode,
+    )
+
+    logger.debug("[8] YouTube Analytics API 호출 완료")
+
+    # 9. TopContent 조립 (Analytics top_videos + DB lookup)
+    processor = DataProcessor()
+    top_content_rows = raw_data.get("top_videos", {}).get("rows", [])
+    top_content: list[TopContent] = []
+    for row in top_content_rows[:4]:
+        if len(row) < 4:
+            continue
+        video_id, views, likes, comments = row[0], row[1], row[2], row[3]
+        meta = video_lookup.get(video_id)
+        if not meta:
+            continue
+        title, uploaded_at = meta
+        engagement_rate = ((likes + comments) / views * 100) if views > 0 else 0
+        top_content.append(
+            TopContent(
+                id=video_id,
+                title=title,
+                thumbnail=f"https://i.ytimg.com/vi/{video_id}/mqdefault.jpg",
+                platform="youtube",
+                views=DataProcessor._format_number(views),
+                engagement=f"{engagement_rate:.1f}%",
+                date=uploaded_at.strftime("%Y.%m.%d"),
+            )
+        )
+
+    logger.debug(f"[9] TopContent 조립 완료 - count={len(top_content)}")
+
+    # 10. 데이터 가공 (period_video_count=0 — API 무관 DB 집계값, 캐시에 포함하지 않음)
+    dashboard_data = processor.process(
+        raw_data, top_content, 0, mode=mode, end_date=end_date
+    )
+
+    logger.debug("[10] 데이터 가공 완료")
+
+    # 11. Redis 캐싱 (TTL: 12시간)
+    # YouTube Analytics는 하루 1회 갱신 (PT 자정, 한국 시간 오후 5~8시)
+    # 48시간 지연된 데이터이므로 12시간 캐싱으로 API 호출 최소화
+    # period_video_count는 캐시에 포함하지 않음 (DB 직접 집계, API 미사용)
+    cache_payload = json.dumps(
+        {"response": json.loads(dashboard_data.model_dump_json())}
+    )
+    cache_success = await set_cache(
+        cache_key,
+        cache_payload,
+        ttl=43200,  # 12시간
+    )
+
+    if cache_success:
+        logger.debug(f"[CACHE SET] 캐시 저장 성공 - key={cache_key}")
+    else:
+        logger.warning(f"[CACHE SET] 캐시 저장 실패 - key={cache_key}")
+
+    # 12. 업로드 영상 수 주입 (캐시 저장 후 — 항상 DB에서 직접 집계)
+    for metric in dashboard_data.content_metrics:
+        if metric.id == "uploaded-videos":
+            metric.value = str(period_video_count)
+            break
+
+    logger.info(
+        f"[DASHBOARD] 통계 조회 완료 - "
+        f"user_uuid={current_user.user_uuid}, "
+        f"mode={mode}, period={period_desc}, videos={len(video_ids)}"
+    )
+
+    return dashboard_data
+
+
+@router.delete(
+    "/cache",
+    response_model=CacheDeleteResponse,
+    summary="대시보드 캐시 삭제",
+    description="""
+대시보드 Redis 캐시를 삭제합니다. 인증 없이 호출 가능합니다.
+
+삭제 후 다음 `/stats` 요청 시 YouTube Analytics API를 새로 호출하여 최신 데이터를 반환합니다.
+
+## 사용 시나리오
+- 코드 배포 후 즉시 최신 데이터 반영이 필요할 때
+- 데이터 이상 발생 시 캐시 강제 갱신
+
+## 캐시 키 구조
+`dashboard:{user_uuid}:{mode}` (mode: day 또는 month)
+
+## 파라미터
+- `user_uuid`: 특정 사용자 캐시만 삭제. 미입력 시 전체 삭제
+- `mode`: day / month / all (기본값: all)
+""",
+)
+async def delete_dashboard_cache(
+    mode: Literal["day", "month", "all"] = Query(
+        default="all",
+        description="삭제할 캐시 모드: day, month, all(기본값, 모두 삭제)",
+    ),
+    user_uuid: str | None = Query(
+        default=None,
+        description="대상 사용자 UUID. 미입력 시 전체 사용자 캐시 삭제",
+    ),
+) -> CacheDeleteResponse:
+    """
+    대시보드 캐시 삭제
+
+    Args:
+        mode: 삭제할 캐시 모드 (day / month / all)
+        user_uuid: 대상 사용자 UUID (없으면 전체 삭제)
+
+    Returns:
+        CacheDeleteResponse: 삭제된 캐시 키 개수 및 메시지
+    """
+    if user_uuid:
+        if mode == "all":
+            deleted = await delete_cache_pattern(f"dashboard:{user_uuid}:*")
+            message = f"전체 캐시 삭제 완료 ({deleted}개)"
+        else:
+            cache_key = f"dashboard:{user_uuid}:{mode}"
+            success = await delete_cache(cache_key)
+            deleted = 1 if success else 0
+            message = f"{mode} 캐시 삭제 {'완료' if success else '실패 (키 없음)'}"
+    else:
+        deleted = await delete_cache_pattern("dashboard:*")
+        message = f"전체 사용자 캐시 삭제 완료 ({deleted}개)"
+
+    logger.info(
+        f"[CACHE DELETE] user_uuid={user_uuid or 'ALL'}, mode={mode}, deleted={deleted}"
+    )
+
+    return CacheDeleteResponse(deleted_count=deleted, message=message)

app/dashboard/exceptions.py

@@ -0,0 +1,195 @@
+"""
+Dashboard Exceptions
+
+Dashboard API 관련 예외 클래스를 정의합니다.
+"""
+
+from fastapi import status
+
+
+class DashboardException(Exception):
+    """Dashboard 기본 예외"""
+
+    def __init__(
+        self,
+        message: str,
+        status_code: int = status.HTTP_500_INTERNAL_SERVER_ERROR,
+        code: str = "DASHBOARD_ERROR",
+    ):
+        self.message = message
+        self.status_code = status_code
+        self.code = code
+        super().__init__(self.message)
+
+
+# =============================================================================
+# YouTube Analytics API 관련 예외
+# =============================================================================
+
+
+class YouTubeAPIException(DashboardException):
+    """YouTube Analytics API 관련 예외 기본 클래스"""
+
+    def __init__(
+        self,
+        message: str = "YouTube Analytics API 호출 중 오류가 발생했습니다.",
+        status_code: int = status.HTTP_502_BAD_GATEWAY,
+        code: str = "YOUTUBE_API_ERROR",
+    ):
+        super().__init__(message, status_code, code)
+
+
+class YouTubeAPIError(YouTubeAPIException):
+    """YouTube Analytics API 일반 오류"""
+
+    def __init__(self, detail: str = ""):
+        error_message = "YouTube Analytics API 호출에 실패했습니다."
+        if detail:
+            error_message += f" ({detail})"
+        super().__init__(
+            message=error_message,
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            code="YOUTUBE_API_FAILED",
+        )
+
+
+class YouTubeAuthError(YouTubeAPIException):
+    """YouTube 인증 실패"""
+
+    def __init__(self, detail: str = ""):
+        error_message = "YouTube 인증에 실패했습니다. 계정 재연동이 필요합니다."
+        if detail:
+            error_message += f" ({detail})"
+        super().__init__(
+            message=error_message,
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            code="YOUTUBE_AUTH_FAILED",
+        )
+
+
+class YouTubeQuotaExceededError(YouTubeAPIException):
+    """YouTube API 할당량 초과"""
+
+    def __init__(self):
+        super().__init__(
+            message="YouTube API 일일 할당량이 초과되었습니다. 내일 다시 시도해주세요.",
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            code="YOUTUBE_QUOTA_EXCEEDED",
+        )
+
+
+class YouTubeDataNotFoundError(YouTubeAPIException):
+    """YouTube Analytics 데이터 없음"""
+
+    def __init__(self, detail: str = ""):
+        error_message = "YouTube Analytics 데이터를 찾을 수 없습니다."
+        if detail:
+            error_message += f" ({detail})"
+        super().__init__(
+            message=error_message,
+            status_code=status.HTTP_404_NOT_FOUND,
+            code="YOUTUBE_DATA_NOT_FOUND",
+        )
+
+
+# =============================================================================
+# 계정 연동 관련 예외
+# =============================================================================
+
+
+class YouTubeAccountNotConnectedError(DashboardException):
+    """YouTube 계정 미연동"""
+
+    def __init__(self):
+        super().__init__(
+            message="YouTube 계정이 연동되어 있지 않습니다. 먼저 YouTube 계정을 연동해주세요.",
+            status_code=status.HTTP_403_FORBIDDEN,
+            code="YOUTUBE_NOT_CONNECTED",
+        )
+
+
+class YouTubeAccountSelectionRequiredError(DashboardException):
+    """여러 YouTube 계정이 연동된 경우 계정 선택 필요"""
+
+    def __init__(self):
+        super().__init__(
+            message="연결된 YouTube 계정이 여러 개입니다. social_account_id 파라미터로 사용할 계정을 선택해주세요.",
+            status_code=status.HTTP_400_BAD_REQUEST,
+            code="YOUTUBE_ACCOUNT_SELECTION_REQUIRED",
+        )
+
+
+class YouTubeAccountNotFoundError(DashboardException):
+    """지정한 YouTube 계정을 찾을 수 없음"""
+
+    def __init__(self):
+        super().__init__(
+            message="지정한 YouTube 계정을 찾을 수 없습니다.",
+            status_code=status.HTTP_404_NOT_FOUND,
+            code="YOUTUBE_ACCOUNT_NOT_FOUND",
+        )
+
+
+class YouTubeTokenExpiredError(DashboardException):
+    """YouTube 토큰 만료"""
+
+    def __init__(self):
+        super().__init__(
+            message="YouTube 인증이 만료되었습니다. 계정을 재연동해주세요.",
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            code="YOUTUBE_TOKEN_EXPIRED",
+        )
+
+
+# =============================================================================
+# 데이터 관련 예외
+# =============================================================================
+
+
+class NoVideosFoundError(DashboardException):
+    """업로드된 영상 없음"""
+
+    def __init__(self):
+        super().__init__(
+            message="업로드된 YouTube 영상이 없습니다. 먼저 영상을 업로드해주세요.",
+            status_code=status.HTTP_404_NOT_FOUND,
+            code="NO_VIDEOS_FOUND",
+        )
+
+
+class DashboardDataError(DashboardException):
+    """대시보드 데이터 처리 오류"""
+
+    def __init__(self, detail: str = ""):
+        error_message = "대시보드 데이터 처리 중 오류가 발생했습니다."
+        if detail:
+            error_message += f" ({detail})"
+        super().__init__(
+            message=error_message,
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            code="DASHBOARD_DATA_ERROR",
+        )
+
+
+# =============================================================================
+# 캐싱 관련 예외 (경고용, 실제로는 raise하지 않고 로깅만 사용)
+# =============================================================================
+
+
+class CacheError(DashboardException):
+    """캐시 작업 오류
+
+    Note:
+        이 예외는 실제로 raise되지 않고,
+        캐시 실패 시 로깅만 하고 원본 데이터를 반환합니다.
+    """
+
+    def __init__(self, operation: str, detail: str = ""):
+        error_message = f"캐시 {operation} 작업 중 오류가 발생했습니다."
+        if detail:
+            error_message += f" ({detail})"
+        super().__init__(
+            message=error_message,
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            code="CACHE_ERROR",
+        )

app/dashboard/migration.py

@@ -0,0 +1,112 @@
+"""
+Dashboard Migration
+
+dashboard 테이블 초기화 및 기존 데이터 마이그레이션을 담당합니다.
+서버 기동 시 create_db_tables() 이후 호출됩니다.
+"""
+
+import logging
+
+from sqlalchemy import func, select, text
+from sqlalchemy.dialects.mysql import insert
+
+from app.dashboard.models import Dashboard
+from app.database.session import AsyncSessionLocal, engine
+from app.social.models import SocialUpload
+from app.user.models import SocialAccount
+
+logger = logging.getLogger(__name__)
+
+
+async def _dashboard_table_exists() -> bool:
+    """dashboard 테이블 존재 여부 확인"""
+    async with engine.connect() as conn:
+        result = await conn.execute(
+            text(
+                "SELECT COUNT(*) FROM information_schema.tables "
+                "WHERE table_schema = DATABASE() AND table_name = 'dashboard'"
+            )
+        )
+        return result.scalar() > 0
+
+
+async def _dashboard_is_empty() -> bool:
+    """dashboard 테이블 데이터 존재 여부 확인"""
+    async with AsyncSessionLocal() as session:
+        result = await session.execute(
+            select(func.count()).select_from(Dashboard)
+        )
+        return result.scalar() == 0
+
+
+async def _migrate_existing_data() -> None:
+    """
+    SocialUpload(status=completed) → Dashboard 마이그레이션.
+    INSERT IGNORE로 중복 안전하게 삽입.
+    """
+    async with AsyncSessionLocal() as session:
+        result = await session.execute(
+            select(
+                SocialUpload.user_uuid,
+                SocialUpload.platform,
+                SocialUpload.platform_video_id,
+                SocialUpload.platform_url,
+                SocialUpload.title,
+                SocialUpload.uploaded_at,
+                SocialAccount.platform_user_id,
+            )
+            .join(SocialAccount, SocialUpload.social_account_id == SocialAccount.id)
+            .where(
+                SocialUpload.status == "completed",
+                SocialUpload.platform_video_id.isnot(None),
+                SocialUpload.uploaded_at.isnot(None),
+                SocialAccount.platform_user_id.isnot(None),
+            )
+        )
+        rows = result.all()
+
+    if not rows:
+        logger.info("[DASHBOARD_MIGRATE] 마이그레이션 대상 없음")
+        return
+
+    async with AsyncSessionLocal() as session:
+        for row in rows:
+            stmt = (
+                insert(Dashboard)
+                .values(
+                    user_uuid=row.user_uuid,
+                    platform=row.platform,
+                    platform_user_id=row.platform_user_id,
+                    platform_video_id=row.platform_video_id,
+                    platform_url=row.platform_url,
+                    title=row.title,
+                    uploaded_at=row.uploaded_at,
+                )
+                .prefix_with("IGNORE")
+            )
+            await session.execute(stmt)
+        await session.commit()
+
+    logger.info(f"[DASHBOARD_MIGRATE] 마이그레이션 완료 - {len(rows)}건 삽입")
+
+
+async def init_dashboard_table() -> None:
+    """
+    dashboard 테이블 초기화 진입점.
+
+    - 테이블이 없으면 생성 후 마이그레이션
+    - 테이블이 있지만 비어있으면 마이그레이션 (DEBUG 모드에서 create_db_tables()가 빈 테이블 생성한 경우)
+    - 테이블이 있고 데이터도 있으면 스킵
+    """
+    if not await _dashboard_table_exists():
+        logger.info("[DASHBOARD_MIGRATE] dashboard 테이블 없음 - 생성 및 마이그레이션 시작")
+        async with engine.begin() as conn:
+            await conn.run_sync(
+                lambda c: Dashboard.__table__.create(c, checkfirst=True)
+            )
+        await _migrate_existing_data()
+    elif await _dashboard_is_empty():
+        logger.info("[DASHBOARD_MIGRATE] dashboard 테이블 비어있음 - 마이그레이션 시작")
+        await _migrate_existing_data()
+    else:
+        logger.info("[DASHBOARD_MIGRATE] dashboard 테이블 이미 존재 - 스킵")

app/dashboard/models.py

@@ -0,0 +1,134 @@
+"""
+Dashboard Models
+
+대시보드 전용 SQLAlchemy 모델을 정의합니다.
+"""
+
+from datetime import datetime
+from typing import Optional
+
+from sqlalchemy import BigInteger, DateTime, Index, String, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database.session import Base
+
+
+class Dashboard(Base):
+    """
+    채널별 영상 업로드 기록 테이블
+
+    YouTube 업로드 완료 시 채널 ID(platform_user_id)와 함께 기록합니다.
+    SocialUpload.social_account_id는 재연동 시 변경되므로,
+    이 테이블로 채널 기준 안정적인 영상 필터링을 제공합니다.
+
+    Attributes:
+        id: 고유 식별자 (자동 증가)
+        user_uuid: 사용자 UUID (User.user_uuid 참조)
+        platform: 플랫폼 (youtube/instagram)
+        platform_user_id: 채널 ID (재연동 후에도 불변)
+        platform_video_id: 영상 ID
+        platform_url: 영상 URL
+        title: 영상 제목
+        uploaded_at: SocialUpload 완료 시각
+        created_at: 레코드 생성 시각
+    """
+
+    __tablename__ = "dashboard"
+    __table_args__ = (
+        UniqueConstraint(
+            "platform_video_id",
+            "platform_user_id",
+            name="uq_vcu_video_channel",
+        ),
+        Index("idx_vcu_user_platform", "user_uuid", "platform_user_id"),
+        Index("idx_vcu_uploaded_at", "uploaded_at"),
+        {
+            "mysql_engine": "InnoDB",
+            "mysql_charset": "utf8mb4",
+            "mysql_collate": "utf8mb4_unicode_ci",
+        },
+    )
+
+    # ==========================================================================
+    # 기본 식별자
+    # ==========================================================================
+    id: Mapped[int] = mapped_column(
+        BigInteger,
+        primary_key=True,
+        nullable=False,
+        autoincrement=True,
+        comment="고유 식별자",
+    )
+
+    # ==========================================================================
+    # 관계 필드
+    # ==========================================================================
+    user_uuid: Mapped[str] = mapped_column(
+        String(36),
+        nullable=False,
+        comment="사용자 UUID (User.user_uuid 참조)",
+    )
+
+    # ==========================================================================
+    # 플랫폼 정보
+    # ==========================================================================
+    platform: Mapped[str] = mapped_column(
+        String(20),
+        nullable=False,
+        comment="플랫폼 (youtube/instagram)",
+    )
+
+    platform_user_id: Mapped[str] = mapped_column(
+        String(100),
+        nullable=False,
+        comment="채널 ID (재연동 후에도 불변)",
+    )
+
+    # ==========================================================================
+    # 플랫폼 결과
+    # ==========================================================================
+    platform_video_id: Mapped[str] = mapped_column(
+        String(100),
+        nullable=False,
+        comment="영상 ID",
+    )
+
+    platform_url: Mapped[Optional[str]] = mapped_column(
+        String(500),
+        nullable=True,
+        comment="영상 URL",
+    )
+
+    # ==========================================================================
+    # 메타데이터
+    # ==========================================================================
+    title: Mapped[str] = mapped_column(
+        String(200),
+        nullable=False,
+        comment="영상 제목",
+    )
+
+    # ==========================================================================
+    # 시간 정보
+    # ==========================================================================
+    uploaded_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        comment="SocialUpload 완료 시각",
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        nullable=False,
+        server_default=func.now(),
+        comment="레코드 생성 시각",
+    )
+
+    def __repr__(self) -> str:
+        return (
+            f"<Dashboard("
+            f"id={self.id}, "
+            f"platform_user_id='{self.platform_user_id}', "
+            f"platform_video_id='{self.platform_video_id}'"
+            f")>"
+        )

app/dashboard/redis_cache.py

@@ -0,0 +1,173 @@
+"""
+Redis 캐싱 유틸리티
+
+Dashboard API 성능 최적화를 위한 Redis 캐싱 기능을 제공합니다.
+YouTube Analytics API 호출 결과를 캐싱하여 중복 요청을 방지합니다.
+"""
+
+from typing import Optional
+
+from redis.asyncio import Redis
+
+from app.utils.logger import get_logger
+from config import db_settings
+
+logger = get_logger("redis_cache")
+
+# Dashboard 전용 Redis 클라이언트 (db=3 사용)
+_cache_client = Redis(
+    host=db_settings.REDIS_HOST,
+    port=db_settings.REDIS_PORT,
+    db=3,
+    decode_responses=True,
+)
+
+
+async def get_cache(key: str) -> Optional[str]:
+    """
+    Redis 캐시에서 값을 조회합니다.
+
+    Args:
+        key: 캐시 키
+
+    Returns:
+        캐시된 값 (문자열) 또는 None (캐시 미스)
+
+    Example:
+        >>> cached_data = await get_cache("dashboard:user123:2026-01-01:2026-12-31")
+        >>> if cached_data:
+        >>>     return json.loads(cached_data)
+    """
+    try:
+        logger.debug(f"[GET_CACHE] 캐시 조회 시작 - key: {key}")
+        value = await _cache_client.get(key)
+
+        if value:
+            logger.debug(f"[GET_CACHE] 캐시 HIT - key: {key}")
+        else:
+            logger.debug(f"[GET_CACHE] 캐시 MISS - key: {key}")
+
+        return value
+    except Exception as e:
+        logger.error(f"[GET_CACHE] 캐시 조회 실패 - key: {key}, error: {e}")
+        return None  # 캐시 실패 시 None 반환 (원본 데이터 조회하도록 유도)
+
+
+async def set_cache(key: str, value: str, ttl: int = 43200) -> bool:
+    """
+    Redis 캐시에 값을 저장합니다.
+
+    Args:
+        key: 캐시 키
+        value: 저장할 값 (문자열)
+        ttl: 캐시 만료 시간 (초). 기본값: 43200초 (12시간)
+
+    Returns:
+        성공 여부
+
+    Example:
+        >>> import json
+        >>> data = {"views": 1000, "likes": 50}
+        >>> await set_cache("dashboard:user123:2026-01-01:2026-12-31", json.dumps(data), ttl=3600)
+    """
+    try:
+        logger.debug(f"[SET_CACHE] 캐시 저장 시작 - key: {key}, ttl: {ttl}s")
+        await _cache_client.setex(key, ttl, value)
+        logger.debug(f"[SET_CACHE] 캐시 저장 성공 - key: {key}")
+        return True
+    except Exception as e:
+        logger.error(f"[SET_CACHE] 캐시 저장 실패 - key: {key}, error: {e}")
+        return False
+
+
+async def delete_cache(key: str) -> bool:
+    """
+    Redis 캐시에서 값을 삭제합니다.
+
+    Args:
+        key: 삭제할 캐시 키
+
+    Returns:
+        성공 여부
+
+    Example:
+        >>> await delete_cache("dashboard:user123:2026-01-01:2026-12-31")
+    """
+    try:
+        logger.debug(f"[DELETE_CACHE] 캐시 삭제 시작 - key: {key}")
+        deleted_count = await _cache_client.delete(key)
+        logger.debug(
+            f"[DELETE_CACHE] 캐시 삭제 완료 - key: {key}, deleted: {deleted_count}"
+        )
+        return deleted_count > 0
+    except Exception as e:
+        logger.error(f"[DELETE_CACHE] 캐시 삭제 실패 - key: {key}, error: {e}")
+        return False
+
+
+async def delete_cache_pattern(pattern: str) -> int:
+    """
+    패턴에 매칭되는 모든 캐시 키를 삭제합니다.
+
+    Args:
+        pattern: 삭제할 키 패턴 (예: "dashboard:user123:*")
+
+    Returns:
+        삭제된 키 개수
+
+    Example:
+        >>> # 특정 사용자의 모든 대시보드 캐시 삭제
+        >>> deleted = await delete_cache_pattern("dashboard:user123:*")
+        >>> print(f"{deleted}개의 캐시 삭제됨")
+
+    Note:
+        대량의 키 삭제 시 성능에 영향을 줄 수 있으므로 주의해서 사용하세요.
+    """
+    try:
+        logger.debug(f"[DELETE_CACHE_PATTERN] 패턴 캐시 삭제 시작 - pattern: {pattern}")
+
+        # 패턴에 매칭되는 모든 키 조회
+        keys = []
+        async for key in _cache_client.scan_iter(match=pattern):
+            keys.append(key)
+
+        if not keys:
+            logger.debug(f"[DELETE_CACHE_PATTERN] 삭제할 키 없음 - pattern: {pattern}")
+            return 0
+
+        # 모든 키 삭제
+        deleted_count = await _cache_client.delete(*keys)
+        logger.debug(
+            f"[DELETE_CACHE_PATTERN] 패턴 캐시 삭제 완료 - "
+            f"pattern: {pattern}, deleted: {deleted_count}"
+        )
+        return deleted_count
+
+    except Exception as e:
+        logger.error(
+            f"[DELETE_CACHE_PATTERN] 패턴 캐시 삭제 실패 - pattern: {pattern}, error: {e}"
+        )
+        return 0
+
+
+async def close_cache_client():
+    """
+    Redis 클라이언트 연결을 종료합니다.
+
+    애플리케이션 종료 시 호출되어야 합니다.
+    main.py의 shutdown 이벤트 핸들러에서 사용하세요.
+
+    Example:
+        >>> # main.py
+        >>> @app.on_event("shutdown")
+        >>> async def shutdown_event():
+        >>>     await close_cache_client()
+    """
+    try:
+        logger.info("[CLOSE_CACHE_CLIENT] Redis 캐시 클라이언트 종료 중...")
+        await _cache_client.close()
+        logger.info("[CLOSE_CACHE_CLIENT] Redis 캐시 클라이언트 종료 완료")
+    except Exception as e:
+        logger.error(
+            f"[CLOSE_CACHE_CLIENT] Redis 캐시 클라이언트 종료 실패 - error: {e}"
+        )

app/dashboard/schemas/__init__.py

@@ -0,0 +1,33 @@
+"""
+Dashboard Schemas
+
+Dashboard API의 요청/응답 스키마를 정의합니다.
+"""
+
+from app.dashboard.schemas.dashboard_schema import (
+    AudienceData,
+    CacheDeleteResponse,
+    ConnectedAccount,
+    ConnectedAccountsResponse,
+    ContentMetric,
+    DailyData,
+    DashboardResponse,
+    MonthlyData,
+    PlatformData,
+    PlatformMetric,
+    TopContent,
+)
+
+__all__ = [
+    "ConnectedAccount",
+    "ConnectedAccountsResponse",
+    "ContentMetric",
+    "DailyData",
+    "MonthlyData",
+    "TopContent",
+    "AudienceData",
+    "PlatformMetric",
+    "PlatformData",
+    "DashboardResponse",
+    "CacheDeleteResponse",
+]

app/dashboard/schemas/dashboard_schema.py

@@ -0,0 +1,352 @@
+"""
+Dashboard API Schemas
+
+대시보드 API의 요청/응답 Pydantic 스키마를 정의합니다.
+YouTube Analytics API 데이터를 프론트엔드에 전달하기 위한 모델입니다.
+
+사용 예시:
+    from app.dashboard.schemas import DashboardResponse, ContentMetric
+
+    # 라우터에서 response_model로 사용
+    @router.get("/dashboard/stats", response_model=DashboardResponse)
+    async def get_dashboard_stats():
+        ...
+"""
+
+from datetime import datetime
+from typing import Any, Literal, Optional
+from pydantic import BaseModel, ConfigDict, Field
+
+
+def to_camel(string: str) -> str:
+    """snake_case를 camelCase로 변환
+
+    Args:
+        string: snake_case 문자열 (예: "label_en")
+
+    Returns:
+        camelCase 문자열 (예: "labelEn")
+
+    Example:
+        >>> to_camel("content_metrics")
+        "contentMetrics"
+        >>> to_camel("this_year")
+        "thisYear"
+    """
+    components = string.split("_")
+    return components[0] + "".join(x.capitalize() for x in components[1:])
+
+
+class ContentMetric(BaseModel):
+    """KPI 지표 카드
+
+    대시보드 상단에 표시되는 핵심 성과 지표(KPI) 카드입니다.
+
+    Attributes:
+        id: 지표 고유 ID (예: "total-views", "total-engagement")
+        label: 한글 라벨 (예: "총 조회수")
+        label_en: 영문 라벨 (예: "Total Views")
+        value: 포맷팅된 값 (예: "1.2M", "8.2%")
+        trend: 증감 추세 값 (예: 12.5 → 12.5% 증가)
+        trend_direction: 추세 방향 ("up" 또는 "down")
+
+    Example:
+        >>> metric = ContentMetric(
+        ...     id="total-views",
+        ...     label="총 조회수",
+        ...     label_en="Total Views",
+        ...     value="1.2M",
+        ...     trend=12.5,
+        ...     trend_direction="up"
+        ... )
+    """
+
+    id: str
+    label: str
+    label_en: str
+    value: str
+    trend: float
+    trend_direction: Literal["up", "down", "-"] = Field(alias="trendDirection")
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class MonthlyData(BaseModel):
+    """월별 추이 데이터
+
+    전년 대비 월별 조회수 비교 데이터입니다.
+
+    Attributes:
+        month: 월 표시 (예: "1월", "2월")
+        this_year: 올해 해당 월 조회수
+        last_year: 작년 해당 월 조회수
+
+    Example:
+        >>> data = MonthlyData(
+        ...     month="1월",
+        ...     this_year=150000,
+        ...     last_year=120000
+        ... )
+    """
+
+    month: str
+    this_year: int = Field(alias="thisYear")
+    last_year: int = Field(alias="lastYear")
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class DailyData(BaseModel):
+    """일별 추이 데이터 (mode=day 전용)
+
+    최근 30일과 이전 30일의 일별 조회수 비교 데이터입니다.
+
+    Attributes:
+        date: 날짜 표시 (예: "1/18", "1/19")
+        this_period: 최근 30일 조회수
+        last_period: 이전 30일 동일 요일 조회수
+    """
+
+    date: str
+    this_period: int = Field(alias="thisPeriod")
+    last_period: int = Field(alias="lastPeriod")
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class TopContent(BaseModel):
+    """인기 영상
+
+    조회수 기준 상위 인기 영상 정보입니다.
+
+    Attributes:
+        id: YouTube 영상 ID
+        title: 영상 제목
+        thumbnail: 썸네일 이미지 URL
+        platform: 플랫폼 ("youtube" 또는 "instagram")
+        views: 포맷팅된 조회수 (예: "125K")
+        engagement: 참여율 (예: "8.2%")
+        date: 업로드 날짜 (예: "2026.01.15")
+
+    Example:
+        >>> content = TopContent(
+        ...     id="video-id-1",
+        ...     title="힐링 영상",
+        ...     thumbnail="https://i.ytimg.com/...",
+        ...     platform="youtube",
+        ...     views="125K",
+        ...     engagement="8.2%",
+        ...     date="2026.01.15"
+        ... )
+    """
+
+    id: str
+    title: str
+    thumbnail: str
+    platform: Literal["youtube", "instagram"]
+    views: str
+    engagement: str
+    date: str
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class AudienceData(BaseModel):
+    """시청자 분석 데이터
+
+    시청자의 연령대, 성별, 지역 분포 데이터입니다.
+
+    Attributes:
+        age_groups: 연령대별 비율 리스트
+            [{"label": "18-24", "percentage": 35}, ...]
+        gender: 성별 조회수
+            {"male": 45000, "female": 55000}
+        top_regions: 상위 지역 리스트
+            [{"region": "서울", "percentage": 42}, ...]
+
+    Example:
+        >>> data = AudienceData(
+        ...     age_groups=[{"label": "18-24", "percentage": 35}],
+        ...     gender={"male": 45, "female": 55},
+        ...     top_regions=[{"region": "서울", "percentage": 42}]
+        ... )
+    """
+
+    age_groups: list[dict[str, Any]] = Field(alias="ageGroups")
+    gender: dict[str, int]
+    top_regions: list[dict[str, Any]] = Field(alias="topRegions")
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class PlatformMetric(BaseModel):
+    """플랫폼별 메트릭
+
+    플랫폼별 세부 지표입니다 (예: 구독자 증가, 구독 취소).
+
+    Attributes:
+        id: 메트릭 고유 ID (예: "subscribers-gained")
+        label: 지표명 (예: "구독자 증가")
+        value: 포맷팅된 값 (예: "1.2K")
+        unit: 단위 (선택, 예: "명", "%")
+        trend: 증감 추세 값
+        trend_direction: 추세 방향 ("up" 또는 "down")
+
+    Example:
+        >>> metric = PlatformMetric(
+        ...     id="subscribers-gained",
+        ...     label="구독자 증가",
+        ...     value="1.2K",
+        ...     trend=8.5,
+        ...     trend_direction="up"
+        ... )
+    """
+
+    id: str
+    label: str
+    value: str
+    unit: Optional[str] = None
+    trend: float
+    trend_direction: Literal["up", "down", "-"] = Field(alias="trendDirection")
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class PlatformData(BaseModel):
+    """플랫폼별 데이터
+
+    플랫폼별로 그룹화된 메트릭 데이터입니다.
+
+    Attributes:
+        platform: 플랫폼 종류 ("youtube" 또는 "instagram")
+        display_name: 표시할 플랫폼 이름 (예: "YouTube")
+        metrics: 플랫폼별 메트릭 리스트
+
+    Example:
+        >>> data = PlatformData(
+        ...     platform="youtube",
+        ...     display_name="YouTube",
+        ...     metrics=[
+        ...         PlatformMetric(
+        ...             id="subscribers",
+        ...             label="구독자 증가",
+        ...             value="1.2K",
+        ...             trend=8.5,
+        ...             trend_direction="up"
+        ...         )
+        ...     ]
+        ... )
+    """
+
+    platform: Literal["youtube", "instagram"]
+    display_name: str = Field(alias="displayName")
+    metrics: list[PlatformMetric]
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class DashboardResponse(BaseModel):
+    """대시보드 전체 응답
+
+    GET /api/dashboard/stats 엔드포인트의 전체 응답 스키마입니다.
+    모든 대시보드 데이터를 포함합니다.
+
+    Attributes:
+        content_metrics: KPI 지표 카드 리스트
+        monthly_data: 월별 추이 데이터 (전년 대비)
+        top_content: 인기 영상 리스트
+        audience_data: 시청자 분석 데이터
+        platform_data: 플랫폼별 메트릭 데이터
+
+    Example:
+        >>> response = DashboardResponse(
+        ...     content_metrics=[...],
+        ...     monthly_data=[...],
+        ...     top_content=[...],
+        ...     audience_data=AudienceData(...),
+        ...     platform_data=[...]
+        ... )
+        >>> json_str = response.model_dump_json()  # JSON 직렬화
+    """
+
+    content_metrics: list[ContentMetric] = Field(alias="contentMetrics")
+    monthly_data: list[MonthlyData] = Field(default=[], alias="monthlyData")
+    daily_data: list[DailyData] = Field(default=[], alias="dailyData")
+    top_content: list[TopContent] = Field(alias="topContent")
+    audience_data: AudienceData = Field(alias="audienceData")
+    platform_data: list[PlatformData] = Field(default=[], alias="platformData")
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class ConnectedAccount(BaseModel):
+    """연결된 소셜 계정 정보
+
+    Attributes:
+        id: SocialAccount 테이블 ID (계정 선택 시 이 값을 social_account_id로 전달)
+        platform: 플랫폼 (예: "youtube")
+        platform_username: 플랫폼 사용자명 (예: "@channelname")
+        platform_user_id: 플랫폼 사용자 고유 ID
+        channel_title: YouTube 채널 제목 (platform_data에서 추출)
+        connected_at: 연동 일시
+        is_active: 활성화 상태
+    """
+
+    id: int
+    platform: str
+    platform_user_id: str
+    platform_username: Optional[str] = None
+    channel_title: Optional[str] = None
+    connected_at: datetime
+    is_active: bool
+
+    model_config = ConfigDict(
+        alias_generator=to_camel,
+        populate_by_name=True,
+    )
+
+
+class ConnectedAccountsResponse(BaseModel):
+    """연결된 소셜 계정 목록 응답
+
+    Attributes:
+        accounts: 연결된 계정 목록
+    """
+
+    accounts: list[ConnectedAccount]
+
+
+class CacheDeleteResponse(BaseModel):
+    """캐시 삭제 응답
+
+    Attributes:
+        deleted_count: 삭제된 캐시 키 개수
+        message: 처리 결과 메시지
+    """
+
+    deleted_count: int
+    message: str

app/dashboard/services/__init__.py

@@ -0,0 +1,13 @@
+"""
+Dashboard Services
+
+YouTube Analytics API 연동 및 데이터 가공 서비스를 제공합니다.
+"""
+
+from app.dashboard.services.data_processor import DataProcessor
+from app.dashboard.services.youtube_analytics import YouTubeAnalyticsService
+
+__all__ = [
+    "YouTubeAnalyticsService",
+    "DataProcessor",
+]

app/dashboard/services/data_processor.py

@@ -0,0 +1,565 @@
+"""
+YouTube Analytics 데이터 가공 프로세서
+
+YouTube Analytics API의 원본 데이터를 프론트엔드용 Pydantic 스키마로 변환합니다.
+"""
+
+from datetime import datetime, timedelta
+from typing import Any, Literal
+
+from app.dashboard.schemas import (
+    AudienceData,
+    ContentMetric,
+    DailyData,
+    DashboardResponse,
+    MonthlyData,
+    TopContent,
+)
+from app.utils.logger import get_logger
+
+logger = get_logger("dashboard")
+
+
+class DataProcessor:
+    """YouTube Analytics 데이터 가공 프로세서
+
+    YouTube Analytics API의 원본 JSON 데이터를 DashboardResponse 스키마로 변환합니다.
+    각 섹션별로 데이터 가공 로직을 분리하여 유지보수성을 향상시켰습니다.
+    """
+
+    def process(
+        self,
+        raw_data: dict[str, Any],
+        top_content: list[TopContent],
+        period_video_count: int = 0,
+        mode: Literal["day", "month"] = "month",
+        end_date: str = "",
+    ) -> DashboardResponse:
+        """YouTube Analytics API 원본 데이터를 DashboardResponse로 변환
+
+        Args:
+            raw_data: YouTube Analytics API 응답 데이터 (mode에 따라 키 구성 다름)
+                공통:
+                - kpi: KPI 메트릭 (조회수, 좋아요, 댓글, 시청시간 등)
+                - top_videos: 인기 영상 데이터
+                - demographics: 연령/성별 데이터
+                - region: 지역별 데이터
+                mode="month" 추가:
+                - trend_recent: 최근 12개월 월별 조회수
+                - trend_previous: 이전 12개월 월별 조회수
+                mode="day" 추가:
+                - trend_recent: 최근 30일 일별 조회수
+                - trend_previous: 이전 30일 일별 조회수
+            top_content: TopContent 리스트 (라우터에서 Analytics + DB lookup으로 생성)
+            period_video_count: 조회 기간 내 업로드된 영상 수 (DB에서 집계)
+            mode: 조회 모드 ("month" | "day")
+
+        Returns:
+            DashboardResponse: 프론트엔드용 대시보드 응답 스키마
+                - mode="month": monthly_data 채움, daily_data=[]
+                - mode="day": daily_data 채움, monthly_data=[]
+
+        Example:
+            >>> processor = DataProcessor()
+            >>> response = processor.process(
+            ...     raw_data={
+            ...         "kpi": {...},
+            ...         "monthly_recent": {...},
+            ...         "monthly_previous": {...},
+            ...         "top_videos": {...},
+            ...         "demographics": {...},
+            ...         "region": {...},
+            ...     },
+            ...     top_content=[TopContent(...)],
+            ...     mode="month",
+            ... )
+        """
+        logger.debug(
+            f"[DataProcessor.process] START - "
+            f"top_content_count={len(top_content)}"
+        )
+
+        # 각 섹션별 데이터 가공 (안전한 딕셔너리 접근)
+        content_metrics = self._build_content_metrics(
+            raw_data.get("kpi", {}),
+            raw_data.get("kpi_previous", {}),
+            period_video_count,
+        )
+
+        if mode == "month":
+            monthly_data = self._merge_monthly_data(
+                raw_data.get("trend_recent", {}),
+                raw_data.get("trend_previous", {}),
+            )
+            daily_data: list[DailyData] = []
+        else:  # mode == "day"
+            daily_data = self._build_daily_data(
+                raw_data.get("trend_recent", {}),
+                raw_data.get("trend_previous", {}),
+                end_date=end_date,
+            )
+            monthly_data = []
+
+        audience_data = self._build_audience_data(
+            raw_data.get("demographics", {}),
+            raw_data.get("region", {}),
+        )
+        logger.debug(
+            f"[DataProcessor.process] SUCCESS - "
+            f"mode={mode}, metrics={len(content_metrics)}, "
+            f"top_content={len(top_content)}"
+        )
+
+        return DashboardResponse(
+            content_metrics=content_metrics,
+            monthly_data=monthly_data,
+            daily_data=daily_data,
+            top_content=top_content,
+            audience_data=audience_data,
+        )
+
+    def _build_content_metrics(
+        self,
+        kpi_data: dict[str, Any],
+        kpi_previous_data: dict[str, Any],
+        period_video_count: int = 0,
+    ) -> list[ContentMetric]:
+        """KPI 데이터를 ContentMetric 리스트로 변환
+
+        Args:
+            kpi_data: 최근 기간 KPI 응답
+                rows[0] = [views, likes, comments, shares,
+                           estimatedMinutesWatched, averageViewDuration,
+                           subscribersGained]
+            kpi_previous_data: 이전 기간 KPI 응답 (증감률 계산용)
+            period_video_count: 조회 기간 내 업로드된 영상 수 (DB에서 집계)
+
+        Returns:
+            list[ContentMetric]: KPI 지표 카드 리스트 (8개)
+                순서: 조회수, 시청시간, 평균 시청시간, 신규 구독자, 좋아요, 댓글, 공유, 업로드 영상
+        """
+        logger.info(
+            f"[DataProcessor._build_content_metrics] START - "
+            f"kpi_keys={list(kpi_data.keys())}"
+        )
+
+        rows = kpi_data.get("rows", [])
+        if not rows or not rows[0]:
+            logger.warning(
+                f"[DataProcessor._build_content_metrics] NO_DATA - " f"rows={rows}"
+            )
+            return []
+
+        row = rows[0]
+        prev_rows = kpi_previous_data.get("rows", [])
+        prev_row = prev_rows[0] if prev_rows else []
+
+        def _get(r: list, i: int, default: float = 0.0) -> float:
+            return r[i] if len(r) > i else default
+
+        def _trend(recent: float, previous: float) -> tuple[float, str]:
+            pct = recent - previous
+            if pct > 0:
+                direction = "up"
+            elif pct < 0:
+                direction = "down"
+            else:
+                direction = "-"
+            return pct, direction
+
+        # 최근 기간
+        views = _get(row, 0)
+        likes = _get(row, 1)
+        comments = _get(row, 2)
+        shares = _get(row, 3)
+        estimated_minutes_watched = _get(row, 4)
+        average_view_duration = _get(row, 5)
+        subscribers_gained = _get(row, 6)
+
+        # 이전 기간
+        prev_views = _get(prev_row, 0)
+        prev_likes = _get(prev_row, 1)
+        prev_comments = _get(prev_row, 2)
+        prev_shares = _get(prev_row, 3)
+        prev_minutes_watched = _get(prev_row, 4)
+        prev_avg_duration = _get(prev_row, 5)
+        prev_subscribers = _get(prev_row, 6)
+
+        views_trend, views_dir = _trend(views, prev_views)
+        watch_trend, watch_dir = _trend(estimated_minutes_watched, prev_minutes_watched)
+        duration_trend, duration_dir = _trend(average_view_duration, prev_avg_duration)
+        subs_trend, subs_dir = _trend(subscribers_gained, prev_subscribers)
+        likes_trend, likes_dir = _trend(likes, prev_likes)
+        comments_trend, comments_dir = _trend(comments, prev_comments)
+        shares_trend, shares_dir = _trend(shares, prev_shares)
+
+        logger.info(
+            f"[DataProcessor._build_content_metrics] SUCCESS - "
+            f"views={views}({views_trend:+.1f}%), "
+            f"watch_time={estimated_minutes_watched}min({watch_trend:+.1f}%), "
+            f"subscribers={subscribers_gained}({subs_trend:+.1f}%)"
+        )
+
+        return [
+            ContentMetric(
+                id="total-views",
+                label="조회수",
+                label_en="Total Views",
+                value=self._format_number(int(views)),
+                trend=views_trend,
+                trend_direction=views_dir,
+            ),
+            ContentMetric(
+                id="total-watch-time",
+                label="시청시간",
+                label_en="Watch Time",
+                value=f"{round(estimated_minutes_watched / 60, 1)}시간",
+                trend=watch_trend,
+                trend_direction=watch_dir,
+            ),
+            ContentMetric(
+                id="avg-view-duration",
+                label="평균 시청시간",
+                label_en="Avg. View Duration",
+                value=f"{round(average_view_duration / 60)}분",
+                trend=duration_trend,
+                trend_direction=duration_dir,
+            ),
+            ContentMetric(
+                id="new-subscribers",
+                label="신규 구독자",
+                label_en="New Subscribers",
+                value=self._format_number(int(subscribers_gained)),
+                trend=subs_trend,
+                trend_direction=subs_dir,
+            ),
+            ContentMetric(
+                id="likes",
+                label="좋아요",
+                label_en="Likes",
+                value=self._format_number(int(likes)),
+                trend=likes_trend,
+                trend_direction=likes_dir,
+            ),
+            ContentMetric(
+                id="comments",
+                label="댓글",
+                label_en="Comments",
+                value=self._format_number(int(comments)),
+                trend=comments_trend,
+                trend_direction=comments_dir,
+            ),
+            ContentMetric(
+                id="shares",
+                label="공유",
+                label_en="Shares",
+                value=self._format_number(int(shares)),
+                trend=shares_trend,
+                trend_direction=shares_dir,
+            ),
+            ContentMetric(
+                id="uploaded-videos",
+                label="업로드 영상",
+                label_en="Uploaded Videos",
+                value=str(period_video_count),
+                trend=0.0,
+                trend_direction="up",
+            ),
+        ]
+
+    def _merge_monthly_data(
+        self,
+        data_recent: dict[str, Any],
+        data_previous: dict[str, Any],
+    ) -> list[MonthlyData]:
+        """최근 12개월과 이전 12개월의 월별 데이터를 병합
+
+        최근 12개월 대비 이전 12개월의 월별 조회수 비교 차트를 위한 데이터를 생성합니다.
+        실제 API 응답의 월 데이터를 기준으로 매핑합니다.
+
+        Args:
+            data_recent: 최근 12개월 월별 조회수 데이터
+                rows = [["2026-01", 150000], ["2026-02", 180000], ...]
+            data_previous: 이전 12개월 월별 조회수 데이터
+                rows = [["2025-01", 120000], ["2025-02", 140000], ...]
+
+        Returns:
+            list[MonthlyData]: 월별 비교 데이터 (최대 12개)
+        """
+        logger.debug("[DataProcessor._merge_monthly_data] START")
+
+        rows_recent = data_recent.get("rows", [])
+        rows_previous = data_previous.get("rows", [])
+
+        # 월별 맵 생성: {"2025-02": 150000, "2025-03": 180000}
+        map_recent = {row[0]: row[1] for row in rows_recent if len(row) >= 2}
+        map_previous = {row[0]: row[1] for row in rows_previous if len(row) >= 2}
+
+        # 최근 기간의 월 키만 기준으로 정렬 (24개 합집합 방지)
+        # 각 월의 이전 연도 키는 1년 전으로 계산: "2025-02" → "2024-02"
+        recent_months = sorted(map_recent.keys())
+
+        # 월별 데이터 생성
+        result = []
+        for month_key in recent_months:
+            year, month = month_key.split("-")
+            month_num = int(month)
+            month_label = f"{month_num}월"
+
+            # 이전 연도 동일 월: "2025-02" → "2024-02"
+            prev_year_key = f"{int(year) - 1}-{month}"
+
+            result.append(
+                MonthlyData(
+                    month=month_label,
+                    this_year=map_recent.get(month_key, 0),
+                    last_year=map_previous.get(prev_year_key, 0),
+                )
+            )
+
+        logger.debug(
+            f"[DataProcessor._merge_monthly_data] SUCCESS - count={len(result)}"
+        )
+        return result
+
+    def _build_daily_data(
+        self,
+        data_recent: dict[str, Any],
+        data_previous: dict[str, Any],
+        end_date: str = "",
+        num_days: int = 30,
+    ) -> list[DailyData]:
+        """최근 30일과 이전 30일의 일별 데이터를 병합
+
+        end_date 기준 num_days개 날짜를 직접 생성하여 YouTube API 응답에
+        해당 날짜 row가 없어도 0으로 채웁니다 (X축 누락 방지).
+
+        Args:
+            data_recent: 최근 30일 일별 조회수 데이터
+                rows = [["2026-01-20", 5000], ["2026-01-21", 6200], ...]
+            data_previous: 이전 30일 일별 조회수 데이터
+                rows = [["2025-12-21", 4500], ["2025-12-22", 5100], ...]
+            end_date: 최근 기간의 마지막 날 (YYYY-MM-DD). 미전달 시 rows 마지막 날 사용
+            num_days: 표시할 일수 (기본 30)
+
+        Returns:
+            list[DailyData]: 일별 비교 데이터 (num_days개, 데이터 없는 날은 0)
+        """
+        logger.debug("[DataProcessor._build_daily_data] START")
+
+        rows_recent = data_recent.get("rows", [])
+        rows_previous = data_previous.get("rows", [])
+
+        # 날짜 → 조회수 맵
+        map_recent = {row[0]: row[1] for row in rows_recent if len(row) >= 2}
+        map_previous = {row[0]: row[1] for row in rows_previous if len(row) >= 2}
+
+        # end_date 결정: 전달된 값 우선, 없으면 rows 마지막 날짜 사용
+        if end_date:
+            end_dt = datetime.strptime(end_date, "%Y-%m-%d").date()
+        elif rows_recent:
+            end_dt = datetime.strptime(rows_recent[-1][0], "%Y-%m-%d").date()
+        else:
+            logger.warning(
+                "[DataProcessor._build_daily_data] NO_DATA - rows_recent 비어있음"
+            )
+            return []
+
+        start_dt = end_dt - timedelta(days=num_days - 1)
+
+        # 날짜 범위를 직접 생성하여 누락된 날짜도 0으로 채움
+        result = []
+        current = start_dt
+        while current <= end_dt:
+            date_str = current.strftime("%Y-%m-%d")
+            date_label = f"{current.month}/{current.day}"
+
+            this_views = map_recent.get(date_str, 0)
+
+            # 이전 기간: 동일 인덱스 날짜 (current - 30일)
+            prev_date_str = (current - timedelta(days=num_days)).strftime("%Y-%m-%d")
+            last_views = map_previous.get(prev_date_str, 0)
+
+            result.append(
+                DailyData(
+                    date=date_label,
+                    this_period=int(this_views),
+                    last_period=int(last_views),
+                )
+            )
+            current += timedelta(days=1)
+
+        logger.debug(f"[DataProcessor._build_daily_data] SUCCESS - count={len(result)}")
+        return result
+
+    def _build_audience_data(
+        self,
+        demographics_data: dict[str, Any],
+        geography_data: dict[str, Any],
+    ) -> AudienceData:
+        """시청자 분석 데이터 생성
+
+        연령대별, 성별, 지역별 시청자 분포를 분석합니다.
+
+        Args:
+            demographics_data: 연령/성별 API 응답
+                rows = [["age18-24", "male", 45000], ["age18-24", "female", 55000], ...]
+            geography_data: 지역별 API 응답
+                rows = [["KR", 1000000], ["US", 500000], ...]
+
+        Returns:
+            AudienceData: 시청자 분석 데이터
+                - age_groups: 연령대별 비율
+                - gender: 성별 조회수
+                - top_regions: 상위 지역 (5개)
+        """
+        logger.debug("[DataProcessor._build_audience_data] START")
+
+        # === 연령/성별 데이터 처리 ===
+        demo_rows = demographics_data.get("rows", [])
+
+        age_map: dict[str, int] = {}
+        gender_map = {"male": 0, "female": 0}
+
+        for row in demo_rows:
+            if len(row) < 3:
+                continue
+
+            age_group = row[0]  # "age18-24"
+            gender = row[1]  # "male" or "female"
+            views = row[2]
+
+            # 연령대별 집계 (age18-24 → 18-24)
+            age_label = age_group.replace("age", "")
+            age_map[age_label] = age_map.get(age_label, 0) + views
+
+            # 성별 집계
+            if gender in gender_map:
+                gender_map[gender] += views
+
+        # 연령대별 비율 계산
+        total_demo_views = sum(age_map.values())
+        age_groups = [
+            {
+                "label": age,
+                "percentage": int(
+                    (count / total_demo_views * 100) if total_demo_views > 0 else 0
+                ),
+            }
+            for age, count in sorted(age_map.items())
+        ]
+
+        # === 지역 데이터 처리 ===
+        geo_rows = geography_data.get("rows", [])
+        total_geo_views = sum(row[1] for row in geo_rows if len(row) >= 2)
+
+        top_regions = [
+            {
+                "region": self._translate_country_code(row[0]),
+                "percentage": int(
+                    (row[1] / total_geo_views * 100) if total_geo_views > 0 else 0
+                ),
+            }
+            for row in geo_rows[:5]  # 상위 5개
+            if len(row) >= 2
+        ]
+
+        logger.debug(
+            f"[DataProcessor._build_audience_data] SUCCESS - "
+            f"age_groups={len(age_groups)}, regions={len(top_regions)}"
+        )
+
+        return AudienceData(
+            age_groups=age_groups,
+            gender=gender_map,
+            top_regions=top_regions,
+        )
+
+    @staticmethod
+    def _format_number(num: int) -> str:
+        """숫자 포맷팅 (1234567 → "1.2M")
+
+        조회수, 구독자 수 등 큰 숫자를 읽기 쉽게 포맷팅합니다.
+
+        Args:
+            num: 원본 숫자
+
+        Returns:
+            str: 포맷팅된 문자열
+                - 1,000,000 이상: "1.2M"
+                - 1,000 이상: "12.5K"
+                - 1,000 미만: "123"
+
+        Example:
+            >>> _format_number(1234567)
+            "1.2M"
+            >>> _format_number(12345)
+            "12.3K"
+            >>> _format_number(123)
+            "123"
+        """
+        if num >= 1_000_000:
+            return f"{num / 1_000_000:.1f}M"
+        elif num >= 1_000:
+            return f"{num / 1_000:.1f}K"
+        else:
+            return str(num)
+
+    @staticmethod
+    def _format_duration(seconds: int) -> str:
+        """초를 M:SS 형식으로 변환 (평균 시청 시간 표시용)
+
+        Args:
+            seconds: 초 단위 시간
+
+        Returns:
+            str: M:SS 형식 문자열
+                - 204초 → "3:24"
+                - 65초  → "1:05"
+                - 45초  → "0:45"
+        """
+        minutes = seconds // 60
+        secs = seconds % 60
+        return f"{minutes}:{secs:02d}"
+
+    @staticmethod
+    def _translate_country_code(code: str) -> str:
+        """국가 코드를 한국어로 변환
+
+        ISO 3166-1 alpha-2 국가 코드를 한국어 국가명으로 변환합니다.
+
+        Args:
+            code: ISO 3166-1 alpha-2 국가 코드 (예: "KR", "US")
+
+        Returns:
+            str: 한국어 국가명 (매핑되지 않은 경우 원본 코드 반환)
+
+        Example:
+            >>> _translate_country_code("KR")
+            "대한민국"
+            >>> _translate_country_code("US")
+            "미국"
+        """
+        country_map = {
+            "KR": "대한민국",
+            "US": "미국",
+            "JP": "일본",
+            "CN": "중국",
+            "GB": "영국",
+            "DE": "독일",
+            "FR": "프랑스",
+            "CA": "캐나다",
+            "AU": "호주",
+            "IN": "인도",
+            "ID": "인도네시아",
+            "TH": "태국",
+            "VN": "베트남",
+            "PH": "필리핀",
+            "MY": "말레이시아",
+            "SG": "싱가포르",
+            "TW": "대만",
+            "HK": "홍콩",
+            "BR": "브라질",
+            "MX": "멕시코",
+        }
+        return country_map.get(code, code)

app/dashboard/services/youtube_analytics.py

@@ -0,0 +1,497 @@
+"""
+YouTube Analytics API 서비스
+
+YouTube Analytics API v2를 호출하여 채널 및 영상 통계를 조회합니다.
+"""
+
+import asyncio
+from datetime import datetime, timedelta
+from typing import Any, Literal
+
+import httpx
+
+from app.dashboard.exceptions import (
+    YouTubeAPIError,
+    YouTubeAuthError,
+    YouTubeQuotaExceededError,
+)
+from app.utils.logger import get_logger
+
+logger = get_logger("dashboard")
+
+
+class YouTubeAnalyticsService:
+    """YouTube Analytics API 호출 서비스
+
+    YouTube Analytics API v2를 사용하여 채널 통계, 영상 성과,
+    시청자 분석 데이터를 조회합니다.
+
+    API 문서:
+        https://developers.google.com/youtube/analytics/reference
+    """
+
+    BASE_URL = "https://youtubeanalytics.googleapis.com/v2/reports"
+
+    async def fetch_all_metrics(
+        self,
+        video_ids: list[str],
+        start_date: str,
+        end_date: str,
+        access_token: str,
+        mode: Literal["day", "month"] = "month",
+        kpi_end_date: str = "",
+    ) -> dict[str, Any]:
+        """YouTube Analytics API 호출을 병렬로 실행
+
+        Args:
+            video_ids: YouTube 영상 ID 리스트 (최대 30개)
+            start_date: 조회 시작일 (YYYY-MM-DD)
+            end_date: 조회 종료일 (YYYY-MM-DD)
+            access_token: YouTube OAuth 2.0 액세스 토큰
+
+        Returns:
+            dict[str, Any]: API 응답 데이터
+                - kpi: KPI 메트릭 (조회수, 좋아요, 댓글 등)
+                - monthly_recent: 최근 12개월 월별 조회수
+                - monthly_previous: 이전 12개월 월별 조회수
+                - top_videos: 인기 영상 TOP 4
+                - demographics: 연령/성별 분포
+                - region: 지역별 조회수
+
+        Raises:
+            YouTubeAPIError: API 호출 실패
+            YouTubeQuotaExceededError: 할당량 초과
+            YouTubeAuthError: 인증 실패
+
+        Example:
+            >>> service = YouTubeAnalyticsService()
+            >>> data = await service.fetch_all_metrics(
+            ...     video_ids=["dQw4w9WgXcQ", "jNQXAC9IVRw"],
+            ...     start_date="2026-01-01",
+            ...     end_date="2026-12-31",
+            ...     access_token="ya29.a0..."
+            ... )
+        """
+        logger.debug(
+            f"[1/6] YouTube Analytics API 병렬 호출 시작 - "
+            f"video_count={len(video_ids)}, period={start_date}~{end_date}, mode={mode}"
+        )
+
+        end_dt = datetime.strptime(end_date, "%Y-%m-%d")
+        # kpi_end_date: KPI/top_videos/demographics/region 호출에 사용
+        # month 모드에서는 현재 월 전체 데이터를 포함하기 위해 end_date(YYYY-MM-01)보다 늦은 날짜 사용
+        # day 모드 또는 미전달 시 end_date와 동일
+        _kpi_end = kpi_end_date if kpi_end_date else end_date
+
+        if mode == "month":
+            # 월별 차트: 라우터에서 이미 YYYY-MM-01 형식으로 계산된 날짜 그대로 사용
+            # recent: start_date ~ end_date (ex. 2025-03-01 ~ 2026-02-01)
+            # previous: 1년 전 동일 기간 (ex. 2024-03-01 ~ 2025-02-01)
+            recent_start = start_date
+            recent_end = end_date
+            previous_start = f"{int(start_date[:4]) - 1}{start_date[4:]}"
+            previous_end = f"{int(end_date[:4]) - 1}{end_date[4:]}"
+            # KPI 이전 기간: _kpi_end 기준 1년 전 (ex. 2026-02-22 → 2025-02-22)
+            previous_kpi_end = f"{int(_kpi_end[:4]) - 1}{_kpi_end[4:]}"
+            logger.debug(
+                f"[월별 데이터] 최근 12개월: {recent_start}~{recent_end}, "
+                f"이전 12개월: {previous_start}~{previous_end}, "
+                f"KPI 조회 종료일: {_kpi_end}"
+            )
+        else:
+            # 일별 차트: end_date 기준 최근 30일 / 이전 30일
+            day_recent_end = end_date
+            day_recent_start = (end_dt - timedelta(days=29)).strftime("%Y-%m-%d")
+            day_previous_end = (end_dt - timedelta(days=30)).strftime("%Y-%m-%d")
+            day_previous_start = (end_dt - timedelta(days=59)).strftime("%Y-%m-%d")
+            logger.debug(
+                f"[일별 데이터] 최근 30일: {day_recent_start}~{day_recent_end}, "
+                f"이전 30일: {day_previous_start}~{day_previous_end}"
+            )
+
+        # 7개 API 호출 태스크 생성 (mode별 선택적)
+        # [0] KPI(최근), [1] KPI(이전), [2] 추이(최근), [3] 추이(이전), [4] 인기영상, [5] 인구통계, [6] 지역
+        # mode=month: [2][3] = 월별 데이터 (YYYY-MM-01 형식 필요)
+        # mode=day:   [2][3] = 일별 데이터
+        if mode == "month":
+            tasks = [
+                self._fetch_kpi(video_ids, start_date, _kpi_end, access_token),
+                self._fetch_kpi(video_ids, previous_start, previous_kpi_end, access_token),
+                self._fetch_monthly_data(video_ids, recent_start, recent_end, access_token),
+                self._fetch_monthly_data(video_ids, previous_start, previous_end, access_token),
+                self._fetch_top_videos(video_ids, start_date, _kpi_end, access_token),
+                self._fetch_demographics(start_date, _kpi_end, access_token),
+                self._fetch_region(video_ids, start_date, _kpi_end, access_token),
+            ]
+        else:  # mode == "day"
+            tasks = [
+                self._fetch_kpi(video_ids, start_date, end_date, access_token),
+                self._fetch_kpi(video_ids, day_previous_start, day_previous_end, access_token),
+                self._fetch_daily_data(video_ids, day_recent_start, day_recent_end, access_token),
+                self._fetch_daily_data(video_ids, day_previous_start, day_previous_end, access_token),
+                self._fetch_top_videos(video_ids, start_date, end_date, access_token),
+                self._fetch_demographics(start_date, end_date, access_token),
+                self._fetch_region(video_ids, start_date, end_date, access_token),
+            ]
+
+        # 병렬 실행
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        # 에러 체크 (YouTubeAuthError, YouTubeQuotaExceededError는 원형 그대로 전파)
+        for i, result in enumerate(results):
+            if isinstance(result, Exception):
+                logger.error(
+                    f"[YouTubeAnalyticsService] API 호출 {i+1}/7 실패: {result.__class__.__name__}"
+                )
+                if isinstance(result, (YouTubeAuthError, YouTubeQuotaExceededError)):
+                    raise result
+                raise YouTubeAPIError(f"데이터 조회 실패: {result.__class__.__name__}")
+
+        logger.debug(
+            f"[7/7] YouTube Analytics API 병렬 호출 완료 - mode={mode}, 성공률 100%"
+        )
+
+        # 각 API 호출 결과 디버그 로깅
+        labels = [
+            "kpi",
+            "kpi_previous",
+            "trend_recent",
+            "trend_previous",
+            "top_videos",
+            "demographics",
+            "region",
+        ]
+        for label, result in zip(labels, results):
+            rows = result.get("rows") if isinstance(result, dict) else None
+            row_count = len(rows) if rows else 0
+            preview = rows[:2] if rows else []
+            logger.debug(
+                f"[fetch_all_metrics] {label}: row_count={row_count}, preview={preview}"
+            )
+
+        return {
+            "kpi": results[0],
+            "kpi_previous": results[1],
+            "trend_recent": results[2],
+            "trend_previous": results[3],
+            "top_videos": results[4],
+            "demographics": results[5],
+            "region": results[6],
+        }
+
+    async def _fetch_kpi(
+        self,
+        video_ids: list[str],
+        start_date: str,
+        end_date: str,
+        access_token: str,
+    ) -> dict[str, Any]:
+        """전체 KPI 메트릭 조회 (contentMetrics용)
+
+        조회수, 좋아요, 댓글, 공유, 시청 시간, 구독자 증감 등
+        핵심 성과 지표를 조회합니다.
+
+        Args:
+            video_ids: YouTube 영상 ID 리스트
+            start_date: 조회 시작일 (YYYY-MM-DD)
+            end_date: 조회 종료일 (YYYY-MM-DD)
+            access_token: OAuth 2.0 액세스 토큰
+
+        Returns:
+            dict[str, Any]: YouTube Analytics API 응답
+                rows[0] = [views, likes, comments, shares,
+                           estimatedMinutesWatched, averageViewDuration,
+                           subscribersGained]
+
+        Note:
+            annotationClickThroughRate는 2019년 annotations 기능 제거로 deprecated.
+        """
+        logger.debug(
+            f"[YouTubeAnalyticsService._fetch_kpi] START - video_count={len(video_ids)}"
+        )
+
+        params = {
+            "ids": "channel==MINE",
+            "startDate": start_date,
+            "endDate": end_date,
+            "metrics": "views,likes,comments,shares,estimatedMinutesWatched,averageViewDuration,subscribersGained",
+            # "filters": f"video=={','.join(video_ids)}",
+        }
+
+        result = await self._call_api(params, access_token)
+        logger.debug("[YouTubeAnalyticsService._fetch_kpi] SUCCESS")
+        return result
+
+    async def _fetch_monthly_data(
+        self,
+        video_ids: list[str],
+        start_date: str,
+        end_date: str,
+        access_token: str,
+    ) -> dict[str, Any]:
+        """월별 조회수 데이터 조회
+
+        지정된 기간의 월별 조회수를 조회합니다.
+        최근 12개월과 이전 12개월을 각각 조회하여 비교합니다.
+
+        Args:
+            video_ids: YouTube 영상 ID 리스트
+            start_date: 조회 시작일 (YYYY-MM-DD)
+            end_date: 조회 종료일 (YYYY-MM-DD)
+            access_token: OAuth 2.0 액세스 토큰
+
+        Returns:
+            dict[str, Any]: YouTube Analytics API 응답
+                rows = [["2026-01", 150000], ["2026-02", 180000], ...]
+        """
+        logger.debug(
+            f"[YouTubeAnalyticsService._fetch_monthly_data] START - "
+            f"period={start_date}~{end_date}"
+        )
+
+        params = {
+            "ids": "channel==MINE",
+            "startDate": start_date,
+            "endDate": end_date,
+            "dimensions": "month",
+            "metrics": "views",
+            # "filters": f"video=={','.join(video_ids)}",
+            "sort": "month",
+        }
+
+        result = await self._call_api(params, access_token)
+        logger.debug(
+            f"[YouTubeAnalyticsService._fetch_monthly_data] SUCCESS - "
+            f"period={start_date}~{end_date}"
+        )
+        return result
+
+    async def _fetch_daily_data(
+        self,
+        video_ids: list[str],
+        start_date: str,
+        end_date: str,
+        access_token: str,
+    ) -> dict[str, Any]:
+        """일별 조회수 데이터 조회
+
+        지정된 기간의 일별 조회수를 조회합니다.
+        최근 30일과 이전 30일을 각각 조회하여 비교합니다.
+
+        Args:
+            video_ids: YouTube 영상 ID 리스트
+            start_date: 조회 시작일 (YYYY-MM-DD)
+            end_date: 조회 종료일 (YYYY-MM-DD)
+            access_token: OAuth 2.0 액세스 토큰
+
+        Returns:
+            dict[str, Any]: YouTube Analytics API 응답
+                rows = [["2026-01-18", 5000], ["2026-01-19", 6200], ...]
+        """
+        logger.debug(
+            f"[YouTubeAnalyticsService._fetch_daily_data] START - "
+            f"period={start_date}~{end_date}"
+        )
+
+        params = {
+            "ids": "channel==MINE",
+            "startDate": start_date,
+            "endDate": end_date,
+            "dimensions": "day",
+            "metrics": "views",
+            # "filters": f"video=={','.join(video_ids)}",
+            "sort": "day",
+        }
+
+        result = await self._call_api(params, access_token)
+        logger.debug(
+            f"[YouTubeAnalyticsService._fetch_daily_data] SUCCESS - "
+            f"period={start_date}~{end_date}"
+        )
+        return result
+
+    async def _fetch_top_videos(
+        self,
+        video_ids: list[str],
+        start_date: str,
+        end_date: str,
+        access_token: str,
+    ) -> dict[str, Any]:
+        """영상별 조회수 조회 (topContent용)
+
+        조회수 기준 상위 4개 영상의 성과 데이터를 조회합니다.
+
+        Args:
+            video_ids: YouTube 영상 ID 리스트
+            start_date: 조회 시작일 (YYYY-MM-DD)
+            end_date: 조회 종료일 (YYYY-MM-DD)
+            access_token: OAuth 2.0 액세스 토큰
+
+        Returns:
+            dict[str, Any]: YouTube Analytics API 응답
+                rows = [["video_id", views, likes, comments], ...]
+                조회수 내림차순으로 정렬된 상위 4개 영상
+        """
+        logger.debug("[YouTubeAnalyticsService._fetch_top_videos] START")
+
+        params = {
+            "ids": "channel==MINE",
+            "startDate": start_date,
+            "endDate": end_date,
+            "dimensions": "video",
+            "metrics": "views,likes,comments",
+            # "filters": f"video=={','.join(video_ids)}",
+            "sort": "-views",
+            "maxResults": "4",
+        }
+
+        result = await self._call_api(params, access_token)
+        logger.debug("[YouTubeAnalyticsService._fetch_top_videos] SUCCESS")
+        return result
+
+    async def _fetch_demographics(
+        self,
+        start_date: str,
+        end_date: str,
+        access_token: str,
+    ) -> dict[str, Any]:
+        """연령/성별 분포 조회 (채널 전체 기준)
+
+        시청자의 연령대별, 성별 시청 비율을 조회합니다.
+
+        Note:
+            YouTube Analytics API 제약: ageGroup/gender 차원은 video 필터와 혼용 불가.
+            채널 전체 시청자 기준 데이터를 반환합니다.
+
+        Args:
+            start_date: 조회 시작일 (YYYY-MM-DD)
+            end_date: 조회 종료일 (YYYY-MM-DD)
+            access_token: OAuth 2.0 액세스 토큰
+
+        Returns:
+            dict[str, Any]: YouTube Analytics API 응답
+                rows = [["age18-24", "female", 45.5], ["age18-24", "male", 32.1], ...]
+        """
+        logger.debug("[YouTubeAnalyticsService._fetch_demographics] START")
+
+        # Demographics 보고서는 video 필터 미지원 → 채널 전체 기준 데이터
+        # 지원 filters: country, province, continent, subContinent, liveOrOnDemand, subscribedStatus
+        params = {
+            "ids": "channel==MINE",
+            "startDate": start_date,
+            "endDate": end_date,
+            "dimensions": "ageGroup,gender",
+            "metrics": "viewerPercentage",
+        }
+
+        result = await self._call_api(params, access_token)
+        logger.debug("[YouTubeAnalyticsService._fetch_demographics] SUCCESS")
+        return result
+
+    async def _fetch_region(
+        self,
+        video_ids: list[str],
+        start_date: str,
+        end_date: str,
+        access_token: str,
+    ) -> dict[str, Any]:
+        """지역별 조회수 조회
+
+        지역별 조회수 분포를 조회합니다 (상위 5개).
+
+        Args:
+            video_ids: YouTube 영상 ID 리스트
+            start_date: 조회 시작일 (YYYY-MM-DD)
+            end_date: 조회 종료일 (YYYY-MM-DD)
+            access_token: OAuth 2.0 액세스 토큰
+
+        Returns:
+            dict[str, Any]: YouTube Analytics API 응답
+                rows = [["KR", 1000000], ["US", 500000], ...]
+                조회수 내림차순으로 정렬된 상위 5개 국가
+        """
+        logger.debug("[YouTubeAnalyticsService._fetch_country] START")
+
+        params = {
+            "ids": "channel==MINE",
+            "startDate": start_date,
+            "endDate": end_date,
+            "dimensions": "country",
+            "metrics": "views",
+            # "filters": f"video=={','.join(video_ids)}",
+            "sort": "-views",
+            "maxResults": "5",
+        }
+
+        result = await self._call_api(params, access_token)
+        logger.debug("[YouTubeAnalyticsService._fetch_region] SUCCESS")
+        return result
+
+    async def _call_api(
+        self,
+        params: dict[str, str],
+        access_token: str,
+    ) -> dict[str, Any]:
+        """YouTube Analytics API 호출 공통 로직
+
+        모든 API 호출에 공통적으로 사용되는 HTTP 요청 로직입니다.
+        인증 헤더 추가, 에러 처리, 응답 파싱을 담당합니다.
+
+        Args:
+            params: API 요청 파라미터 (dimensions, metrics, filters 등)
+            access_token: OAuth 2.0 액세스 토큰
+
+        Returns:
+            dict[str, Any]: YouTube Analytics API JSON 응답
+
+        Raises:
+            YouTubeQuotaExceededError: 할당량 초과 (429)
+            YouTubeAuthError: 인증 실패 (401, 403)
+            YouTubeAPIError: 기타 API 오류
+
+        Note:
+            - 타임아웃: 30초
+            - 할당량 초과 시 자동으로 YouTubeQuotaExceededError 발생
+            - 인증 실패 시 자동으로 YouTubeAuthError 발생
+        """
+        headers = {"Authorization": f"Bearer {access_token}"}
+
+        try:
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                response = await client.get(
+                    self.BASE_URL,
+                    params=params,
+                    headers=headers,
+                )
+
+                # 할당량 초과 체크
+                if response.status_code == 429:
+                    logger.warning("[YouTubeAnalyticsService._call_api] QUOTA_EXCEEDED")
+                    raise YouTubeQuotaExceededError()
+
+                # 인증 실패 체크
+                if response.status_code in (401, 403):
+                    logger.warning(
+                        f"[YouTubeAnalyticsService._call_api] AUTH_FAILED - status={response.status_code}"
+                    )
+                    raise YouTubeAuthError(f"YouTube 인증 실패: {response.status_code}")
+
+                # HTTP 에러 체크
+                response.raise_for_status()
+
+                return response.json()
+
+        except (YouTubeAuthError, YouTubeQuotaExceededError):
+            raise  # 이미 처리된 예외는 그대로 전파
+        except httpx.HTTPStatusError as e:
+            logger.error(
+                f"[YouTubeAnalyticsService._call_api] HTTP_ERROR - "
+                f"status={e.response.status_code}, body={e.response.text[:500]}"
+            )
+            raise YouTubeAPIError(f"HTTP {e.response.status_code}")
+        except httpx.RequestError as e:
+            logger.error(f"[YouTubeAnalyticsService._call_api] REQUEST_ERROR - {e}")
+            raise YouTubeAPIError(f"네트워크 오류: {e}")
+        except Exception as e:
+            logger.error(f"[YouTubeAnalyticsService._call_api] UNEXPECTED_ERROR - {e}")
+            raise YouTubeAPIError(f"알 수 없는 오류: {e}")

app/dashboard/tasks.py

@@ -0,0 +1,71 @@
+"""
+Dashboard Background Tasks
+
+업로드 완료 시 Dashboard 테이블에 레코드를 삽입하는 백그라운드 태스크입니다.
+"""
+
+import logging
+
+from sqlalchemy import select
+from sqlalchemy.dialects.mysql import insert
+
+from app.dashboard.models import Dashboard
+from app.database.session import BackgroundSessionLocal
+from app.social.models import SocialUpload
+from app.user.models import SocialAccount
+
+logger = logging.getLogger(__name__)
+
+
+async def insert_dashboard(upload_id: int) -> None:
+    """
+    Dashboard 레코드 삽입
+
+    SocialUpload(id=upload_id) 완료 데이터를 DB에서 조회하여 Dashboard에 삽입합니다.
+    UniqueConstraint(platform_video_id, platform_user_id) 충돌 시 스킵(INSERT IGNORE).
+    """
+    try:
+        async with BackgroundSessionLocal() as session:
+            result = await session.execute(
+                select(
+                    SocialUpload.user_uuid,
+                    SocialUpload.platform,
+                    SocialUpload.platform_video_id,
+                    SocialUpload.platform_url,
+                    SocialUpload.title,
+                    SocialUpload.uploaded_at,
+                    SocialAccount.platform_user_id,
+                )
+                .join(SocialAccount, SocialUpload.social_account_id == SocialAccount.id)
+                .where(SocialUpload.id == upload_id)
+            )
+            row = result.one_or_none()
+
+            if not row:
+                logger.warning(f"[dashboard] upload_id={upload_id} 데이터 없음")
+                return
+
+            stmt = (
+                insert(Dashboard)
+                .values(
+                    user_uuid=row.user_uuid,
+                    platform=row.platform,
+                    platform_user_id=row.platform_user_id,
+                    platform_video_id=row.platform_video_id,
+                    platform_url=row.platform_url,
+                    title=row.title,
+                    uploaded_at=row.uploaded_at,
+                )
+                .prefix_with("IGNORE")
+            )
+            await session.execute(stmt)
+            await session.commit()
+
+        logger.info(
+            f"[dashboard] 삽입 완료 - "
+            f"upload_id={upload_id}, platform_video_id={row.platform_video_id}"
+        )
+    except Exception as e:
+        logger.error(
+            f"[dashboard] 삽입 실패 - upload_id={upload_id}, error={e}"
+        )

app/database/session.py

@@ -79,6 +79,7 @@ async def create_db_tables():
     from app.video.models import Video  # noqa: F401
     from app.sns.models import SNSUploadTask  # noqa: F401
     from app.social.models import SocialUpload  # noqa: F401
+    from app.dashboard.models import Dashboard  # noqa: F401
 
     # 생성할 테이블 목록
     tables_to_create = [
@@ -94,6 +95,7 @@ async def create_db_tables():
         SNSUploadTask.__table__,
         SocialUpload.__table__,
         MarketingIntel.__table__,
+        Dashboard.__table__,
     ]
 
     logger.info("Creating database tables...")
@@ -126,7 +128,9 @@ async def get_session() -> AsyncGenerator[AsyncSession, None]:
         try:
             yield session
         except Exception as e:
+            import traceback
             await session.rollback()
+            logger.error(traceback.format_exc())
             logger.error(
                 f"[get_session] ROLLBACK - error: {type(e).__name__}: {e}, "
                 f"duration: {(time.perf_counter() - start_time)*1000:.1f}ms"

app/lyric/api/routers/v1/lyric.py

@@ -292,7 +292,6 @@ async def generate_lyric(
             "promotional_expression_example" : promotional_expressions[request_body.language],
             "timing_rules" : timing_rules["60s"], # 아직은 선택지 하나
         }
-        estimated_prompt = lyric_prompt.build_prompt(lyric_input_data) 
 
         step1_elapsed = (time.perf_counter() - step1_start) * 1000
         #logger.debug(f"[generate_lyric] Step 1 완료 - 프롬프트 {len(prompt)}자 ({step1_elapsed:.1f}ms)")

app/social/api/routers/v1/__init__.py

@@ -4,5 +4,5 @@ Social API Routers v1
 
 from app.social.api.routers.v1.oauth import router as oauth_router
 from app.social.api.routers.v1.upload import router as upload_router
-
-__all__ = ["oauth_router", "upload_router"]
+from app.social.api.routers.v1.seo import router as seo_router
+__all__ = ["oauth_router", "upload_router", "seo_router"]

app/social/api/routers/v1/seo.py

@@ -0,0 +1,131 @@
+
+import logging, json
+
+from redis.asyncio import Redis
+
+from config import social_oauth_settings, db_settings
+from app.social.constants import YOUTUBE_SEO_HASH
+from sqlalchemy import select, func
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.social.schemas import (
+    YoutubeDescriptionRequest,
+    YoutubeDescriptionResponse, 
+)
+
+from app.database.session import get_session
+from app.user.dependencies import get_current_user
+from app.user.models import User
+from app.home.models import Project, MarketingIntel
+from fastapi import APIRouter, BackgroundTasks, Depends, Query
+from fastapi import HTTPException, status
+from app.utils.prompts.prompts import yt_upload_prompt
+from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError
+
+redis_seo_client = Redis(
+    host=db_settings.REDIS_HOST,
+    port=db_settings.REDIS_PORT,
+    db=0,
+    decode_responses=True,
+)
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/seo", tags=["Social SEO"])
+
+@router.post(
+    "/youtube",
+    response_model=YoutubeDescriptionResponse,
+    summary="유튜브 SEO descrption 생성",
+    description="유튜브 업로드 시 사용할 descrption을 SEO 적용하여 생성",
+)
+async def youtube_seo_description(
+        request_body: YoutubeDescriptionRequest,
+        current_user: User = Depends(get_current_user),
+        session: AsyncSession = Depends(get_session),
+        ) -> YoutubeDescriptionResponse:
+    
+    # TODO : 나중에 Session Task_id 검증 미들웨어 만들면 추가해주세요.
+
+    logger.info(
+        f"[youtube_seo_description] Try Cache - user: {current_user.user_uuid} / task_id : {request_body.task_id}"
+    )
+    cached = await get_yt_seo_in_redis(request_body.task_id)
+    if cached: # redis hit
+        return cached
+    
+    logger.info(
+        f"[youtube_seo_description] Cache miss - user: {current_user.user_uuid} "
+    )
+    updated_seo = await make_youtube_seo_description(request_body.task_id, current_user, session)
+    await set_yt_seo_in_redis(request_body.task_id, updated_seo)
+
+    return updated_seo
+
+async def make_youtube_seo_description(
+        task_id: str,
+        current_user: User,
+        session: AsyncSession,
+    ) -> YoutubeDescriptionResponse:
+    
+    logger.info(
+        f"[make_youtube_seo_description] START - user: {current_user.user_uuid} "
+    )
+    try:
+        project_query = await session.execute(
+            select(Project)
+            .where(
+                Project.task_id == task_id,
+                Project.user_uuid == current_user.user_uuid)
+               .order_by(Project.created_at.desc())
+            .limit(1)
+        )
+
+        project = project_query.scalar_one_or_none()
+        marketing_query = await session.execute(
+            select(MarketingIntel)
+            .where(MarketingIntel.id == project.marketing_intelligence)
+        )
+        marketing_intelligence = marketing_query.scalar_one_or_none()
+
+        hashtags = marketing_intelligence.intel_result["target_keywords"]
+        
+        yt_seo_input_data = {
+            "customer_name" : project.store_name,
+            "detail_region_info" : project.detail_region_info,
+            "marketing_intelligence_summary" : json.dumps(marketing_intelligence.intel_result, ensure_ascii=False),
+            "language" : project.language,
+            "target_keywords" : hashtags
+        }
+        chatgpt = ChatgptService()
+        yt_seo_output = await chatgpt.generate_structured_output(yt_upload_prompt, yt_seo_input_data)
+        result_dict = {
+            "title" : yt_seo_output.title,
+            "description" : yt_seo_output.description,
+            "keywords": hashtags
+        }
+
+        result = YoutubeDescriptionResponse(**result_dict)
+        return result
+
+    except Exception as e:
+        logger.error(f"[youtube_seo_description] EXCEPTION - error: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
+        )
+    
+
+async def get_yt_seo_in_redis(task_id:str) -> YoutubeDescriptionResponse | None:
+    field = f"task_id:{task_id}"
+    yt_seo_info = await redis_seo_client.hget(YOUTUBE_SEO_HASH, field)
+    if yt_seo_info:
+        yt_seo = json.loads(yt_seo_info)
+    else:
+        return None
+    return YoutubeDescriptionResponse(**yt_seo)
+
+async def set_yt_seo_in_redis(task_id:str, yt_seo : YoutubeDescriptionResponse) -> None:
+    field = f"task_id:{task_id}"
+    yt_seo_info = json.dumps(yt_seo.model_dump(), ensure_ascii=False)
+    await redis_seo_client.hsetex(YOUTUBE_SEO_HASH, field, yt_seo_info, ex=3600)
+    return
+

app/social/api/routers/v1/upload.py

@@ -4,10 +4,11 @@
 소셜 미디어 영상 업로드 관련 엔드포인트를 제공합니다.
 """
 
-import logging
+import logging, json
 from typing import Optional
 
 from fastapi import APIRouter, BackgroundTasks, Depends, Query
+from fastapi import HTTPException, status
 from sqlalchemy import select, func
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -403,16 +404,12 @@ async def cancel_upload(
     upload = result.scalar_one_or_none()
 
     if not upload:
-        from fastapi import HTTPException, status
-
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail="업로드 정보를 찾을 수 없습니다.",
         )
 
     if upload.status != UploadStatus.PENDING.value:
-        from fastapi import HTTPException, status
-
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail="대기 중인 업로드만 취소할 수 있습니다.",
@@ -424,4 +421,4 @@ async def cancel_upload(
     return MessageResponse(
         success=True,
         message="업로드가 취소되었습니다.",
-    )
+    )

app/social/constants.py

@@ -91,9 +91,12 @@ PLATFORM_CONFIG = {
 YOUTUBE_SCOPES = [
     "https://www.googleapis.com/auth/youtube.upload",  # 영상 업로드
     "https://www.googleapis.com/auth/youtube.readonly",  # 채널 정보 읽기
+    "https://www.googleapis.com/auth/yt-analytics.readonly",  # 대시보드
     "https://www.googleapis.com/auth/userinfo.profile",  # 사용자 프로필
 ]
 
+YOUTUBE_SEO_HASH = "SEO_Describtion_YT"
+
 # =============================================================================
 # Instagram/Facebook OAuth Scopes (추후 구현)
 # =============================================================================

app/social/schemas.py

@@ -276,6 +276,32 @@ class SocialUploadHistoryResponse(BaseModel):
         }
     )
 
+class YoutubeDescriptionRequest(BaseModel):
+    """유튜브 SEO Description 제안 (자동완성) Request 모델"""
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "task_id" :  "019c739f-65fc-7d15-8c88-b31be00e588e"
+            }
+        }
+    )
+    task_id: str = Field(..., description="작업 고유 식별자")
+
+class YoutubeDescriptionResponse(BaseModel):
+    """유튜브 SEO Description 제안 (자동완성) Response 모델"""
+    title:str = Field(..., description="유튜브 영상 제목 - SEO/AEO 최적화")
+    description : str = Field(..., description="제안된 유튜브 SEO Description")
+    keywords : list[str] = Field(..., description="해시태그 리스트")
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "title" : "여기에 더미 타이틀",
+                "description": "여기에 더미 텍스트",
+                "keywords": ["여기에", "더미", "해시태그"]
+            }
+        }
+    )
 
 # =============================================================================
 # 공통 응답 스키마

app/social/worker/upload_task.py

@@ -18,6 +18,7 @@ from sqlalchemy import select
 from sqlalchemy.exc import SQLAlchemyError
 
 from config import social_upload_settings
+from app.dashboard.tasks import insert_dashboard
 from app.database.session import BackgroundSessionLocal
 from app.social.constants import SocialPlatform, UploadStatus
 from app.social.exceptions import TokenExpiredError, UploadError, UploadQuotaExceededError
@@ -318,6 +319,7 @@ async def process_social_upload(upload_id: int) -> None:
                 f"platform_video_id: {result.platform_video_id}, "
                 f"url: {result.platform_url}"
             )
+            await insert_dashboard(upload_id)
         else:
             retry_count = await _increment_retry_count(upload_id)
 

app/utils/chatgpt_prompt.py

@@ -82,7 +82,7 @@ class ChatgptService:
         self,
         prompt : Prompt,
         input_data : dict,
-    ) -> str:
+    ) -> BaseModel:
         prompt_text = prompt.build_prompt(input_data)
 
         logger.debug(f"[ChatgptService] Generated Prompt (length: {len(prompt_text)})")

app/utils/prompts/prompts.py

@@ -52,6 +52,14 @@ lyric_prompt = Prompt(
     prompt_model = prompt_settings.LYRIC_PROMPT_MODEL
 )
 
+yt_upload_prompt = Prompt(
+    prompt_template_path=os.path.join(prompt_settings.PROMPT_FOLDER_ROOT, prompt_settings.YOUTUBE_PROMPT_FILE_NAME),
+    prompt_input_class = YTUploadPromptInput,
+    prompt_output_class = YTUploadPromptOutput,
+    prompt_model = prompt_settings.YOUTUBE_PROMPT_MODEL
+)
+
 def reload_all_prompt():
     marketing_prompt._reload_prompt()
-    lyric_prompt._reload_prompt()
+    lyric_prompt._reload_prompt()
+    yt_upload_prompt._reload_prompt()

app/utils/prompts/schemas/__init__.py

@@ -1,2 +1,3 @@
 from .lyric import LyricPromptInput, LyricPromptOutput
-from .marketing import MarketingPromptInput, MarketingPromptOutput
+from .marketing import MarketingPromptInput, MarketingPromptOutput
+from .youtube import YTUploadPromptInput, YTUploadPromptOutput

app/utils/prompts/schemas/youtube.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel, Field
+from typing import List, Optional
+
+# Input 정의
+class YTUploadPromptInput(BaseModel):
+    customer_name : str = Field(..., description = "마케팅 대상 사업체 이름")
+    detail_region_info : str = Field(..., description = "마케팅 대상 지역 상세")
+    marketing_intelligence_summary : Optional[str] = Field(None, description = "마케팅 분석 정보 보고서")
+    language : str= Field(..., description = "영상 언어")
+    target_keywords: List[str] = Field(..., description="태그 키워드 리스트")
+
+# Output 정의
+class YTUploadPromptOutput(BaseModel):
+    title:str = Field(..., description="유튜브 영상 제목 - SEO/AEO 최적화")
+    description: str = Field(..., description = "유튜브 영상 설명 - SEO/AEO 최적화")
+    

config.py

@@ -184,6 +184,8 @@ class PromptSettings(BaseSettings):
     LYRIC_PROMPT_FILE_NAME : str = Field(default="lyric_prompt.txt")
     LYRIC_PROMPT_MODEL : str = Field(default="gpt-5-mini")
 
+    YOUTUBE_PROMPT_FILE_NAME : str = Field(default="yt_upload_prompt.txt")
+    YOUTUBE_PROMPT_MODEL : str = Field(default="gpt-5-mini")
     model_config = _base_config
 
 
@@ -598,9 +600,9 @@ class SocialUploadSettings(BaseSettings):
 
 
 prj_settings = ProjectSettings()
+cors_settings = CORSSettings()
 apikey_settings = APIKeySettings()
 db_settings = DatabaseSettings()
-cors_settings = CORSSettings()
 crawler_settings = CrawlerSettings()
 naver_api_settings = NaverAPISettings()
 azure_blob_settings = AzureBlobSettings()

main.py

@@ -12,6 +12,7 @@ from app.database.session import engine
 from app.user.models import User, RefreshToken  # noqa: F401
 
 from app.archive.api.routers.v1.archive import router as archive_router
+from app.dashboard.api.routers.v1.dashboard import router as dashboard_router
 from app.home.api.routers.v1.home import router as home_router
 from app.user.api.routers.v1.auth import router as auth_router, test_router as auth_test_router
 from app.user.api.routers.v1.social_account import router as social_account_router
@@ -21,6 +22,7 @@ from app.sns.api.routers.v1.sns import router as sns_router
 from app.video.api.routers.v1.video import router as video_router
 from app.social.api.routers.v1.oauth import router as social_oauth_router
 from app.social.api.routers.v1.upload import router as social_upload_router
+from app.social.api.routers.v1.seo import router as social_seo_router
 from app.utils.cors import CustomCORSMiddleware
 from config import prj_settings
 
@@ -219,6 +221,34 @@ tags_metadata = [
 - `processing`: 플랫폼에서 처리 중
 - `completed`: 업로드 완료
 - `failed`: 업로드 실패
+""",
+    },
+    {
+        "name": "Dashboard",
+        "description": """YouTube Analytics 대시보드 API
+
+## 주요 기능
+
+- `GET /dashboard/stats` - YouTube 영상 성과 통계 조회
+
+## 제공 데이터
+
+- **KPI 지표**: 총 조회수, 참여율, 시청 시간, 평균 시청 시간
+- **월별 추이**: 최근 12개월 vs 이전 12개월 비교
+- **인기 영상**: 조회수 TOP 10
+- **시청자 분석**: 연령/성별/지역 분포
+- **플랫폼 메트릭**: 구독자 증감
+
+## 성능 최적화
+
+- 6개 YouTube Analytics API 병렬 호출
+- Redis 캐싱 (TTL: 1시간)
+- 최근 30개 업로드 영상 기준 분석
+
+## 사전 조건
+
+- YouTube 계정이 연동되어 있어야 합니다
+- 최소 1개 이상의 영상이 업로드되어 있어야 합니다
 """,
     },
     {
@@ -360,7 +390,9 @@ app.include_router(video_router)
 app.include_router(archive_router)  # Archive API 라우터 추가
 app.include_router(social_oauth_router, prefix="/social")  # Social OAuth 라우터 추가
 app.include_router(social_upload_router, prefix="/social")  # Social Upload 라우터 추가
+app.include_router(social_seo_router, prefix="/social")  # Social Upload 라우터 추가
 app.include_router(sns_router)  # SNS API 라우터 추가
+app.include_router(dashboard_router)  # Dashboard API 라우터 추가
 
 # DEBUG 모드에서만 테스트 라우터 등록
 if prj_settings.DEBUG: