.gitignore

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

README.md

@@ -69,9 +69,6 @@ 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_API_BASE_URL=  # 공유 OG 페이지의 외부 공개 API URL (예: https://dev-ssul.castad.net/api). 프록시가 /api 를 떼면 필수
-SHARE_DEFAULT_IMAGE_URL=  # 포스터 없을 때 OG 이미지 (비우면 API /static/images/ado2_image.png)
 DEBUG=True                             # 디버그 모드 (True: 개발, False: 운영)
 
 # ================================

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

@@ -16,13 +16,8 @@ from app.user.dependencies.auth import get_current_user
 from app.user.models import User
 from app.utils.logger import get_logger
 from app.utils.pagination import PaginatedResponse
-from app.utils.upload_blob_as_request import to_playback_url
 from app.comment.models import Comment
-from app.database.like_cache import (
-    bulk_is_user_liked,
-    get_like_counts,
-    mset_like_counts,
-)
+from app.database.like_cache import get_like_counts, mset_like_counts
 from app.video.models import Video, VideoReaction
 from app.video.schemas.video_schema import VideoListItem
 
@@ -154,24 +149,6 @@ async def get_videos(
                 if vid not in db_found_ids:
                     like_count_map[vid] = 0
 
-        # is_liked_by_me: Redis user-set 기준, 캐시 미스 시 현재 사용자 상태만 DB 조회
-        raw_liked = await bulk_is_user_liked(video_ids, current_user.user_uuid)
-        needs_db_lookup = [
-            vid for vid, liked in raw_liked.items()
-            if liked is None and like_count_map.get(vid, 0) > 0
-        ]
-        if needs_db_lookup:
-            liked_video_ids = set((await session.execute(
-                select(VideoReaction.video_id).where(
-                    VideoReaction.video_id.in_(needs_db_lookup),
-                    VideoReaction.user_uuid == current_user.user_uuid,
-                )
-            )).scalars().all())
-            for vid in needs_db_lookup:
-                raw_liked[vid] = vid in liked_video_ids
-
-        liked_map = {vid: bool(liked) for vid, liked in raw_liked.items()}
-
         # VideoListItem으로 변환
         items = [
             VideoListItem(
@@ -179,15 +156,10 @@ async def get_videos(
                 store_name=project.store_name,
                 region=project.region,
                 task_id=video.task_id,
-                result_movie_url=to_playback_url(video.result_movie_url),
-                poster_url=video.poster_url,
-                title=video.title,
-                description=video.description,
-                hashtags=video.hashtags,
+                result_movie_url=video.result_movie_url,
                 created_at=video.created_at,
                 like_count=like_count_map.get(video.id) or 0,
                 comment_count=comment_count or 0,
-                is_liked_by_me=liked_map.get(video.id, False),
             )
             for video, project, comment_count in rows
         ]

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

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

app/comment/models.py

@@ -17,8 +17,7 @@ class Comment(Base):
 
     2-depth 구조 (최상위 댓글 + 대댓글 1단계).
     parent_id가 NULL이면 최상위 댓글, 값이 있으면 대댓글.
-    작성자 닉네임은 카카오 로그인 정보를 작성 시점에 그대로 저장한 스냅샷이며,
-    프로필 이미지는 별도 컬럼 없이 응답 시 User 테이블을 조인해 최신값을 조회한다.
+    작성자(user_uuid)는 DB에 저장하지만 API 응답에는 미노출 (익명 정책).
     """
 
     __tablename__ = "comment"
@@ -55,7 +54,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,6 +5,7 @@ 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")
 
@@ -13,8 +14,7 @@ class ReplyItem(BaseModel):
     """대댓글 응답"""
 
     id: int = Field(..., description="댓글 고유 ID")
-    nickname: str = Field(..., description="작성자 닉네임 (카카오 닉네임, 미보유 시 '익명')")
-    profile_image_url: Optional[str] = Field(None, description="작성자 프로필 이미지 URL (카카오 프로필, 로그인 시점 기준 최신값)")
+    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
     content: Optional[str] = Field(None, description="본문 (소프트 삭제된 경우 null)")
     is_deleted: bool = Field(..., description="삭제 여부")
     is_mine: bool = Field(..., description="현재 로그인 사용자의 댓글 여부")
@@ -25,8 +25,7 @@ class CommentItem(BaseModel):
     """최상위 댓글 응답 — replies 포함"""
 
     id: int = Field(..., description="댓글 고유 ID")
-    nickname: str = Field(..., description="작성자 닉네임 (카카오 닉네임, 미보유 시 '익명')")
-    profile_image_url: Optional[str] = Field(None, description="작성자 프로필 이미지 URL (카카오 프로필, 로그인 시점 기준 최신값)")
+    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
     content: Optional[str] = Field(None, description="본문 (소프트 삭제된 경우 null)")
     is_deleted: bool = Field(..., description="삭제 여부")
     is_mine: bool = Field(..., description="현재 로그인 사용자의 댓글 여부")
@@ -36,8 +35,7 @@ class CommentItem(BaseModel):
 
 class CommentCreateResponse(BaseModel):
     id: int = Field(..., description="생성된 댓글 고유 ID")
-    nickname: str = Field(..., description="작성자 닉네임 (카카오 닉네임, 미보유 시 '익명')")
-    profile_image_url: Optional[str] = Field(None, description="작성자 프로필 이미지 URL (카카오 프로필)")
+    nickname: str = Field(..., description="작성자 닉네임 (미입력 시 '익명')")
     parent_id: Optional[int] = Field(None, description="부모 댓글 id (대댓글인 경우)")
     content: str = Field(..., description="댓글 본문")
     created_at: datetime = Field(..., description="작성 일시")

app/comment/services/comment.py

@@ -7,7 +7,6 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.comment.models import Comment
 from app.comment.schemas.comment_schema import CommentItem, ReplyItem
-from app.user.models import User
 from app.utils.pagination import PaginatedResponse
 from app.video.models import Video
 
@@ -38,7 +37,6 @@ 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:
@@ -47,7 +45,6 @@ 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,
@@ -59,7 +56,6 @@ 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,
@@ -74,7 +70,7 @@ async def create_comment(
     session: AsyncSession,
     video_id: int,
     user_uuid: str,
-    nickname: Optional[str],
+    nickname: str,
     content: str,
     parent_id: Optional[int],
 ) -> Comment:
@@ -147,7 +143,6 @@ 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 = (
@@ -162,16 +157,7 @@ async def list_comments(
         for r in replies:
             replies_map[r.parent_id].append(r)
 
-    # 작성자 프로필 이미지는 스냅샷을 저장하지 않고, 응답 시 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)
+    items = _build_comment_items(list(parents), replies_map, current_user_uuid)
 
     return PaginatedResponse.create(
         items=items,

app/database/session.py

@@ -80,8 +80,7 @@ async def create_db_tables():
     from app.home.models import Image, Project, MarketingIntel, ImageTag  # noqa: F401
     from app.lyric.models import Lyric  # noqa: F401
     from app.song.models import Song, SongTimestamp  # noqa: F401
-    from app.video.models import Video, VideoReaction  # noqa: F401
-    from app.comment.models import Comment  # noqa: F401
+    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
@@ -99,8 +98,6 @@ async def create_db_tables():
         Song.__table__,
         SongTimestamp.__table__,
         Video.__table__,
-        VideoReaction.__table__,
-        Comment.__table__,
         SNSUploadTask.__table__,
         SocialUpload.__table__,
         MarketingIntel.__table__,

app/home/api/routers/v1/home.py

@@ -417,16 +417,8 @@ async def _crawling_logic(url: str, session: AsyncSession):
             )
 
             # Step 4-3: 분석 결과 DB 저장 (industry는 Project로 흐르므로 여기엔 미저장)
-            # 공식 링크: 플레이스 홈페이지 항목 우선, 없으면 유저가 입력한 크롤링 소스 URL
-            # (컬럼 길이를 넘는 긴 검색 URL은 place_id 기반 표준 플레이스 URL로 대체)
-            official_site_url = scraper.official_site_url or url
-            if len(official_site_url) > 2048:
-                official_site_url = (
-                    f"https://map.naver.com/p/entry/place/{scraper.place_id[2:]}"
-                )
             marketing_intel = MarketingIntel(
                 place_id=scraper.place_id,
-                official_site_url=official_site_url,
                 intel_result=marketing_analysis.model_dump(),
             )
             session.add(marketing_intel)
@@ -494,7 +486,6 @@ async def _crawling_logic(url: str, session: AsyncSession):
 - **customer_name**: 업체명 / 브랜드명 (필수)
 - **address**: 도로명 또는 지번 주소 (필수)
 - **category**: 업종/카테고리 자유 입력 (선택, 예: 펜션, 카페). 비우면 업체명 기반 AI 분류
-- **official_site_url**: 업체 공식 홈페이지 링크 (선택, http/https만 허용, 최대 2048자). 영상 응답의 official_site_url로 노출
 
 ## 반환 정보
 - **processed_info**: 가공된 장소 정보 (customer_name, region, detail_region_info)
@@ -538,7 +529,6 @@ async def manual_marketing(
         # Step 3: 분석 결과 DB 저장 (place_id=None — 네이버 장소와 연결되지 않음)
         marketing_intel = MarketingIntel(
             place_id=None,
-            official_site_url=request_body.official_site_url,
             intel_result=marketing_analysis.model_dump(),
         )
         session.add(marketing_intel)

app/home/models.py

@@ -274,7 +274,6 @@ class MarketingIntel(Base):
     Attributes:
         id: 고유 식별자 (자동 증가)
         place_id : 데이터 소스별 식별자
-        official_site_url : 업체 공식 링크 (플레이스 홈페이지 항목, 없으면 크롤링 소스 URL)
         intel_result : 마케팅 분석 결과물 json
         created_at: 생성 일시 (자동 설정)
     """
@@ -303,12 +302,6 @@ class MarketingIntel(Base):
         comment="매장 소스별 고유 식별자 (네이버 크롤링 시 'nv{id}' 형식; 직접 입력 시 NULL)",
     )
 
-    official_site_url: Mapped[Optional[str]] = mapped_column(
-        String(2048),
-        nullable=True,
-        comment="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 시 NULL)",
-    )
-
     intel_result : Mapped[dict[str, Any]] = mapped_column(
         JSON,
         nullable=False,

app/home/schemas/home_schema.py

@@ -1,6 +1,6 @@
 from typing import Literal, Optional
 
-from pydantic import BaseModel, ConfigDict, Field, field_validator
+from pydantic import BaseModel, ConfigDict, Field
 from app.utils.prompts.schemas import MarketingPromptOutput
 
 class CrawlingRequest(BaseModel):
@@ -261,7 +261,6 @@ class ManualMarketingRequest(BaseModel):
                 "store_name": "스테이 머뭄",
                 "address": "전북특별자치도 군산시 절골길 18",
                 "category": "펜션",
-                "official_site_url": "https://www.staymeomoom.com",
             }
         }
     )
