fix: 썸네일 슬롯 매칭 개선, 가사 스키마 오류 수정, 롤백 로그 레벨 조정

- creatomate: 썸네일 슬롯 한정 subject 일치 시 space_type 페널티 면제,
  subject 가중치 상향 (2.0 → 3.0)
- lyric 스키마: 트레일링 콤마로 tuple이 되어버린 필드 정의 수정,
  중복 genre 필드 제거
- session: 4xx 도메인 예외 롤백 로그를 traceback 없는 WARNING으로 완화

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -157,7 +157,6 @@ async def get_videos(
                 region=project.region,
                 task_id=video.task_id,
                 result_movie_url=video.result_movie_url,
-                thumbnail_url=video.video_thumbnail_url,
                 created_at=video.created_at,
                 like_count=like_count_map.get(video.id) or 0,
                 comment_count=comment_count or 0,

app/database/session.py

@@ -139,11 +139,18 @@ async def get_session() -> AsyncGenerator[AsyncSession, None]:
             raise
         except Exception as e:
             await session.rollback()
-            logger.error(traceback.format_exc())
-            logger.error(
-                f"[get_session] ROLLBACK - error: {type(e).__name__}: {e}, "
-                f"duration: {(time.perf_counter() - start_time)*1000:.1f}ms"
-            )
+            # status_code < 500인 도메인 예외(계정 미연동 등)는 정상적인 비즈니스 흐름이므로 ERROR로 남기지 않음
+            if getattr(e, "status_code", 500) < 500:
+                logger.warning(
+                    f"[get_session] ROLLBACK - client error: {type(e).__name__}: {e}, "
+                    f"duration: {(time.perf_counter() - start_time)*1000:.1f}ms"
+                )
+            else:
+                logger.error(traceback.format_exc())
+                logger.error(
+                    f"[get_session] ROLLBACK - error: {type(e).__name__}: {e}, "
+                    f"duration: {(time.perf_counter() - start_time)*1000:.1f}ms"
+                )
             raise
         finally:
             total_time = time.perf_counter() - start_time

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

@@ -946,6 +946,11 @@ async def tagging_images(
         async with AsyncSessionLocal() as session:
             for tag, tag_data in zip(null_imts, tag_datas):
                 if isinstance(tag_data, Exception):
+                    # 태깅 실패 이미지는 img_tag가 NULL로 남아 영상 생성 풀에서 제외됨
+                    logger.warning(
+                        f"[tagging_images] 이미지 태깅 최종 실패 - url: {tag.img_url}, "
+                        f"error: {type(tag_data).__name__}: {tag_data}"
+                    )
                     continue
                 tag.img_tag = tag_data.model_dump(mode="json")
                 session.add(tag)

app/home/models.py

@@ -370,8 +370,10 @@ class ImageTag(Base):
     )
 
     img_tag: Mapped[dict[str, Any]] = mapped_column(
-        JSON,
+        # none_as_null=True: ORM에서 None 대입 시 JSON 리터럴 null이 아닌 SQL NULL로 저장.
+        # 미지정 시 JSON null이 저장되어 `img_tag.is_not(None)` 필터를 통과하는 버그가 있었음.
+        JSON(none_as_null=True),
         nullable=True,
-        default=False,
+        default=None,
         comment="태그 JSON",
     )

app/lyric/schemas/lyric.py

@@ -70,11 +70,11 @@ class GenerateLyricRequest(BaseModel):
     language: str = Field(
         default="Korean",
         description="가사 출력 언어 (Korean, English, Chinese, Japanese, Thai, Vietnamese)",
-    ),
+    )
     orientation: Literal["horizontal", "vertical"] = Field(
         default="vertical",
         description="영상 방향 (horizontal: 가로형, vertical: 세로형)",
-    ),
+    )
     m_id : Optional[int] = Field(None, description="마케팅 인텔리전스 ID 값")
     industry: str = Field(default="", description="업종 분류 (stay|restaurant|cafe|salon|clinic|fitness|academy|attraction|general)")
     instrumental: bool = Field(default=False, description="BGM 전용 모드 (가사 생성 안 함)")
@@ -83,11 +83,6 @@ class GenerateLyricRequest(BaseModel):
         description="사용자가 수동으로 고른 음악 장르 (kpop|pop|ballad|hip-hop|rnb|edm|jazz|rock). "
         "'자동 선택'인 경우 None으로 전달하면 GPT가 가사 무드에 맞춰 추천",
     )
-    genre: Optional[str] = Field(
-        None,
-        description="사용자가 수동으로 고른 음악 장르 (kpop|pop|ballad|hip-hop|rnb|edm|jazz|rock). "
-        "'자동 선택'인 경우 None으로 전달하면 GPT가 가사 무드에 맞춰 추천",
-    )
 
 
 class GenerateLyricResponse(BaseModel):

app/song/api/routers/v1/song.py

@@ -35,7 +35,7 @@ logger = get_logger("song")
 
 router = APIRouter(prefix="/song", tags=["Song"])
 
