feat(video): 제목·설명·해시태그를 영상 완료 시 DB에 저장

업로드 시점 Redis SEO 캐시 대신 video 행을 기준으로 두고, 게시 시에만 변경분을 갱신한다.

app/archive/api/routers/v1/archive.py

@@ -180,6 +180,9 @@ async def get_videos(
                 task_id=video.task_id,
                 result_movie_url=video.result_movie_url,
                 poster_url=video.poster_url,
+                title=video.title,
+                description=video.description,
+                hashtags=video.hashtags,
                 created_at=video.created_at,
                 like_count=like_count_map.get(video.id) or 0,
                 comment_count=comment_count or 0,

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

@@ -33,5 +33,5 @@ async def youtube_seo_description(
     session: AsyncSession = Depends(get_session),
 ) -> YoutubeDescriptionResponse:
     return await seo_service.get_youtube_seo_description(
-        request_body.task_id, current_user, session
+        request_body.video_id, current_user, session
     )

app/social/constants.py

@@ -95,8 +95,6 @@ YOUTUBE_SCOPES = [
     "https://www.googleapis.com/auth/userinfo.profile",  # 사용자 프로필
 ]
 
-YOUTUBE_SEO_HASH = "SEO_Describtion_YT"
-
 # =============================================================================
 # Instagram/Facebook OAuth Scopes (추후 구현)
 # =============================================================================

app/social/schemas/seo_schema.py

@@ -8,12 +8,12 @@ from pydantic import BaseModel, ConfigDict, Field
 class YoutubeDescriptionRequest(BaseModel):
     """유튜브 SEO Description 제안 요청"""
 
-    task_id: str = Field(..., description="작업 고유 식별자")
+    video_id: int = Field(..., description="영상 고유 ID")
 
     model_config = ConfigDict(
         json_schema_extra={
             "example": {
-                "task_id": "019c739f-65fc-7d15-8c88-b31be00e588e"
+                "video_id": 123
             }
         }
     )

app/social/services/seo_service.py

@@ -1,88 +1,132 @@
 """
 유튜브 SEO 서비스
 
-SEO description 생성 및 Redis 캐싱 로직을 처리합니다.
+영상 제목/설명/해시태그를 생성하고 video 테이블에 저장합니다.
 """
 
 import json
 import logging
 
-from fastapi import HTTPException
+from fastapi import HTTPException, status
-from redis.asyncio import Redis
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from config import db_settings
 from app.home.models import MarketingIntel, Project
-from app.social.constants import YOUTUBE_SEO_HASH
 from app.social.schemas import YoutubeDescriptionResponse
+from app.social.services.sns_metadata import apply_sns_metadata, has_stored_sns_metadata
 from app.user.models import User
-from app.utils.prompts.chatgpt_prompt import ChatgptService
+from app.video.models import Video
-from app.utils.prompts.prompts import yt_upload_prompt
 
 logger = logging.getLogger(__name__)
 
-redis_seo_client = Redis(
-    host=db_settings.REDIS_HOST,
-    port=db_settings.REDIS_PORT,
-    db=0,
-    decode_responses=True,
-)
-
 
 class SeoService:
     """유튜브 SEO 비즈니스 로직 서비스"""
 
     async def get_youtube_seo_description(
         self,
-        task_id: str,
+        video_id: int,
         current_user: User,
         session: AsyncSession,
     ) -> YoutubeDescriptionResponse:
         """
-        유튜브 SEO description 생성
+        저장된 SNS 메타데이터를 반환하거나, 없으면 생성 후 video에 저장합니다.
-
-        Redis 캐시 확인 후 miss이면 GPT로 생성하고 캐싱.
         """
         logger.info(
-            f"[SEO_SERVICE] Try Cache - user: {current_user.user_uuid} / task_id: {task_id}"
+            f"[SEO_SERVICE] Load metadata - user: {current_user.user_uuid} / video_id: {video_id}"
         )
 
-        cached = await self._get_from_redis(task_id)
+        video = await self._get_owned_video(video_id, current_user.user_uuid, session)
-        if cached:
+        if video is None:
-            return cached
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"video_id '{video_id}'에 해당하는 영상을 찾을 수 없습니다.",
+            )
 
-        logger.info(f"[SEO_SERVICE] Cache miss - user: {current_user.user_uuid}")
+        if has_stored_sns_metadata(video):
-        result = await self._generate_seo_description(task_id, current_user, session)
+            return self._response_from_video(video)
-        await self._set_to_redis(task_id, result)
 
