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/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

@@ -120,7 +120,7 @@ async def _resolve_industry(category: str, customer_name: str = "") -> str:
         "위 정보를 바탕으로 이 업체를 다음 업종 중 가장 적합한 하나로 분류하세요: "
         "stay(숙박/펜션/호텔), restaurant(음식점), cafe(카페/디저트), "
         "salon(미용실/네일/뷰티), clinic(병원/의원/치과), fitness(헬스/필라테스/요가), "
-        "academy(학원/교습소), attraction(관광/체험/액티비티). "
+        "academy(학원/교습소), attraction(관광/체험/액티비티/축제/행사). "
         "위 8개 중 어느 것에도 명확히 해당하지 않는 경우에만 general(기타/범용 업종)로 분류하세요. "
         "유사한 업종이 있으면 general 대신 그 업종을 우선 선택하세요."
     )
@@ -304,11 +304,14 @@ async def _crawling_logic(
         extra_photo_urls = scraper.extra_photo_urls or []
 
         if len(owner_images) >= NvMapScraper.SUPPLEMENT_THRESHOLD:
-            scraper.image_link_list = owner_images[: NvMapScraper.MAX_IMAGES]
+            # 업체 제공 사진은 필터링 면제 대상이므로 MAX_IMAGES 상한 없이 수집분을 전부 사용한다
+            # (MAX_IMAGES 상한은 방문자 사진이 섞이는 보충 경로에만 적용.
+            #  수집 자체는 NvMapScraper.BIZ_MAX_PAGES가 상한이며 도달 시 scraper가 warning을 남긴다).
+            scraper.image_link_list = list(owner_images)
             step3_elapsed = (time.perf_counter() - step3_start) * 1000
             logger.info(
                 f"[crawling] Step 3 SKIP - 업체 사진 {len(owner_images)}장 "
-                f"≥ {NvMapScraper.SUPPLEMENT_THRESHOLD}장 → 방문자 사진 필터링 생략"
+                f"≥ {NvMapScraper.SUPPLEMENT_THRESHOLD}장 → 방문자 사진 필터링 생략, 업체 사진만 사용"
             )
         else:
             try:
@@ -943,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/autotag.py

@@ -7,7 +7,7 @@ from app.utils.prompts.schemas import SpaceType, Subject, Camera, MotionRecommen
 import asyncio
 
 async def autotag_image(image_url : str, industry: str = "") -> list[str]: #tag_list
-    chatgpt = ChatgptService(model_type="gemini")
+    chatgpt = ChatgptService(model_type="gpt")
     image_input_data = {
         "img_url" : image_url,
         "industry" : industry,
@@ -21,7 +21,7 @@ async def autotag_image(image_url : str, industry: str = "") -> list[str]: #tag_
     return image_result
 
 async def autotag_images(image_url_list : list[str], industry: str = "") -> list[dict]: #tag_list
-    chatgpt = ChatgptService(model_type="gemini")
+    chatgpt = ChatgptService(model_type="gpt")
     image_input_data_list = [{
         "img_url" : image_url,
         "industry" : industry,

app/utils/creatomate.py

@@ -228,34 +228,98 @@ autotext_template_h_1 = {
 DVST0001 = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6"
 DVRT0001 = "3c4b74c4-82d7-4742-9b05-4b999fef4cbd"
 DVCF0001 = "ebb532ec-e49a-4a7e-b0e2-a8132f1bf1f7"
+DVAT0001 = "14d4fa22-49b8-49ca-a233-ac0556436f60"
+DVSL0001 = "20c526a7-b184-46f3-9138-9dfaff2fa342"
+DVCL0001 = "b57dca80-167d-430a-8905-30f4674ff72b"
+DVFT0001 = "03b6d521-4864-4e69-8e6c-1deab9f0ce6e"
+DVAC0001 = "4db8f5a3-4fd1-42cf-86b3-d997f0713d99"
+
 DHST0001 = "660be601-080a-43ea-bf0f-adcf4596fa98"
-DHST0002 = "3f194cc7-464e-4581-9db2-179d42d3e40f"
-DHST0003 = "f45df555-2956-4a13-9004-ead047070b3d"
-HST_LIST = [DHST0001,DHST0002,DHST0003]
-VST_LIST = [DVST0001,DVRT0001,DVCF0001]
+
+HST_LIST = [DHST0001]
+VST_LIST = [DVST0001,DVRT0001,DVCF0001,DVAT0001,DVSL0001,DVCL0001,DVFT0001,DVAC0001]
 
 # industry → 세로형 템플릿 매핑 (신규 세로형 템플릿 추가 시 여기에만 등록)
 VERTICAL_INDUSTRY_TEMPLATE_MAP: dict[str, str] = {
     "stay": DVST0001,
     "restaurant": DVRT0001,
     "cafe": DVCF0001,
+    "attraction": DVAT0001,
+    "salon": DVSL0001,
+    "clinic": DVCL0001,
+    "fitness": DVFT0001,
+    "academy": DVAC0001,
 }
 
-# 매핑에 없는 업종(salon, clinic 등)의 폴백 템플릿
+# 매핑에 없는 업종(general)의 폴백 템플릿
 DEFAULT_VERTICAL_TEMPLATE = DVST0001
 
+# 템플릿 슬롯명의 모션 동의어 → MotionRecommended 표준값 정규화 맵.
+# 동의어를 enum에 추가하면 LLM 태그 후보가 갈라져 매칭 정확도가 떨어지므로,
+# enum은 표준 어휘만 유지하고 슬롯명 파싱 경계에서 흡수한다.
+MOTION_TOKEN_NORMALIZATION: dict[str, MotionRecommended] = {
+    "zoom_in": MotionRecommended.slow_zoom_in,
+    "zoom_out": MotionRecommended.slow_zoom_out,
+    "pan_left": MotionRecommended.slow_pan,
+    "pan_right": MotionRecommended.slow_pan,
+    "tilt_up": MotionRecommended.slow_pan,
+    "tilt_down": MotionRecommended.slow_pan,
+    "push_diag": MotionRecommended.dolly,
+    "montage": MotionRecommended.static,
+}
+
 SCENE_TRACK = 1
 AUDIO_TRACK = 2
 SUBTITLE_TRACK = 3
 KEYWORD_TRACK = 4
 
-def select_template(orientation: OrientationType, industry: str | None = None) -> str:
+# 썸네일 컴포지션의 이미지 슬롯명 접미사 (8개 업종 템플릿 전수 확인, 예:
+# "exterior_front-architecture_detail-wide_angle-slow_zoom_in-intro-core-9999").
+# 일반 씬 이미지는 "-0001"~"-0030"류 4자리 순번을 쓰는 반면 썸네일만 "-9999"라
+# 슬롯명 접미사만으로 안전하게 식별 가능(가로 템플릿처럼 썸네일 컴포지션이
+# 없는 템플릿도 있음).
+THUMBNAIL_SLOT_MARKER = "-9999"
+
+
+def is_fixed_slot_name(name: str) -> bool:
+    """이름이 '-fixed'로 끝나는 요소인지 판별합니다.
+
+    템플릿 명명 규칙상 마지막 토큰이 'fixed'인 요소(예: 'brand-cta-bi_logo-
+    factual-fixed', 'brand-cta-contact_phone-factual-fixed')는 회사 연락처
+    문구·로고 등 매 영상마다 절대 바뀌면 안 되는 고정 콘텐츠다. 이런 요소는
+    이미지 배정/자막 생성 대상에서 제외하고 템플릿에 이미 설정된 값을 그대로
+    보존해야 한다.
+    """
+    if not name:
+        return False
+    return name.rsplit("-", 1)[-1] == "fixed"
+
+# 언어별 대체 폰트 (Google Fonts — Creatomate 기본 지원).
+# 템플릿 기본 폰트(GyeonggiTitleOTF/Pretendard)는 한글·라틴 전용이라 여기 있는
+# 언어는 렌더 직전 전체 텍스트 폰트를 교체한다. 매핑에 없는 언어(Korean,
+# English 등)는 템플릿 원본 폰트를 유지한다. 중국어는 간체(SC) 기준.
+LANGUAGE_FONT_MAP = {
+    "Japanese": "Noto Sans JP",
+    "Chinese": "Noto Sans SC",
+    "Thai": "Noto Sans Thai",
+    "Vietnamese": "Noto Sans",
+}
+
+def select_template(
+    orientation: OrientationType,
+    industry: str | None = None,
+    project_id: int | None = None,
+) -> str:
     """orientation과 industry에 따라 Creatomate 템플릿 ID를 선택합니다.
 
     Args:
         orientation: 영상 방향 ("horizontal" 또는 "vertical")
-        industry: 업종 분류 (stay|restaurant|cafe 등).
-                  세로형에서만 사용되며, 매핑에 없으면 기본 템플릿으로 폴백합니다.
+        industry: 업종 분류 (stay|restaurant|cafe|attraction 등).
+                  세로형에서 매핑에 있으면 전용 템플릿을 반환합니다.
+        project_id: 미매핑 업종(general 등)의 세로형 분배용 키.
+                    project_id % len(VST_LIST)로 결정적으로 선택하므로, 같은
+                    프로젝트는 몇 번 호출해도 동일 템플릿을 받아 사전/렌더 단계가
+                    일치한다. 없으면 DEFAULT_VERTICAL_TEMPLATE로 폴백한다.
 
     Returns:
         선택된 템플릿 ID
@@ -263,10 +327,17 @@ def select_template(orientation: OrientationType, industry: str | None = None) -
     if orientation == "horizontal":
         template_id = DHST0001
     elif orientation == "vertical":
-        template_id = VERTICAL_INDUSTRY_TEMPLATE_MAP.get(industry, DEFAULT_VERTICAL_TEMPLATE)
+        mapped = VERTICAL_INDUSTRY_TEMPLATE_MAP.get(industry)
+        if mapped is not None:
+            template_id = mapped
+        elif project_id is not None:
+            # 미매핑 업종: project_id 나머지로 VST_LIST 결정적 분배 (공유 상태 없음)
+            template_id = VST_LIST[project_id % len(VST_LIST)]
+        else:
+            template_id = DEFAULT_VERTICAL_TEMPLATE
     else:
         raise
-    logger.info(f"[select_template] orientation={orientation}, industry={industry}, template_id={template_id}")
+    logger.info(f"[select_template] orientation={orientation}, industry={industry}, project_id={project_id}, template_id={template_id}")
     return template_id
 
 async def get_shared_client() -> httpx.AsyncClient:
@@ -317,20 +388,23 @@ class CreatomateService:
         api_key: str | None = None,
         orientation: OrientationType = "vertical",
         industry: str | None = None,
+        project_id: int | None = None,
     ):
         """
         Args:
             api_key: Creatomate API 키 (Bearer token으로 사용)
                      None일 경우 config에서 자동으로 가져옴
             orientation: 영상 방향 ("horizontal" 또는 "vertical", 기본값: "vertical")
-            industry: 업종 분류 (stay|restaurant|cafe 등). 세로형 템플릿 선택에 사용되며,
-                      매핑에 없거나 None이면 기본 세로형 템플릿으로 폴백
+            industry: 업종 분류 (stay|restaurant|cafe|attraction 등). 세로형 템플릿 선택에 사용되며,
+                      매핑에 있으면 전용 템플릿을 사용
+            project_id: 미매핑 업종(general 등)의 세로형 분배용 키. 매핑에 없을 때
+                        project_id % len(VST_LIST)로 결정적 선택
         """
         self.api_key = api_key or apikey_settings.CREATOMATE_API_KEY
         self.orientation = orientation
 
         # orientation·industry에 따른 템플릿 설정 가져오기
-        self.template_id = select_template(orientation, industry=industry)
+        self.template_id = select_template(orientation, industry=industry, project_id=project_id)
         self.headers = {
             "Content-Type": "application/json",
             "Authorization": f"Bearer {self.api_key}",
@@ -466,22 +540,98 @@ class CreatomateService:
         template : dict,
         target_template_type : str
     ) -> list:
+        """target_template_type 요소 개수를 셉니다.
+
+        target_template_type이 "image"인 경우, 고정 자산('-fixed' 명명) 및
+        5토큰 명명 규칙을 따르지 않는 요소(고정 워터마크 등)는 콘텐츠 슬롯이
+        아니므로 카운트에서 제외합니다 (template_matching_taged_image의
+        image_slots 필터링과 동일 기준).
+        """
         source_elements = template["source"]["elements"]
         template_component_data = self.parse_template_component_name(source_elements)
         count = 0
 
-        for _, (_, template_type) in enumerate(template_component_data.items()):
-            if template_type == target_template_type:
-                count += 1
+        for name, template_type in template_component_data.items():
+            if template_type != target_template_type:
+                continue
+            if target_template_type == "image" and (is_fixed_slot_name(name) or self.parse_slot_name_to_tag(name) is None):
+                continue
+            count += 1
         return count
 
+    def _slot_scores_with_fitness(
+        self,
+        pool_subset: list[dict],
+        slot: str,
+        thumbnail_fitness_map: dict | None,
+    ) -> list[float]:
+        """슬롯 점수에 썸네일 픽셀 적합도를 결합합니다.
+
+        썸네일 슬롯(-9999)에 한해 태그 점수에 적합도를 반영한다:
+        - reject(저해상도/극단 가로형) → 0점 (하드 배제)
+        - 정상 → 태그 점수 × 적합도 배율(0.25~1.0)
+        - 적합도 데이터 없음(헤더 확보/파싱 실패) → 태그 점수 그대로(중립).
+          다운로드 실패를 배제로 오판해 풀 전체가 날아가는 것을 막기 위함.
+        일반 씬 슬롯은 태그 점수를 그대로 반환한다.
+        """
+        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 = []
+        for item, score in zip(pool_subset, scores):
+            fitness = thumbnail_fitness_map.get(item.get("image_url"))
+            if fitness is None:
+                adjusted.append(score)
+            elif fitness["reject"]:
+                adjusted.append(0.0)
+            else:
+                adjusted.append(score * fitness["score"])
+        return adjusted
+
+    def rank_thumbnail_candidates(
+        self,
+        template: dict,
+        taged_image_list: list,
+        thumbnail_fitness_map: dict | None = None,
+        top_n: int = 3,
+    ) -> dict[str, list[dict]]:
+        """썸네일 슬롯별로 상위 후보(태그×픽셀 점수 내림차순)를 반환합니다.
+
+        비전 LLM 최종 선택(Phase 2)에 넘길 후보 압축용. 읽기 전용 — pool을
+        변형하지 않으며 배정도 하지 않는다. 점수 0 이하(픽셀 reject 포함)는
+        제외한다.
+
+        Returns:
+            {슬롯명: [pool item, ...]} — 슬롯당 최대 top_n개, 점수 내림차순.
+            후보가 없는 슬롯은 키 자체를 포함하지 않는다.
+        """
+        component = self.parse_template_component_name(template["source"]["elements"])
+        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)
+            ranked = sorted(
+                range(len(scores)), key=lambda i: scores[i], reverse=True
+            )
+            candidates = [taged_image_list[i] for i in ranked if scores[i] > 0][:top_n]
+            if candidates:
+                result[slot] = candidates
+        return result
+
     def template_matching_taged_image(
         self,
         template: dict,
         taged_image_list: list,  # [{"image_url": str, "image_tag": dict}] — 이미 마케팅 적합성 필터를 통과한 이미지만 전달할 것
         music_url: str,
         address: str,
-        duplicate: bool = False
+        duplicate: bool = False,
+        thumbnail_fitness_map: dict | None = None,  # {image_url: {"score", "reject"}} — 썸네일 슬롯 픽셀 적합도
+        thumbnail_choice: dict | None = None,  # {슬롯명: image_url} — 비전 LLM 최종 선택(있으면 결정론 최고점 대신 사용)
     ) -> tuple[dict, dict]:
         """템플릿 슬롯에 이미지를 배정합니다.
 
@@ -515,9 +665,50 @@ class CreatomateService:
         assigned: dict = {}   # {슬롯명: image_tag} — 자막 컨텍스트용
 
         # 이미지 슬롯과 텍스트 슬롯 분리
-        image_slots = [name for name, t in template_component_data.items() if t == "image"]
+        # 고정 자산(이름이 '-fixed'로 끝나는 로고 등) 및 5토큰 명명 규칙을 따르지
+        # 않는 image 요소는 콘텐츠 슬롯이 아니므로 배정 대상에서 제외 — 그렇지
+        # 않으면 파싱 실패로 0점 처리되어 "가장 까다로운 슬롯"으로 취급되고
+        # 무작위 이미지로 덮어써진다.
+        image_slots = [
+            name for name, t in template_component_data.items()
+            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"]
 
+        # ── 썸네일 슬롯(-9999) 우선 배정 ─────────────────────────────────
+        # 썸네일은 영상에서 가장 노출이 큰 표면이므로 씬 슬롯과 다르게 취급한다:
+        # 1) 씬 슬롯이 pool에서 좋은 컷을 pop해 가기 전에 먼저 배정 (같은 태그
+        #    계열 씬 슬롯이 여러 개면 썸네일 차례에 적합 컷이 소진되는 문제 방지)
+        # 2) "상위 3장 가중 랜덤" 없이 결정론적 최고점 선택 — 랜덤 다양화가
+        #    저점수 컷(예: space_type 불일치 ×0.1 페널티 컷)을 확률적으로
+        #    썸네일에 앉히는 사고 방지
+        # duplicate(이미지 부족) 시에는 pop하지 않아 씬 커버리지를 깎지 않는다.
+        # thumbnail_choice(비전 LLM 최종 선택)가 해당 슬롯에 있으면 그 이미지를,
+        # 없으면(선택 실패/미제공) 결정론적 최고점 컷을 사용한다 — 폴백 안전.
+        thumbnail_choice = thumbnail_choice or {}
+        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}")
+                break
+            scores = self._slot_scores_with_fitness(pool, slot, thumbnail_fitness_map)
+            chosen_url = thumbnail_choice.get(slot)
+            chosen_idx = next(
+                (i for i, item in enumerate(pool) if item["image_url"] == chosen_url), None
+            ) if chosen_url else None
+            if chosen_idx is not None:
+                sel_idx, sel_source = chosen_idx, "vision"
+            else:
+                sel_idx, sel_source = max(range(len(scores)), key=lambda i: scores[i]), "결정론"
+            selected = pool[sel_idx] if duplicate else pool.pop(sel_idx)
+            modifications[slot] = selected["image_url"]
+            assigned[slot] = selected["image_tag"]
+            logger.info(
+                f"[template_matching_taged_image] 썸네일 배정({sel_source}) — slot: {slot}, "
+                f"score: {scores[sel_idx]:.3f}, url: {selected['image_url']}"
+            )
+
         if not pool:
             logger.warning("[template_matching_taged_image] 태그된 이미지가 없음 — 이미지 슬롯 배정 불가")
         elif duplicate:
@@ -528,7 +719,7 @@ class CreatomateService:
             def _best_slot_for_image(image: dict, slots: list) -> tuple[str | None, float]:
                 best_slot, best_score = None, -1.0
                 for slot in slots:
-                    score = self.calculate_image_slot_score_multi([image], slot)[0]
+                    score = self._slot_scores_with_fitness([image], slot, thumbnail_fitness_map)[0]
                     if score > best_score:
                         best_slot, best_score = slot, score
                 return best_slot, best_score
@@ -556,7 +747,7 @@ class CreatomateService:
 
             # Step 2: 커버리지 배정 후 남은 슬롯은 pool 전체에서 최고점 이미지 배정 (재사용 불가피)
             for slot in remaining_slots:
-                scores = self.calculate_image_slot_score_multi(pool, slot)
+                scores = self._slot_scores_with_fitness(pool, slot, thumbnail_fitness_map)
                 if not scores:
                     continue
                 best_idx = scores.index(max(scores))
@@ -566,7 +757,7 @@ class CreatomateService:
         else:
             # 이미지 충분(슬롯 수 <= 이미지 수): 난이도 우선 greedy 배정
             # Step 1: 정렬용 사전 점수 계산 (최고 달성 점수가 낮을수록 배정이 어려운 슬롯)
-            prelim = {slot: self.calculate_image_slot_score_multi(pool, slot) for slot in image_slots}
+            prelim = {slot: self._slot_scores_with_fitness(pool, slot, thumbnail_fitness_map) for slot in image_slots}
             ordered_slots = sorted(
                 image_slots,
                 key=lambda s: max(prelim[s]) if prelim[s] else 0.0
@@ -579,7 +770,7 @@ class CreatomateService:
                     logger.warning(f"[template_matching_taged_image] 이미지 풀 소진 — 남은 슬롯 배정 불가: {slot}")
                     break
                 # pop 후 인덱스 변동이 있으므로 현재 pool로 재계산 (사전 점수 재사용 금지)
-                scores = self.calculate_image_slot_score_multi(pool, slot)
+                scores = self._slot_scores_with_fitness(pool, slot, thumbnail_fitness_map)
                 if not scores:
                     continue
 
@@ -609,14 +800,20 @@ class CreatomateService:
     _NARR_BONUS = 1.5    # narrative 독립 보너스: base=0이어도 narrative 신호가 점수에 남음
     _NARR_MIN   = 0.0    # narrative_preference 값 clamp 하한 (무경계 스키마 방어)
     _NARR_MAX   = 100.0  # narrative_preference 값 clamp 상한. NarrativePreference 스키마(0.0~100.0)와 스케일 일치
+    _SPACE_TYPE_MISMATCH_PENALTY = 0.1  # 슬롯이 요구하는 space_type과 다른 이미지에 적용하는 페널티 배수 (하드 필터에 가깝게)
 
     def calculate_image_slot_score_multi(self, taged_image_list : list[dict], slot_name : str):
         """이미지 슬롯에 대한 각 이미지의 매칭 점수를 계산합니다.
 
-        점수 = base * narr_mod + NARR_BONUS * narr
-        - base: 태그 카테고리 매칭 합산 (최대 5.5)
+        점수 = (base * narr_mod + NARR_BONUS * narr) * space_type_penalty
+        - base: 태그 카테고리 매칭 합산 (최대 6.5 = space_type 2.0 + subject 3.0 + camera 1.0 + motion 0.5)
         - narr_mod: [NARR_FLOOR, 1.0] 구간으로 완화된 narrative 계수
         - NARR_BONUS * narr: narrative 독립 보너스 (base=0이어도 신호 유지)
+        - space_type_penalty: 슬롯이 요구하는 space_type과 이미지가 불일치하면
+          SPACE_TYPE_MISMATCH_PENALTY(0.1)를 곱해 사실상 탈락시킴. narrative_preference
+          점수가 아무리 높아도(예: bathroom 이미지가 intro narr=100) 슬롯이 지정한
+          space_type(예: exterior_front)과 다르면 우선 배정되지 않도록 하기 위함 —
+          같은 space_type 이미지가 pool에 하나도 없을 때만 상대적으로 선택됨(fallback 유지).
         이 구조로 "base=0, narrative가 높음"과 "완전히 틀린 이미지(0,0)"를 구별할 수 있음.
         """
         image_tag_list = [taged_image["image_tag"] for taged_image in taged_image_list]
@@ -627,6 +824,25 @@ class CreatomateService:
 
         base_score_list = [0.0] * len(image_tag_list)
         # slot_tag_narrative = NarrativePhase.accent  # 기본값 (narrative 토큰 없는 슬롯 대비)
+        slot_space_type = slot_tag_dict.get("space_type")
+        space_type_match_list = [
+            slot_space_type is not None and slot_space_type.value in image_tag.get("space_type", [])
+            for image_tag in image_tag_list
+        ]
+
+        # 썸네일 슬롯 한정: subject가 슬롯 요구와 일치하면 space_type 불일치
+        # 페널티를 면제한다. 배경: 일부 업종(예: 레스토랑 dining_hall-food_dish)은
+        # 태깅 관행상 슬롯이 요구하는 subject(음식 클로즈업)가 실제로는 다른
+        # space_type(table_setting/detail_plating 등)으로 붙는다 — space_type과
+        # subject가 한 사진에 공존하기 어려운 조합. 이런 슬롯은 space_type 하드
+        # 필터가 정작 원하는 subject 이미지를 전부 걸러내는 역효과를 낸다.
+        # 템플릿 슬롯명을 바꾸지 않고 여기서 흡수(썸네일 한정이라 일반 씬 슬롯의
+        # space_type 하드 필터는 그대로 유지).
+        slot_subject = slot_tag_dict.get("subject")
+        if slot_name.endswith(THUMBNAIL_SLOT_MARKER) and slot_subject is not None:
+            for idx, image_tag in enumerate(image_tag_list):
+                if slot_subject.value in image_tag.get("subject", []):
+                    space_type_match_list[idx] = True
 
         for slot_tag_cate, slot_tag_item in slot_tag_dict.items():
             if slot_tag_cate == "narrative_preference":
@@ -637,7 +853,7 @@ class CreatomateService:
                 case "space_type":
                     weight = 2.0
                 case "subject":
-                    weight = 2.0
+                    weight = 3.0
                 case "camera":
                     weight = 1.0
                 case "motion_recommended":
@@ -659,6 +875,8 @@ class CreatomateService:
             narr = clamped_narr / self._NARR_MAX  # 0.0~1.0로 정규화
             narr_mod = self._NARR_FLOOR + (1.0 - self._NARR_FLOOR) * narr
             score = base_score_list[idx] * narr_mod + self._NARR_BONUS * narr
+            if not space_type_match_list[idx]:
+                score *= self._SPACE_TYPE_MISMATCH_PENALTY
             image_score_list.append(score)
 
         return image_score_list
@@ -677,7 +895,8 @@ class CreatomateService:
             space_type = SpaceType(tag_list[0])
             subject = Subject(tag_list[1])
             camera = Camera(tag_list[2])
-            motion = MotionRecommended(tag_list[3])
+            # 모션 동의어(zoom_in, pan_left 등)는 표준 enum 값으로 정규화 후 변환
+            motion = MOTION_TOKEN_NORMALIZATION.get(tag_list[3]) or MotionRecommended(tag_list[3])
             narrative = NarrativePhase(tag_list[4])
             tag_dict = {
                 "space_type": space_type,
@@ -720,20 +939,26 @@ class CreatomateService:
         return modifications
 
     def modify_element(self, elements: list, modification: dict) -> list:
-        """elements의 source를 modification에 따라 수정합니다."""
+        """elements의 source를 modification에 따라 수정합니다.
+
+        modification에 없는 image/text 요소(예: '-fixed' 명명의 고정 로고·
+        연락처 문구처럼 배정/자막 생성 대상에서 제외된 고정 콘텐츠)는 템플릿에
+        이미 설정된 source/text를 그대로 보존한다 (KeyError·빈 문자열로
+        렌더가 깨지지 않도록).
+        """
 
         def recursive_modify(element: dict) -> None:
             if "name" in element:
                 match element["type"]:
                     case "image":
-                        element["source"] = modification[element["name"]]
+                        element["source"] = modification.get(element["name"], element.get("source"))
                     case "audio":
                         element["source"] = modification.get(element["name"], "")
                     case "video":
                         element["source"] = modification[element["name"]]
                     case "text":
                         #element["source"] = modification[element["name"]]
-                        element["text"] = modification.get(element["name"], "")
+                        element["text"] = modification.get(element["name"], element.get("text", ""))
                     case "composition":
                         for minor in element["elements"]:
                             recursive_modify(minor)
@@ -1071,12 +1296,38 @@ class CreatomateService:
             case "horizontal":
                 return autotext_template_h_1
 
+    def apply_language_font(self, template: dict, language: str) -> dict:
+        """출력 언어에 맞춰 템플릿 내 모든 텍스트 요소의 폰트를 교체합니다.
+
+        템플릿 기본 폰트(GyeonggiTitleOTF/Pretendard)는 한글·라틴 전용이라
+        CJK/태국어 글리프가 없다. LANGUAGE_FONT_MAP에 있는 언어는 전체 텍스트
+        폰트를 해당 언어 지원 폰트(Google Fonts)로 교체하고, 매핑에 없는
+        언어(Korean, English 등)는 템플릿 원본 폰트를 유지한다.
+        """
+        font_family = LANGUAGE_FONT_MAP.get(language)
+        if not font_family:
+            return template
+
+        def recursive_apply(element: dict) -> None:
+            if element.get("type") == "text":
+                element["font_family"] = font_family
+            for minor in element.get("elements") or []:
+                recursive_apply(minor)
+
+        for elem in template["source"]["elements"]:
+            recursive_apply(elem)
+
+        logger.info(
+            f"[apply_language_font] 텍스트 폰트 교체 완료 - language: {language}, font: {font_family}"
+        )
+        return template
+
     def extract_text_format_from_template(self, template:dict):
         keyword_list = []
         subtitle_list = []
         for elem in template["source"]["elements"]:
             try: #최상위 내 텍스트만 검사
-                if elem["type"] == "text":
+                if elem["type"] == "text" and not is_fixed_slot_name(elem["name"]):
                     if elem["track"] == SUBTITLE_TRACK:
                         subtitle_list.append(elem["name"])
                     elif elem["track"] == KEYWORD_TRACK:
@@ -1090,21 +1341,41 @@ class CreatomateService:
             assert(len(keyword_list)==len(subtitle_list))
         except Exception as E:
             logger.error("this template does not have same amount of keyword and subtitle.")
-        pitching_list = keyword_list + subtitle_list
+
+        # 썸네일 텍스트(thumb-*)는 전용 트랙이 아닌 thumbnail-composition 내부에
+        # 중첩되어 있어 위 최상위 트랙 스캔에 잡히지 않는다. 다국어 자막 생성에
+        # 함께 포함되도록 재귀 파싱으로 수집한다.
+        component_data = self.parse_template_component_name(
+            template["source"]["elements"]
+        )
+        thumbnail_list = [
+            name
+            for name, elem_type in component_data.items()
+            if elem_type == "text" and name.startswith("thumb-")
+        ]
+
+        pitching_list = keyword_list + subtitle_list + thumbnail_list
         return pitching_list
 
 
-    def make_thumbnail_modification(self, brand_name : str, region : str, brand_concept : str, category_definition : str, target_keywords : list[str]):
-        
+    def make_thumbnail_modification(self, brand_name : str, region : str, category_definition : str, target_keywords : list[str], detail_region_info : str = ""):
+
         len_keywords = len(target_keywords) if len(target_keywords) < 3 else 3
 
         hashtaged_target_keywords = [f"#{tk}" for tk in target_keywords[:len_keywords]]
-        
+
+        # 지역 표기: 상세주소의 공백 기준 앞 두 마디 사용
+        # (예: "전북 군산시 절골길 18" → "전북 군산시").
+        # 상세주소가 없으면 기존 region 정규화 표기로 폴백.
+        if detail_region_info and detail_region_info.strip():
+            region_display = " ".join(detail_region_info.split()[:2])
+        else:
+            region_display = normalize_location(region)
+
         mod_dict = {
             "thumb-hashtag-primary" : ' '.join(hashtaged_target_keywords),
-            "thumb-brand-wordmark" : brand_name,
-            "thumb-subheadline-selling_point" : f"{brand_name} · {normalize_location(region)}",
-            "thumb-headline-hook_claim-aspirational" : brand_concept,
+            "thumb-headline-brand_name-factual" : brand_name,
+            "thumb-subheadline-local_info-factual" : region_display,
             "thumb-badge-category" : category_definition,
         }
         return mod_dict

app/utils/image_filter.py

@@ -21,7 +21,7 @@ from app.utils.prompts.schemas import MarketingFilterOutput
 logger = get_logger("image_filter")
 
 # 크롤링 단계 필터링에 사용할 Gemini 모델. 시트(prompts.py)와 달리 코드에 고정한다.
-MARKETING_FILTER_MODEL = "gemini-3.5-flash"
+MARKETING_FILTER_MODEL = "gpt-5-mini"
 
 MAX_RETRY = 2
 
@@ -61,7 +61,7 @@ async def filter_marketing_images(image_url_list: list[str], industry: str = "")
     if not image_url_list:
         return []
 
-    chatgpt = ChatgptService(model_type="gemini")
+    chatgpt = ChatgptService(model_type="gpt")
     prompt_text = _build_prompt(industry)
 
     async def _call(url: str):
@@ -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/nvMapPwScraper.py

@@ -48,6 +48,30 @@ class NvMapPwScraper():
     ]
     _current_profile = None
 
+    # 헤드리스 자동화 탐지 신호(navigator.webdriver, 빈 plugins/languages, window.chrome 부재,
+    # permissions.query 이상 동작)를 정상 브라우저처럼 위장하는 스텔스 패치.
+    # create_page()와 fetch_graphql() 양쪽에서 생성하는 모든 page에 반드시 적용해야 한다 —
+    # 한쪽이라도 빠지면 그 경로만 헤드리스로 노출되어 안티봇 캡차 트리거 확률이 올라간다.
+    _STEALTH_INIT_SCRIPT = '''
+Object.defineProperty(Navigator.prototype, "webdriver", {
+    set: undefined,
+    enumerable: true,
+    configurable: true,
+    get: () => false,
+});
+Object.defineProperty(navigator, "plugins", { get: () => [1, 2, 3, 4, 5] });
+Object.defineProperty(navigator, "languages", { get: () => ["ko-KR", "ko"] });
+window.chrome = window.chrome || { runtime: {} };
+const originalQuery = window.navigator.permissions && window.navigator.permissions.query;
+if (originalQuery) {
+    window.navigator.permissions.query = (parameters) => (
+        parameters.name === "notifications"
+            ? Promise.resolve({ state: Notification.permission })
+            : originalQuery(parameters)
+    );
+}
+'''
+
     # instance var
     page = None
 
@@ -89,6 +113,17 @@ class NvMapPwScraper():
         if old_context:
             await old_context.close()
 
+    @classmethod
+    async def _new_stealth_page(cls):
+        """스텔스 패치(webdriver 위장 등)와 sec-ch-ua 헤더가 적용된 새 page를 생성한다."""
+        page = await cls._context.new_page()
+        await page.add_init_script(cls._STEALTH_INIT_SCRIPT)
+        if cls._current_profile:
+            await page.set_extra_http_headers({
+                'sec-ch-ua': cls._current_profile['sec_ch_ua']
+            })
+        return page
+
     GRAPHQL_URL = "https://pcmap-api.place.naver.com/graphql"
 
     @classmethod
@@ -116,7 +151,7 @@ class NvMapPwScraper():
             logger.warning("[NvMapPwScraper] fetch_graphql: scraper가 초기화되지 않았습니다")
             return None
 
-        page = await cls._context.new_page()
+        page = await cls._new_stealth_page()
         captured: dict = {}
 
         def on_request(req):
@@ -205,35 +240,7 @@ class NvMapPwScraper():
         await self.page.close()
         
     async def create_page(self):
-        self.page = await self._context.new_page()
-        await self.page.add_init_script(
-'''const defaultGetter = Object.getOwnPropertyDescriptor(
-    Navigator.prototype,
-    "webdriver"
-).get;
-defaultGetter.apply(navigator);
-defaultGetter.toString();
-Object.defineProperty(Navigator.prototype, "webdriver", {
-    set: undefined,
-    enumerable: true,
-    configurable: true,
-    get: new Proxy(defaultGetter, {
-    apply: (target, thisArg, args) => {
-        Reflect.apply(target, thisArg, args);
-        return false;
-    },
-    }),
-});
-const patchedGetter = Object.getOwnPropertyDescriptor(
-    Navigator.prototype,
-    "webdriver"
-).get;
-patchedGetter.apply(navigator);
-patchedGetter.toString();''')
-
-        await self.page.set_extra_http_headers({
-            'sec-ch-ua': self._current_profile['sec_ch_ua']
-        })
+        self.page = await self._new_stealth_page()
         await self.page.goto("http://google.com")
         
     async def goto_url(self, url, wait_until="domcontentloaded", timeout=20000):
@@ -415,23 +422,24 @@ patchedGetter.toString();''')
             return best['place_url']
 
         # 2순위(안전망): allSearch 캡처 실패 시 기존 iframe HTML 파싱 방식으로 폴백
-        logger.warning("[FALLBACK] allSearch 캡처 실패 → iframe 파싱 방식 시도")
-        iframe_candidates = []
-        for _ in range(3):  # 500ms × 3 = 최대 1.5초
-            if "/place/" in self.page.url:
-                return self.page.url
-            iframe_candidates = await self._extract_candidates_from_list_page()
-            if iframe_candidates:
-                break
-            await self.page.wait_for_timeout(500)
-
-        if iframe_candidates:
-            best = self._select_best_candidate(iframe_candidates, title, address)
-            logger.info(
-                f"[AUTO-SELECT-IFRAME] '{title}' → '{best['title']}' "
-                f"(name={best['_name_score']:.2f}, addr={best['_addr_score']:.2f}) {best['place_url']}"
-            )
-            return best['place_url']
+        # ── 잠시 비활성화 ──
+        # logger.warning("[FALLBACK] allSearch 캡처 실패 → iframe 파싱 방식 시도")
+        # iframe_candidates = []
+        # for _ in range(3):  # 500ms × 3 = 최대 1.5초
+        #     if "/place/" in self.page.url:
+        #         return self.page.url
+        #     iframe_candidates = await self._extract_candidates_from_list_page()
+        #     if iframe_candidates:
+        #         break
+        #     await self.page.wait_for_timeout(500)
+        #
+        # if iframe_candidates:
+        #     best = self._select_best_candidate(iframe_candidates, title, address)
+        #     logger.info(
+        #         f"[AUTO-SELECT-IFRAME] '{title}' → '{best['title']}' "
+        #         f"(name={best['_name_score']:.2f}, addr={best['_addr_score']:.2f}) {best['place_url']}"
+        #     )
+        #     return best['place_url']
 
         # isCorrectAnswer=true 로 강제 단일결과 재시도 (원본 로직 유지)
         correct_url = self.page.url.replace("?", "?isCorrectAnswer=true&")

app/utils/nvMapScraper.py

@@ -36,7 +36,12 @@ class NvMapScraper:
     REQUEST_TIMEOUT = 120  # 초
     data_source_identifier = "nv"
     SUPPLEMENT_THRESHOLD = 30  # 업체 사진이 이 수 미만일 때 방문자 사진으로 보충
-    MAX_IMAGES = 50             # 업체+방문자 사진 합산 최대 장수
+    # 방문자 사진 보충 시의 합산 상한 (필터링·정책상 제한).
+    # 업체 제공 사진은 필터링 면제 대상이므로 이 상한과 무관하게 수집분을 전부 사용한다
+    # (수집 자체는 BIZ_MAX_PAGES가 상한 — 초과 시 warning 로그 발생).
+    MAX_IMAGES = 50
+    BIZ_PAGE_SIZE = 20          # getPhotoViewerItems 'biz' 커서의 페이지당 사진 수
+    BIZ_MAX_PAGES = 5          # 업체 사진 수집 상한 (5×20=100장, 도달 시 warning)
     OVERVIEW_QUERY: str = """
 query getAccommodation($id: String!, $deviceType: String) {
     business: placeDetail(input: {id: $id, isNx: true, deviceType: $deviceType}) {
@@ -151,6 +156,42 @@ query getVisitorReviewStats($id: String!) {
             if p.get("originalUrl") and not NvMapScraper._is_gif_url(p["originalUrl"])
         ]
 
+    @classmethod
+    def _build_biz_payload(cls, place_id: str, page: int) -> dict:
+        """'biz' 커서(업체 등록 사진)의 page번째(0-base) 페이지 요청 payload를 만든다."""
+        start_index = page * cls.BIZ_PAGE_SIZE
+        cursor: dict = {"id": "biz"}
+        if start_index > 0:
+            cursor.update({
+                "startIndex": start_index,
+                "hasNext": True,
+                "lastCursor": str(start_index),
+            })
+        return {
+            "operationName": "getPhotoViewerItems",
+            "variables": {
+                "input": {
+                    "businessId": place_id,
+                    "cursors": [cursor],
+                    "dateRange": "",
+                    "excludeAuthorIds": [],
+                    "excludeClipIds": [],
+                    "excludeSection": [],
+                }
+            },
+            "query": cls.PHOTO_VIEWER_QUERY,
+        }
+
+    @staticmethod
+    def _raw_photo_count(photo_viewer_raw: dict | None) -> int:
+        """getPhotoViewerItems 응답의 사진 수(gif 필터링 전 원본 기준)를 반환한다.
+
+        페이지네이션 계속 여부 판단용 — gif가 걸러진 뒤의 수로 판단하면
+        마지막 페이지가 아닌데도 조기 종료할 수 있어 원본 개수를 사용한다.
+        """
+        photos = ((photo_viewer_raw or {}).get("data") or {}).get("photoViewer") or {}
+        return len(photos.get("photos") or [])
+
     @staticmethod
     def _dedup_by_original(images: list[dict]) -> list[dict]:
         """{"preview","original"} dict 리스트를 original URL 기준으로 순서를 유지한 채 중복 제거한다."""
@@ -215,7 +256,7 @@ query getVisitorReviewStats($id: String!) {
         #     self.scrap_type = "GraphQL-Browser"
 
         # ── 실제 브라우저로 WTM 캡차 우회 ──
-        data, stats_data, extra_photo_urls = 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)
         fac_data = await self._get_facility_string(place_id)
         self.scrap_type = "GraphQL-Browser"
@@ -226,8 +267,11 @@ query getVisitorReviewStats($id: String!) {
         self.rawdata["facilities"] = fac_data
         business = data["data"]["business"]
 
-        # 업체 등록 이미지 (마케팅 적합성 필터 제외 대상 — 자체 dedup)
-        self.owner_images = self._dedup_by_original(self._extract_origins(business.get("images") or {}))
+        # 업체 등록 이미지 (마케팅 적합성 필터 제외 대상 — 자체 dedup).
+        # placeDetail.images가 대표 사진 일부만 반환하는 경우가 있어 biz 커서 결과를 병합한다.
+        self.owner_images = self._dedup_by_original(
+            self._extract_origins(business.get("images") or {}) + biz_photo_urls
+        )
 
         # 방문자/AI View 보충 사진 (마케팅 적합성 필터 대상). owner에 이미 있는 원본은 제외.
         owner_originals = {img["original"] for img in self.owner_images}
@@ -238,16 +282,17 @@ query getVisitorReviewStats($id: String!) {
         # 업체 사진이 임계값 미만이면 내부/외부/리뷰 사진으로 보충
         # (필터링 없는 기본 조립. 마케팅 적합성 필터를 적용하려면 호출측에서
         #  owner_images / extra_photo_urls를 직접 사용해 image_filter.assemble_images로 재조립할 것.)
+        # MAX_IMAGES 상한은 방문자 사진이 섞이는 보충 경로에만 적용하며(필터링·정책상 제한),
+        # 업체 제공 사진만으로 구성되는 경우는 수집분(최대 BIZ_MAX_PAGES 페이지)을 전부 사용한다.
         if len(self.owner_images) < self.SUPPLEMENT_THRESHOLD:
             combined = self._dedup_by_original(self.owner_images + self.extra_photo_urls)
             logger.info(
                 f"[NvMapScraper] 업체 사진 {len(self.owner_images)}장 < {self.SUPPLEMENT_THRESHOLD}장 "
                 f"→ 보충 사진 {len(self.extra_photo_urls)}장 추가 (합산 {len(combined)}장, 상한 {self.MAX_IMAGES}장)"
             )
+            self.image_link_list = combined[: self.MAX_IMAGES]
         else:
-            combined = self.owner_images
-
-        self.image_link_list = combined[: self.MAX_IMAGES]
+            self.image_link_list = list(self.owner_images)
         self.base_info = data["data"]["business"]["base"]
         self.facility_info = fac_data
         self.voted_keyword_stats = stats_data
@@ -255,11 +300,11 @@ query getVisitorReviewStats($id: String!) {
 
         return
 
-    async def _scrap_via_browser(self, place_id: str) -> tuple[dict, list[dict] | None, list[str]]:
+    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)
+            (overview_data, review_stats_details, extra_photo_urls, biz_photo_urls)
 
         Raises:
             GraphQLException: 브라우저 폴백마저 실패한 경우
@@ -322,8 +367,19 @@ query getVisitorReviewStats($id: String!) {
             },
             "query": self.PHOTO_VIEWER_QUERY,
         }
+        # 업체 등록 사진. placeDetail.images는 대표 사진 일부(1~수 장)만 반환하는
+        # 경우가 있어, 사진 탭이 실제 사용하는 'biz' 커서로 별도 조회해 보강한다.
+        # biz 커서는 페이지당 BIZ_PAGE_SIZE(20)장이며 lastCursor가 단순 인덱스 문자열
+        # ("20", "40")이라 이전 응답 없이도 모든 페이지를 미리 만들 수 있고, 범위를
+        # 벗어난 페이지는 빈 목록을 반환하므로 블라인드 요청해도 안전하다.
+        # 따라서 상한(BIZ_MAX_PAGES)까지의 전 페이지를 첫 배치에 한꺼번에 실어 보낸다
+        # — 추가 왕복(페이지 재탐색 + WTM 토큰 재캡처)이 없고, 개별 페이지 실패가
+        # 이후 페이지 수집을 막지 못한다.
+        biz_payloads = [
+            self._build_biz_payload(place_id, page) for page in range(self.BIZ_MAX_PAGES)
+        ]
 
-        payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload]
+        payloads = [overview_payload, stats_payload, interior_payload, exterior_payload, review_payload, *biz_payloads]
         MAX_RETRY = 3
         results = None
         last_error: Exception | None = None
@@ -360,13 +416,31 @@ query getVisitorReviewStats($id: String!) {
         interior_urls = self._extract_photo_viewer_urls(results[2] if len(results) > 2 else None)
         exterior_urls = self._extract_photo_viewer_urls(results[3] if len(results) > 3 else None)
         review_urls   = self._extract_photo_viewer_urls(results[4] if len(results) > 4 else None)
+
+        biz_pages = results[5:]
+        biz_urls = [url for page_result in biz_pages for url in self._extract_photo_viewer_urls(page_result)]
+        # 개별 페이지 실패(None)는 해당 20장만 누락되고 나머지 페이지는 영향 없다 (best-effort).
+        failed_biz_pages = sum(1 for r in biz_pages if r is None)
+        if failed_biz_pages:
+            logger.warning(
+                f"[NvMapScraper] 업체 사진 {failed_biz_pages}개 페이지 응답 실패 "
+                f"— 페이지당 최대 {self.BIZ_PAGE_SIZE}장 누락 가능 (수집분으로 진행)"
+            )
+        # 마지막 페이지까지 가득 차 있으면 상한 밖에 사진이 더 있을 수 있다.
+        if biz_pages and self._raw_photo_count(biz_pages[-1]) >= self.BIZ_PAGE_SIZE:
+            logger.warning(
+                f"[NvMapScraper] 업체 사진이 수집 상한(BIZ_MAX_PAGES={self.BIZ_MAX_PAGES}, "
+                f"{self.BIZ_MAX_PAGES * self.BIZ_PAGE_SIZE}장)까지 가득 참 — 초과분은 수집되지 않음"
+            )
+
         extra_photo_urls = self._interleave(interior_urls, exterior_urls) + review_urls
         logger.info(
-            f"[NvMapScraper] 보충 이미지 - 내부:{len(interior_urls)} 외부:{len(exterior_urls)} 리뷰:{len(review_urls)}"
+            f"[NvMapScraper] 보충 이미지 - 내부:{len(interior_urls)} 외부:{len(exterior_urls)} "
+            f"리뷰:{len(review_urls)} / 업체(biz):{len(biz_urls)}"
         )
 
         logger.info(f"[NvMapScraper] 브라우저 폴백 SUCCESS - place_id: {place_id}")
-        return data, stats_data, extra_photo_urls
+        return data, stats_data, extra_photo_urls, biz_urls
 
     async def _call_get_accommodation(self, place_id: str) -> dict:
         """GraphQL API를 호출하여 숙소 정보를 가져옵니다.

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/image.py

@@ -42,6 +42,61 @@ class SpaceType(StrEnum):
     detail_lighting  = auto()
     detail_decor = auto()
     detail_tableware = auto()
+    # 레스토랑(DVRT0001) 템플릿 어휘
+    dining_hall = auto()
+    table_setting = auto()
+    private_room = auto()
+    detail_cooking = auto()
+    detail_plating = auto()
+    detail_menu = auto()
+    detail_dish_main = auto()
+    detail_dish_side = auto()
+    brand_summary = auto()
+    # 카페(DVCF0001) 템플릿 어휘
+    hall = auto()
+    counter_bar = auto()
+    seating_window = auto()
+    seating_lounge = auto()
+    signage = auto()
+    detail_dessert = auto()
+    detail_signature_drink = auto()
+    # 관광지(DVAT0001) 템플릿 어휘
+    entrance_gate = auto()
+    exhibition_hall = auto()
+    landmark_main = auto()
+    landmark_detail = auto()
+    night_view = auto()
+    panorama_view = auto()
+    photo_spot = auto()
+    seasonal_scene = auto()
+    walking_path = auto()
+    # 미용실(DVSL0001) 템플릿 어휘
+    mirror_detail = auto()
+    styling_zone = auto()
+    waiting_lounge = auto()
+    # 병원(DVCL0001) 템플릿 어휘
+    consultation_room = auto()
+    detail_amenity = auto()
+    equipment_zone = auto()
+    recovery_room = auto()
+    treatment_room = auto()
+    waiting_area = auto()
+    # 피트니스(DVFT0001) 템플릿 어휘
+    apparatus_zone = auto()
+    brand_sign = auto()
+    detail_equipment = auto()
+    locker_room = auto()
+    powder_room = auto()
+    pt_zone = auto()
+    reformer_zone = auto()
+    # 학원(DVAC0001) 템플릿 어휘
+    classroom = auto()
+    counseling_room = auto()
+    detail_facility = auto()
+    detail_interior_prop = auto()
+    detail_materials = auto()
+    library_corner = auto()
+    study_room = auto()
 
 class Subject(StrEnum):
     """이미지 내 주요 피사체 유형. 화면에 무엇이 담겨 있는지를 분류하는 태그."""
@@ -55,6 +110,28 @@ class Subject(StrEnum):
     signage = auto()
     amenity_item = auto()
     person = auto()
+    # 레스토랑(DVRT0001) 템플릿 어휘
+    cooking_action = auto()
+    menu_board = auto()
+    triptych = auto()
+    # 카페(DVCF0001) 템플릿 어휘
+    beverage = auto()
+    brewing_action = auto()
+    dessert = auto()
+    # 관광지(DVAT0001) 템플릿 어휘
+    scenery = auto()
+    structure = auto()
+    # 병원(DVCL0001) 템플릿 어휘
+    interior_clean = auto()
+    medical_equipment = auto()
+    # 피트니스(DVFT0001) 템플릿 어휘
+    interior_scale = auto()
+    pilates_apparatus = auto()
+    training_action = auto()
+    # 학원(DVAC0001) 템플릿 어휘
+    facility_equipment = auto()
+    learning_material = auto()
+    study_scene = auto()
 
 class Camera(StrEnum):
     """이미지의 촬영 기법·구도·조명 특성. 영상 편집 시 컷의 시각적 스타일을 구분하는 태그."""

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_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,7 +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
+from app.utils.creatomate import CreatomateService, LANGUAGE_FONT_MAP
 
 from app.comment.models import Comment
 from app.database.like_cache import (
@@ -192,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(
@@ -221,11 +222,6 @@ async def generate_video(
             category_definition = marketing_intelligence.intel_result["market_positioning"]["category_definition"]
             target_keywords = marketing_intelligence.intel_result["target_keywords"]
 
-            brand_concept = ""
-            for sp in marketing_intelligence.intel_result["selling_points"]:
-                if "concept" in sp["english_category"].lower():
-                    brand_concept = sp["description"]
-
             # Lyric 조회
             lyric_result = await session.execute(
                 select(Lyric)
@@ -387,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(
@@ -420,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):
@@ -445,6 +444,14 @@ async def generate_video(
                     )
                     final_template["source"]["elements"].append(caption)
         # END - LYRIC AUTO 결정부
+
+        # 언어별 폰트 교체: 템플릿 기본 폰트는 한글·라틴 전용이라 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)}"
         # )

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
@@ -72,7 +76,7 @@ async def generate_creative_assets_background(
                 )
                 marketing_intelligence = marketing_result.scalar_one_or_none()
 
-            creatomate_service = CreatomateService(orientation=orientation, industry=project.industry)
+            creatomate_service = CreatomateService(orientation=orientation, industry=project.industry, project_id=project.id)
             template = await creatomate_service.get_one_template_data(creatomate_service.template_id)
 
             store_address = project.detail_region_info
@@ -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"}