@@ -269,23 +268,6 @@ class ManualMarketingRequest(BaseModel):
     store_name: str = Field(..., description="업체명 / 브랜드명")
     address: str = Field(..., description="도로명 또는 지번 주소")
     category: str = Field(default="", description="업체 업종/카테고리 자유 입력 (예: 펜션, 카페). 크롤링 경로와 동일하게 AI가 8개 industry enum으로 자동 분류하는 데 사용. 비우면 업체명 기반 AI 분류")
-    official_site_url: Optional[str] = Field(
-        default=None,
-        max_length=2048,
-        description="업체 공식 홈페이지 링크 (선택, http/https만 허용). 영상 응답의 official_site_url로 노출됨",
-    )
-
-    @field_validator("official_site_url")
-    @classmethod
-    def _validate_official_site_url(cls, v: Optional[str]) -> Optional[str]:
-        if v is None:
-            return None
-        v = v.strip()
-        if not v:
-            return None
-        if not v.startswith(("http://", "https://")):
-            raise ValueError("official_site_url은 http:// 또는 https://로 시작해야 합니다.")
-        return v
 
 
 class ErrorResponse(BaseModel):

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.video_id, current_user, session
+        request_body.task_id, current_user, session
     )

app/social/constants.py

@@ -95,6 +95,8 @@ 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 제안 요청"""
 
-    video_id: int = Field(..., description="영상 고유 ID")
+    task_id: str = Field(..., description="작업 고유 식별자")
 
     model_config = ConfigDict(
         json_schema_extra={
             "example": {
-                "video_id": 123
+                "task_id": "019c739f-65fc-7d15-8c88-b31be00e588e"
             }
         }
     )

app/social/services/seo_service.py

@@ -1,132 +1,88 @@
 """
 유튜브 SEO 서비스
 
-영상 제목/설명/해시태그를 생성하고 video 테이블에 저장합니다.
+SEO description 생성 및 Redis 캐싱 로직을 처리합니다.
 """
 
 import json
 import logging
 
-from fastapi import HTTPException, status
+from fastapi import HTTPException
+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.video.models import Video
+from app.utils.prompts.chatgpt_prompt import ChatgptService
+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,
-        video_id: int,
+        task_id: str,
         current_user: User,
         session: AsyncSession,
     ) -> YoutubeDescriptionResponse:
         """
-        저장된 SNS 메타데이터를 반환하거나, 없으면 생성 후 video에 저장합니다.
+        유튜브 SEO description 생성
+
+        Redis 캐시 확인 후 miss이면 GPT로 생성하고 캐싱.
         """
         logger.info(
-            f"[SEO_SERVICE] Load metadata - user: {current_user.user_uuid} / video_id: {video_id}"
+            f"[SEO_SERVICE] Try Cache - user: {current_user.user_uuid} / task_id: {task_id}"
         )
 
-        video = await self._get_owned_video(video_id, current_user.user_uuid, session)
-        if video is None:
-            raise HTTPException(
-                status_code=status.HTTP_404_NOT_FOUND,
-                detail=f"video_id '{video_id}'에 해당하는 영상을 찾을 수 없습니다.",
-            )
+        cached = await self._get_from_redis(task_id)
+        if cached:
+            return cached
 
-        if has_stored_sns_metadata(video):
-            return self._response_from_video(video)
+        logger.info(f"[SEO_SERVICE] Cache miss - user: {current_user.user_uuid}")
+        result = await self._generate_seo_description(task_id, current_user, session)
+        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 생성"""
-        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}")
+        logger.info(f"[SEO_SERVICE] Generating SEO - user: {current_user.user_uuid}")
 
         try:
             project_result = await session.execute(
                 select(Project)
-                .where(Project.task_id == task_id)
+                .where(
+                    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"]
 
@@ -138,13 +94,12 @@ class SeoService:
                 ),
                 "language": project.language,
                 "target_keywords": hashtags,
-                "industry": project.industry or "",
+                "industry": project.industry or "",  # 크롤 시 분류해 Project에 저장한 업종 enum
             }
 
+            # 업종 분기는 프롬프트 내부 {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,
@@ -152,8 +107,6 @@ class SeoService:
                 keywords=hashtags,
             )
 
-        except HTTPException:
-            raise
         except Exception as e:
             logger.error(f"[SEO_SERVICE] EXCEPTION - error: {e}")
             raise HTTPException(
@@ -161,12 +114,18 @@ class SeoService:
                 detail=f"유튜브 SEO 생성에 실패했습니다. : {str(e)}",
             )
 