-TARGET_SONG_DURATION_SECONDS = 60.0
+TARGET_SONG_DURATION_SECONDS = 40.0
 
 
 def _select_clip_by_duration(
@@ -99,7 +99,7 @@ curl -X POST "http://localhost:8000/song/generate/019123ab-cdef-7890-abcd-ef1234
 ```
 
 ## 참고
-- 생성되는 노래는 약 1분 이내 길이입니다.
+- 생성되는 노래는 약 40초 내외 길이입니다.
 - song_id를 사용하여 /status/{song_id} 엔드포인트에서 생성 상태를 확인할 수 있습니다.
 - Song 테이블에 데이터가 저장되며, project_id와 lyric_id가 자동으로 연결됩니다.
     """,

app/utils/image_filter.py

@@ -70,7 +70,7 @@ async def filter_marketing_images(image_url_list: list[str], industry: str = "")
             output_format=MarketingFilterOutput,
             model=MARKETING_FILTER_MODEL,
             img_url=url,
-            image_detail_high=False,
+            image_detail_high=True,
         )
 
     results: list[MarketingFilterOutput | BaseException] = await asyncio.gather(

app/utils/prompts/chatgpt_prompt.py

@@ -182,6 +182,60 @@ class ChatgptService:
         logger.error(f"[ChatgptService({self.model_type})] All retries exhausted. Last error: {last_error}")
         raise last_error
     
+    async def generate_structured_output_multi_image(
+        self,
+        prompt_text: str,
+        output_format: BaseModel,
+        model: str,
+        img_urls: List[str],
+        image_detail_high: bool = True,
+    ) -> BaseModel:
+        """여러 이미지를 한 번에 보고 구조화 출력을 생성합니다 (썸네일 비전 선택용).
+
+        sheet 기반 Prompt를 거치지 않고 코드에서 조립한 프롬프트 텍스트와
+        이미지 URL 리스트를 직접 받는다. 이미지들은 프롬프트에 나열된 순서와
+        동일하게 첨부되므로, 프롬프트에서 "N번째 이미지"로 지칭할 수 있다.
+        """
+        content = []
+        for url in img_urls:
+            content.append({
+                "type": "image_url",
+                "image_url": {
+                    "url": url,
+                    "detail": "high" if image_detail_high else "low",
+                },
+            })
+        content.append({"type": "text", "text": prompt_text})
+
+        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,
+                )
+            except (ValidationError, json.JSONDecodeError) as e:
+                logger.warning(
+                    f"[ChatgptService({self.model_type})] multi-image parse failed "
+                    f"(attempt {attempt + 1}/{self.max_retries + 1}): {e}"
+                )
+                last_error = ChatGPTResponseError("parse_error", type(e).__name__, str(e))
+                if attempt < self.max_retries:
+                    continue
+                raise last_error
+
+            choice = response.choices[0]
+            if choice.finish_reason == "stop":
+                return choice.message.parsed
+            logger.warning(
+                f"[ChatgptService({self.model_type})] multi-image unexpected finish_reason "
+                f"(attempt {attempt + 1}/{self.max_retries + 1}): {choice.finish_reason}"
+            )
+            last_error = ChatGPTResponseError("failed", choice.finish_reason, "multi-image call failed")
+
+        raise last_error
+
     async def generate_structured_output(
         self,
         prompt : Prompt,

app/utils/prompts/schemas/marketing.py

@@ -44,4 +44,3 @@ class MarketingPromptOutput(BaseModel):
     target_persona: List[TargetPersona] = Field(..., description="타겟 페르소나")
     selling_points: List[SellingPoint] = Field(..., description="셀링 포인트")
     target_keywords: List[str] = Field(..., description="타겟 키워드 리스트")
-    hook_headline: str = Field(..., description="썸네일 헤드라인용 훅 문구. 브랜드 컨셉/핵심 매력을 5자 이상 15자 이하의 한국어 한 구절로 축약 (마침표·해시태그 없이, 예: '바다 앞 프라이빗 휴식')")  # min/max constraint는 openai json schema에서 미작동 보고가 있어 description으로 유도, 소비처에서 길이 안전망 적용

app/utils/subtitles.py

@@ -1,4 +1,5 @@
 import copy
+import re
 import time
 import json
 from typing import Literal, Any
@@ -12,10 +13,31 @@ from app.utils.prompts.prompts import *
 
 logger = get_logger("subtitle")
 
+# 비한국어 출력에서 번역 누락(한글 잔존)을 탐지하기 위한 패턴
+_HANGUL_RE = re.compile(r"[가-힣]")
+
+# 한글 잔존 시 GPT 재호출 최대 횟수 (최초 호출 포함)
+_MAX_LANGUAGE_ATTEMPTS = 3
+
+
 class SubtitleContentsGenerator():
     def __init__(self):
         self.chatgpt_service = ChatgptService(timeout=60.0)
 
+    @staticmethod
+    def _find_hangul_leftovers(output_data: SubtitlePromptOutput) -> list[str]:
+        """비한국어 출력에서 한글이 남아 있는 pitching_tag 목록을 반환합니다.
+
+        프롬프트가 '한국어로 생성 후 {language}로 번역'하는 2단계 구조라서,
+        항목 수가 많으면 일부가 번역되지 않은 채(한국어/혼합 문장) 돌아오는
+        사례가 있어 코드 레벨에서 검증한다.
+        """
+        return [
+            result.pitching_tag
+            for result in output_data.pitching_results
+            if _HANGUL_RE.search(result.pitching_data)
+        ]
+
     async def generate_subtitle_contents(self, marketing_intelligence : dict[str, Any], pitching_label_list : list[Any], customer_name : str, detail_region_info : str, language : str = "Korean", industry: str = "") -> SubtitlePromptOutput:
         start = time.perf_counter()
         logger.info(
@@ -41,7 +63,39 @@ class SubtitleContentsGenerator():
         logger.info(
             f"[SubtitleContentsGenerator] GPT 호출 시작 - model: {dynamic_subtitle_prompt.prompt_model}"
         )
-        output_data = await self.chatgpt_service.generate_structured_output(dynamic_subtitle_prompt, input_data)
+
+        # 비한국어 언어는 번역 누락(한글 잔존) 검증 후 필요 시 재호출.
+        # 전부 실패하면 잔존 건수가 가장 적은 시도를 채택한다 (영상 생성 자체는 진행).
+        output_data = None
+        best_output = None
+        best_leftover_count: int | None = None
+        for lang_attempt in range(1, _MAX_LANGUAGE_ATTEMPTS + 1):
+            candidate = await self.chatgpt_service.generate_structured_output(dynamic_subtitle_prompt, input_data)
+
+            if language == "Korean":
+                output_data = candidate
+                break
+
+            leftovers = self._find_hangul_leftovers(candidate)
+            if not leftovers:
+                output_data = candidate
+                break
+
+            logger.warning(
+                f"[SubtitleContentsGenerator] 번역 누락(한글 잔존) {len(leftovers)}건 "
+                f"(attempt {lang_attempt}/{_MAX_LANGUAGE_ATTEMPTS}) - language: {language}, "
+                f"tags: {leftovers}"
+            )
+            if best_leftover_count is None or len(leftovers) < best_leftover_count:
+                best_output = candidate
+                best_leftover_count = len(leftovers)
+
+        if output_data is None:
+            logger.error(
+                f"[SubtitleContentsGenerator] 모든 시도에서 한글 잔존 - "
+                f"최소 잔존 {best_leftover_count}건 결과 채택 - language: {language}"
+            )
+            output_data = best_output
 
         elapsed = (time.perf_counter() - start) * 1000
         logger.info(

app/utils/suno.py

@@ -113,11 +113,11 @@ class SunoService:
 
         Args:
             prompt: 가사 (customMode=true일 때 가사로 사용)
-                    1분 이내 길이의 노래에 적합한 가사여야 함
+                    40초 이내 길이의 노래에 적합한 가사여야 함
             genre: 음악 장르 (예: "K-Pop", "Pop", "R&B", "Hip-Hop", "Ballad", "EDM", "Rock", "Jazz")
                    None일 경우 style 파라미터를 전송하지 않음
             callback_url: 생성 완료 시 알림 받을 URL (None일 경우 config에서 기본값 사용)
-            instrumental: True이면 BGM 전용 — 더미 가사로 60초 길이를 유도하고 보컬 없이 생성
+            instrumental: True이면 BGM 전용 — 더미 가사로 40초 길이를 유도하고 보컬 없이 생성
 
         Returns:
             task_id: 작업 추적용 ID
@@ -125,7 +125,7 @@ class SunoService:
         Note:
             - 스트림 URL: 30-40초 내 생성
             - 다운로드 URL: 2-3분 내 생성
-            - 생성되는 노래는 약 1분 이내의 길이
+            - 생성되는 노래는 약 40초 이내의 길이
         """
         actual_callback_url = callback_url or apikey_settings.SUNO_CALLBACK_URL
 
@@ -133,11 +133,11 @@ class SunoService:
 
         if instrumental:
             bgm_lyrics = get_bgm_lyrics(genre)
-            formatted_prompt = f"[Song Duration: Around 1 minute - Must be around 60 seconds]\n{bgm_lyrics}"
+            formatted_prompt = f"[Song Duration: Around 40 seconds]\n{bgm_lyrics}"
             logger.info(f"[Suno] BGM 더미 가사 장르 {normalized_genre} 선택됨")
         else:
             formatted_prompt = (
-                f"[Song Duration: Around 1 minute - Must be around 60 seconds]\n{prompt}"
+                f"[Song Duration: Around 40 seconds]\n{prompt}"
             )
 
         payload: dict[str, Any] = {
@@ -148,7 +148,7 @@ class SunoService:
             "callBackUrl": actual_callback_url,
         }
         if normalized_genre:
-            payload["style"] = f"{normalized_genre}, around 60 seconds" if instrumental else normalized_genre
+            payload["style"] = f"{normalized_genre}, around 40 seconds" if instrumental else normalized_genre
 
         last_error: Exception | None = None
 

app/utils/thumbnail.py

@@ -1,197 +0,0 @@
-"""세로형 썸네일 정지 이미지 렌더용 컴포지션 및 source 조립.
-
-CreatomateService에 의존하지 않는 코드 측 자산이다. 실제 렌더 요청·이미지 선택은
-CreatomateService.render_thumbnail / select_thumbnail_image가 담당한다.
-"""
-
-import copy
-
-# 세로형 썸네일(1080×1920) 정지 이미지 렌더용 컴포지션.
-# 운영 템플릿에 의존하지 않는 코드 측 자산 — 디자인 수정 시 이 상수만 변경.
-# 텍스트 슬롯 이름은 CreatomateService.make_thumbnail_modification()의 반환 키와 일치해야 한다.
-THUMBNAIL_COMPOSITION_V: dict = {
-    "name": "thumbnail-composition",
-    "type": "composition",
-    "track": 1,
-    "time": 0,
-    "elements": [
-        {
-            "name": "thumb-background",
-            "type": "image",
-            "track": 1,
-            "time": 0,
-            "width": "105%",
-            "height": "105%",
-            "smart_crop": True,
-            "color_overlay": "rgba(0,0,0,0.22)",
-        },
-        {
-            "name": "thumb-badge-category",
-            "type": "text",
-            "track": 2,
-            "time": 0,
-            "y": "8%",
-            "width": "75%",
-            "height": "7%",
-            "x_alignment": "50%",
-            "y_alignment": "50%",
-            "font_family": "Pretendard",
-            "font_weight": "700",
-            "font_size_maximum": "4.8 vmin",
-            "letter_spacing": "12%",
-            "line_height": "110%",
-            "background_color": "#FFFFFF",
-            "background_x_padding": "50%",
-            "background_y_padding": "65%",
-            "background_border_radius": "100%",
-            "fill_color": "#1A1A1A",
-        },
-        {
-            "name": "thumb-headline-hook_claim-aspirational",
-            "type": "text",
-            "track": 3,
-            "time": 0,
-            "y": "45%",
-            "width": "85%",
-            "height": "22%",
-            "x_alignment": "50%",
-            "y_alignment": "50%",
-            "font_family": "Pretendard",
-            "font_weight": "900",
-            "font_size_maximum": "11 vmin",
-            "letter_spacing": "-2%",
-            "fill_color": "#FFFFFF",
-            "stroke_color": "rgba(0,0,0,0.65)",
-            "stroke_width": "0.8 vmin",
-            "shadow_color": "rgba(0,0,0,0.55)",
-            "shadow_blur": "2.5 vmin",
-            "shadow_y": "0.5 vmin",
-        },
-        {
-            "name": "thumb-subheadline-selling_point",
-            "type": "text",
-            "track": 4,
-            "time": 0,
-            "y": "64%",
-            "width": "88%",
-            "height": "9%",
-            "x_alignment": "50%",
-            "y_alignment": "50%",
-            "font_family": "Pretendard",
-            "font_weight": "500",
-            "font_size_maximum": "6 vmin",
-            "letter_spacing": "6%",
-            "line_height": "130%",
-            "fill_color": "#FFFFFF",
-            "stroke_color": "rgba(0,0,0,0.55)",
-            "stroke_width": "0.4 vmin",
-        },
-        {
-            "name": "thumb-brand-wordmark",
-            "type": "text",
-            "track": 5,
-            "time": 0,
-            "y": "89%",
-            "width": "75%",
-            "height": "9%",
-            "x_alignment": "50%",
-            "y_alignment": "50%",
-            "font_family": "Pretendard",
-            "font_weight": "900",
-            "font_size_maximum": "7.5 vmin",
-            "letter_spacing": "8%",
-            "line_height": "120%",
-            "fill_color": "#FFFFFF",
-            "stroke_color": "rgba(0,0,0,0.6)",
-            "stroke_width": "0.5 vmin",
-            "shadow_color": "rgba(0,0,0,0.45)",
-            "shadow_blur": "1.5 vmin",
-            "shadow_y": "0.3 vmin",
-        },
-        {
-            "name": "thumb-hashtag-primary",
-            "type": "text",
-            "track": 6,
-            "time": 0,
-            "y": "95.5%",
-            "width": "92%",
-            "height": "6%",
-            "x_alignment": "50%",
-            "y_alignment": "50%",
-            "font_family": "Pretendard",
-            "font_weight": "600",
-            "font_size_maximum": "4.2 vmin",
-            "letter_spacing": "4%",
-            "line_height": "120%",
-            "fill_color": "#FFFFFF",
-            "stroke_color": "rgba(0,0,0,0.5)",
-            "stroke_width": "0.3 vmin",
-        },
-    ],
-}
-
-
-# 영상 첫 화면 썸네일 오버레이 노출 시간(초). 30fps 기준 3프레임 — SNS 첫 프레임
-# 캡처에는 충분하면서 시청 흐름에는 거의 인지되지 않는 안전 하한선.
-THUMBNAIL_INTRO_DURATION = 0.1
-
-# 오버레이가 기존 최상위 트랙(1: 씬, 2: 오디오, 3: 자막, 4: 키워드)과
-# 렌더 직전 append되는 가사 캡션(track 3~4) 위에 오도록 여유 있게 잡은 트랙 번호.
-THUMBNAIL_INTRO_TRACK = 9
-
-
-def _fill_thumbnail_slots(composition: dict, image_url: str, text_modifications: dict[str, str]) -> None:
-    """썸네일 컴포지션의 이미지/텍스트 슬롯을 in-place로 채웁니다."""
-    for elem in composition["elements"]:
-        name = elem.get("name")
-        if elem["type"] == "image" and name == "thumb-background":
-            elem["source"] = image_url
-        elif elem["type"] == "text" and name in text_modifications:
-            elem["text"] = text_modifications[name]
-
-
-def build_thumbnail_source(image_url: str, text_modifications: dict[str, str]) -> dict:
-    """썸네일 렌더용 Creatomate source를 조립합니다.
-
-    Args:
-        image_url: 배경 이미지 URL
-        text_modifications: make_thumbnail_modification() 반환값 (thumb-* 텍스트 슬롯)
-
-    Returns:
-        1080×1920 정지 이미지 렌더 source 딕셔너리
-    """
-    composition = copy.deepcopy(THUMBNAIL_COMPOSITION_V)
-    _fill_thumbnail_slots(composition, image_url, text_modifications)
-    return {
-        "width": 1080,
-        "height": 1920,
-        "elements": [composition],
-    }
-
-
-def build_thumbnail_intro_element(
-    image_url: str,
-    text_modifications: dict[str, str],
-    duration: float = THUMBNAIL_INTRO_DURATION,
-) -> dict:
-    """영상 첫 화면(SNS 첫 프레임용)에 얹을 썸네일 오버레이 요소를 조립합니다.
-
-    duration 스케일링(extend_template_duration) 이후, 렌더 요청 직전에
-    final_template["source"]["elements"]에 append해야 노출 시간이 그대로 유지된다.
-    기존 씬·자막·오디오 타이밍은 건드리지 않는 순수 오버레이이다.
-
-    Args:
-        image_url: 배경 이미지 URL
-        text_modifications: make_thumbnail_modification() 반환값 (thumb-* 텍스트 슬롯)
-        duration: 오버레이 노출 시간(초)
-
-    Returns:
-        time=0, 최상위 트랙에 배치된 composition 요소 딕셔너리
-    """
-    composition = copy.deepcopy(THUMBNAIL_COMPOSITION_V)
-    composition["name"] = "thumbnail-intro"
-    composition["track"] = THUMBNAIL_INTRO_TRACK
-    composition["time"] = 0
-    composition["duration"] = duration
-    _fill_thumbnail_slots(composition, image_url, text_modifications)
-    return composition

app/utils/thumbnail_fitness.py

@@ -0,0 +1,231 @@
+# -*- coding: utf-8 -*-
+"""
+썸네일 슬롯 한정 픽셀(헤더) 룰 스코어러 — Pillow/numpy 의존성 없이 구현.
+
+배경 (zzz/ado2-thumbnail-selection 파일럿 대비 축소 이식):
+  기존 태그 매칭(calculate_image_slot_score_multi)은 의미 태그 교집합만 보고
+  실제 화질·프레이밍을 보지 않는다. 파일럿(수원화성, 2026-07-20)에서 검증된
+  6축 스코어러 중 "실질 가치는 최고 컷 찾기보다 명백히 나쁜 컷의 자동 배제"
+  (참고: references/scoring-logic.md)라는 결론에 따라, 하드 리젝트에 직결되는
+  두 축(해상도 업스케일 배율, 종횡비)만 이식한다.
+
+  이 두 축은 실제 픽셀 색상이 필요 없고 원본 가로/세로 픽셀 수만 알면 되므로,
+  이미지를 전체 디코드하지 않고 파일 헤더 몇십 KB만 읽어(HTTP Range) 포맷별
+  고정 위치에서 크기를 파싱한다(JPEG는 SOF 마커 스캔). Pillow의 픽셀 디코드가
+  없으므로 서버(Standard_B2als_v2, 2 vCPU 버스터블) CPU 부담이 사실상 0에
+  가깝다.
+
+  텍스트 안전영역 밀도·명암 대비(원 스코어러의 safe_area/contrast 축)는 실제
+  픽셀 값이 있어야 계산되는 축이라 이 방식으로는 커버하지 못한다 — 의도적으로
+  범위 밖으로 남겨둔 트레이드오프.
+"""
+from __future__ import annotations
+
+import asyncio
+import struct
+
+import httpx
+
+from app.utils.logger import get_logger
+
+logger = get_logger("thumbnail_fitness")
+
+# ── 커버 스펙 (thumbnail-composition 렌더 기준) ──────────────────────────
+TARGET_H = 1920          # 9:16 출력 캔버스 기준 세로 px
+_CROP_RATIO = 9 / 16      # tight_crop(fit=cover) 목표 종횡비
+
+REJECT_UPSCALE = 2.5      # 9:16 크롭 후 업스케일 배율 초과 → 배제 (OG 1200×630 ≈ 3.05배)
+REJECT_ASPECT = 2.2       # 원본 종횡비(w/h) 초과 극단 가로형 → 배제
+
+_RANGE_BYTES = 131072     # 헤더만 읽기 위한 최대 다운로드 크기 (128KB, SOF 스캔 여유분)
+_DOWNLOAD_TIMEOUT = 8.0
+_DEFAULT_CONCURRENCY = 4  # 헤더만 받으므로 CPU 부담 없음 — I/O 대기 기준으로 넉넉히
+
+
+# ── 포맷별 헤더 파싱 (순수 struct, 전체 디코드 없음) ──────────────────────
+def parse_image_size(data: bytes) -> tuple[int, int] | None:
+    """이미지 바이트(파일 앞부분)에서 원본 (width, height)를 추출합니다.
+
+    지원: JPEG(SOF 마커 스캔), PNG(IHDR), GIF(고정 오프셋), WebP(VP8X/VP8L).
+    단순 손실 WebP(VP8)는 비트 단위 파싱이 필요해 범위 밖 — 실패 시 None
+    (호출자가 중립 처리, 배제하지 않음).
+    """
+    if len(data) < 12:
+        return None
+    if data[:2] == b"\xff\xd8":
+        return _parse_jpeg_size(data)
+    if data[:8] == b"\x89PNG\r\n\x1a\n":
+        return _parse_png_size(data)
+    if data[:6] in (b"GIF87a", b"GIF89a"):
+        return _parse_gif_size(data)
+    if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
+        return _parse_webp_size(data)
+    return None
+
+
+def _parse_png_size(data: bytes) -> tuple[int, int] | None:
+    if len(data) < 24:
+        return None
+    width, height = struct.unpack(">II", data[16:24])
+    return width, height
+
+
+def _parse_gif_size(data: bytes) -> tuple[int, int] | None:
+    if len(data) < 10:
+        return None
+    width, height = struct.unpack("<HH", data[6:10])
+    return width, height
+
+
+_JPEG_SOF_MARKERS = {
+    0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7,
+    0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF,
+}
+_JPEG_NO_LENGTH_MARKERS = {0xD8, 0xD9, 0x01} | set(range(0xD0, 0xD8))  # SOI/EOI/RST0-7/TEM
+
+
+def _parse_jpeg_size(data: bytes) -> tuple[int, int] | None:
+    n = len(data)
+    pos = 2  # SOI(0xFFD8) 이후부터 마커 스캔
+    while pos < n:
+        if data[pos] != 0xFF:
+            pos += 1
+            continue
+        # 0xFF 뒤 연속된 fill byte(0xFF) 스킵 후 실제 마커 바이트 탐색
+        marker_pos = pos
+        while marker_pos < n and data[marker_pos] == 0xFF:
+            marker_pos += 1
+        if marker_pos >= n:
+            return None
+        marker = data[marker_pos]
+        pos = marker_pos + 1
+        if marker in _JPEG_NO_LENGTH_MARKERS:
+            continue
+        if pos + 2 > n:
+            return None
+        seg_len = struct.unpack(">H", data[pos:pos + 2])[0]
+        if marker in _JPEG_SOF_MARKERS:
+            if pos + 7 > n:
+                return None  # SOF가 헤더 범위 밖에 있음 — 파싱 실패(중립 처리)
+            height, width = struct.unpack(">HH", data[pos + 3:pos + 7])
+            return width, height
+        pos += seg_len
+    return None
+
+
+def _parse_webp_size(data: bytes) -> tuple[int, int] | None:
+    if len(data) < 30:
+        return None
+    fourcc = data[12:16]
+    if fourcc == b"VP8X":
+        width = 1 + (data[24] | (data[25] << 8) | (data[26] << 16))
+        height = 1 + (data[27] | (data[28] << 8) | (data[29] << 16))
+        return width, height
+    if fourcc == b"VP8L":
+        if len(data) < 25 or data[20] != 0x2F:
+            return None
+        b0, b1, b2, b3 = data[21:25]
+        bits = b0 | (b1 << 8) | (b2 << 16) | (b3 << 24)
+        width = (bits & 0x3FFF) + 1
+        height = ((bits >> 14) & 0x3FFF) + 1
+        return width, height
+    # 단순 손실 WebP(VP8) — 비트 단위 파싱 범위 밖, 중립 처리
+    return None
+
+
+# ── 점수화 (실제 픽셀 없이 w,h만으로 산출) ─────────────────────────────
+def score_pixel_fitness(width: int, height: int) -> dict:
+    """해상도(업스케일 배율)·종횡비만으로 썸네일 적합도를 판정합니다.
+
+    Returns:
+        {"score": 0.0~1.0 (배율, 태그 점수에 곱해 쓴다), "reject": str}
+        reject가 비어있지 않으면 하드 배제 대상.
+    """
+    if width <= 0 or height <= 0:
+        return {"score": 0.0, "reject": "invalid_dimensions"}
+
+    ratio = width / height
+    if ratio > REJECT_ASPECT:
+        return {"score": 0.0, "reject": f"극단 가로형({ratio:.2f}:1)"}
+
+    cropped_h = height if ratio > _CROP_RATIO else int(width / _CROP_RATIO)
+    if cropped_h <= 0:
+        return {"score": 0.0, "reject": "invalid_dimensions"}
+
+    upscale = TARGET_H / cropped_h
+    if upscale > REJECT_UPSCALE:
+        return {"score": 0.0, "reject": f"저해상도(업스케일 {upscale:.2f}x)"}
+
+    soft_score = 1.0 if upscale <= 1.0 else max(0.0, 1.0 - (upscale - 1.0) / 2.0)
+    return {"score": round(soft_score, 4), "reject": ""}
+
+
+# ── 네트워크: 헤더 바이트만 받기 (HTTP Range, 미지원 서버도 조기 종료로 방어) ──
+async def _fetch_header_bytes(
+    client: httpx.AsyncClient, url: str, max_bytes: int = _RANGE_BYTES, timeout: float = _DOWNLOAD_TIMEOUT
+) -> bytes | None:
+    """이미지 URL에서 앞부분 max_bytes만 받아옵니다.
+
+    Range 헤더를 무시하고 전체를 돌려주는 서버가 있어도, 스트리밍을 max_bytes
+    수신 즉시 중단해 실제 다운로드량을 상한선 이내로 강제한다.
+    """
+    try:
+        async with client.stream(
+            "GET", url, headers={"Range": f"bytes=0-{max_bytes - 1}"}, timeout=timeout
+        ) as response:
+            if response.status_code not in (200, 206):
+                return None
+            buf = bytearray()
+            async for chunk in response.aiter_bytes():
+                buf.extend(chunk)
+                if len(buf) >= max_bytes:
+                    break
+            return bytes(buf[:max_bytes])
+    except Exception as e:
+        logger.warning(f"[thumbnail_fitness] 헤더 다운로드 실패: {url} - {e}")
+        return None
+
+
+async def score_pool_thumbnail_fitness(
+    pool: list[dict],
+    client: httpx.AsyncClient,
+    concurrency: int = _DEFAULT_CONCURRENCY,
+    total_timeout: float = 20.0,
+) -> dict[str, dict]:
+    """이미지 풀(URL 중복 제거)의 썸네일 픽셀 적합도를 병렬로 계산합니다.
+
+    다운로드/파싱 실패 항목은 결과 dict에서 아예 빠진다 — 호출자가 "데이터
+    없음 = 판정 보류(중립)"로 처리하도록 유도(하드 배제로 오판하지 않기 위함).
+    전체 배치에 total_timeout 상한을 걸어, 네트워크 불량 시 이 부가 기능이
+    영상 생성 사전 준비 전체를 지연시키지 않도록 한다(초과 시 빈 dict).
+
+    Returns:
+        {image_url: {"score": float, "reject": str}} — 성공한 URL만 포함.
+    """
+    semaphore = asyncio.Semaphore(concurrency)
+    urls = list({item["image_url"] for item in pool if item.get("image_url")})
+
+    async def _score_one(url: str) -> tuple[str, dict | None]:
+        async with semaphore:
+            header = await _fetch_header_bytes(client, url)
+        if header is None:
+            return url, None
+        size = parse_image_size(header)
+        if size is None:
+            logger.warning(f"[thumbnail_fitness] 이미지 크기 파싱 실패(포맷 미지원/헤더 부족): {url}")
+            return url, None
+        width, height = size
+        return url, score_pixel_fitness(width, height)
+
+    try:
+        results = await asyncio.wait_for(
+            asyncio.gather(*[_score_one(u) for u in urls]),
+            timeout=total_timeout,
+        )
+    except asyncio.TimeoutError:
+        logger.warning(
+            f"[thumbnail_fitness] 전체 스코어링 타임아웃({total_timeout}s, urls={len(urls)}) — "
+            "픽셀 적합도 없이(태그 점수만) 진행"
+        )
+        return {}
+    return {url: fitness for url, fitness in results if fitness is not None}

app/utils/thumbnail_vision.py

@@ -0,0 +1,110 @@
+# -*- coding: utf-8 -*-
+"""
+썸네일 최종 선택 — 규칙/픽셀로 압축한 top-N 후보 중 비전 LLM이 1개 선택 (Phase 2).
+
+배경 (하이브리드 설계):
+  규칙(태그 매칭) + 픽셀 헤더 필터로 썸네일 후보를 소수(top-3)로 압축한 뒤,
+  그 소수 중 최종 1컷만 비전 LLM이 이미지를 실제로 보고 고른다. 선택지가
+  "1슬롯 × 3후보"로 갇혀 있어 전면 LLM 배정의 위험(제약 위반·중복 배정·큰
+  블라스트 반경)이 없고, 실패 시 규칙 1위로 폴백한다.
+
+  이 단계의 실질 가치: Pillow를 쓰지 않아 포기했던 지각적 축을 비전으로 되찾음.
+  - 이미지 속 글자/간판/워터마크가 커버 텍스트 4슬롯과 겹치는지
+  - 중앙 9:16 크롭 후 주제가 잘리거나 어중간해지는지
+  - 업종 대표성·클릭 유인
+"""
+from __future__ import annotations
+
+import asyncio
+
+from pydantic import BaseModel, Field
+
+from app.utils.logger import get_logger
+from app.utils.prompts.chatgpt_prompt import ChatgptService
+
+logger = get_logger("thumbnail_vision")
+
+# 비전 판정 모델·타임아웃 — 부가 기능이므로 실패해도 규칙 폴백, 파이프라인은 진행
+_VISION_MODEL = "gpt-5-mini"
+_VISION_TIMEOUT = 20.0
+
+
+class ThumbnailPickOutput(BaseModel):
+    """비전 LLM의 썸네일 선택 출력."""
+    choice_index: int = Field(..., description="선택한 이미지의 0-기반 인덱스 (첨부 이미지 순서와 동일)")
+    reason: str = Field(..., description="선택 근거 한 줄 (텍스트 충돌/크롭 구도/대표성 관점)")
+
+
+def _build_prompt(candidate_count: int, industry: str, business_name: str) -> str:
+    return f"""당신은 숏폼 광고 영상의 **썸네일(커버) 배경 이미지**를 고르는 전문가입니다.
+
+첨부된 {candidate_count}장의 이미지는 이미 태그·화질 필터를 통과한 후보들입니다.
+이 중 커버로 가장 적합한 **1장**을 골라 0-기반 인덱스로 반환하세요.
+(첫 번째 이미지 = 0, 두 번째 = 1, ...)
+
+업체 정보: {business_name} ({industry} 업종)
+
+썸네일 위에는 아래 4개의 텍스트가 흰 글자로 얹힙니다:
+- 상단(약 8% 높이): 카테고리 뱃지
+- 중앙(약 50%): 업체명 (큰 글자)
+- 중앙 하단(약 62%): 지역
+- 최하단(약 94%): 해시태그
+
+선택 기준 (중요도 순):
+0. **업종 대표성·클릭 유인**: 한눈에 어떤 곳인지 전달되고 매력적일 것
+1. **텍스트 충돌 회피**: 이미지 속 간판·안내판 글자·워터마크가 위 텍스트 영역과 겹치지 않을 것
+2. **크롭 후 구도**: 세로 9:16 중앙 크롭 시 핵심 주제가 잘리지 않고 살아있을 것
+3. **가독성**: 텍스트가 얹히는 영역(상/중/하단)이 너무 밝거나 복잡하지 않아 흰 글자가 잘 보일 것
+"""
+
+
+async def pick_thumbnail_by_vision(
+    candidates: list[dict],
+    industry: str,
+    business_name: str,
+) -> dict | None:
+    """후보 이미지 중 비전 LLM이 최종 1컷을 선택합니다.
+
+    Args:
+        candidates: [{"image_url": str, "image_tag": dict}, ...] — 규칙/픽셀로
+            압축한 top-N 후보 (점수 내림차순, 즉 index 0이 규칙 1위).
+
+    Returns:
+        선택된 candidate dict. 후보가 1개 이하이거나 호출 실패/타임아웃/무효
+        인덱스면 None (호출자가 규칙 1위 폴백).
+    """
+    if len(candidates) < 2:
+        # 선택할 게 없음 — 규칙 1위(있으면)로 폴백
+        return None
+
+    urls = [c["image_url"] for c in candidates]
+    prompt = _build_prompt(len(candidates), industry, business_name)
+    chatgpt = ChatgptService(model_type="gpt", timeout=_VISION_TIMEOUT)
+
+    try:
+        result: ThumbnailPickOutput = await asyncio.wait_for(
+            chatgpt.generate_structured_output_multi_image(
+                prompt_text=prompt,
+                output_format=ThumbnailPickOutput,
+                model=_VISION_MODEL,
+                img_urls=urls,
+            ),
+            timeout=_VISION_TIMEOUT,
+        )
+    except Exception as e:
+        logger.warning(f"[thumbnail_vision] 비전 선택 실패 — 규칙 폴백: {e}")
+        return None
+
+    idx = result.choice_index
+    if not (0 <= idx < len(candidates)):
+        logger.warning(
+            f"[thumbnail_vision] 무효 인덱스({idx}, 후보 {len(candidates)}개) — 규칙 폴백"
+        )
+        return None
+
+    logger.info(
+        f"[thumbnail_vision] 비전 선택 — index={idx}"
+        f"{' (규칙 1위와 동일)' if idx == 0 else ' (규칙 1위 아님)'}, "
+        f"url={candidates[idx]['image_url']}, reason={result.reason}"
+    )
+    return candidates[idx]

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

@@ -31,10 +31,7 @@ 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.lyric.models import Lyric
 from app.song.models import Song, SongTimestamp
-from app.utils.creatomate import CreatomateService
-# 썸네일 인트로 오버레이 잠정 비활성화 — 재활성화 시 주석 해제 (generate_video 내 블록 참조)
-# from app.utils.thumbnail import build_thumbnail_intro_element
-# from app.video.services.video import get_image_tags_by_task_id
+from app.utils.creatomate import CreatomateService, LANGUAGE_FONT_MAP
 
 from app.comment.models import Comment
 from app.database.like_cache import (
@@ -60,7 +57,6 @@ from app.video.schemas.video_schema import (
     VideoRenderData,
     VideoThumbnailItem,
 )
-from app.video.worker.thumbnail_task import generate_thumbnail_background
 from app.video.worker.video_task import download_and_upload_video_to_blob
 
 
@@ -129,7 +125,6 @@ curl -X GET "http://localhost:8000/video/generate/0694b716-dbff-7219-8000-d08cb5
 )
 async def generate_video(
     task_id: str,
-    background_tasks: BackgroundTasks,
     orientation: Literal["horizontal", "vertical"] = Query(
         default="vertical",
         description="영상 방향 (horizontal: 가로형, vertical: 세로형)",
@@ -197,6 +192,7 @@ async def generate_video(
             brand_name = project.store_name
             region = project.region
             industry = project.industry
+            output_language = project.language or "Korean"
 
             # MarketingIntel 조회
             marketing_result = await session.execute(
@@ -226,25 +222,6 @@ async def generate_video(
             category_definition = marketing_intelligence.intel_result["market_positioning"]["category_definition"]
             target_keywords = marketing_intelligence.intel_result["target_keywords"]
 
-            # 썸네일 헤드라인(thumb-headline-hook_claim-aspirational)용 문구 추출.
-            # 1순위: 마케팅 분석 시 LLM이 5~15자로 축약 생성한 hook_headline (신규 데이터)
-            # 폴백: hook_headline이 없는 구 데이터는 concept 셀링포인트 → core_value →
-            #        최고점 셀링포인트 순으로 추출 (make_thumbnail_modification에서 15자 절단 안전망 적용)
-            brand_concept = marketing_intelligence.intel_result.get("hook_headline", "") or ""
-            selling_points = marketing_intelligence.intel_result["selling_points"]
-            if not brand_concept:
-                for sp in selling_points:
-                    if "concept" in sp["english_category"].lower():
-                        brand_concept = sp["description"]
-                        break
-            if not brand_concept:
-                brand_concept = marketing_intelligence.intel_result["market_positioning"].get("core_value", "")
-                if brand_concept:
-                    logger.info(f"[generate_video] brand_concept 폴백(core_value) 사용 - task_id: {task_id}")
-            if not brand_concept and selling_points:
-                brand_concept = max(selling_points, key=lambda sp: sp.get("score", 0))["description"]
-                logger.info(f"[generate_video] brand_concept 폴백(최고점 셀링포인트) 사용 - task_id: {task_id}")
-
             # Lyric 조회
             lyric_result = await session.execute(
                 select(Lyric)
@@ -355,19 +332,6 @@ async def generate_video(
             )
         # 세션이 여기서 자동으로 닫힘 (async with 블록 종료)
 
-        # 썸네일 백그라운드 생성 (세로형만, 응답 비차단) — 실패해도 영상 생성에 영향 없음
-        background_tasks.add_task(
-            generate_thumbnail_background,
-            video_id=video_id,
-            task_id=task_id,
-            orientation=orientation,
-            brand_name=brand_name,
-            region=region,
-            brand_concept=brand_concept,
-            category_definition=category_definition,
-            target_keywords=target_keywords,
-        )
-
     except HTTPException:
         raise
     except Exception as e:
@@ -392,7 +356,6 @@ async def generate_video(
         creatomate_service = CreatomateService(
             orientation=orientation,
             industry=industry,
-            project_id=project_id,
         )
         logger.debug(
             f"[generate_video] Using template_id: {creatomate_service.template_id}, (song duration: {song_duration})"
@@ -420,15 +383,22 @@ async def generate_video(
 
         modifications.update(subtitle_modifications)
 
-        # revert thumbnail scene
-        thumbnail_modifications = creatomate_service.make_thumbnail_modification(
+        # 썸네일 텍스트: 사전 자막 생성(LLM, 다국어) 결과에 thumb-* 슬롯이 포함되면
+        # 그대로 사용한다. 과거 파이프라인 산출물(subtitle에 thumb-* 없음)은
+        # 기존 팩트값 조립(한국어)으로 폴백.
+        thumbnail_fallback = creatomate_service.make_thumbnail_modification(
             brand_name =brand_name,
             region = region,
-            brand_concept = brand_concept,
             category_definition= category_definition,
-            target_keywords=target_keywords)
-        
-        modifications.update(thumbnail_modifications)
+            target_keywords=target_keywords,
+            detail_region_info=store_address)
+
+        for slot_name, fallback_value in thumbnail_fallback.items():
+            if not modifications.get(slot_name):
+                modifications[slot_name] = fallback_value
+                logger.info(
+                    f"[generate_video] thumbnail slot fallback(factual) 적용: {slot_name} - task_id: {task_id}"
+                )
 
         # 6-3. elements 수정
         new_elements = creatomate_service.modify_element(
@@ -453,12 +423,8 @@ async def generate_video(
         for i, ts in enumerate(song_timestamp_list):
             logger.debug(f"[generate_video] timestamp[{i}]: lyric_line={ts.lyric_line}, start_time={ts.start_time}, end_time={ts.end_time}")
 
-        match lyric_language:
-            case "English" : 
-                lyric_font = "Noto Sans"
-                # lyric_font = "Pretendard" # 없어요
-            case _ : 
-                lyric_font = "Noto Sans"
+        # 가사 자막 폰트: CJK/태국어는 글리프 지원 폰트로, 그 외는 Noto Sans
+        lyric_font = LANGUAGE_FONT_MAP.get(lyric_language, "Noto Sans")
 
         # LYRIC AUTO 결정부
         if (creatomate_settings.LYRIC_SUBTITLE):
@@ -479,23 +445,13 @@ async def generate_video(
                     final_template["source"]["elements"].append(caption)
         # END - LYRIC AUTO 결정부
 
-        # 썸네일 인트로 오버레이 (세로형만) — SNS 첫 프레임용 0.1초 정지 화면.
-        # duration 스케일링·자막 append 이후에 붙여야 노출 시간이 유지되며,
-        # 실패해도 영상 생성 자체에는 영향을 주지 않는다.
-        # NOTE: 잠정 비활성화 — 재활성화 시 아래 블록과 상단 import 주석 해제
-        # if orientation == "vertical":
-        #     try:
-        #         taged_image_list = await get_image_tags_by_task_id(task_id)
-        #         thumb_image_url = creatomate_service.select_thumbnail_image(taged_image_list)
-        #         if thumb_image_url:
-        #             final_template["source"]["elements"].append(
-        #                 build_thumbnail_intro_element(thumb_image_url, thumbnail_modifications)
-        #             )
-        #             logger.info(f"[generate_video] Thumbnail intro overlay added - task_id: {task_id}")
-        #         else:
-        #             logger.warning(f"[generate_video] 썸네일 인트로 배경 이미지 없음 — skip - task_id: {task_id}")
-        #     except Exception as e:
-        #         logger.warning(f"[generate_video] 썸네일 인트로 추가 실패(무시) - task_id: {task_id}, error: {e}")
+        # 언어별 폰트 교체: 템플릿 기본 폰트는 한글·라틴 전용이라 CJK/태국어는
+        # 지원 폰트로 전체 텍스트(자막·키워드·썸네일·가사 캡션 포함)를 교체한다.
+        # 가사 캡션 append 이후에 실행해야 캡션까지 커버된다.
+        final_template = creatomate_service.apply_language_font(
+            final_template, output_language
+        )
+
         # logger.debug(
         #     f"[generate_video] final_template: {json.dumps(final_template, indent=2, ensure_ascii=False)}"
         # )
@@ -1028,7 +984,6 @@ async def get_all_videos(
                 video_id=v.id,
                 store_name=p.store_name,
                 result_movie_url=v.result_movie_url,
-                thumbnail_url=v.video_thumbnail_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),

app/video/models.py

@@ -106,12 +106,6 @@ class Video(Base):
         comment="생성된 영상 URL",
     )
 
-    video_thumbnail_url: Mapped[Optional[str]] = mapped_column(
-        String(2048),
-        nullable=True,
-        comment="영상 생성 시 백그라운드로 렌더된 썸네일 이미지 URL (실패/미생성 시 NULL)",
-    )
-
     is_deleted: Mapped[bool] = mapped_column(
         Boolean,
         nullable=False,

app/video/schemas/video_schema.py

@@ -157,7 +157,6 @@ class VideoListItem(BaseModel):
     region: Optional[str] = Field(None, description="지역명")
     task_id: str = Field(..., description="작업 고유 식별자")
     result_movie_url: Optional[str] = Field(None, description="영상 결과 URL")
-    thumbnail_url: Optional[str] = Field(None, description="백그라운드 렌더된 썸네일 이미지 URL (세로형만, 실패/미생성 시 null)")
     created_at: Optional[datetime] = Field(None, description="생성 일시")
     like_count: int = Field(0, description="좋아요 수")
     comment_count: int = Field(0, description="댓글 수 (대댓글 포함)")
@@ -172,8 +171,7 @@ class VideoThumbnailItem(BaseModel):
 
     video_id: int = Field(..., description="영상 고유 ID (상세 페이지 라우팅 키)")
     store_name: str = Field(..., description="업체명")
-    result_movie_url: str = Field(..., description="영상 URL — thumbnail_url 없을 때 프론트에서 <video> 태그 첫 프레임을 썸네일로 사용")
-    thumbnail_url: Optional[str] = Field(None, description="백그라운드 렌더된 썸네일 이미지 URL (세로형만, 실패/미생성 시 null)")
+    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)")

app/video/services/video.py

@@ -857,6 +857,16 @@ async def get_image_tags_by_task_id(task_id: str) -> list[dict]:
         # print(stmt.compile(dialect=mysql.dialect(), compile_kwargs={"literal_binds": True}))
         rows = (await session.execute(stmt)).all()
         # print("rows", rows)
-    # print(rows)
-    # print("image" , [{"image_url": row.img_url, "image_tag": row.img_tag} for row in rows])
-    return [{"image_url": row.img_url, "image_tag": row.img_tag} for row in rows]
+    # img_tag가 JSON 리터럴 null로 저장된 행(태깅 실패 잔재)은 SQL IS NOT NULL 필터를
+    # 통과하므로 파이썬 레벨에서 한 번 더 걸러낸다.
+    null_tag_urls = [row.img_url for row in rows if row.img_tag is None]
+    if null_tag_urls:
+        logger.warning(
+            f"[get_image_tags_by_task_id] img_tag가 null인 이미지 {len(null_tag_urls)}개 제외 "
+            f"- task_id: {task_id}, urls: {null_tag_urls}"
+        )
+    return [
+        {"image_url": row.img_url, "image_tag": row.img_tag}
+        for row in rows
+        if row.img_tag is not None
+    ]

app/video/worker/creative_assets_task.py

@@ -9,11 +9,15 @@ from sqlalchemy import select
 from app.database.session import BackgroundSessionLocal
 from app.home.models import Project, MarketingIntel
 from app.utils.subtitles import SubtitleContentsGenerator
+from app.utils import thumbnail_fitness
+from app.utils.thumbnail_vision import pick_thumbnail_by_vision
 from app.utils.creatomate import (
     CreatomateService,
     SCENE_TRACK,
     SUBTITLE_TRACK,
     KEYWORD_TRACK,
+    THUMBNAIL_SLOT_MARKER,
+    get_shared_client,
 )
 from app.utils.logger import get_logger
 from app.video.services.video import get_image_tags_by_task_id
@@ -97,12 +101,65 @@ async def generate_creative_assets_background(
                 f"duplicate={duplicate} - task_id: {task_id}"
             )
 
+            # 썸네일 슬롯(-9999)이 있는 템플릿이면 픽셀(헤더) 적합도 사전 계산.
+            # 헤더 바이트만 받아 크기를 파싱하므로 CPU 부담 없음 — 실패/타임아웃 시
+            # 빈 dict로 진행(태그 점수만 사용, 배정 자체는 막지 않음).
+            thumbnail_fitness_map: dict = {}
+            has_thumbnail_slot = any(
+                elem_type == "image" and name.endswith(THUMBNAIL_SLOT_MARKER)
+                for name, elem_type in creatomate_service.parse_template_component_name(
+                    template["source"]["elements"]
+                ).items()
+            )
+            if has_thumbnail_slot and taged_image_list:
+                client = await get_shared_client()
+                thumbnail_fitness_map = await thumbnail_fitness.score_pool_thumbnail_fitness(
+                    taged_image_list, client
+                )
+                rejected = {
+                    url: fit["reject"]
+                    for url, fit in thumbnail_fitness_map.items()
+                    if fit["reject"]
+                }
+                logger.info(
+                    f"[generate_creative_assets_background] thumbnail fitness — "
+                    f"scored={len(thumbnail_fitness_map)}/{len(taged_image_list)}, "
+                    f"rejected={len(rejected)} {rejected if rejected else ''} - task_id: {task_id}"
+                )
+
+            # ── Step 1-b: 썸네일 최종 선택 (비전 LLM, Phase 2) ──────────────
+            # 규칙+픽셀로 압축한 top-3 후보를 비전 LLM이 실제로 보고 1컷 선택.
+            # 실패/후보 부족 시 thumbnail_choice가 비어 규칙 1위로 자동 폴백.
+            thumbnail_choice: dict = {}
+            if has_thumbnail_slot and taged_image_list:
+                candidates_by_slot = creatomate_service.rank_thumbnail_candidates(
+                    template=template,
+                    taged_image_list=taged_image_list,
+                    thumbnail_fitness_map=thumbnail_fitness_map,
+                    top_n=3,
+                )
+                for slot, candidates in candidates_by_slot.items():
+                    chosen = await pick_thumbnail_by_vision(
+                        candidates=candidates,
+                        industry=project.industry,
+                        business_name=customer_name,
+                    )
+                    if chosen is not None:
+                        thumbnail_choice[slot] = chosen["image_url"]
+                logger.info(
+                    f"[generate_creative_assets_background] thumbnail vision pick — "
+                    f"{ {s: u.rsplit('/', 1)[-1] for s, u in thumbnail_choice.items()} } "
+                    f"- task_id: {task_id}"
+                )
+
             image_modifications, assigned_image_tags = creatomate_service.template_matching_taged_image(
                 template=template,
                 taged_image_list=taged_image_list,
                 music_url="",       # 음악 URL은 영상 생성 시점에 결정 — 사전 단계에서는 빈 값
                 address=store_address,
                 duplicate=duplicate,
+                thumbnail_fitness_map=thumbnail_fitness_map,
+                thumbnail_choice=thumbnail_choice,
             )
             # audio-music은 영상 생성 시점에 덮어씌워지므로 image_match에서 제거
             image_match = {k: v for k, v in image_modifications.items() if k != "audio-music"}

app/video/worker/thumbnail_task.py

@@ -1,69 +0,0 @@
-"""썸네일 백그라운드 생성 태스크.
-
-/video/generate 요청 시 응답을 막지 않도록 BackgroundTasks로 실행되며,
-세로형 영상에 한해 썸네일 정지 이미지를 렌더해 Video.video_thumbnail_url에 저장한다.
-모든 실패는 경고 로그 후 무시하며 예외를 전파하지 않는다.
-"""
-
-from sqlalchemy import select
-
-from app.database.session import BackgroundSessionLocal
-from app.utils.creatomate import CreatomateService
-from app.utils.logger import get_logger
-from app.video.models import Video
-from app.video.services.video import get_image_tags_by_task_id
-
-logger = get_logger("video")
-
-
-async def _save_thumbnail_url(video_id: int, thumbnail_url: str) -> None:
-    """렌더된 썸네일 URL을 Video 행에 저장한다."""
-    async with BackgroundSessionLocal() as session:
-        result = await session.execute(select(Video).where(Video.id == video_id))
-        video = result.scalar_one_or_none()
-        if video is None:
-            logger.warning(f"[generate_thumbnail_background] Video 없음 - video_id: {video_id}")
-            return
-        video.video_thumbnail_url = thumbnail_url
-        await session.commit()
-
-
-async def generate_thumbnail_background(
-    video_id: int,
-    task_id: str,
-    orientation: str,
-    brand_name: str,
-    region: str,
-    brand_concept: str,
-    category_definition: str,
-    target_keywords: list[str],
-) -> None:
-    """세로형 영상의 썸네일을 렌더해 Video.video_thumbnail_url에 저장한다.
-
-    보조 자산이므로 어떤 실패(이미지 없음, API 오류, 렌더 실패)도 예외를
-    전파하지 않고 경고 로그 후 종료한다. 가로형은 대상이 아니다.
-    """
-    if orientation != "vertical":
-        return
-    try:
-        service = CreatomateService(orientation=orientation)
-        taged_image_list = await get_image_tags_by_task_id(task_id)
-        image_url = service.select_thumbnail_image(taged_image_list)
-        if not image_url:
-            logger.warning(f"[generate_thumbnail_background] 썸네일 배경 이미지 없음 — skip - task_id: {task_id}")
-            return
-        text_mods = service.make_thumbnail_modification(
-            brand_name=brand_name,
-            region=region,
-            brand_concept=brand_concept,
-            category_definition=category_definition,
-            target_keywords=target_keywords,
-        )
-        thumbnail_url = await service.render_thumbnail(image_url, text_mods)
-        if thumbnail_url:
-            await _save_thumbnail_url(video_id, thumbnail_url)
-            logger.info(f"[generate_thumbnail_background] 저장 완료 - video_id: {video_id}, url: {thumbnail_url}")
-        else:
-            logger.warning(f"[generate_thumbnail_background] 렌더 실패 — skip - task_id: {task_id}")
-    except Exception as e:
-        logger.warning(f"[generate_thumbnail_background] 실패(무시) - task_id: {task_id}, error: {e}")

docs/database-schema/migration_2026_07_16_add_video_thumbnail_url.sql

@@ -1,11 +0,0 @@
--- ============================================================
--- Migration: video 테이블에 video_thumbnail_url 컬럼 추가
--- Date: 2026-07-16
--- Description: 영상 생성 시 백그라운드로 렌더된 썸네일 이미지 URL 저장
---   - 실패/미생성 시 NULL
--- ============================================================
-
-ALTER TABLE video
-  ADD COLUMN video_thumbnail_url VARCHAR(2048) NULL
-    COMMENT '영상 생성 시 백그라운드로 렌더된 썸네일 이미지 URL (실패/미생성 시 NULL)'
-  AFTER result_movie_url;