diff --git a/app/utils/creatomate.py b/app/utils/creatomate.py index 8ef2192..c0662f0 100644 --- a/app/utils/creatomate.py +++ b/app/utils/creatomate.py @@ -273,6 +273,13 @@ AUDIO_TRACK = 2 SUBTITLE_TRACK = 3 KEYWORD_TRACK = 4 +# 썸네일 컴포지션의 이미지 슬롯명 접미사 (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'로 끝나는 요소인지 판별합니다. @@ -552,13 +559,79 @@ class CreatomateService: 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]: """템플릿 슬롯에 이미지를 배정합니다. @@ -602,6 +675,40 @@ class CreatomateService: ] 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: @@ -612,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 @@ -640,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)) @@ -650,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 @@ -663,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 diff --git a/app/utils/image_filter.py b/app/utils/image_filter.py index 6e7dc10..a802750 100644 --- a/app/utils/image_filter.py +++ b/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( diff --git a/app/utils/prompts/chatgpt_prompt.py b/app/utils/prompts/chatgpt_prompt.py index 224b7a0..893a8f9 100644 --- a/app/utils/prompts/chatgpt_prompt.py +++ b/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, diff --git a/app/utils/thumbnail_fitness.py b/app/utils/thumbnail_fitness.py new file mode 100644 index 0000000..e10b2fd --- /dev/null +++ b/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(" 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} diff --git a/app/utils/thumbnail_vision.py b/app/utils/thumbnail_vision.py new file mode 100644 index 0000000..486f753 --- /dev/null +++ b/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] diff --git a/app/video/worker/creative_assets_task.py b/app/video/worker/creative_assets_task.py index b94729c..7be92b0 100644 --- a/app/video/worker/creative_assets_task.py +++ b/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"}