-    def _response_from_video(self, video: Video) -> YoutubeDescriptionResponse:
-        return YoutubeDescriptionResponse(
-            title=video.title or "",
-            description=video.description or "",
-            keywords=list(video.hashtags or []),
-        )
+    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)
 
 
 seo_service = SeoService()

app/social/services/sns_metadata.py

@@ -1,36 +0,0 @@
-"""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,7 +26,6 @@ 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
@@ -77,12 +76,6 @@ 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/utils/autotag.py

@@ -6,10 +6,6 @@ from app.utils.prompts.schemas import SpaceType, Subject, Camera, MotionRecommen
 
 import asyncio
 
-# medium 추론은 출력 비용의 80%를 차지하고, minimal은 A/B 비교(4회)에서 narrative 점수가
-# welcome 단계로 편향되고 태그를 과다 선택하는 패턴이 반복돼 low로 고정한다.
-IMAGE_TAG_REASONING_EFFORT = "low"
-
 async def autotag_image(image_url : str, industry: str = "") -> list[str]: #tag_list
     chatgpt = ChatgptService(model_type="gpt")
     image_input_data = {
@@ -21,7 +17,7 @@ async def autotag_image(image_url : str, industry: str = "") -> list[str]: #tag_
         "motion_recommended" : list(MotionRecommended)
     }
 
-    image_result = await chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_url, True, reasoning_effort=IMAGE_TAG_REASONING_EFFORT)
+    image_result = await chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_url, False)   
     return image_result
 
 async def autotag_images(image_url_list : list[str], industry: str = "") -> list[dict]: #tag_list
@@ -35,7 +31,7 @@ async def autotag_images(image_url_list : list[str], industry: str = "") -> list
         "motion_recommended" : list(MotionRecommended)
     }for image_url in image_url_list]
     
-    image_result_tasks = [chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_input_data['img_url'], True, silent = True, reasoning_effort=IMAGE_TAG_REASONING_EFFORT) for image_input_data in image_input_data_list]
+    image_result_tasks = [chatgpt.generate_structured_output(image_autotag_prompt, image_input_data, image_input_data['img_url'], False, silent = True) for image_input_data in image_input_data_list] 
     image_result_list: list[BaseModel | BaseException] = await asyncio.gather(*image_result_tasks, return_exceptions=True)
     MAX_RETRY = 2
     for _ in range(MAX_RETRY):
@@ -44,7 +40,7 @@ async def autotag_images(image_url_list : list[str], industry: str = "") -> list
         if not failed_idx:
             break
         retried = await asyncio.gather(
-            *[chatgpt.generate_structured_output(image_autotag_prompt, image_input_data_list[i], image_input_data_list[i]['img_url'], True, silent=True, reasoning_effort=IMAGE_TAG_REASONING_EFFORT) for i in failed_idx],
+            *[chatgpt.generate_structured_output(image_autotag_prompt, image_input_data_list[i], image_input_data_list[i]['img_url'], False, silent=True) for i in failed_idx],
             return_exceptions=True
         )
         for i, result in zip(failed_idx, retried):

app/utils/creatomate.py

@@ -574,15 +574,8 @@ class CreatomateService:
           다운로드 실패를 배제로 오판해 풀 전체가 날아가는 것을 막기 위함.
         일반 씬 슬롯은 태그 점수를 그대로 반환한다.
         """
-        is_thumbnail = slot.endswith(THUMBNAIL_SLOT_MARKER)
-        if is_thumbnail and self.parse_slot_name_to_tag(slot) is None:
-            # 슬롯명이 명명 규칙을 어겨 태그 매칭이 불가능한 썸네일 슬롯.
-            # 태그 점수를 0으로 두면 pool 첫 컷이 뽑혀 사실상 무작위가 되므로
-            # 전 이미지 중립(1.0)으로 두고 아래 픽셀 적합도만으로 순위를 가른다.
-            scores = [1.0] * len(pool_subset)
-        else:
-            scores = self.calculate_image_slot_score_multi(pool_subset, slot)
-        if not thumbnail_fitness_map or not is_thumbnail:
+        scores = self.calculate_image_slot_score_multi(pool_subset, slot)
+        if not thumbnail_fitness_map or not slot.endswith(THUMBNAIL_SLOT_MARKER):
             return scores
 
         adjusted = []
@@ -596,31 +589,6 @@ class CreatomateService:
                 adjusted.append(score * fitness["score"])
         return adjusted
 