+        result = await self.generate_and_save_for_video(video.id, session)
+        if result is None:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail=f"video_id '{video_id}'에 해당하는 영상을 찾을 수 없습니다.",
+            )
+        await session.commit()
         return result
 
+    async def generate_and_save_for_video(
+        self,
+        video_id: int,
+        session: AsyncSession,
+    ) -> YoutubeDescriptionResponse | None:
+        """GPT로 SEO를 생성해 지정한 video 행에 저장합니다. 워커/온디맨드 공용."""
+        video_result = await session.execute(select(Video).where(Video.id == video_id))
+        video = video_result.scalar_one_or_none()
+        if video is None:
+            logger.warning(f"[SEO_SERVICE] Video NOT FOUND - video_id: {video_id}")
+            return None
+
+        if has_stored_sns_metadata(video):
+            return self._response_from_video(video)
+
+        result = await self._generate_seo_description(video.task_id, session)
+        apply_sns_metadata(video, result.title, result.description, result.keywords)
+        await session.flush()
+        logger.info(f"[SEO_SERVICE] Saved metadata - video_id: {video_id}")
+        return result
+
+    async def _get_owned_video(
+        self,
+        video_id: int,
+        user_uuid: str,
+        session: AsyncSession,
+    ) -> Video | None:
+        result = await session.execute(
+            select(Video)
+            .join(Project, Project.id == Video.project_id)
+            .where(
+                Video.id == video_id,
+                Project.user_uuid == user_uuid,
+                Video.is_deleted.is_(False),
+            )
+        )
+        return result.scalar_one_or_none()
+
     async def _generate_seo_description(
         self,
         task_id: str,
-        current_user: User,
         session: AsyncSession,
     ) -> YoutubeDescriptionResponse:
         """GPT를 사용하여 SEO description 생성"""
-        logger.info(f"[SEO_SERVICE] Generating SEO - user: {current_user.user_uuid}")
+        from app.utils.prompts.chatgpt_prompt import ChatgptService
+        from app.utils.prompts.prompts import yt_upload_prompt
+
+        logger.info(f"[SEO_SERVICE] Generating SEO - task_id: {task_id}")
 
         try:
             project_result = await session.execute(
                 select(Project)
-                .where(
+                .where(Project.task_id == task_id)
-                    Project.task_id == task_id,
-                    Project.user_uuid == current_user.user_uuid,
-                )
                 .order_by(Project.created_at.desc())
                 .limit(1)
             )
             project = project_result.scalar_one_or_none()
+            if project is None:
+                raise HTTPException(
+                    status_code=status.HTTP_404_NOT_FOUND,
+                    detail=f"task_id '{task_id}'에 해당하는 Project를 찾을 수 없습니다.",
+                )
 
             marketing_result = await session.execute(
                 select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
             )
             marketing_intelligence = marketing_result.scalar_one_or_none()
+            if marketing_intelligence is None:
+                raise HTTPException(
+                    status_code=status.HTTP_404_NOT_FOUND,
+                    detail="마케팅 인텔리전스를 찾을 수 없습니다.",
+                )
 
             hashtags = marketing_intelligence.intel_result["target_keywords"]
 
@@ -94,12 +138,13 @@ class SeoService:
                 ),
                 "language": project.language,
                 "target_keywords": hashtags,
-                "industry": project.industry or "",  # 크롤 시 분류해 Project에 저장한 업종 enum
+                "industry": project.industry or "",
             }
 
-            # 업종 분기는 프롬프트 내부 {industry}로 처리하므로 단일 프롬프트 사용
             chatgpt = ChatgptService(timeout=180)
-            yt_seo_output = await chatgpt.generate_structured_output(yt_upload_prompt, yt_seo_input_data)
+            yt_seo_output = await chatgpt.generate_structured_output(
+                yt_upload_prompt, yt_seo_input_data
+            )
 
             return YoutubeDescriptionResponse(
                 title=yt_seo_output.title,
@@ -107,6 +152,8 @@ class SeoService:
                 keywords=hashtags,
             )
 
+        except HTTPException:
+            raise
         except Exception as e:
             logger.error(f"[SEO_SERVICE] EXCEPTION - error: {e}")
             raise HTTPException(
@@ -114,18 +161,12 @@ class SeoService:
                 detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
             )
 
