Merge branch 'main' into feature-ssulbox

Co-authored-by: Cursor <cursoragent@cursor.com>

.gitignore

@@ -32,8 +32,11 @@ media/
 
 
 *.ipynb_checkpoint*
-# Static files
-static/
+# Static files (공유 기본 이미지는 예외로 추적)
+static/*
+!static/images/
+static/images/*
+!static/images/ado2_image.png
 
 # Log files
 *.log

README.md

@@ -69,6 +69,8 @@ PROJECT_DOMAIN=localhost:8000          # 프로젝트 도메인 (호스트:포
 PROJECT_VERSION=0.1.0                  # 프로젝트 버전
 DESCRIPTION=FastAPI 기반 CastAD 프로젝트  # 프로젝트 설명
 ADMIN_BASE_URL=/admin                  # 관리자 페이지 기본 URL
+SHARE_FRONTEND_URL=https://ado2.o2osolution.ai  # 공유 페이지 → 영상 상세 이동 프론트 URL (로컬: http://localhost:3000, 테스트: https://dev.castad.net)
+SHARE_DEFAULT_IMAGE_URL=  # 포스터 없을 때 OG 이미지 (비우면 API /static/images/ado2_image.png)
 DEBUG=True                             # 디버그 모드 (True: 개발, False: 운영)
 
 # ================================

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

@@ -94,9 +94,11 @@ async def get_videos(
                 # 프론트는 반드시 (type, video_id) 쌍으로 식별할 것.
                 task_id=it.task_id,
                 result_movie_url=it.movie_url,
+                poster_url=it.poster_url,
                 created_at=it.created_at,
                 like_count=it.like_count,
                 comment_count=it.comment_count,
+                is_liked_by_me=it.is_liked_by_me,
             )
             for it in items
         ]

app/comment/api/routers/v1/comment.py

@@ -48,7 +48,7 @@ router = APIRouter(prefix="/comment", tags=["Comment"])
 - **parent_id**: 대댓글일 때만 부모 댓글 id (생략 시 최상위 댓글)
 
 ## 참고
-- 작성자 정보는 응답에 포함되지 않습니다 (익명 정책).
+- 작성자 닉네임/프로필 이미지는 카카오 로그인 정보를 그대로 사용합니다 (클라이언트에서 지정 불가).
 - 대댓글에 또 대댓글을 다는 것은 불가합니다 (최대 2-depth).
     """,
     response_model=CommentCreateResponse,
@@ -77,7 +77,7 @@ async def post_comment(
         session=session,
         video_id=video_id,
         user_uuid=current_user.user_uuid,
-        nickname=body.nickname,
+        nickname=current_user.nickname,
         content=body.content,
         parent_id=body.parent_id,
         content_type=type,
@@ -86,6 +86,7 @@ async def post_comment(
     return CommentCreateResponse(
         id=comment.id,
         nickname=comment.nickname or "익명",
+        profile_image_url=current_user.profile_image_url,
         parent_id=comment.parent_id,
         content=comment.content,
         created_at=comment.created_at,
@@ -108,7 +109,7 @@ async def post_comment(
 
 ## 참고
 - 최상위 댓글만 페이지네이션됩니다. 각 댓글의 대댓글은 전부 포함됩니다.
-- 작성자 정보는 노출되지 않으며, is_mine으로 본인 댓글 여부만 확인 가능합니다.
+- 작성자 닉네임/프로필 이미지는 카카오 로그인 정보 기준이며, is_mine으로 본인 댓글 여부도 확인 가능합니다.
 - 삭제된 댓글은 content=null로 노출됩니다 (대댓글이 있는 경우).
     """,
     response_model=PaginatedResponse[CommentItem],

app/comment/models.py

@@ -28,7 +28,8 @@ class Comment(Base):
 
     2-depth 구조 (최상위 댓글 + 대댓글 1단계).
     parent_id가 NULL이면 최상위 댓글, 값이 있으면 대댓글.
-    작성자(user_uuid)는 DB에 저장하지만 API 응답에는 미노출 (익명 정책).
+    작성자 닉네임은 카카오 로그인 정보를 작성 시점에 그대로 저장한 스냅샷이며,
+    프로필 이미지는 별도 컬럼 없이 응답 시 User 테이블을 조인해 최신값을 조회한다.
 
     **ADO2 영상과 썰박스 콘텐츠를 모두 담는다.** 대상은 `video_id` 또는 `content_id`
     중 **정확히 하나**만 채워지며, 이를 DB `CHECK` 로 강제한다. MySQL 은 하나의 FK 가
@@ -84,7 +85,7 @@ class Comment(Base):
         comment="NULL=최상위 댓글, 값=대댓글의 부모 id",
     )
     nickname: Mapped[Optional[str]] = mapped_column(
-        String(50), nullable=True, comment="댓글 작성자 닉네임 (null이면 익명)"
+        String(50), nullable=True, comment="댓글 작성자 카카오 닉네임 스냅샷 (null이면 익명)"
     )
     content: Mapped[str] = mapped_column(
         String(100), nullable=False, comment="댓글 본문 (한글 기준 100자 이내)"

app/comment/schemas/comment_schema.py

@@ -5,7 +5,6 @@ from pydantic import BaseModel, Field
 
 
 class CommentCreateRequest(BaseModel):
-    nickname: Optional[str] = Field(None, min_length=1, max_length=50, description="작성자 닉네임 (미입력 시 익명)")
     content: str = Field(..., min_length=1, max_length=100, description="댓글 본문 (한글 기준 100자 이내)")
     parent_id: Optional[int] = Field(None, description="대댓글일 때만 부모 댓글 id")
 
@@ -14,7 +13,8 @@ class ReplyItem(BaseModel):
     """대댓글 응답"""
 
     id: int = Field(..., description="댓글 고유 ID")
-    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
+    nickname: str = Field(..., description="작성자 닉네임 (카카오 닉네임, 미보유 시 '익명')")
+    profile_image_url: Optional[str] = Field(None, description="작성자 프로필 이미지 URL (카카오 프로필, 로그인 시점 기준 최신값)")
     content: Optional[str] = Field(None, description="본문 (소프트 삭제된 경우 null)")
     is_deleted: bool = Field(..., description="삭제 여부")
     is_mine: bool = Field(..., description="현재 로그인 사용자의 댓글 여부")
@@ -25,7 +25,8 @@ class CommentItem(BaseModel):
     """최상위 댓글 응답 — replies 포함"""
 
     id: int = Field(..., description="댓글 고유 ID")
-    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
+    nickname: str = Field(..., description="작성자 닉네임 (카카오 닉네임, 미보유 시 '익명')")
+    profile_image_url: Optional[str] = Field(None, description="작성자 프로필 이미지 URL (카카오 프로필, 로그인 시점 기준 최신값)")
     content: Optional[str] = Field(None, description="본문 (소프트 삭제된 경우 null)")
     is_deleted: bool = Field(..., description="삭제 여부")
     is_mine: bool = Field(..., description="현재 로그인 사용자의 댓글 여부")
@@ -35,7 +36,8 @@ class CommentItem(BaseModel):
 
 class CommentCreateResponse(BaseModel):
     id: int = Field(..., description="생성된 댓글 고유 ID")
-    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
+    nickname: str = Field(..., description="작성자 닉네임 (카카오 닉네임, 미보유 시 '익명')")
+    profile_image_url: Optional[str] = Field(None, description="작성자 프로필 이미지 URL (카카오 프로필)")
     parent_id: Optional[int] = Field(None, description="부모 댓글 id (대댓글인 경우)")
     content: str = Field(..., description="댓글 본문")
     created_at: datetime = Field(..., description="작성 일시")

app/comment/services/comment.py

@@ -8,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from app.comment.models import Comment
 from app.comment.schemas.comment_schema import CommentItem, ReplyItem
 from app.ssulbox.models import SsulContent
+from app.user.models import User
 from app.utils.pagination import PaginatedResponse
 from app.video.models import Video
 
@@ -78,6 +79,7 @@ def _build_comment_items(
     parents: list,
     replies_map: dict,
     current_user_uuid: Optional[str],
+    profile_image_map: dict,
 ) -> List[CommentItem]:
     items = []
     for c in parents:
@@ -86,6 +88,7 @@ def _build_comment_items(
             ReplyItem(
                 id=r.id,
                 nickname=r.nickname or "익명",
+                profile_image_url=profile_image_map.get(r.user_uuid),
                 content=None if r.is_deleted else r.content,
                 is_deleted=r.is_deleted,
                 is_mine=(current_user_uuid == r.user_uuid) if current_user_uuid else False,
@@ -97,6 +100,7 @@ def _build_comment_items(
             CommentItem(
                 id=c.id,
                 nickname=c.nickname or "익명",
+                profile_image_url=profile_image_map.get(c.user_uuid),
                 content=None if c.is_deleted else c.content,
                 is_deleted=c.is_deleted,
                 is_mine=(current_user_uuid == c.user_uuid) if current_user_uuid else False,
@@ -111,7 +115,7 @@ async def create_comment(
     session: AsyncSession,
     video_id: int,
     user_uuid: str,
-    nickname: str,
+    nickname: Optional[str],
     content: str,
     parent_id: Optional[int],
     content_type: ContentType = "video",
@@ -181,6 +185,7 @@ async def list_comments(
     parents = (await session.execute(parents_q)).scalars().all()
 
     replies_map: dict = defaultdict(list)
+    replies: list = []
     if parents:
         parent_ids = [c.id for c in parents]
         replies_q = (
@@ -195,7 +200,16 @@ async def list_comments(
         for r in replies:
             replies_map[r.parent_id].append(r)
 
-    items = _build_comment_items(list(parents), replies_map, current_user_uuid)
+    # 작성자 프로필 이미지는 스냅샷을 저장하지 않고, 응답 시 User 테이블을 조인해 최신값을 조회한다.
+    user_uuids = {c.user_uuid for c in parents} | {r.user_uuid for r in replies}
+    profile_image_map: dict = {}
+    if user_uuids:
+        profile_q = select(User.user_uuid, User.profile_image_url).where(
+            User.user_uuid.in_(user_uuids)
+        )
+        profile_image_map = {uuid: url for uuid, url in (await session.execute(profile_q)).all()}
+
+    items = _build_comment_items(list(parents), replies_map, current_user_uuid, profile_image_map)
 
     return PaginatedResponse.create(
         items=items,

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

@@ -7,7 +7,7 @@ SEO 관련 엔드포인트를 제공합니다.
 
 import logging
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.database.session import get_session
@@ -32,9 +32,21 @@ async def youtube_seo_description(
     current_user: User = Depends(get_current_user),
     session: AsyncSession = Depends(get_session),
 ) -> YoutubeDescriptionResponse:
+    if request_body.content_type == "ssul":
+        content_id = request_body.video_id
+        if content_id is None:
+            try:
+                content_id = int(request_body.task_id)
+            except (TypeError, ValueError):
+                raise HTTPException(
+                    status_code=status.HTTP_400_BAD_REQUEST,
+                    detail="썰박스 SEO 에는 video_id 또는 숫자 task_id 가 필요합니다.",
+                )
+        return await seo_service.get_ssul_seo(content_id, current_user, session)
+
     return await seo_service.get_youtube_seo_description(
-        request_body.task_id,
         current_user,
         session,
-        content_type=request_body.content_type,
+        video_id=request_body.video_id,
+        task_id=request_body.task_id,
     )

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

@@ -2,25 +2,36 @@
 소셜 SEO 관련 Pydantic 스키마
 """
 
-from typing import Literal
+from typing import Literal, Optional
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 
 class YoutubeDescriptionRequest(BaseModel):
     """유튜브 SEO Description 제안 요청"""
 
-    # 썰박스 콘텐츠의 SEO 를 요청할 때는 "ssul" + task_id 자리에 ssul_content.id.
-    # 종류를 안 밝히면 video 로 간주된다 (기존 호출 호환).
     content_type: Literal["video", "ssul"] = Field(
         default="video", description="콘텐츠 종류"
     )
-    task_id: str = Field(..., description="작업 고유 식별자 (ssul 이면 ssul_content.id)")
+    video_id: Optional[int] = Field(
+        None, description="ADO2 video.id 또는 썰박스 ssul_content.id"
+    )
+    task_id: Optional[str] = Field(
+        None,
+        description="ADO2 작업 UUID. 썰박스는 ssul_content.id 문자열도 허용",
+    )
+
+    @model_validator(mode="after")
+    def require_identifier(self) -> "YoutubeDescriptionRequest":
+        if self.video_id is None and not self.task_id:
+            raise ValueError("video_id 또는 task_id 가 필요합니다.")
+        return self
 
     model_config = ConfigDict(
         json_schema_extra={
             "example": {
-                "task_id": "019c739f-65fc-7d15-8c88-b31be00e588e"
+                "content_type": "video",
+                "video_id": 123,
             }
         }
     )

app/social/services/seo_service.py

@@ -1,93 +1,84 @@
 """
 유튜브 SEO 서비스
 
-SEO description 생성 및 Redis 캐싱 로직을 처리합니다.
+ADO2 영상은 제목/설명/해시태그를 video 테이블에 저장합니다.
+썰박스는 별도 프롬프트로 생성합니다.
 """
 
 import json
 import logging
 
-from fastapi import HTTPException
-from redis.asyncio import Redis
+from fastapi import HTTPException, status
 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.ssulbox.models import SsulContent
 from app.user.models import User
-from app.utils.prompts.chatgpt_prompt import ChatgptService
-from app.utils.prompts.prompts import yt_upload_prompt
+from app.video.models import Video
 
 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,
         current_user: User,
         session: AsyncSession,
-        content_type: str = "video",
+        video_id: int | None = None,
+        task_id: str | None = None,
     ) -> YoutubeDescriptionResponse:
-        """
-        유튜브 SEO description 생성
+        """저장된 SNS 메타데이터를 반환하거나, 없으면 생성 후 video에 저장합니다."""
+        video = None
+        if video_id is not None:
+            video = await self._get_owned_video(video_id, current_user.user_uuid, session)
+        elif task_id:
+            video = await self._get_owned_video_by_task(
+                task_id, current_user.user_uuid, session
+            )
 
-        Redis 캐시 확인 후 miss이면 GPT로 생성하고 캐싱.
+        if video is None:
+            raise HTTPException(
+                status_code=status.HTTP_404_NOT_FOUND,
+                detail="해당하는 영상을 찾을 수 없습니다.",
+            )
 
-        content_type="ssul" 이면 task_id 자리에 ssul_content.id(문자열)가 온다.
-        캐시 키에 종류 접두를 붙인다 — ADO2 task_id(UUID)와 썰박스 id(숫자)는
-        형식이 달라 실제로 겹치진 않지만, 형식 우연에 기대지 않는다.
-        """
-        cache_key = f"ssul:{task_id}" if content_type == "ssul" else task_id
         logger.info(
-            f"[SEO_SERVICE] Try Cache - user: {current_user.user_uuid} / key: {cache_key}"
+            f"[SEO_SERVICE] Load metadata - user: {current_user.user_uuid} / video_id: {video.id}"
         )
 
-        cached = await self._get_from_redis(cache_key)
-        if cached:
-            return cached
-
-        logger.info(f"[SEO_SERVICE] Cache miss - user: {current_user.user_uuid}")
-        if content_type == "ssul":
-            result = await self._generate_ssul_seo(task_id, current_user, session)
-        else:
-            result = await self._generate_seo_description(task_id, current_user, session)
-        await self._set_to_redis(cache_key, result)
+        if has_stored_sns_metadata(video):
+            return self._response_from_video(video)
 
+        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_ssul_seo(
+    async def get_ssul_seo(
         self,
-        content_id: str,
+        content_id: int,
         current_user: User,
         session: AsyncSession,
     ) -> YoutubeDescriptionResponse:
-        """썰박스 콘텐츠용 SEO 생성 — ADO2 와 **다른 프롬프트**(시트 ssul_upload)를 쓴다.
-
-        ADO2 는 업장 마케팅 분석 보고서 기반의 광고 영상 SEO 지만, 썰박스는
-        병맛 역사 썰툰이라 톤이 완전히 다르다. 태그도 GPT 가 함께 만든다
-        (ADO2 처럼 재사용할 마케팅 분석 target_keywords 가 없다).
-        """
+        """썰박스 콘텐츠용 SEO 생성 — ADO2 와 다른 프롬프트(시트 ssul_upload)를 쓴다."""
         from app.ssulbox.constants import SCENARIO_NAMES
-        from app.ssulbox.models import SsulContent
+        from app.utils.prompts.chatgpt_prompt import ChatgptService
         from app.utils.prompts.prompts import get_ssul_upload_prompt
 
         try:
             content = (
                 await session.execute(
                     select(SsulContent).where(
-                        SsulContent.id == int(content_id),
+                        SsulContent.id == content_id,
                         SsulContent.user_uuid == current_user.user_uuid,
                         SsulContent.is_deleted.is_(False),
                     )
@@ -125,31 +116,97 @@ class SeoService:
                 detail=f"썰박스 SEO 생성에 실패했습니다. : {str(e)}",
             )
 
+    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 _get_owned_video_by_task(
+        self,
+        task_id: str,
+        user_uuid: str,
+        session: AsyncSession,
+    ) -> Video | None:
+        result = await session.execute(
+            select(Video)
+            .join(Project, Project.id == Video.project_id)
+            .where(
+                Video.task_id == task_id,
+                Project.user_uuid == user_uuid,
+                Video.is_deleted.is_(False),
+            )
+            .order_by(Video.created_at.desc())
+            .limit(1)
+        )
+        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(
-                    Project.task_id == task_id,
-                    Project.user_uuid == current_user.user_uuid,
-                )
+                .where(Project.task_id == task_id)
                 .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"]
 
@@ -161,12 +218,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,
@@ -174,6 +232,8 @@ class SeoService:
                 keywords=hashtags,
             )
 
+        except HTTPException:
+            raise
         except Exception as e:
             logger.error(f"[SEO_SERVICE] EXCEPTION - error: {e}")
             raise HTTPException(
@@ -181,18 +241,12 @@ class SeoService:
                 detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
             )
 
-    async def _get_from_redis(self, 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:
-            return YoutubeDescriptionResponse(**json.loads(yt_seo_info))
-        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)
+    def _response_from_video(self, video: Video) -> YoutubeDescriptionResponse:
+        return YoutubeDescriptionResponse(
+            title=video.title or "",
+            description=video.description or "",
+            keywords=list(video.hashtags or []),
+        )
 
 
 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.home.models import Project
@@ -113,6 +114,14 @@ class SocialUploadService:
                     detail="영상이 아직 준비되지 않았습니다. 영상 생성이 완료된 후 시도해주세요.",
                 )
 
+        if body.content_type != "ssul" and 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/utils/video_poster.py

@@ -0,0 +1,173 @@
+"""영상 파일에서 SNS 공유용 포스터 이미지를 생성하고 저장합니다."""
+
+import asyncio
+from pathlib import Path
+
+from app.utils.logger import get_logger
+from app.utils.upload_blob_as_request import AzureBlobUploader
+
+logger = get_logger("video_poster")
+
+FFMPEG_TIMEOUT_SECONDS = 30.0
+FFMPEG_CLEANUP_TIMEOUT_SECONDS = 5.0
+_STDERR_LOG_LIMIT = 500
+
+
+async def _kill_and_wait(process: asyncio.subprocess.Process) -> None:
+    """실행 중인 ffmpeg 프로세스를 종료하고 자원을 회수합니다."""
+    try:
+        process.kill()
+    except ProcessLookupError:
+        pass
+    except Exception as exc:
+        logger.warning(
+            "[video_poster] ffmpeg 프로세스 종료에 실패했습니다: %s",
+            exc,
+        )
+
+    try:
+        await asyncio.wait_for(
+            process.wait(),
+            timeout=FFMPEG_CLEANUP_TIMEOUT_SECONDS,
+        )
+    except TimeoutError:
+        logger.warning(
+            "[video_poster] 종료한 ffmpeg 프로세스 회수 시간이 초과되었습니다 "
+            "(timeout=%ss)",
+            FFMPEG_CLEANUP_TIMEOUT_SECONDS,
+        )
+    except Exception as exc:
+        logger.warning(
+            "[video_poster] ffmpeg 프로세스 회수에 실패했습니다: %s",
+            exc,
+        )
+
+
+def _format_stderr(stderr: bytes) -> str:
+    """ffmpeg 표준 오류를 로그에 안전한 길이의 문자열로 변환합니다."""
+    return stderr.decode("utf-8", errors="replace").strip()[-_STDERR_LOG_LIMIT:]
+
+
+async def extract_first_frame(video_path: str | Path) -> bytes | None:
+    """로컬 영상의 첫 프레임을 JPEG 바이트로 추출하고 실패 시 ``None``을 반환합니다."""
+    process: asyncio.subprocess.Process | None = None
+
+    try:
+        process = await asyncio.create_subprocess_exec(
+            "ffmpeg",
+            "-nostdin",
+            "-hide_banner",
+            "-loglevel",
+            "error",
+            "-i",
+            str(video_path),
+            "-frames:v",
+            "1",
+            "-f",
+            "image2pipe",
+            "-c:v",
+            "mjpeg",
+            "pipe:1",
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+
+        try:
+            stdout, stderr = await asyncio.wait_for(
+                process.communicate(),
+                timeout=FFMPEG_TIMEOUT_SECONDS,
+            )
+        except TimeoutError:
+            await _kill_and_wait(process)
+            logger.warning(
+                "[video_poster] ffmpeg 첫 프레임 추출 시간이 초과되었습니다 "
+                "(path=%s, timeout=%ss)",
+                video_path,
+                FFMPEG_TIMEOUT_SECONDS,
+            )
+            return None
+
+        if process.returncode != 0:
+            logger.warning(
+                "[video_poster] ffmpeg 첫 프레임 추출에 실패했습니다 "
+                "(path=%s, returncode=%s, stderr=%s)",
+                video_path,
+                process.returncode,
+                _format_stderr(stderr),
+            )
+            return None
+
+        if not stdout:
+            logger.warning(
+                "[video_poster] ffmpeg가 빈 이미지를 반환했습니다 (path=%s)",
+                video_path,
+            )
+            return None
+
+        return stdout
+
+    except asyncio.CancelledError:
+        if process is not None:
+            await _kill_and_wait(process)
+        raise
+    except Exception as exc:
+        if process is not None:
+            await _kill_and_wait(process)
+        logger.warning(
+            "[video_poster] 첫 프레임 추출 중 오류가 발생했습니다 "
+            "(path=%s, error=%s: %s)",
+            video_path,
+            type(exc).__name__,
+            exc,
+        )
+        return None
+
+
+async def generate_and_store_poster(
+    *,
+    video_path: str | Path,
+    user_uuid: str,
+    task_id: str,
+    file_stem: str,
+) -> str | None:
+    """첫 프레임을 Blob에 저장하고 공개 URL을 반환하며, 실패 시 ``None``을 반환합니다."""
+    try:
+        image_bytes = await extract_first_frame(video_path)
+        if image_bytes is None:
+            return None
+
+        uploader = AzureBlobUploader(user_uuid=user_uuid, task_id=task_id)
+        uploaded = await uploader.upload_image_bytes(
+            image_bytes,
+            f"{file_stem}.jpg",
+        )
+        if not uploaded:
+            logger.warning(
+                "[video_poster] 포스터 Blob 업로드에 실패했습니다 "
+                "(path=%s, task_id=%s)",
+                video_path,
+                task_id,
+            )
+            return None
+
+        if not uploader.public_url:
+            logger.warning(
+                "[video_poster] 포스터 업로드 후 공개 URL이 비어 있습니다 "
+                "(path=%s, task_id=%s)",
+                video_path,
+                task_id,
+            )
+            return None
+
+        return uploader.public_url
+
+    except Exception as exc:
+        logger.warning(
+            "[video_poster] 포스터 생성 또는 저장 중 오류가 발생했습니다 "
+            "(path=%s, task_id=%s, error=%s: %s)",
+            video_path,
+            task_id,
+            type(exc).__name__,
+            exc,
+        )
+        return None

app/video/api/routers/v1/video.py

@@ -15,7 +15,8 @@ Video API Router
 
 from typing import Literal
 
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request
+from fastapi.responses import HTMLResponse
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -60,9 +61,10 @@ from app.video.worker.video_task import (
     _fail_and_refund,
     download_and_upload_video_to_blob,
 )
+from app.video.services.share_page import build_video_share_html, get_video_share_data
 
 
-from config import creatomate_settings
+from config import creatomate_settings, prj_settings
 
 logger = get_logger("video")
 
@@ -956,6 +958,7 @@ async def get_all_videos(
                     video_id=it.id,
                     store_name=it.store_name,
                     result_movie_url=it.movie_url,
+                    poster_url=it.poster_url,
                     created_at=it.created_at,
                     like_count=it.like_count,
                     is_liked_by_me=it.is_liked_by_me,
@@ -1077,6 +1080,43 @@ async def toggle_like(
         raise HTTPException(status_code=500, detail=f"좋아요 처리에 실패했습니다: {str(e)}")
 
 
+@router.get(
+    "/share/{video_id}",
+    response_class=HTMLResponse,
+    summary="영상 공유용 Open Graph 페이지",
+    description="영상별 제목, 설명, 포스터 메타데이터가 포함된 공개 HTML을 반환합니다.",
+    responses={
+        200: {"description": "공유 메타데이터 HTML 반환"},
+        404: {"description": "공유 가능한 완료 영상을 찾을 수 없음"},
+    },
+)
+async def get_video_share_page(
+    video_id: int,
+    request: Request,
+    session: AsyncSession = Depends(get_session),
+) -> HTMLResponse:
+    """공개 공유 페이지를 반환하고 일반 브라우저는 영상 상세로 이동시킵니다."""
+    share_data = await get_video_share_data(session, video_id)
+    if share_data is None:
+        raise HTTPException(status_code=404, detail="공유 가능한 영상을 찾을 수 없습니다.")
+
+    share_url = str(request.url).split("?", maxsplit=1)[0]
+    html = build_video_share_html(
+        share_data,
+        share_url=share_url,
+        frontend_base_url=prj_settings.SHARE_FRONTEND_URL,
+        configured_default_image_url=prj_settings.SHARE_DEFAULT_IMAGE_URL,
+    )
+    return HTMLResponse(
+        content=html,
+        headers={
+            "Cache-Control": "public, max-age=300",
+            "Referrer-Policy": "no-referrer",
+            "X-Content-Type-Options": "nosniff",
+        },
+    )
+
+
 @router.get(
     "/{video_id}",
     summary="단일 영상 상세 조회",
@@ -1153,6 +1193,7 @@ async def get_video_detail(
         return VideoDetailResponse(
             video_id=video.id,
             result_movie_url=video.result_movie_url,
+            poster_url=video.poster_url,
             store_name=project.store_name,
             region=project.region or _extract_region_from_address(project.detail_region_info),
             created_at=video.created_at,

app/video/models.py

@@ -10,9 +10,11 @@ from sqlalchemy import (
     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
@@ -40,6 +42,10 @@ class Video(Base):
         task_id: 영상 생성 작업의 고유 식별자 (UUID7 형식)
         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:
@@ -117,6 +123,30 @@ class Video(Base):
         comment="생성된 영상 URL",
     )
 
+    poster_url: Mapped[Optional[str]] = mapped_column(
+        String(2048),
+        nullable=True,
+        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, Literal, Optional
+from typing import Any, Dict, List, Literal, Optional
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -148,6 +148,7 @@ class VideoListItem(BaseModel):
             "region": "군산",
             "task_id": "019123ab-cdef-7890-abcd-ef1234567890",
             "result_movie_url": "http://localhost:8000/media/2025-01-15/video.mp4",
+            "poster_url": "http://localhost:8000/media/2025-01-15/video.jpg",
             "created_at": "2025-01-15T12:00:00"
         }
     """
@@ -169,9 +170,17 @@ class VideoListItem(BaseModel):
         description="작업 고유 식별자 (ADO2 전용. 썰박스는 개념이 없어 빈 문자열)",
     )
     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="댓글 수 (대댓글 포함)")
+    is_liked_by_me: bool = Field(
+        False,
+        description="현재 로그인 사용자가 좋아요를 눌렀는지",
+    )
 
 
 class VideoThumbnailItem(BaseModel):
@@ -190,7 +199,8 @@ class VideoThumbnailItem(BaseModel):
     )
     video_id: int = Field(..., description="콘텐츠 고유 ID (type 안에서만 유일)")
     store_name: str = Field(..., description="업체명")
-    result_movie_url: str = Field(..., description="영상 URL — 프론트에서 <video> 태그 첫 프레임을 썸네일로 사용")
+    result_movie_url: str = Field(..., description="영상 URL")
+    poster_url: Optional[str] = Field(None, description="영상 첫 프레임 포스터 이미지 URL (썸네일 표시용)")
     created_at: datetime = Field(..., description="생성 일시")
     like_count: int = Field(..., description="좋아요 수")
     is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지 (비로그인은 항상 false)")
@@ -206,6 +216,7 @@ class VideoDetailResponse(BaseModel):
 
     video_id: int = Field(..., description="영상 고유 ID")
     result_movie_url: str = Field(..., description="영상 URL")
+    poster_url: Optional[str] = Field(None, description="영상 첫 프레임 포스터 이미지 URL")
     store_name: Optional[str] = Field(None, description="업체명")
     region: Optional[str] = Field(None, description="지역명")
     created_at: datetime = Field(..., description="생성 일시")

app/video/services/share_page.py

@@ -0,0 +1,188 @@
+"""영상 공유 링크용 Open Graph HTML 생성 서비스."""
+
+from dataclasses import dataclass
+from html import escape
+from urllib.parse import urlsplit, urlunsplit
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.home.models import Project
+from app.video.models import Video
+
+FALLBACK_FRONTEND_URL = "https://ado2.o2osolution.ai"
+DEFAULT_SHARE_IMAGE_PATH = "/assets/images/ado2_image.png"
+DEFAULT_SHARE_IMAGE_STATIC_PATH = "/static/images/ado2_image.png"
+
+@dataclass(frozen=True, slots=True)
+class VideoShareData:
+    """공유 페이지에 필요한 영상 및 프로젝트 정보."""
+
+    video_id: int
+    poster_url: str | None
+    store_name: str
+    region: str
+
+
+async def get_video_share_data(
+    session: AsyncSession,
+    video_id: int,
+) -> VideoShareData | None:
+    """공유 가능한 완료 영상을 프로젝트 정보와 함께 조회합니다."""
+    result = await session.execute(
+        select(
+            Video.id,
+            Video.poster_url,
+            Project.store_name,
+            Project.region,
+        )
+        .join(Project, Video.project_id == Project.id)
+        .where(
+            Video.id == video_id,
+            Video.status == "completed",
+            Video.is_deleted.is_(False),
+            Project.is_deleted.is_(False),
+        )
+    )
+    row = result.one_or_none()
+    if row is None:
+        return None
+
+    return VideoShareData(
+        video_id=row.id,
+        poster_url=row.poster_url,
+        store_name=row.store_name,
+        region=row.region,
+    )
+
+
+def build_video_share_html(
+    data: VideoShareData,
+    *,
+    share_url: str,
+    frontend_base_url: str,
+    configured_default_image_url: str = "",
+) -> str:
+    """영상별 OG 메타데이터와 상세 화면 이동 기능을 포함한 HTML을 생성합니다."""
+    frontend_base = _normalise_frontend_base_url(frontend_base_url)
+    detail_url = f"{frontend_base}/video/{data.video_id}"
+    fallback_image_url = _resolve_default_image_url(
+        configured_default_image_url,
+        frontend_base,
+        share_url=share_url,
+    )
+    image_url = _absolute_http_url(data.poster_url) or fallback_image_url
+
+    store_name = _normalise_text(data.store_name, "ADO2 영상")
+    region = _normalise_text(data.region, "")
+    title = f"{store_name} | ADO2"
+    description = (
+        f"{region} · {store_name} 어떤 내용이 들어가야할지 정해야 합니다. 기존 영상 업로드시에 생성되는 내용을 사용하려면 db에 저장하고 업로드 할때마다 변경되는 부분을 전부 수정해야 합니다."
+        if region
+        else "ADO2 AI 마케팅 영상"
+    )
+
+    escaped_title = escape(title, quote=True)
+    escaped_description = escape(description, quote=True)
+    escaped_image_url = escape(image_url, quote=True)
+    escaped_share_url = escape(_absolute_http_url(share_url) or detail_url, quote=True)
+    escaped_detail_url = escape(detail_url, quote=True)
+
+    return f"""<!doctype html>
+<html lang="ko">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>{escaped_title}</title>
+  <meta name="description" content="{escaped_description}">
+  <link rel="canonical" href="{escaped_share_url}">
+
+  <meta property="og:title" content="{escaped_title}">
+  <meta property="og:description" content="{escaped_description}">
+  <meta property="og:image" content="{escaped_image_url}">
+  <meta property="og:image:alt" content="{escaped_title}">
+  <meta property="og:url" content="{escaped_share_url}">
+  <meta property="og:type" content="video.other">
+
+  <meta name="twitter:card" content="summary_large_image">
+  <meta name="twitter:title" content="{escaped_title}">
+  <meta name="twitter:description" content="{escaped_description}">
+  <meta name="twitter:image" content="{escaped_image_url}">
+
+</head>
+<body>
+  <main>
+    <h1>{escaped_title}</h1>
+    <p>{escaped_description}</p>
+    <a id="continue-link" href="{escaped_detail_url}">영상 보기</a>
+  </main>
+  <script>
+    window.location.replace(document.getElementById("continue-link").href);
+  </script>
+</body>
+</html>
+"""
+
+
+def _normalise_text(value: str | None, fallback: str) -> str:
+    """메타데이터용 텍스트에서 불필요한 공백을 제거합니다."""
+    normalised = " ".join((value or "").split())
+    return normalised or fallback
+
+
+def _normalise_frontend_base_url(value: str) -> str:
+    """프론트엔드 기준 URL을 안전한 절대 HTTP(S) URL로 정규화합니다."""
+    absolute_url = _absolute_http_url(value) or FALLBACK_FRONTEND_URL
+    parts = urlsplit(absolute_url)
+    path = parts.path.rstrip("/")
+    return urlunsplit((parts.scheme, parts.netloc, path, "", ""))
+
+
+def _resolve_default_image_url(
+    configured_url: str,
+    frontend_base: str,
+    *,
+    share_url: str = "",
+) -> str:
+    """포스터가 없을 때 사용할 기본 OG 이미지 URL을 반환합니다.
+
+    우선순위:
+    1. ``SHARE_DEFAULT_IMAGE_URL`` (.env)
+    2. 공유 API 호스트 ``/static/images/ado2_image.png``
+    3. ``SHARE_FRONTEND_URL`` + ``/assets/images/ado2_image.png``
+    """
+    configured_absolute_url = _absolute_http_url(configured_url)
+    if configured_absolute_url:
+        return configured_absolute_url
+
+    share_origin = _origin_from_url(share_url)
+    if share_origin:
+        return f"{share_origin}{DEFAULT_SHARE_IMAGE_STATIC_PATH}"
+
+    return f"{frontend_base}{DEFAULT_SHARE_IMAGE_PATH}"
+
+
+def _origin_from_url(value: str) -> str | None:
+    """URL에서 scheme + host(+port) origin만 추출합니다."""
+    absolute_url = _absolute_http_url(value)
+    if not absolute_url:
+        return None
+
+    parts = urlsplit(absolute_url)
+    return urlunsplit((parts.scheme, parts.netloc, "", "", ""))
+
+
+def _absolute_http_url(value: str | None) -> str | None:
+    """값이 절대 HTTP(S) URL인 경우에만 정리된 문자열을 반환합니다."""
+    candidate = (value or "").strip()
+    if not candidate:
+        return None
+
+    try:
+        parts = urlsplit(candidate)
+    except ValueError:
+        return None
+
+    if parts.scheme.lower() not in {"http", "https"} or not parts.hostname:
+        return None
+    return candidate

app/video/services/unified_list.py

@@ -73,6 +73,7 @@ class UnifiedItem:
     like_count: int = 0
     comment_count: int = 0
     is_liked_by_me: bool = False
+    poster_url: Optional[str] = None
 
 
 # ──────────────────────────────────────────────
@@ -190,6 +191,7 @@ def _video_branch(where: list, sort_by: str) -> Select:
         Video.result_movie_url.label("movie_url"),
         Video.created_at.label("created_at"),
         Video.task_id.label("task_id"),
+        Video.poster_url.label("poster_url"),
     ]
     if sort_by == SORT_LIKE:
         cols.append(_video_like_subq().label("sort_value"))
@@ -208,6 +210,7 @@ def _ssul_branch(where: list, sort_by: str) -> Select:
         SsulContent.created_at.label("created_at"),
         # UNION 은 컬럼 수·순서가 양쪽 같아야 한다. 썰박스에는 task_id 가 없다.
         literal("").label("task_id"),
+        SsulContent.poster_url.label("poster_url"),
     ]
     if sort_by == SORT_LIKE:
         cols.append(_ssul_like_subq().label("sort_value"))
@@ -399,6 +402,7 @@ def _to_items(rows) -> list[UnifiedItem]:
             movie_url=r.movie_url,
             created_at=r.created_at,
             task_id=r.task_id or "",
+            poster_url=r.poster_url,
         )
         for r in rows
     ]

app/video/worker/video_task.py

@@ -17,6 +17,7 @@ from app.ssulbox.constants import JOB_TYPE_VIDEO as CREDIT_JOB_TYPE_VIDEO
 from app.video.models import Video
 from app.utils.upload_blob_as_request import AzureBlobUploader
 from app.utils.logger import get_logger
+from app.utils.video_poster import generate_and_store_poster
 
 # 로거 설정
 logger = get_logger("video")
@@ -81,7 +82,8 @@ async def _update_video_status(
     status: str,
     video_url: str | None = None,
     creatomate_render_id: str | None = None,
-) -> bool:
+    poster_url: str | None = None,
+) -> int | None:
     """Video 테이블의 상태를 업데이트합니다.
 
     Args:
@@ -89,9 +91,10 @@ async def _update_video_status(
         status: 변경할 상태 ("processing", "completed", "failed")
         video_url: 영상 URL
         creatomate_render_id: Creatomate render ID (선택)
+        poster_url: 영상 첫 프레임 포스터 URL (선택)
 
     Returns:
-        bool: 업데이트 성공 여부
+        int | None: 업데이트된 Video id. 대상이 없거나 실패하면 None.
     """
     try:
         async with BackgroundSessionLocal() as session:
@@ -116,19 +119,58 @@ async def _update_video_status(
                 video.status = status
                 if video_url is not None:
                     video.result_movie_url = video_url
+                if poster_url is not None:
+                    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(
+    temp_file_path: Path,
+    user_uuid: str,
+    task_id: str,
+    render_id: str,
+) -> str | None:
+    """포스터 생성 실패가 영상 생성 완료 처리에 영향을 주지 않도록 격리합니다."""
+    try:
+        return await generate_and_store_poster(
+            video_path=temp_file_path,
+            user_uuid=user_uuid,
+            task_id=task_id,
+            file_stem=render_id,
+        )
+    except Exception as e:
+        logger.warning(
+            f"[VideoPoster] Failed to generate poster - task_id: {task_id}, render_id: {render_id}, error: {e}",
+            exc_info=True,
+        )
+        return None
 
 
 async def _download_video(url: str, task_id: str) -> bytes:
@@ -204,8 +246,20 @@ async def download_and_upload_video_to_blob(
         blob_url = uploader.public_url
         logger.info(f"[download_and_upload_video_to_blob] Uploaded to Blob - task_id: {task_id}, url: {blob_url}")
 
+        poster_url = await _try_generate_poster(
+            temp_file_path, user_uuid, task_id, creatomate_render_id
+        )
+
         # Video 테이블 업데이트 (creatomate_render_id로 특정 Video 식별)
-        await _update_video_status(task_id, "completed", blob_url, creatomate_render_id)
+        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)
 
         # 크레딧은 generate_video 에서 이미 선차감했다. 여기서 차감하지 않는다.
 
@@ -307,13 +361,20 @@ async def download_and_upload_video_by_creatomate_render_id(
         blob_url = uploader.public_url
         logger.info(f"[download_and_upload_video_by_creatomate_render_id] Uploaded to Blob - creatomate_render_id: {creatomate_render_id}, url: {blob_url}")
 
+        poster_url = await _try_generate_poster(
+            temp_file_path, user_uuid, task_id, 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:

config.py

@@ -33,6 +33,17 @@ class ProjectSettings(BaseSettings):
     ADMIN_BASE_URL: str = Field(default="/admin")
     ADMIN_SESSION_SECRET: str = Field(default="dev-secret-change-me-in-production")
     ADMIN_SESSION_MAX_AGE: int = Field(default=60 * 60 * 8)
+    SHARE_FRONTEND_URL: str = Field(
+        default="https://ado2.o2osolution.ai",
+        description="공유 페이지에서 영상 상세로 이동할 프론트엔드 공개 기준 URL (.env: SHARE_FRONTEND_URL)",
+    )
+    SHARE_DEFAULT_IMAGE_URL: str = Field(
+        default="",
+        description=(
+            "포스터가 없는 영상 공유 시 사용할 절대 이미지 URL (.env: SHARE_DEFAULT_IMAGE_URL). "
+            "비우면 공유 API /static/images/ado2_image.png, 없으면 SHARE_FRONTEND_URL/assets 경로 사용"
+        ),
+    )
     DEBUG: bool = Field(default=True)
     TIMEZONE: str = Field(
         default="Asia/Seoul",
@@ -174,8 +185,8 @@ class AzureBlobSettings(BaseSettings):
         ge=1,
         description="worker별 이미지 upload named lock 동시 점유 상한",
     )
-    IMAGE_UPLOAD_MAX_TASK_IMAGES: int = Field(
-        default=100,
+    IMAGE_UPLOAD_MAX_TASK_IMAGES: int = Field(
+        default=100,
         ge=1,
         description="한 task에 누적할 수 있는 활성 이미지 최대 개수",
     )

docs/database-schema/migration_2026_08_13_add_video_poster_url.sql

@@ -0,0 +1,10 @@
+-- ============================================================
+-- Migration: video 테이블에 poster_url 컬럼 추가
+-- Date: 2026-08-13
+-- Description: 영상 첫 프레임 포스터 이미지 URL. SNS 공유 og:image 용도.
+--              관련 코드: app/utils/video_poster.py, app/video/worker/video_task.py
+-- ============================================================
+
+ALTER TABLE `video`
+    ADD COLUMN `poster_url` VARCHAR(2048) NULL
+    COMMENT '영상 첫 프레임 포스터 이미지 URL (SNS 공유 og:image용)' AFTER `result_movie_url`;

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`;