-    def _collect_thumbnail_slots(self, template_component_data: dict) -> list[str]:
-        """배정 대상 썸네일 슬롯(-9999)을 수집합니다.
-
-        일반 씬 슬롯과 달리 슬롯명 파싱에 실패해도 제외하지 않는다. 썸네일은
-        노출 면적이 가장 큰 표면이라, 미배정 시 modify_element가 템플릿 원본
-        (샘플 이미지)을 그대로 남겨 완성 영상에 그대로 나가기 때문이다. 태그
-        매칭이 불가능한 슬롯은 _slot_scores_with_fitness가 픽셀 적합도만으로
-        고른다. 단 '-fixed' 고정 자산은 여기서도 제외한다.
-
-        파싱 실패는 템플릿 슬롯명 오타이므로 ERROR로 남겨 드러나게 한다
-        (조용히 넘어가면 샘플 이미지가 나가도 아무도 알아채지 못한다).
-        """
-        slots = [
-            name for name, t in template_component_data.items()
-            if t == "image" and name.endswith(THUMBNAIL_SLOT_MARKER)
-            and not is_fixed_slot_name(name)
-        ]
-        for name in slots:
-            if self.parse_slot_name_to_tag(name) is None:
-                logger.error(
-                    f"[_collect_thumbnail_slots] 썸네일 슬롯명이 명명 규칙 위반 — "
-                    f"'{name}' — 템플릿 슬롯명 수정 필요. 픽셀 적합도 기준으로 폴백 배정합니다."
-                )
-        return slots
-
     def rank_thumbnail_candidates(
         self,
         template: dict,
@@ -639,7 +607,11 @@ class CreatomateService:
             후보가 없는 슬롯은 키 자체를 포함하지 않는다.
         """
         component = self.parse_template_component_name(template["source"]["elements"])
-        thumbnail_slots = self._collect_thumbnail_slots(component)
+        thumbnail_slots = [
+            name for name, t in component.items()
+            if t == "image" and name.endswith(THUMBNAIL_SLOT_MARKER)
+            and not is_fixed_slot_name(name) and self.parse_slot_name_to_tag(name) is not None
+        ]
         result: dict[str, list[dict]] = {}
         for slot in thumbnail_slots:
             scores = self._slot_scores_with_fitness(taged_image_list, slot, thumbnail_fitness_map)
@@ -697,13 +669,9 @@ class CreatomateService:
         # 않는 image 요소는 콘텐츠 슬롯이 아니므로 배정 대상에서 제외 — 그렇지
         # 않으면 파싱 실패로 0점 처리되어 "가장 까다로운 슬롯"으로 취급되고
         # 무작위 이미지로 덮어써진다.
-        # 썸네일 슬롯(-9999)은 파싱 실패해도 배정해야 하므로 여기서 제외하고
-        # _collect_thumbnail_slots가 별도로 수집한다.
         image_slots = [
             name for name, t in template_component_data.items()
-            if t == "image" and not is_fixed_slot_name(name)
-            and not name.endswith(THUMBNAIL_SLOT_MARKER)
-            and self.parse_slot_name_to_tag(name) is not None
+            if t == "image" and not is_fixed_slot_name(name) and self.parse_slot_name_to_tag(name) is not None
         ]
         text_slots  = [(name, t) for name, t in template_component_data.items() if t == "text"]
 
@@ -718,7 +686,8 @@ class CreatomateService:
         # thumbnail_choice(비전 LLM 최종 선택)가 해당 슬롯에 있으면 그 이미지를,
         # 없으면(선택 실패/미제공) 결정론적 최고점 컷을 사용한다 — 폴백 안전.
         thumbnail_choice = thumbnail_choice or {}
-        thumbnail_slots = self._collect_thumbnail_slots(template_component_data)
+        thumbnail_slots = [s for s in image_slots if s.endswith(THUMBNAIL_SLOT_MARKER)]
+        image_slots = [s for s in image_slots if not s.endswith(THUMBNAIL_SLOT_MARKER)]
         for slot in thumbnail_slots:
             if not pool:
                 logger.warning(f"[template_matching_taged_image] 이미지 풀 없음 — 썸네일 슬롯 배정 불가: {slot}")
@@ -916,13 +885,7 @@ class CreatomateService:
         """슬롯 이름을 파싱하여 태그 딕셔너리를 반환합니다.
 
         슬롯 이름 형식: {space_type}-{subject}-{camera}-{motion}-{narrative}
-
-        위치 기반 파싱에 실패하면 토큰 위치를 무시한 대조로 한 번 더 시도한다
-        (_parse_slot_name_loosely). 슬롯명 오타로 슬롯이 배정에서 빠지면
-        modify_element가 템플릿 원본(샘플 이미지)을 그대로 남겨 완성 영상에
-        그대로 나가기 때문이다.
-
-        둘 다 실패하면 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리).
+        파싱 실패 시 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리).
         """
         try:
             tag_list = slot_name.split("-")
@@ -944,64 +907,9 @@ class CreatomateService:
             }
             return tag_dict
         except (ValueError, IndexError) as e:
-            loose = self._parse_slot_name_loosely(tag_list)
-            if loose is not None:
-                logger.warning(
-                    f"[parse_slot_name_to_tag] 슬롯명이 명명 규칙 위반: '{slot_name}' — {e} — "
-                    f"위치 무시 대조로 복구: { {k: v.value for k, v in loose.items()} } — 템플릿 슬롯명 수정 권장"
-                )
-                return loose
             logger.warning(f"[parse_slot_name_to_tag] 슬롯명 파싱 실패: '{slot_name}' — {e} — 슬롯 skip")
             return None
 
-    def _parse_slot_name_loosely(self, tag_list: list[str]) -> dict[str, StrEnum] | None:
-        """토큰 위치를 무시하고 각 enum에 대조해 태그를 복구합니다.
-
-        위치 기반 파싱이 실패했을 때만 호출한다. 토큰 하나는 한 카테고리에만
-        쓰이며, 앞선 토큰부터 순서대로 소비한다(중복 후보가 있으면 앞선 것 채택
-        — 위치 기반과 같은 값을 고르게 된다).
-
-        space_type/subject/narrative는 필수다. 이 셋을 못 채우면 슬롯명이 아닌
-        것으로 보고 None을 반환한다(고정 자산·비규칙 요소가 배정 대상에 섞여
-        무작위 이미지로 덮어써지는 것을 막기 위함). camera/motion은 선택이며
-        못 찾으면 키 자체를 넣지 않는다 — 점수 계산은 태그 딕셔너리를 순회하므로
-        없는 키는 자연히 가중치에서 빠진다.
-        """
-        used: set[int] = set()
-
-        def take(converter) -> StrEnum | None:
-            for idx, token in enumerate(tag_list):
-                if idx in used:
-                    continue
-                try:
-                    value = converter(token)
-                except ValueError:
-                    continue
-                if value is not None:
-                    used.add(idx)
-                    return value
-            return None
-
-        space_type = take(SpaceType)
-        subject = take(Subject)
-        narrative = take(NarrativePhase)
-        if space_type is None or subject is None or narrative is None:
-            return None
-
-        camera = take(Camera)
-        motion = take(lambda t: MOTION_TOKEN_NORMALIZATION.get(t) or MotionRecommended(t))
-
-        tag_dict: dict[str, StrEnum] = {
-            "space_type": space_type,
-            "subject": subject,
-            "narrative_preference": narrative,
-        }
-        if camera is not None:
-            tag_dict["camera"] = camera
-        if motion is not None:
-            tag_dict["motion_recommended"] = motion
-        return tag_dict
-
     def elements_connect_resource_blackbox(
         self,
         elements: list,

app/utils/nvMapPwScraper.py

@@ -131,8 +131,7 @@ if (originalQuery) {
         cls,
         place_id: str,
         payloads: list[dict],
-        capture_apollo: bool = False,
-    ) -> list[dict | None] | None | tuple[list[dict | None] | None, str | None]:
+    ) -> list[dict | None] | None:
         """실제 브라우저로 네이버 WTM 안티봇 캡차를 통과해 GraphQL 쿼리를 실행한다.
 
         네이버 pcmap GraphQL은 두 헤더를 검사한다:
@@ -145,19 +144,12 @@ if (originalQuery) {
         Args:
             place_id: 네이버 place ID
             payloads: GraphQL POST 본문 목록
-            capture_apollo: True면 place 페이지에 인라인된 __APOLLO_STATE__
-                JSON 문자열도 함께 캡처해 (results, apollo_json) 튜플로 반환.
-                (GraphQL base가 노출하지 않는 homepages 등 SSR 캐시 전용 필드용)
         Returns:
             payload별 파싱 JSON 목록(실패 항목은 None). 토큰 캡처 실패 시 None.
-            capture_apollo=True면 (위 결과, apollo_json 또는 None) 튜플.
         """
-        def _ret(results, apollo=None):
-            return (results, apollo) if capture_apollo else results
-
         if not cls.is_ready:
             logger.warning("[NvMapPwScraper] fetch_graphql: scraper가 초기화되지 않았습니다")
-            return _ret(None)
+            return None
 
         page = await cls._new_stealth_page()
         captured: dict = {}
@@ -193,17 +185,7 @@ if (originalQuery) {
 
             if not captured.get("tok"):
                 logger.warning("[NvMapPwScraper] WTM 토큰 캡처 실패")
-                return _ret(None)
-
-            apollo_json: str | None = None
-            if capture_apollo:
-                try:
-                    apollo_json = await page.evaluate(
-                        "() => { try { return JSON.stringify(window.__APOLLO_STATE__ || null); }"
-                        " catch (e) { return null; } }"
-                    )
-                except Exception as e:
-                    logger.warning(f"[NvMapPwScraper] __APOLLO_STATE__ 캡처 실패: {e}")
+                return None
 
             headers = {
                 "Content-Type": "application/json",
@@ -238,11 +220,11 @@ if (originalQuery) {
                     results.append(None)
                 else:
                     results.append(r)
-            return _ret(results, apollo_json)
+            return results
 
         except Exception as e:
             logger.error(f"[NvMapPwScraper] fetch_graphql 오류: {e}")
-            return _ret(None)
+            return None
         finally:
             await page.close()
     

app/utils/nvMapScraper.py

@@ -112,7 +112,6 @@ query getVisitorReviewStats($id: String!) {
         self.facility_info: str | None = None
         self.voted_keyword_stats: list[dict] | None = None  # 키워드 투표 집계 (displayName, count)
         self.menu_info: list[dict] | None = None  # 메뉴 목록 (name, price, description, recommend)
-        self.official_site_url: str | None = None  # 업체 공식 링크 (base.homepages 대표 URL)
 
     def _get_request_headers(self) -> dict:
         headers = self.DEFAULT_HEADERS.copy()
@@ -257,11 +256,9 @@ query getVisitorReviewStats($id: String!) {
         #     self.scrap_type = "GraphQL-Browser"
 
         # ── 실제 브라우저로 WTM 캡차 우회 ──
-        data, stats_data, extra_photo_urls, biz_photo_urls, homepage_url = await self._scrap_via_browser(place_id)
+        data, stats_data, extra_photo_urls, biz_photo_urls = await self._scrap_via_browser(place_id)
         # 편의시설은 HTML 페이지 파싱이라 GraphQL(WTM) 차단과 별개로 직접 시도 (best-effort, 실패 시 None)
-        # 홈페이지 링크는 브라우저 캡처가 실패한 경우에만 HTML 경로로 보충한다.
-        fac_data, html_homepage = await self._get_facility_and_homepage(place_id)
-        homepage_url = homepage_url or html_homepage
+        fac_data = await self._get_facility_string(place_id)
         self.scrap_type = "GraphQL-Browser"
 
         self.rawdata = data
@@ -300,43 +297,14 @@ query getVisitorReviewStats($id: String!) {
         self.facility_info = fac_data
         self.voted_keyword_stats = stats_data
         self.menu_info = business.get("menus") or None
-        self.official_site_url = homepage_url
 
         return
 
-    @staticmethod
-    def _extract_homepage_from_html(html: str) -> str | None:
-        """플레이스 페이지 HTML의 __APOLLO_STATE__에서 홈페이지 링크를 추출한다.
-
-        GraphQL placeDetail(base)는 homepages 필드를 노출하지 않으므로(400),
-        SSR 페이지에 인라인된 Apollo 캐시의 "homepages" 객체를 직접 파싱한다.
-        대표(repr) 링크 우선, 죽은 링크(isDeadUrl)는 제외. 홈페이지 항목은
-        자체 홈페이지 외에 인스타그램/블로그 등일 수도 있다 — 업체가 대표로
-        등록한 링크를 그대로 신뢰한다.
-        """
-        decoder = json.JSONDecoder()
-        search_from = 0
-        while True:
-            idx = html.find('"homepages":', search_from)
-            if idx == -1:
-                return None
-            search_from = idx + 1
-            try:
-                homepages, _ = decoder.raw_decode(html, idx + len('"homepages":'))
-            except ValueError:
-                continue
-            if not isinstance(homepages, dict):
-                continue
-            candidates = [homepages.get("repr"), *(homepages.get("etc") or [])]
-            for item in candidates:
-                if isinstance(item, dict) and item.get("url") and not item.get("isDeadUrl"):
-                    return item["url"]
-
-    async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[dict], list[dict], str | None]:
+    async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[dict], list[dict]]:
         """직접 호출이 WTM 캡차에 막힌 경우, 실제 브라우저로 GraphQL을 호출한다.
 
         Returns:
-            (overview_data, review_stats_details, extra_photo_urls, biz_photo_urls, homepage_url)
+            (overview_data, review_stats_details, extra_photo_urls, biz_photo_urls)
 
         Raises:
             GraphQLException: 브라우저 폴백마저 실패한 경우
@@ -414,14 +382,10 @@ query getVisitorReviewStats($id: String!) {
         payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload, *biz_payloads]
         MAX_RETRY = 3
         results = None
-        apollo_json: str | None = None
         last_error: Exception | None = None
         for attempt in range(1, MAX_RETRY + 1):
             try:
-                results, captured_apollo = await NvMapPwScraper.fetch_graphql(
-                    place_id, payloads, capture_apollo=True
-                )
-                apollo_json = apollo_json or captured_apollo
+                results = await NvMapPwScraper.fetch_graphql(place_id, payloads)
             except Exception as e:
                 last_error = e
                 logger.warning(f"[NvMapScraper] 브라우저 폴백 시도 {attempt}/{MAX_RETRY} 오류: {e}")
@@ -475,17 +439,8 @@ query getVisitorReviewStats($id: String!) {
             f"리뷰:{len(review_urls)} / 업체(biz):{len(biz_urls)}"
         )
 
-        # 홈페이지 링크: GraphQL base는 homepages를 노출하지 않으므로(400),
-        # 브라우저가 로드한 place 페이지의 __APOLLO_STATE__ JSON에서 추출한다.
-        homepage_url = (
-            self._extract_homepage_from_html(apollo_json) if apollo_json else None
-        )
-
-        logger.info(
-            f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}, "
-            f"homepage: {homepage_url or '없음'}"
-        )
-        return data, stats_data, extra_photo_urls, biz_urls, homepage_url
+        logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}")
+        return data, stats_data, extra_photo_urls, biz_urls
 
     async def _call_get_accommodation(self, place_id: str) -> dict:
         """GraphQL API를 호출하여 숙소 정보를 가져옵니다.
@@ -566,39 +521,29 @@ query getVisitorReviewStats($id: String!) {
             logger.warning(f"[NvMapScraper] Failed to get review stats: {e}")
             return None
 
-    async def _get_facility_and_homepage(self, place_id: str) -> tuple[str | None, str | None]:
-        """장소 페이지에서 편의시설 정보와 홈페이지 링크를 크롤링합니다. 숙소, 음식점 순으로 시도합니다.
+    async def _get_facility_string(self, place_id: str) -> str | None:
+        """장소 페이지에서 편의시설 정보를 크롤링합니다. 숙소, 음식점 순으로 시도합니다.
 
         Args:
             place_id: 네이버 지도 장소 ID
 
         Returns:
-            (편의시설 정보 문자열 또는 None, 홈페이지 링크 또는 None)
+            편의시설 정보 문자열 또는 None
         """
-        facility: str | None = None
-        homepage: str | None = None
         place_types = ["place", "accommodation", "restaurant"]
         try:
             async with aiohttp.ClientSession() as session:
                 for place_type in place_types:
                     url = f"https://pcmap.place.naver.com/{place_type}/{place_id}/home"
                     async with session.get(url, headers=self._get_request_headers()) as response:
-                        raw = await response.read()
-                    if homepage is None:
-                        homepage = self._extract_homepage_from_html(
-                            raw.decode("utf-8", errors="ignore")
-                        )
-                    if facility is None:
-                        soup = bs4.BeautifulSoup(raw, "html.parser")
+                        soup = bs4.BeautifulSoup(await response.read(), "html.parser")
                         c_elem = soup.find("span", "place_blind", string="편의")
                         if c_elem:
-                            facility = c_elem.parent.parent.find("div").string
-                    if facility is not None and homepage is not None:
-                        break
-            return facility, homepage
+                            return c_elem.parent.parent.find("div").string
+            return None
         except Exception as e:
-            logger.warning(f"[NvMapScraper] Failed to get facility/homepage info: {e}")
-            return facility, homepage
+            logger.warning(f"[NvMapScraper] Failed to get facility info: {e}")
+            return None
 
 
 # if __name__ == "__main__":

app/utils/prompts/chatgpt_prompt.py

@@ -47,19 +47,6 @@ class ChatgptService:
             case _:
                 raise NotImplementedError(f"Unknown Provider : {model_type}")
     
-    def _log_usage(self, response, model: str, output_format: type[BaseModel]) -> None:
-        usage = getattr(response, "usage", None)
-        if usage is None:
-            return
-        # 토큰 소모량 로깅 (필요 시 주석 해제)
-        # cached = getattr(getattr(usage, "prompt_tokens_details", None), "cached_tokens", None) or 0
-        # reasoning = getattr(getattr(usage, "completion_tokens_details", None), "reasoning_tokens", None) or 0
-        # logger.info(
-        #     f"[ChatgptService({self.model_type})] usage model={model} output={output_format.__name__} "
-        #     f"prompt={usage.prompt_tokens} cached={cached} "
-        #     f"completion={usage.completion_tokens} reasoning={reasoning} total={usage.total_tokens}"
-        # )
-
     async def _call_pydantic_output(
             self, 
             prompt : str, 
@@ -128,8 +115,7 @@ class ChatgptService:
             output_format : BaseModel,  #입력 output_format의 경우 Pydantic BaseModel Class를 상속한 Class 자체임에 유의할 것
             model : str, 
             img_url : str,
-            image_detail_high : bool,
-            reasoning_effort : Optional[str] = None) -> BaseModel:
+            image_detail_high : bool) -> BaseModel: 
         content = []
         if img_url:
             content.append({
@@ -143,16 +129,13 @@ class ChatgptService:
             "type": "text",
             "text": prompt
         })