-    async def _get_from_redis(self, task_id: str) -> YoutubeDescriptionResponse | None:
+    def _response_from_video(self, video: Video) -> YoutubeDescriptionResponse:
-        field = f"task_id:{task_id}"
+        return YoutubeDescriptionResponse(
-        yt_seo_info = await redis_seo_client.hget(YOUTUBE_SEO_HASH, field)
+            title=video.title or "",
-        if yt_seo_info:
+            description=video.description or "",
-            return YoutubeDescriptionResponse(**json.loads(yt_seo_info))
+            keywords=list(video.hashtags or []),
-        return None
+        )
-
-    async def _set_to_redis(self, 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.hset(YOUTUBE_SEO_HASH, field, yt_seo_info)
-        await redis_seo_client.expire(YOUTUBE_SEO_HASH, 3600)
 
 
 seo_service = SeoService()

app/social/services/sns_metadata.py

@@ -0,0 +1,36 @@
+"""SNS 업로드용 영상 메타데이터 비교/반영 헬퍼."""
+
+from app.video.models import Video
+
+
+def has_stored_sns_metadata(video: Video) -> bool:
+    """video 행에 SNS 제목이 이미 저장되어 있는지 확인합니다."""
+    return bool(video.title)
+
+
+def sns_metadata_changed(
+    video: Video,
+    title: str,
+    description: str | None,
+    tags: list[str] | None,
+) -> bool:
+    """게시 폼 값이 저장된 SNS 메타데이터와 다른지 비교합니다."""
+    stored_tags = list(video.hashtags or [])
+    incoming_tags = list(tags or [])
+    return (
+        (video.title or "") != title
+        or (video.description or "") != (description or "")
+        or stored_tags != incoming_tags
+    )
+
+
+def apply_sns_metadata(
+    video: Video,
+    title: str,
+    description: str | None,
+    hashtags: list[str] | None,
+) -> None:
+    """video 행에 SNS 메타데이터를 반영합니다."""
+    video.title = title
+    video.description = description
+    video.hashtags = list(hashtags or [])

app/social/services/upload_service.py

@@ -26,6 +26,7 @@ from app.social.schemas import (
     SocialUploadRequest,
 )
 from app.social.services.account_service import SocialAccountService
+from app.social.services.sns_metadata import apply_sns_metadata, sns_metadata_changed
 from app.social.worker.upload_task import process_social_upload
 from app.user.models import User
 from app.video.models import Video
@@ -76,6 +77,12 @@ class SocialUploadService:
                 detail="영상이 아직 준비되지 않았습니다. 영상 생성이 완료된 후 시도해주세요.",
             )
 
+        if sns_metadata_changed(video, body.title, body.description, body.tags):
+            apply_sns_metadata(video, body.title, body.description, body.tags)
+            logger.info(
+                f"[UPLOAD_SERVICE] video SNS 메타데이터 갱신 - video_id: {body.video_id}"
+            )
+
         # 2. 소셜 계정 조회 및 소유권 검증
         account = await self._account_service.get_account_by_id(
             user_uuid=current_user.user_uuid,

app/video/models.py

@@ -1,7 +1,8 @@
 from datetime import datetime
 from typing import TYPE_CHECKING, List, Optional
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, func
+from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint, func
+from sqlalchemy.dialects.mysql import JSON
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database.session import Base
@@ -30,6 +31,9 @@ class Video(Base):
         status: 처리 상태 (pending, processing, completed, failed 등)
         result_movie_url: 생성된 영상 URL (S3, CDN 경로)
         poster_url: 영상 첫 프레임 포스터 이미지 URL (SNS 공유 og:image용)
+        title: SNS 업로드 제목
+        description: SNS 업로드 설명
+        hashtags: SNS 해시태그 목록
         created_at: 생성 일시 (자동 설정)
 
     Relationships:
@@ -113,6 +117,24 @@ class Video(Base):
         comment="영상 첫 프레임 포스터 이미지 URL (SNS 공유용)",
     )
 