-        # gpt-5.4 계열/Gemini 호환 엔드포인트는 허용 값이 다르거나 파라미터를 거부하므로 지정된 경우에만 전달
-        extra_kwargs = {"reasoning_effort": reasoning_effort} if reasoning_effort else {}
         last_error = None
         for attempt in range(self.max_retries + 1):
             try:
                 response = await self.client.beta.chat.completions.parse(
                     model=model,
                     messages=[{"role": "user", "content": content}],
-                    response_format=output_format,
-                    **extra_kwargs,
+                    response_format=output_format
                 )
             except (ValidationError, json.JSONDecodeError) as e:
                 # 모델이 스키마에 맞지 않는 JSON을 반환한 경우 (예: trailing characters).
@@ -165,7 +148,6 @@ class ChatgptService:
                 if attempt < self.max_retries:
                     logger.info(f"[ChatgptService({self.model_type})] Retrying request...")
                 continue
-            self._log_usage(response, model, output_format)
             # Response 디버그 로깅
             # logger.debug(f"[ChatgptService({self.model_type})] attempt: {attempt}")
             # logger.debug(f"[ChatgptService({self.model_type})] Response ID: {response.id}")
@@ -243,7 +225,6 @@ class ChatgptService:
                     continue
                 raise last_error
 
-            self._log_usage(response, model, output_format)
             choice = response.choices[0]
             if choice.finish_reason == "stop":
                 return choice.message.parsed
@@ -261,8 +242,7 @@ class ChatgptService:
         input_data : dict,
         img_url : Optional[str] = None,
         img_detail_high : bool = False,
-        silent : bool = True,
-        reasoning_effort : Optional[str] = None,
+        silent : bool = True
     ) -> BaseModel:
         prompt_text = prompt.build_prompt(input_data, silent)
     
@@ -273,5 +253,5 @@ class ChatgptService:
         # GPT API 호출
         #parsed = await self._call_structured_output_with_response_gpt_api(prompt_text, prompt.prompt_output, prompt.prompt_model) 
         # parsed = await self._call_pydantic_output(prompt_text, prompt.prompt_output_class, prompt.prompt_model, img_url, img_detail_high) 
-        parsed = await self._call_pydantic_output_chat_completion(prompt_text, prompt.prompt_output_class, prompt.prompt_model, img_url, img_detail_high, reasoning_effort)
+        parsed = await self._call_pydantic_output_chat_completion(prompt_text, prompt.prompt_output_class, prompt.prompt_model, img_url, img_detail_high) 
         return parsed

app/utils/upload_blob_as_request.py

@@ -90,23 +90,6 @@ async def close_shared_blob_client() -> None:
         logger.info("[AzureBlobUploader] Shared HTTP client closed")
 
 
-def to_playback_url(blob_url: str | None) -> str | None:
-    """DB에 저장된 SAS 미포함 공개 URL에 읽기용 SAS 토큰을 붙여 반환합니다.
-
-    Azure Blob 익명(공개) 접근은 x-ms-version 헤더가 없어 Range 요청을 지원하지
-    않는 구버전 API로 처리되어 브라우저에서 영상/오디오 seek이 동작하지 않는다.
-    SAS 토큰(sv= 포함)을 붙이면 최신 API 버전으로 처리되어 Range/Accept-Ranges가
-    정상 동작한다. DB에는 SAS 없는 URL을 그대로 저장하고, 응답 시점에만 붙인다.
-    """
-    if not blob_url:
-        return blob_url
-    sas_token = azure_blob_settings.AZURE_BLOB_SAS_TOKEN.strip("?'\"")
-    if not sas_token:
-        return blob_url
-    separator = "&" if "?" in blob_url else "?"
-    return f"{blob_url}{separator}{sas_token}"
-
-
 class AzureBlobUploader:
     """Azure Blob Storage 업로드 클래스
 

app/utils/video_poster.py

@@ -1,173 +0,0 @@
-"""영상 파일에서 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

@@ -17,8 +17,7 @@ import json
 from collections import defaultdict
 from typing import Literal
 
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request
-from fastapi.responses import HTMLResponse
+from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query
 from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -30,7 +29,6 @@ from app.utils.pagination import PaginatedResponse
 from app.home.models import Image, Project, MarketingIntel
 from app.home.api.routers.v1.home import _extract_region_from_address
 from app.utils.address_parser import SIDO_CITIES, SIDO_SEARCH_ALIASES
-from app.utils.upload_blob_as_request import to_playback_url
 from app.lyric.models import Lyric
 from app.song.models import Song, SongTimestamp
 from app.utils.creatomate import CreatomateService, LANGUAGE_FONT_MAP
@@ -59,69 +57,16 @@ from app.video.schemas.video_schema import (
     VideoRenderData,
     VideoThumbnailItem,
 )
-from app.video.services.share_page import (
-    build_video_share_html,
-    get_video_share_data,
-    resolve_frontend_base_url,
-    resolve_share_url,
-)
 from app.video.worker.video_task import download_and_upload_video_to_blob
 
 
-from config import creatomate_settings, prj_settings
+from config import creatomate_settings
 
 logger = get_logger("video")
 
 router = APIRouter(prefix="/video", tags=["Video"])
 
 
-def _place_id_to_site_url(place_id: str | None) -> str | None:
-    """MarketingIntel.place_id("nv{네이버 place ID}")를 네이버 플레이스 URL로 변환한다.
-
-    크롤링 없이 직접 입력된 업체는 place_id가 없으므로 None을 반환한다.
-    """
-    if place_id and place_id.startswith("nv") and place_id[2:].isdigit():
-        return f"https://map.naver.com/p/entry/place/{place_id[2:]}"
-    return None
-
-
-async def _get_official_site_urls(
-    session: AsyncSession, projects: list[Project]
-) -> dict[int, str | None]:
-    """프로젝트 목록에 대해 {project_id: 공식 페이지 URL(or None)}을 일괄 조회한다.
-
-    Project.marketing_intelligence(문자열로 저장된 MarketingIntel.id)를 경유해
-    저장된 official_site_url을 우선 사용하고, 컬럼 도입 전 기존 행은
-    place_id 기반 네이버 플레이스 URL로 폴백한다.
-    """
-    m_id_by_project: dict[int, int] = {}
-    for p in projects:
-        try:
-            if p.marketing_intelligence is not None:
-                m_id_by_project[p.id] = int(p.marketing_intelligence)
-        except (TypeError, ValueError):
-            continue
-
-    url_by_project: dict[int, str | None] = {p.id: None for p in projects}
-    if not m_id_by_project:
-        return url_by_project
-
-    rows = (
-        await session.execute(
-            select(
-                MarketingIntel.id,
-                MarketingIntel.place_id,
-                MarketingIntel.official_site_url,
-            ).where(MarketingIntel.id.in_(set(m_id_by_project.values())))
-        )
-    ).all()
-    intel_by_m_id = {m_id: (place_id, site_url) for m_id, place_id, site_url in rows}
-
-    for project_id, m_id in m_id_by_project.items():
-        place_id, site_url = intel_by_m_id.get(m_id, (None, None))
-        url_by_project[project_id] = site_url or _place_id_to_site_url(place_id)
-    return url_by_project
-
 
 @router.get(
     "/generate/{task_id}",
@@ -861,7 +806,7 @@ async def download_video(
             store_name=project.store_name if project else None,
             region=project.region or _extract_region_from_address(project.detail_region_info) if project else None,
             task_id=task_id,
-            result_movie_url=to_playback_url(video.result_movie_url),
+            result_movie_url=video.result_movie_url,
             created_at=video.created_at,
         )
 
@@ -1037,21 +982,15 @@ async def get_all_videos(
 
             liked_map = {vid: bool(liked) for vid, liked in raw_liked.items()}
 
-        official_site_url_map = await _get_official_site_urls(
-            session, [p for _, p, _ in rows]
-        )
-
         items = [
             VideoThumbnailItem(
                 video_id=v.id,
                 store_name=p.store_name,
-                result_movie_url=to_playback_url(v.result_movie_url),
-                poster_url=v.poster_url,
+                result_movie_url=v.result_movie_url,
                 created_at=v.created_at,
                 like_count=like_count_map.get(v.id) or 0,
                 is_liked_by_me=liked_map.get(v.id, False),
                 comment_count=comment_count or 0,
-                official_site_url=official_site_url_map.get(p.id),
             )
             for v, p, comment_count in rows
         ]
@@ -1151,50 +1090,6 @@ 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 = resolve_share_url(
-        request.headers,
-        str(request.url).split("?", maxsplit=1)[0],
-        prj_settings.SHARE_API_BASE_URL,
-    )
-    html = build_video_share_html(
-        share_data,
-        share_url=share_url,
-        frontend_base_url=resolve_frontend_base_url(
-            request.headers,
-            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="단일 영상 상세 조회",
@@ -1267,21 +1162,15 @@ async def get_video_detail(
                     liked = False
             is_liked_by_me = liked
 
-        official_site_url_map = await _get_official_site_urls(session, [project])
-
         logger.info(f"[get_video_detail] SUCCESS - video_id: {video_id}")
         return VideoDetailResponse(
             video_id=video.id,
-            result_movie_url=to_playback_url(video.result_movie_url),
-            poster_url=video.poster_url,
+            result_movie_url=video.result_movie_url,
             store_name=project.store_name,
             region=project.region or _extract_region_from_address(project.detail_region_info),
-            title=video.title,
-            description=video.description,
             created_at=video.created_at,
             like_count=like_count,
             is_liked_by_me=is_liked_by_me,
-            official_site_url=official_site_url_map.get(project.id),
         )
 
     except HTTPException:

app/video/models.py

@@ -1,8 +1,7 @@
 from datetime import datetime
 from typing import TYPE_CHECKING, List, Optional
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint, func
-from sqlalchemy.dialects.mysql import JSON
+from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String, UniqueConstraint, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from app.database.session import Base
@@ -30,10 +29,6 @@ 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:
@@ -111,30 +106,6 @@ 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, List, Optional
+from typing import Any, Dict, Optional
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -148,7 +148,6 @@ 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"
         }
     """
@@ -158,17 +157,9 @@ class VideoListItem(BaseModel):
     region: Optional[str] = Field(None, description="지역명")
     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="댓글 수 (대댓글 포함)")
-    is_liked_by_me: bool = Field(
-        False,
-        description="현재 로그인 사용자가 좋아요를 눌렀는지",
-    )
 
 
 class VideoThumbnailItem(BaseModel):
@@ -180,16 +171,11 @@ class VideoThumbnailItem(BaseModel):
 
     video_id: int = Field(..., description="영상 고유 ID (상세 페이지 라우팅 키)")
     store_name: str = Field(..., description="업체명")
-    result_movie_url: str = Field(..., description="영상 URL")
-    poster_url: Optional[str] = Field(None, description="영상 첫 프레임 포스터 이미지 URL (썸네일 표시용)")
+    result_movie_url: str = Field(..., description="영상 URL — 프론트에서 <video> 태그 첫 프레임을 썸네일로 사용")
     created_at: datetime = Field(..., description="생성 일시")
     like_count: int = Field(..., description="좋아요 수")
     is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지 (비로그인은 항상 false)")
     comment_count: int = Field(..., description="댓글 수 (대댓글 포함)")
-    official_site_url: Optional[str] = Field(
-        None,
-        description="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 생성 영상만 null)",
-    )
 
 
 class VideoDetailResponse(BaseModel):
@@ -201,18 +187,11 @@ 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="지역명")
-    title: Optional[str] = Field(None, description="SNS 업로드 제목 (공유 시 og:title 및 공유 제목으로 사용)")
-    description: Optional[str] = Field(None, description="SNS 업로드 설명")
     created_at: datetime = Field(..., description="생성 일시")
     like_count: int = Field(..., description="좋아요 수")
     is_liked_by_me: bool = Field(..., description="현재 로그인 사용자가 좋아요를 눌렀는지 (비로그인은 항상 false)")
-    official_site_url: Optional[str] = Field(
-        None,
-        description="업체 공식 링크 (플레이스 홈페이지 항목 우선, 없으면 크롤링 소스 URL; 직접 입력 생성 영상만 null)",
-    )
 
 
 class LikeToggleResponse(BaseModel):

app/video/services/share_page.py

@@ -1,321 +0,0 @@
-"""영상 공유 링크용 Open Graph HTML 생성 서비스."""
-
-from collections.abc import Mapping
-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
-    title: str | None = None
-    description: str | None = None
-
-
-async def get_video_share_data(
-    session: AsyncSession,
-    video_id: int,
-) -> VideoShareData | None:
-    """공유 가능한 완료 영상을 프로젝트 정보와 함께 조회합니다."""
-    result = await session.execute(
-        select(
-            Video.id,
-            Video.poster_url,
-            Video.title,
-            Video.description,
-            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,
-        title=row.title,
-        description=row.description,
-    )
-
-
-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
-
-    title = _share_title(data.title, data.store_name)
-    description = _share_description(data.description)
-    canonical_tags = _canonical_tags(_absolute_http_url(share_url))
-    image_size_tags = _og_image_size_tags(image_url, fallback_image_url)
-
-    escaped_title = escape(title, quote=True)
-    escaped_description = escape(description, quote=True)
-    escaped_image_url = escape(image_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}">
-{canonical_tags}
-  <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}">
-{image_size_tags}  <meta property="og:type" content="website">
-
-  <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>
-    (function () {{
-      var ua = navigator.userAgent || "";
-      // 앱 이름만으로 판별하면 인앱 브라우저(예: KAKAOTALK)까지 크롤러로 잡혀
-      // 사용자가 중간 페이지에 멈춘다. 봇 전용 토큰만 쓴다.
-      if (/bot|crawl|spider|slurp|facebookexternalhit|Facebot|Twitterbot|LinkedInBot|Pinterestbot|Slackbot|TelegramBot|WhatsApp|Discordbot|kakaotalk-scrap|Embedly|redditbot|Applebot/i.test(ua)) {{
-        return;
-      }}
-      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
-
-
-_OG_DESCRIPTION_MAX_LEN = 300
-
-
-def _share_title(title: str | None, store_name: str) -> str:
-    """저장된 SNS 제목을 쓰고, 없으면 가게명 폴백을 사용합니다."""
-    stored = _normalise_text(title, "")
-    if stored:
-        return stored
-    name = _normalise_text(store_name, "ADO2 영상")
-    return f"{name} | ADO2"
-
-
-def _share_description(description: str | None) -> str:
-    """저장된 SNS 설명을 쓰고, 없으면 기본 문구를 사용합니다."""
-    stored = _normalise_text(description, "")
-    if stored:
-        if len(stored) > _OG_DESCRIPTION_MAX_LEN:
-            return stored[: _OG_DESCRIPTION_MAX_LEN - 1].rstrip() + "…"
-        return stored
-    return "ADO2 AI 마케팅 영상"
-
-
-def resolve_frontend_base_url(headers: Mapping[str, str], fallback: str) -> str:
-    """프록시가 넘긴 프론트 호스트를 우선해 canonical 기준 URL을 정합니다."""
-    forwarded_host = (headers.get("x-forwarded-host") or "").split(",")[0].strip()
-    if not forwarded_host:
-        return _normalise_frontend_base_url(fallback)
-
-    forwarded_proto = (headers.get("x-forwarded-proto") or "https").split(",")[0].strip().lower()
-    if forwarded_proto not in {"http", "https"}:
-        forwarded_proto = "https"
-    return _normalise_frontend_base_url(f"{forwarded_proto}://{forwarded_host}")
-
-
-def resolve_share_url(
-    headers: Mapping[str, str],
-    request_url: str,
-    configured_api_base_url: str = "",
-) -> str:
-    """크롤러가 다시 읽어도 같은 OG 페이지가 나오는 공유 URL을 만듭니다.
-
-    nginx가 ``/api`` prefix를 떼고 넘기면 ``request.url``에는 그 prefix가 없어,
-    그대로 쓰면 프론트 SPA 주소가 된다. 복원할 근거가 없으면 빈 문자열을 돌려
-    호출부가 canonical/og:url을 생략하도록 한다.
-    """
-    path = urlsplit(request_url).path
-
-    configured_base = _absolute_http_url(configured_api_base_url)
-    if configured_base:
-        return f"{configured_base.rstrip('/')}{path}"
-
-    origin = _forwarded_origin(headers) or _origin_from_url(request_url)
-    if not origin:
-        return ""
-
-    forwarded_prefix = (headers.get("x-forwarded-prefix") or "").strip().rstrip("/")
-    if forwarded_prefix:
-        return f"{origin}{forwarded_prefix}{path}"
-
-    if headers.get("x-forwarded-host"):
-        # 프록시 뒤인데 prefix 를 못 받았다. 잘못된 URL 을 내보내지 않는다.
-        return ""
-
-    return f"{origin}{path}"
-
-
-def _forwarded_origin(headers: Mapping[str, str]) -> str | None:
-    """프록시가 넘긴 외부 호스트 기준 origin을 만듭니다."""
-    forwarded_host = (headers.get("x-forwarded-host") or "").split(",")[0].strip()
-    if not forwarded_host:
-        return None
-
-    forwarded_proto = (headers.get("x-forwarded-proto") or "https").split(",")[0].strip().lower()
-    if forwarded_proto not in {"http", "https"}:
-        forwarded_proto = "https"
-    return f"{forwarded_proto}://{forwarded_host}"
-
-
-def _canonical_tags(canonical_url: str | None) -> str:
-    """공유 URL을 확신할 때만 canonical/og:url을 붙입니다.
-
-    잘못된 og:url을 내보내면 크롤러가 그 주소를 다시 읽어, OG 메타가 없는
-    프론트 SPA 문서를 미리보기로 쓴다. 확신이 없으면 크롤러가 실제로 받은
-    URL을 쓰도록 태그 자체를 생략한다.
-    """
-    if not canonical_url:
-        return ""
-    escaped = escape(canonical_url, quote=True)
-    return (
-        f'  <link rel="canonical" href="{escaped}">\n'
-        f'  <meta property="og:url" content="{escaped}">\n'
-    )
-
-
-_DEFAULT_OG_IMAGE_SIZE = (385, 385)
-
-
-def _og_image_size_tags(image_url: str, fallback_image_url: str) -> str:
-    """폴백 로고처럼 크기를 아는 이미지에만 width/height 메타를 붙입니다."""
-    if image_url != fallback_image_url:
-        return ""
-    width, height = _DEFAULT_OG_IMAGE_SIZE
-    return (
-        f'  <meta property="og:image:width" content="{width}">\n'
-        f'  <meta property="og:image:height" content="{height}">\n'
-    )
-
-
-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. 공유 URL과 같은 API 베이스 ``.../static/images/ado2_image.png``
-       (``/api/video/share/1`` 이면 ``/api/static/...``)
-    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_api_base = _api_base_from_share_url(share_url)
-    if share_api_base:
-        return f"{share_api_base}{DEFAULT_SHARE_IMAGE_STATIC_PATH}"
-
-    return f"{frontend_base}{DEFAULT_SHARE_IMAGE_PATH}"
-
-
-def _api_base_from_share_url(share_url: str) -> str | None:
-    """공유 URL에서 API 베이스를 만듭니다. ``/api/video/share/1`` → ``https://host/api``."""
-    absolute_url = _absolute_http_url(share_url)
-    if not absolute_url:
-        return None
-
-    parts = urlsplit(absolute_url)
-    origin = urlunsplit((parts.scheme, parts.netloc, "", "", ""))
-    idx = (parts.path or "").find("/video/share/")
-    if idx < 0:
-        return origin
-
-    prefix = parts.path[:idx].rstrip("/")
-    return f"{origin}{prefix}" if prefix else origin
-
-
-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/worker/video_task.py

@@ -4,6 +4,7 @@ Video Background Tasks
 영상 생성 관련 백그라운드 태스크를 정의합니다.
 """
 
+import traceback
 from pathlib import Path
 
 import aiofiles
@@ -16,7 +17,6 @@ from app.user.services.credit import consume_credit
 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")
@@ -30,8 +30,7 @@ async def _update_video_status(
     status: str,
     video_url: str | None = None,
     creatomate_render_id: str | None = None,
-    poster_url: str | None = None,
-) -> int | None:
+) -> bool:
     """Video 테이블의 상태를 업데이트합니다.
 
     Args:
@@ -39,10 +38,9 @@ async def _update_video_status(
         status: 변경할 상태 ("processing", "completed", "failed")
         video_url: 영상 URL
         creatomate_render_id: Creatomate render ID (선택)
-        poster_url: 영상 첫 프레임 포스터 URL (선택)
 
     Returns:
-        int | None: 업데이트된 Video id. 대상이 없거나 실패하면 None.
+        bool: 업데이트 성공 여부
     """
     try:
         async with BackgroundSessionLocal() as session:
@@ -67,58 +65,19 @@ 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 video.id
+                return True
             else:
                 logger.warning(f"[Video] NOT FOUND in DB - task_id: {task_id}")
-                return None
+                return False
 
     except SQLAlchemyError as e:
         logger.error(f"[Video] DB Error while updating status - task_id: {task_id}, error: {e}")
-        return None
+        return False
     except Exception as e:
         logger.error(f"[Video] Unexpected error while updating status - task_id: {task_id}, error: {e}")
-        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
+        return False
 
 
 async def _download_video(url: str, task_id: str) -> bytes:
@@ -194,20 +153,8 @@ 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 식별)
-        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)
+        await _update_video_status(task_id, "completed", blob_url, creatomate_render_id)
 
         # 영상 생성 완료 시 크레딧 1 차감 (credits > 0 조건으로 음수 방지)
         async with BackgroundSessionLocal() as session:
@@ -312,20 +259,13 @@ 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 테이블 업데이트
-        video_id = await _update_video_status(
+        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,25 +33,6 @@ 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_API_BASE_URL: str = Field(
-        default="",
-        description=(
-            "공유 OG 페이지가 외부에 노출되는 API 기준 URL (.env: SHARE_API_BASE_URL). "
-            "예: https://dev-ssul.castad.net/api. 프록시가 /api prefix 를 떼고 넘기면 "
-            "request.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",

docs/database-schema/migration_2026_08_13_add_video_poster_url.sql

@@ -1,10 +0,0 @@
--- ============================================================
--- 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

@@ -1,15 +0,0 @@
--- ============================================================
--- 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`;