+    title: Mapped[Optional[str]] = mapped_column(
+        String(100),
+        nullable=True,
+        comment="SNS 업로드 제목",
+    )
+
+    description: Mapped[Optional[str]] = mapped_column(
+        Text,
+        nullable=True,
+        comment="SNS 업로드 설명",
+    )
+
+    hashtags: Mapped[Optional[list]] = mapped_column(
+        JSON,
+        nullable=True,
+        comment="SNS 해시태그 목록",
+    )
+
     is_deleted: Mapped[bool] = mapped_column(
         Boolean,
         nullable=False,

app/video/schemas/video_schema.py

@@ -5,7 +5,7 @@ Video API Schemas
 """
 
 from datetime import datetime
-from typing import Any, Dict, Optional
+from typing import Any, Dict, List, Optional
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -159,6 +159,9 @@ class VideoListItem(BaseModel):
     task_id: str = Field(..., description="작업 고유 식별자")
     result_movie_url: Optional[str] = Field(None, description="영상 결과 URL")
     poster_url: Optional[str] = Field(None, description="영상 첫 프레임 포스터 이미지 URL")
+    title: Optional[str] = Field(None, description="SNS 업로드 제목")
+    description: Optional[str] = Field(None, description="SNS 업로드 설명")
+    hashtags: Optional[List[str]] = Field(None, description="SNS 해시태그 목록")
     created_at: Optional[datetime] = Field(None, description="생성 일시")
     like_count: int = Field(0, description="좋아요 수")
     comment_count: int = Field(0, description="댓글 수 (대댓글 포함)")

app/video/worker/video_task.py

@@ -31,7 +31,7 @@ async def _update_video_status(
     video_url: str | None = None,
     creatomate_render_id: str | None = None,
     poster_url: str | None = None,
-) -> bool:
+) -> int | None:
     """Video 테이블의 상태를 업데이트합니다.
 
     Args:
@@ -42,7 +42,7 @@ async def _update_video_status(
         poster_url: 영상 첫 프레임 포스터 URL (선택)
 
     Returns:
-        bool: 업데이트 성공 여부
+        int | None: 업데이트된 Video id. 대상이 없거나 실패하면 None.
     """
     try:
         async with BackgroundSessionLocal() as session:
@@ -71,17 +71,32 @@ async def _update_video_status(
                     video.poster_url = poster_url
                 await session.commit()
                 logger.info(f"[Video] Status updated - task_id: {task_id}, status: {status}")
-                return True
+                return video.id
             else:
                 logger.warning(f"[Video] NOT FOUND in DB - task_id: {task_id}")
-                return False
+                return None
 
     except SQLAlchemyError as e:
         logger.error(f"[Video] DB Error while updating status - task_id: {task_id}, error: {e}")
-        return False
+        return None
     except Exception as e:
         logger.error(f"[Video] Unexpected error while updating status - task_id: {task_id}, error: {e}")
-        return False
+        return None
+
+
+async def _try_generate_sns_metadata(video_id: int) -> None:
+    """SEO 생성 실패가 영상 완료 처리에 영향을 주지 않도록 격리합니다."""
+    from app.social.services.seo_service import seo_service
+
+    try:
+        async with BackgroundSessionLocal() as session:
+            await seo_service.generate_and_save_for_video(video_id, session)
+            await session.commit()
+    except Exception as e:
+        logger.warning(
+            f"[VideoSEO] Failed to generate SNS metadata - video_id: {video_id}, error: {e}",
+            exc_info=True,
+        )
 
 
 async def _try_generate_poster(
@@ -184,13 +199,15 @@ async def download_and_upload_video_to_blob(
         )
 
         # Video 테이블 업데이트 (creatomate_render_id로 특정 Video 식별)
-        await _update_video_status(
+        video_id = await _update_video_status(
             task_id,
             "completed",
             blob_url,
             creatomate_render_id,
             poster_url=poster_url,
         )
+        if video_id is not None:
+            await _try_generate_sns_metadata(video_id)
 
         # 영상 생성 완료 시 크레딧 1 차감 (credits > 0 조건으로 음수 방지)
         async with BackgroundSessionLocal() as session:
@@ -300,13 +317,15 @@ async def download_and_upload_video_by_creatomate_render_id(
         )
 
         # Video 테이블 업데이트
-        await _update_video_status(
+        video_id = await _update_video_status(
             task_id=task_id,
             status="completed",
             video_url=blob_url,
             creatomate_render_id=creatomate_render_id,
             poster_url=poster_url,
         )
+        if video_id is not None:
+            await _try_generate_sns_metadata(video_id)
         logger.info(f"[download_and_upload_video_by_creatomate_render_id] SUCCESS - creatomate_render_id: {creatomate_render_id}")
 
     except httpx.HTTPError as e:

docs/database-schema/migration_2026_08_19_add_video_sns_metadata.sql

@@ -0,0 +1,15 @@
+-- ============================================================
+-- Migration: video 테이블에 SNS 메타데이터 컬럼 추가
+-- Date: 2026-08-19
+-- Description: 영상 생성 완료 시 저장하는 제목/설명/해시태그.
+--              관련 코드: app/social/services/seo_service.py,
+--              app/video/worker/video_task.py
+-- ============================================================
+
+ALTER TABLE `video`
+    ADD COLUMN `title` VARCHAR(100) NULL
+    COMMENT 'SNS 업로드 제목' AFTER `poster_url`,
+    ADD COLUMN `description` TEXT NULL
+    COMMENT 'SNS 업로드 설명' AFTER `title`,
+    ADD COLUMN `hashtags` JSON NULL
+    COMMENT 'SNS 해시태그 목록' AFTER `description`;