From 153484f2bde551294c2892195df7e13ca0bf1b0a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=84=B1=EA=B2=BD?= Date: Thu, 16 Jul 2026 13:37:13 +0900 Subject: [PATCH] =?UTF-8?q?=EC=8D=B8=EB=84=A4=EC=9D=BC=20=EC=83=9D?= =?UTF-8?q?=EC=84=B1=20=EA=B8=B0=EB=8A=A5=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/archive/api/routers/v1/archive.py | 1 + app/utils/creatomate.py | 2431 +++++++++-------- app/utils/prompts/schemas/image.py | 38 + app/utils/prompts/schemas/marketing.py | 1 + app/utils/thumbnail.py | 197 ++ app/video/api/routers/v1/video.py | 60 +- app/video/models.py | 6 + app/video/schemas/video_schema.py | 4 +- app/video/worker/creative_assets_task.py | 2 +- app/video/worker/thumbnail_task.py | 69 + ...ion_2026_07_16_add_video_thumbnail_url.sql | 11 + 11 files changed, 1688 insertions(+), 1132 deletions(-) create mode 100644 app/utils/thumbnail.py create mode 100644 app/video/worker/thumbnail_task.py create mode 100644 docs/database-schema/migration_2026_07_16_add_video_thumbnail_url.sql diff --git a/app/archive/api/routers/v1/archive.py b/app/archive/api/routers/v1/archive.py index 17f5817..8545a76 100644 --- a/app/archive/api/routers/v1/archive.py +++ b/app/archive/api/routers/v1/archive.py @@ -157,6 +157,7 @@ async def get_videos( region=project.region, task_id=video.task_id, result_movie_url=video.result_movie_url, + thumbnail_url=video.video_thumbnail_url, created_at=video.created_at, like_count=like_count_map.get(video.id) or 0, comment_count=comment_count or 0, diff --git a/app/utils/creatomate.py b/app/utils/creatomate.py index 7a1554c..940b47b 100644 --- a/app/utils/creatomate.py +++ b/app/utils/creatomate.py @@ -1,1126 +1,1305 @@ -""" -Creatomate API 클라이언트 모듈 - -API 문서: https://creatomate.com/docs/api - -## 사용법 -```python -from app.utils.creatomate import CreatomateService - -# config에서 자동으로 API 키를 가져옴 -creatomate = CreatomateService() - -# 또는 명시적으로 API 키 전달 -creatomate = CreatomateService(api_key="your_api_key") - -# 템플릿 목록 조회 (비동기) -templates = await creatomate.get_all_templates_data() - -# 특정 템플릿 조회 (비동기) -template = await creatomate.get_one_template_data(template_id) - -# 영상 렌더링 요청 (비동기) -response = await creatomate.make_creatomate_call(template_id, modifications) -``` - -## 성능 최적화 -- 템플릿 캐싱: 템플릿 데이터는 메모리에 캐싱되어 반복 조회 시 API 호출을 줄입니다. -- HTTP 클라이언트 재사용: 모듈 레벨의 공유 클라이언트로 커넥션 풀을 재사용합니다. -- 캐시 만료: 기본 5분 후 자동 만료 (CACHE_TTL_SECONDS로 조정 가능) -""" - -import copy -import random -import time -from enum import StrEnum -from typing import Literal -import traceback -import httpx - -from app.utils.logger import get_logger -from app.utils.prompts.schemas.image import SpaceType,Subject,Camera,MotionRecommended,NarrativePhase -from app.utils.common import normalize_location - -from config import apikey_settings, creatomate_settings, recovery_settings - -# 로거 설정 -logger = get_logger("creatomate") - - -class CreatomateResponseError(Exception): - """Creatomate API 응답 오류 시 발생하는 예외 - - Creatomate API 렌더링 실패 또는 비정상 응답 시 사용됩니다. - 재시도 로직에서 이 예외를 catch하여 재시도를 수행합니다. - - Attributes: - message: 에러 메시지 - original_response: 원본 API 응답 (있는 경우) - """ - - def __init__(self, message: str, original_response: dict | None = None): - self.message = message - self.original_response = original_response - super().__init__(self.message) - - -# Orientation 타입 정의 -OrientationType = Literal["horizontal", "vertical"] - -# ============================================================================= -# 모듈 레벨 캐시 및 HTTP 클라이언트 (싱글톤 패턴) -# ============================================================================= - -# 템플릿 캐시: {template_id: {"data": dict, "cached_at": float}} -_template_cache: dict[str, dict] = {} - -# 캐시 TTL (초) - 기본 5분 -CACHE_TTL_SECONDS = 300 - -# 모듈 레벨 공유 HTTP 클라이언트 (커넥션 풀 재사용) -_shared_client: httpx.AsyncClient | None = None - -text_template_v_1 = { - "type": "composition", - "track": 3, - "elements": [ - { - "type": "text", - "time": 0, - "y": "87.9086%", - "width": "100%", - "height": "40%", - "x_alignment": "50%", - "y_alignment": "50%", - "font_family": "Noto Sans", - "font_weight": "700", - "font_size": "8 vmin", - "background_color": "rgba(216,216,216,0)", - "background_x_padding": "33%", - "background_y_padding": "7%", - "background_border_radius": "28%", - "fill_color": "#ffffff", - "stroke_color": "rgba(51,51,51,1)", - "stroke_width": "0.6 vmin", - } - ] -} -text_template_v_2 = { - "type": "composition", - "track": 3, - "elements": [ - { - "type": "text", - "time": 0, - "x": "7.7233%", - "y": "82.9852%", - "width": "84.5534%", - "height": "5.7015%", - "x_anchor": "0%", - "y_anchor": "0%", - "x_alignment": "50%", - "y_alignment": "100%", - "font_family": "Noto Sans", - "font_weight": "700", - "font_size": "6.9999 vmin", - "fill_color": "#ffffff" - } - ] -} -text_template_v_3 = { - "type": "composition", - "track": 3, - "elements": [ - { - "type": "text", - "time": 0, - "x": "0%", - "y": "80%", - "width": "100%", - "height": "15%", - "x_anchor": "0%", - "y_anchor": "0%", - "x_alignment": "50%", - "y_alignment": "50%", - "font_family": "Noto Sans", - "font_weight": "700", - "font_size_maximum": "7 vmin", - "fill_color": "#ffffff", - "animations": [ - { - "time": 0, - "duration": 1, - "easing": "quadratic-out", - "type": "text-wave", - "split": "line", - "overlap": "50%" - } - ] - } - ] -} -text_template_h_1 = { - "type": "composition", - "track": 3, - "elements": [ - { - "type": "text", - "time": 0, - "x": "10%", - "y": "80%", - "width": "80%", - "height": "15%", - "x_anchor": "0%", - "y_anchor": "0%", - "x_alignment": "50%", - "font_family": "Noto Sans", - "font_weight": "700", - "font_size": "5.9998 vmin", - "fill_color": "#ffffff", - "stroke_color": "#333333", - "stroke_width": "0.2 vmin" - } - ] -} - -autotext_template_v_1 = { - "type": "text", - "track": 4, - "time": 0, - "y": "87.9086%", - "width": "100%", - "height": "40%", - "x_alignment": "50%", - "y_alignment": "50%", - "font_family": "Noto Sans", - "font_weight": "700", - "font_size": "8 vmin", - "background_color": "rgba(216,216,216,0)", - "background_x_padding": "33%", - "background_y_padding": "7%", - "background_border_radius": "28%", - "transcript_source": "audio-music", # audio-music과 연동됨 - "transcript_effect": "karaoke", - "fill_color": "#ffffff", - "stroke_color": "rgba(51,51,51,1)", - "stroke_width": "0.6 vmin" -} -autotext_template_h_1 = { - "type": "text", - "track": 4, - "time": 0, - "x": "10%", - "y": "83.2953%", - "width": "80%", - "height": "15%", - "x_anchor": "0%", - "y_anchor": "0%", - "x_alignment": "50%", - "font_family": "Noto Sans", - "font_weight": "700", - "font_size": "5.9998 vmin", - "transcript_source": "audio-music", - "transcript_effect": "karaoke", - "fill_color": "#ffffff", - "stroke_color": "#333333", - "stroke_width": "0.2 vmin" -} -DVST0001 = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6" -DVRT0001 = "3c4b74c4-82d7-4742-9b05-4b999fef4cbd" -DVCF0001 = "ebb532ec-e49a-4a7e-b0e2-a8132f1bf1f7" -DVAT0001 = "14d4fa22-49b8-49ca-a233-ac0556436f60" -DHST0001 = "660be601-080a-43ea-bf0f-adcf4596fa98" - -HST_LIST = [DHST0001] -VST_LIST = [DVST0001,DVRT0001,DVCF0001,DVAT0001] - -# industry → 세로형 템플릿 매핑 (신규 세로형 템플릿 추가 시 여기에만 등록) -VERTICAL_INDUSTRY_TEMPLATE_MAP: dict[str, str] = { - "stay": DVST0001, - "restaurant": DVRT0001, - "cafe": DVCF0001, - "attraction": DVAT0001, -} - -# 매핑에 없는 업종(salon, clinic 등)의 폴백 템플릿 -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: - """orientation과 industry에 따라 Creatomate 템플릿 ID를 선택합니다. - - Args: - orientation: 영상 방향 ("horizontal" 또는 "vertical") - industry: 업종 분류 (stay|restaurant|cafe|attraction 등). - 세로형에서만 사용되며, 매핑에 없으면 기본 템플릿으로 폴백합니다. - - Returns: - 선택된 템플릿 ID - """ - if orientation == "horizontal": - template_id = DHST0001 - elif orientation == "vertical": - template_id = VERTICAL_INDUSTRY_TEMPLATE_MAP.get(industry, DEFAULT_VERTICAL_TEMPLATE) - else: - raise - logger.info(f"[select_template] orientation={orientation}, industry={industry}, template_id={template_id}") - return template_id - -async def get_shared_client() -> httpx.AsyncClient: - """공유 HTTP 클라이언트를 반환합니다. 없으면 생성합니다.""" - global _shared_client - if _shared_client is None or _shared_client.is_closed: - _shared_client = httpx.AsyncClient( - timeout=httpx.Timeout( - recovery_settings.CREATOMATE_RENDER_TIMEOUT, - connect=recovery_settings.CREATOMATE_CONNECT_TIMEOUT, - ), - limits=httpx.Limits(max_keepalive_connections=10, max_connections=20), - ) - return _shared_client - - -async def close_shared_client() -> None: - """공유 HTTP 클라이언트를 닫습니다. 앱 종료 시 호출하세요.""" - global _shared_client - if _shared_client is not None and not _shared_client.is_closed: - await _shared_client.aclose() - _shared_client = None - logger.info("[CreatomateService] Shared HTTP client closed") - - -def clear_template_cache() -> None: - """템플릿 캐시를 전체 삭제합니다.""" - global _template_cache - _template_cache.clear() - logger.info("[CreatomateService] Template cache cleared") - - -def _is_cache_valid(cached_at: float) -> bool: - """캐시가 유효한지 확인합니다.""" - return (time.time() - cached_at) < CACHE_TTL_SECONDS - - -class CreatomateService: - """Creatomate API를 통한 영상 생성 서비스 - - 모든 HTTP 호출 메서드는 비동기(async)로 구현되어 있습니다. - """ - - BASE_URL = "https://api.creatomate.com" - - def __init__( - self, - api_key: str | None = None, - orientation: OrientationType = "vertical", - industry: str | None = None, - ): - """ - Args: - api_key: Creatomate API 키 (Bearer token으로 사용) - None일 경우 config에서 자동으로 가져옴 - orientation: 영상 방향 ("horizontal" 또는 "vertical", 기본값: "vertical") - industry: 업종 분류 (stay|restaurant|cafe|attraction 등). 세로형 템플릿 선택에 사용되며, - 매핑에 없거나 None이면 기본 세로형 템플릿으로 폴백 - """ - 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.headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {self.api_key}", - } - - async def _request( - self, - method: str, - url: str, - timeout: float | None = None, - **kwargs, - ) -> httpx.Response: - """HTTP 요청을 수행합니다. - - Args: - method: HTTP 메서드 ("GET", "POST", etc.) - url: 요청 URL - timeout: 요청 타임아웃 (초). None이면 기본값 사용 - **kwargs: httpx 요청에 전달할 추가 인자 - - Returns: - httpx.Response: 응답 객체 - - Raises: - httpx.HTTPError: 요청 실패 시 - """ - logger.info(f"[Creatomate] {method} {url}") - - # timeout이 None이면 기본 타임아웃 사용 - actual_timeout = timeout if timeout is not None else recovery_settings.CREATOMATE_DEFAULT_TIMEOUT - - client = await get_shared_client() - - if method.upper() == "GET": - response = await client.get( - url, headers=self.headers, timeout=actual_timeout, **kwargs - ) - elif method.upper() == "POST": - response = await client.post( - url, headers=self.headers, timeout=actual_timeout, **kwargs - ) - else: - raise ValueError(f"Unsupported HTTP method: {method}") - - logger.info(f"[Creatomate] Response - Status: {response.status_code}") - return response - - async def get_all_templates_data(self) -> dict: - """모든 템플릿 정보를 조회합니다.""" - url = f"{self.BASE_URL}/v1/templates" - response = await self._request("GET", url) # 기본 타임아웃 사용 - response.raise_for_status() - return response.json() - - async def get_one_template_data( - self, - template_id: str, - use_cache: bool = True, - ) -> dict: - """특정 템플릿 ID로 템플릿 정보를 조회합니다. - - Args: - template_id: 조회할 템플릿 ID - use_cache: 캐시 사용 여부 (기본: True) - - Returns: - 템플릿 데이터 (deep copy) - """ - global _template_cache - - # 캐시 확인 - if use_cache and template_id in _template_cache: - cached = _template_cache[template_id] - if _is_cache_valid(cached["cached_at"]): - logger.debug(f"[CreatomateService] Cache HIT - {template_id}") - return copy.deepcopy(cached["data"]) - else: - # 만료된 캐시 삭제 - del _template_cache[template_id] - logger.debug(f"[CreatomateService] Cache EXPIRED - {template_id}") - - # API 호출 - url = f"{self.BASE_URL}/v1/templates/{template_id}" - response = await self._request("GET", url) # 기본 타임아웃 사용 - response.raise_for_status() - data = response.json() - - # 캐시 저장 - _template_cache[template_id] = { - "data": data, - "cached_at": time.time(), - } - logger.debug(f"[CreatomateService] Cache MISS - {template_id} (cached)") - - return copy.deepcopy(data) - - def parse_template_component_name(self, template_source: list) -> dict: - """템플릿 정보를 파싱하여 리소스 이름을 추출합니다.""" - - def recursive_parse_component(element: dict) -> dict: - if "name" in element: - result_element_name_type = {element["name"]: element["type"]} - else: - result_element_name_type = {} - - if element["type"] == "composition": - minor_component_list = [ - recursive_parse_component(minor) for minor in element["elements"] - ] - # WARNING: Same name component should shroud other component - for minor_component in minor_component_list: - result_element_name_type.update(minor_component) - - return result_element_name_type - - result = {} - for result_element_dict in [ - recursive_parse_component(component) for component in template_source - ]: - result.update(result_element_dict) - - return result - - async def parse_template_name_tag(resource_name : str) -> list: - tag_list = [] - tag_list = resource_name.split("_") - - return tag_list - - - def counting_component( - self, - template : dict, - target_template_type : str - ) -> list: - 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 - return count - - 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 - ) -> tuple[dict, dict]: - """템플릿 슬롯에 이미지를 배정합니다. - - Returns: - (modifications, assigned) 튜플 - - modifications: {슬롯명: image_url, "audio-music": music_url, ...} — modify_element에 전달 - - assigned: {슬롯명: image_tag} — 자막 생성 컨텍스트에 전달 (modify_element에 넣지 않음) - - 배정 전략 (duplicate=False): - - 순수 greedy pop 대신 "난이도 우선 greedy" 적용 - - 사전에 전체 슬롯 점수를 계산해 최고 달성 점수가 낮은(=까다로운) 슬롯을 먼저 배정 - - 이미지 배정 직전 점수를 재계산 (pop 이후 인덱스 변동 대응) - - 배정 전략 (duplicate=True, 이미지 < 슬롯): - - 1단계(최소 커버리지 보장): 이미지별로 가장 잘 맞는 슬롯을 찾아 pool의 모든 이미지를 - 서로 다른 슬롯에 먼저 배정한다 (이미지 수 <= 슬롯 수이므로 전량 배정 가능). - 슬롯을 먼저 소진시키는 것이 아니라 "이미지"를 기준으로 난이도 우선 배정하여, - 특정 이미지 몇 장이 여러 슬롯을 독점하는 것을 방지한다. - - 2단계: 1단계에서 못 채운 나머지 슬롯은 기존처럼 pool 전체에서 최고점 이미지를 - 배정한다 (여기서부터는 중복 재사용 불가피). - - Note: - marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되므로 여기서는 재필터링하지 않는다. - taged_image_list를 그대로 pool로 사용한다. - """ - source_elements = template["source"]["elements"] - template_component_data = self.parse_template_component_name(source_elements) - - pool = taged_image_list - modifications: dict = {} - assigned: dict = {} # {슬롯명: image_tag} — 자막 컨텍스트용 - - # 이미지 슬롯과 텍스트 슬롯 분리 - image_slots = [name for name, t in template_component_data.items() if t == "image"] - text_slots = [(name, t) for name, t in template_component_data.items() if t == "text"] - - if not pool: - logger.warning("[template_matching_taged_image] 태그된 이미지가 없음 — 이미지 슬롯 배정 불가") - elif duplicate: - # 이미지 부족(슬롯 수 > 이미지 수) - # Step 1: 최소 커버리지 보장 — 이미지마다 가장 잘 맞는 슬롯을 찾아 서로 다른 슬롯에 배정 - remaining_slots = list(image_slots) - - 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] - if score > best_score: - best_slot, best_score = slot, score - return best_slot, best_score - - # 사전 점수 계산 (최고 달성 점수가 낮을수록 배정처가 마땅찮은=까다로운 이미지) - prelim_best_score = [ - _best_slot_for_image(image, remaining_slots)[1] for image in pool - ] - ordered_indices = sorted( - range(len(pool)), - key=lambda i: prelim_best_score[i] - ) - - for idx in ordered_indices: - if not remaining_slots: - break - image = pool[idx] - # 슬롯이 이전 배정으로 줄어들었으므로 현재 remaining_slots 기준 재계산 - best_slot, _ = _best_slot_for_image(image, remaining_slots) - if best_slot is None: - continue - modifications[best_slot] = image["image_url"] - assigned[best_slot] = image["image_tag"] - remaining_slots.remove(best_slot) - - # Step 2: 커버리지 배정 후 남은 슬롯은 pool 전체에서 최고점 이미지 배정 (재사용 불가피) - for slot in remaining_slots: - scores = self.calculate_image_slot_score_multi(pool, slot) - if not scores: - continue - best_idx = scores.index(max(scores)) - selected = pool[best_idx] - modifications[slot] = selected["image_url"] - assigned[slot] = selected["image_tag"] - else: - # 이미지 충분(슬롯 수 <= 이미지 수): 난이도 우선 greedy 배정 - # Step 1: 정렬용 사전 점수 계산 (최고 달성 점수가 낮을수록 배정이 어려운 슬롯) - prelim = {slot: self.calculate_image_slot_score_multi(pool, slot) for slot in image_slots} - ordered_slots = sorted( - image_slots, - key=lambda s: max(prelim[s]) if prelim[s] else 0.0 - # 오름차순 → 달성 최대값이 낮은(어려운) 슬롯이 앞으로 - ) - - # Step 2: 난이도 순서로 배정, 배정마다 pool에서 pop + 다음 슬롯은 점수 재계산 - for slot in ordered_slots: - if not pool: - logger.warning(f"[template_matching_taged_image] 이미지 풀 소진 — 남은 슬롯 배정 불가: {slot}") - break - # pop 후 인덱스 변동이 있으므로 현재 pool로 재계산 (사전 점수 재사용 금지) - scores = self.calculate_image_slot_score_multi(pool, slot) - if not scores: - continue - - # 상위 3장 중 점수 비례 확률로 선택 (영상 다양화) - sorted_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True) - top_indices = sorted_indices[:min(3, len(sorted_indices))] - weights = [scores[i] for i in top_indices] - if sum(weights) > 0: - chosen_pool_idx = random.choices(top_indices, weights=weights, k=1)[0] - else: - chosen_pool_idx = random.choice(top_indices) - - selected = pool.pop(chosen_pool_idx) - modifications[slot] = selected["image_url"] - assigned[slot] = selected["image_tag"] - - # 텍스트 슬롯 처리 - for name, _ in text_slots: - if "address_input" in name: - modifications[name] = address - - modifications["audio-music"] = music_url - return modifications, assigned - - # 점수 계산 상수 (모듈 레벨로 빼면 더 좋지만 기존 코드 스타일 따라 클래스 내부에 위치) - _NARR_FLOOR = 0.25 # narrative 최솟값 modulator: base 점수가 완전히 0이 되는 것을 방지 - _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)와 스케일 일치 - - def calculate_image_slot_score_multi(self, taged_image_list : list[dict], slot_name : str): - """이미지 슬롯에 대한 각 이미지의 매칭 점수를 계산합니다. - - 점수 = base * narr_mod + NARR_BONUS * narr - - base: 태그 카테고리 매칭 합산 (최대 5.5) - - narr_mod: [NARR_FLOOR, 1.0] 구간으로 완화된 narrative 계수 - - NARR_BONUS * narr: narrative 독립 보너스 (base=0이어도 신호 유지) - 이 구조로 "base=0, narrative가 높음"과 "완전히 틀린 이미지(0,0)"를 구별할 수 있음. - """ - image_tag_list = [taged_image["image_tag"] for taged_image in taged_image_list] - slot_tag_dict = self.parse_slot_name_to_tag(slot_name) - if slot_tag_dict is None: - # parse 실패 시 전부 0점 반환 (슬롯 skip) - return [0.0] * len(image_tag_list) - - base_score_list = [0.0] * len(image_tag_list) - # slot_tag_narrative = NarrativePhase.accent # 기본값 (narrative 토큰 없는 슬롯 대비) - - for slot_tag_cate, slot_tag_item in slot_tag_dict.items(): - if slot_tag_cate == "narrative_preference": - slot_tag_narrative = slot_tag_item - continue - - match slot_tag_cate: - case "space_type": - weight = 2.0 - case "subject": - weight = 2.0 - case "camera": - weight = 1.0 - case "motion_recommended": - weight = 0.5 - case _: - raise ValueError(f"[calculate_image_slot_score_multi] unknown slot tag category: {slot_tag_cate}") - - for idx, image_tag in enumerate(image_tag_list): - if slot_tag_item.value in image_tag[slot_tag_cate]: # collect! - base_score_list[idx] += weight - - # narrative 점수 적용: 순수 곱셈 대신 "완화 modulator + 독립 보너스" - # - narr_mod: base를 흐릴 수 있지만 NARR_FLOOR 이하로는 안 떨어짐 - # - NARR_BONUS * narr: base=0이어도 narrative가 높으면 양수 점수 확보 - image_score_list = [] - for idx, image_tag in enumerate(image_tag_list): - raw_narr = image_tag.get("narrative_preference", {}).get(slot_tag_narrative, 0.0) - clamped_narr = min(self._NARR_MAX, max(self._NARR_MIN, float(raw_narr))) - 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 - image_score_list.append(score) - - return image_score_list - - def parse_slot_name_to_tag(self, slot_name: str) -> dict[str, StrEnum] | None: - """슬롯 이름을 파싱하여 태그 딕셔너리를 반환합니다. - - 슬롯 이름 형식: {space_type}-{subject}-{camera}-{motion}-{narrative} - 파싱 실패 시 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리). - """ - try: - tag_list = slot_name.split("-") - if len(tag_list) < 5: - logger.warning(f"[parse_slot_name_to_tag] 슬롯명 토큰 부족 ({len(tag_list)}개): '{slot_name}' — 슬롯 skip") - return None - space_type = SpaceType(tag_list[0]) - subject = Subject(tag_list[1]) - camera = Camera(tag_list[2]) - # 모션 동의어(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, - "subject": subject, - "camera": camera, - "motion_recommended": motion, - "narrative_preference": narrative, - } - return tag_dict - except (ValueError, IndexError) as e: - logger.warning(f"[parse_slot_name_to_tag] 슬롯명 파싱 실패: '{slot_name}' — {e} — 슬롯 skip") - return None - - def elements_connect_resource_blackbox( - self, - elements: list, - image_url_list: list[str], - music_url: str, - address: str = None - ) -> dict: - """elements 정보와 이미지/가사/음악 리소스를 매핑합니다.""" - template_component_data = self.parse_template_component_name(elements) - - modifications = {} - - for idx, (template_component_name, template_type) in enumerate( - template_component_data.items() - ): - match template_type: - case "image": - modifications[template_component_name] = image_url_list[ - idx % len(image_url_list) - ] - case "text": - if "address_input" in template_component_name: - modifications[template_component_name] = address - - modifications["audio-music"] = music_url - - return modifications - - def modify_element(self, elements: list, modification: dict) -> list: - """elements의 source를 modification에 따라 수정합니다.""" - - def recursive_modify(element: dict) -> None: - if "name" in element: - match element["type"]: - case "image": - element["source"] = modification[element["name"]] - 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"], "") - case "composition": - for minor in element["elements"]: - recursive_modify(minor) - - for minor in elements: - recursive_modify(minor) - - return elements - - async def make_creatomate_call( - self, template_id: str, modifications: dict - ) -> dict: - """Creatomate에 렌더링 요청을 보냅니다 (재시도 로직 포함). - - Args: - template_id: Creatomate 템플릿 ID - modifications: 수정사항 딕셔너리 - - Returns: - Creatomate API 응답 데이터 - - Raises: - CreatomateResponseError: API 오류 또는 재시도 실패 시 - - Note: - response에 요청 정보가 있으니 폴링 필요 - """ - url = f"{self.BASE_URL}/v2/renders" - payload = { - "template_id": template_id, - "modifications": modifications, - } - - last_error: Exception | None = None - - for attempt in range(recovery_settings.CREATOMATE_MAX_RETRIES + 1): - try: - response = await self._request( - "POST", - url, - timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT, - json=payload, - ) - - # 200 OK, 201 Created, 202 Accepted 모두 성공으로 처리 - if response.status_code in (200, 201, 202): - return response.json() - - # 재시도 불가능한 오류 (4xx 클라이언트 오류) - if 400 <= response.status_code < 500: - raise CreatomateResponseError( - f"Client error: {response.status_code}", - original_response={"status": response.status_code, "text": response.text}, - ) - - # 재시도 가능한 오류 (5xx 서버 오류) - last_error = CreatomateResponseError( - f"Server error: {response.status_code}", - original_response={"status": response.status_code, "text": response.text}, - ) - - except httpx.TimeoutException as e: - logger.warning( - f"[Creatomate] Timeout on attempt {attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES + 1}" - ) - last_error = e - - except httpx.HTTPError as e: - logger.warning(f"[Creatomate] HTTP error on attempt {attempt + 1}: {e}") - last_error = e - - except CreatomateResponseError: - raise # CreatomateResponseError는 재시도하지 않고 즉시 전파 - - # 마지막 시도가 아니면 재시도 - if attempt < recovery_settings.CREATOMATE_MAX_RETRIES: - logger.info( - f"[Creatomate] Retrying... ({attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES})" - ) - - # 모든 재시도 실패 - raise CreatomateResponseError( - f"All {recovery_settings.CREATOMATE_MAX_RETRIES + 1} attempts failed", - original_response={"last_error": str(last_error)}, - ) - - async def make_creatomate_custom_call(self, source: dict) -> dict: - """템플릿 없이 Creatomate에 커스텀 렌더링 요청을 보냅니다 (재시도 로직 포함). - - Args: - source: 렌더링 소스 딕셔너리 - - Returns: - Creatomate API 응답 데이터 - - Raises: - CreatomateResponseError: API 오류 또는 재시도 실패 시 - - Note: - response에 요청 정보가 있으니 폴링 필요 - """ - url = f"{self.BASE_URL}/v2/renders" - - source["frame_rate"] = 30 - - last_error: Exception | None = None - - for attempt in range(recovery_settings.CREATOMATE_MAX_RETRIES + 1): - try: - response = await self._request( - "POST", - url, - timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT, - json=source, - ) - - # 200 OK, 201 Created, 202 Accepted 모두 성공으로 처리 - if response.status_code in (200, 201, 202): - return response.json() - - # 재시도 불가능한 오류 (4xx 클라이언트 오류) - if 400 <= response.status_code < 500: - raise CreatomateResponseError( - f"Client error: {response.status_code}", - original_response={"status": response.status_code, "text": response.text}, - ) - - # 재시도 가능한 오류 (5xx 서버 오류) - last_error = CreatomateResponseError( - f"Server error: {response.status_code}", - original_response={"status": response.status_code, "text": response.text}, - ) - - except httpx.TimeoutException as e: - logger.warning( - f"[Creatomate] Timeout on attempt {attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES + 1}" - ) - last_error = e - - except httpx.HTTPError as e: - logger.warning(f"[Creatomate] HTTP error on attempt {attempt + 1}: {e}") - last_error = e - - except CreatomateResponseError: - raise # CreatomateResponseError는 재시도하지 않고 즉시 전파 - - # 마지막 시도가 아니면 재시도 - if attempt < recovery_settings.CREATOMATE_MAX_RETRIES: - logger.info( - f"[Creatomate] Retrying... ({attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES})" - ) - - # 모든 재시도 실패 - raise CreatomateResponseError( - f"All {recovery_settings.CREATOMATE_MAX_RETRIES + 1} attempts failed", - original_response={"last_error": str(last_error)}, - ) - - async def get_render_status(self, render_id: str) -> dict: - """렌더링 작업의 상태를 조회합니다. - - Args: - render_id: Creatomate 렌더 ID - - Returns: - 렌더링 상태 정보 - - Note: - 상태 값: - - planned: 예약됨 - - waiting: 대기 중 - - transcribing: 트랜스크립션 중 - - rendering: 렌더링 중 - - succeeded: 성공 - - failed: 실패 - """ - url = f"{self.BASE_URL}/v1/renders/{render_id}" - response = await self._request("GET", url) # 기본 타임아웃 사용 - response.raise_for_status() - return response.json() - - @staticmethod - def _entrance_transition_offset(elem: dict) -> float: - """요소의 entrance transition(transition=true, time==0) duration 합을 반환합니다. - - 순차 배치 시 이 값만큼 이전 요소와 겹치도록 시작점이 앞당겨집니다. - """ - offset = 0.0 - for animation in elem.get("animations", []): - if "reversed" in animation: - continue - if "transition" in animation and animation["transition"] and animation.get("time", 0) == 0: - offset += animation["duration"] - return offset - - def calc_scene_duration(self, template: dict) -> float: - """템플릿의 전체 장면 duration을 계산합니다.""" - total_template_duration = 0.0 - track_maximum_duration = { - SCENE_TRACK : 0, - SUBTITLE_TRACK : 0, - KEYWORD_TRACK : 0 - } - for elem in template["source"]["elements"]: - try: - if elem["track"] not in track_maximum_duration: - continue - if "time" not in elem or elem["time"] == 0: # elem is auto / 만약 마지막 elem이 auto인데 그 앞에 time이 있는 elem 일 시 버그 발생 확률 있음 - track_maximum_duration[elem["track"]] += elem["duration"] - self._entrance_transition_offset(elem) - else: - track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"]) - - except Exception as e: - logger.debug(traceback.format_exc()) - logger.error(f"[calc_scene_duration] Error processing element: {elem}, {e}") - - total_template_duration = max(track_maximum_duration.values()) - - return total_template_duration - - def calc_track_time_ranges(self, template: dict, track: int) -> list[tuple[str, float, float]]: - """최상위 요소 중 지정 track의 (name, start, end) 시간 범위를 계산합니다. - - Creatomate 규칙에 따라 `time`이 없거나 0인 요소는 같은 트랙의 이전 요소 뒤에 - 순차 배치되며, entrance transition duration만큼 이전 요소와 겹치도록 - 시작점을 앞당깁니다(calc_scene_duration의 차감 규칙과 동치). - `time`이 명시된 요소는 해당 시각에서 시작하고, 커서는 그 요소의 끝으로 이동합니다. - """ - ranges: list[tuple[str, float, float]] = [] - cursor = 0.0 - for elem in template["source"]["elements"]: - try: - if elem.get("track") != track: - continue - duration = elem["duration"] - - if "time" not in elem or elem["time"] == 0: # 순차 배치(auto) - start = max(0.0, cursor - self._entrance_transition_offset(elem)) - else: - start = elem["time"] - - end = start + duration - cursor = end - ranges.append((elem.get("name", ""), start, end)) - except Exception as e: - logger.debug(traceback.format_exc()) - logger.error(f"[calc_track_time_ranges] Error processing element: {elem}, {e}") - - return ranges - - @staticmethod - def _scale_element(elem: dict, extend_rate: float) -> None: - """element 하나의 time/duration/animations를 extend_rate만큼 스케일링합니다. - - composition의 자식 element까지 재귀적으로 처리합니다. 재귀하지 않으면 - 부모 컴포지션(씬 슬롯)만 늘어나고 내부 leaf 이미지/애니메이션은 원래 - 길이에서 끝나버려, 슬롯이 끝나기 전 컴포지션의 fill_color(회색)가 - 노출되는 버그가 생긴다 — modify_element/parse_template_component_name과 - 동일하게 composition을 재귀 순회해야 함. - - 오디오 판별은 track 번호가 아닌 type == "audio"로 한다. composition - 내부의 track 번호는 그 컴포지션 안에서만 유효한 로컬 레이어 인덱스라 - 전역 AUDIO_TRACK/KEYWORD_TRACK 상수와 우연히 값이 겹칠 수 있다 - (실제 템플릿에도 다중 레이어 씬의 2번째/3번째 이미지가 track=2, track=4인 - 경우가 존재함). - """ - if elem.get("type") == "audio": - return - - # "time"은 숫자 오프셋 대신 "end"(컴포지션 끝에 고정) 같은 특수 키워드일 수 있음 — - # 이런 값은 렌더 시점에 (이미 스케일링된) duration 기준으로 재계산되므로 그대로 둔다. - if "time" in elem and isinstance(elem["time"], (int, float)): - elem["time"] = elem["time"] * extend_rate - if "duration" in elem and isinstance(elem["duration"], (int, float)): - elem["duration"] = elem["duration"] * extend_rate - - for animation in elem.get("animations", []): - anim_time = animation.get("time", 0) - if isinstance(anim_time, (int, float)) and anim_time != 0: - animation["time"] = anim_time * extend_rate - if isinstance(animation.get("duration"), (int, float)): - animation["duration"] = animation["duration"] * extend_rate - - if elem.get("type") == "composition": - for child in elem.get("elements", []): - CreatomateService._scale_element(child, extend_rate) - - def extend_template_duration(self, template: dict, target_duration: float) -> dict: - """템플릿의 duration을 target_duration으로 확장합니다.""" - # template["duration"] = target_duration + 0.5 # 늘린것보단 짧게 - # target_duration += 1 # 수동으로 직접 변경 및 테스트 필요 : 파란박스 생기는것 - total_template_duration = self.calc_scene_duration(template) - extend_rate = target_duration / total_template_duration - new_template = copy.deepcopy(template) - - for elem in new_template["source"]["elements"]: - try: - if elem["track"] == AUDIO_TRACK : # audio track(최상위 음악 트랙)은 패스 - continue - - self._scale_element(elem, extend_rate) - except Exception as e: - logger.error( - f"[extend_template_duration] Error processing element: {elem}, {e}" - ) - - return new_template - - def lining_lyric(self, text_template: dict, lyric_index: int, lyric_text: str, start_sec: float, end_sec: float, font_family: str = "Noto Sans") -> dict: - duration = end_sec - start_sec - text_scene = copy.deepcopy(text_template) - text_scene["name"] = f"Caption-{lyric_index}" - text_scene["duration"] = duration - text_scene["time"] = start_sec - text_scene["elements"][0]["name"] = f"lyric-{lyric_index}" - text_scene["elements"][0]["text"] = lyric_text - text_scene["elements"][0]["font_family"] = font_family - return text_scene - - def auto_lyric(self, auto_text_template : dict): - text_scene = copy.deepcopy(auto_text_template) - return text_scene - - def get_text_template(self): - match self.orientation: - case "vertical": - return text_template_v_3 - case "horizontal": - return text_template_h_1 - - def get_auto_text_template(self): - match self.orientation: - case "vertical": - return autotext_template_v_1 - case "horizontal": - return autotext_template_h_1 - - 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["track"] == SUBTITLE_TRACK: - subtitle_list.append(elem["name"]) - elif elem["track"] == KEYWORD_TRACK: - keyword_list.append(elem["name"]) - except Exception as e: - logger.error( - f"[extend_template_duration] Error processing element: {elem}, {e}" - ) - - try: - 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 - return pitching_list - - - def make_thumbnail_modification(self, brand_name : str, region : str, brand_concept : str, category_definition : str, target_keywords : list[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]] - - 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-badge-category" : category_definition, - } - return mod_dict +""" +Creatomate API 클라이언트 모듈 + +API 문서: https://creatomate.com/docs/api + +## 사용법 +```python +from app.utils.creatomate import CreatomateService + +# config에서 자동으로 API 키를 가져옴 +creatomate = CreatomateService() + +# 또는 명시적으로 API 키 전달 +creatomate = CreatomateService(api_key="your_api_key") + +# 템플릿 목록 조회 (비동기) +templates = await creatomate.get_all_templates_data() + +# 특정 템플릿 조회 (비동기) +template = await creatomate.get_one_template_data(template_id) + +# 영상 렌더링 요청 (비동기) +response = await creatomate.make_creatomate_call(template_id, modifications) +``` + +## 성능 최적화 +- 템플릿 캐싱: 템플릿 데이터는 메모리에 캐싱되어 반복 조회 시 API 호출을 줄입니다. +- HTTP 클라이언트 재사용: 모듈 레벨의 공유 클라이언트로 커넥션 풀을 재사용합니다. +- 캐시 만료: 기본 5분 후 자동 만료 (CACHE_TTL_SECONDS로 조정 가능) +""" + +import asyncio +import copy +import random +import time +from enum import StrEnum +from typing import Literal +import traceback +import httpx + +from app.utils.logger import get_logger +from app.utils.prompts.schemas.image import SpaceType,Subject,Camera,MotionRecommended,NarrativePhase +from app.utils.common import normalize_location +from app.utils.thumbnail import build_thumbnail_source + +from config import apikey_settings, creatomate_settings, recovery_settings + +# 로거 설정 +logger = get_logger("creatomate") + + +class CreatomateResponseError(Exception): + """Creatomate API 응답 오류 시 발생하는 예외 + + Creatomate API 렌더링 실패 또는 비정상 응답 시 사용됩니다. + 재시도 로직에서 이 예외를 catch하여 재시도를 수행합니다. + + Attributes: + message: 에러 메시지 + original_response: 원본 API 응답 (있는 경우) + """ + + def __init__(self, message: str, original_response: dict | None = None): + self.message = message + self.original_response = original_response + super().__init__(self.message) + + +# Orientation 타입 정의 +OrientationType = Literal["horizontal", "vertical"] + +# ============================================================================= +# 모듈 레벨 캐시 및 HTTP 클라이언트 (싱글톤 패턴) +# ============================================================================= + +# 템플릿 캐시: {template_id: {"data": dict, "cached_at": float}} +_template_cache: dict[str, dict] = {} + +# 캐시 TTL (초) - 기본 5분 +CACHE_TTL_SECONDS = 300 + +# 모듈 레벨 공유 HTTP 클라이언트 (커넥션 풀 재사용) +_shared_client: httpx.AsyncClient | None = None + +text_template_v_1 = { + "type": "composition", + "track": 3, + "elements": [ + { + "type": "text", + "time": 0, + "y": "87.9086%", + "width": "100%", + "height": "40%", + "x_alignment": "50%", + "y_alignment": "50%", + "font_family": "Noto Sans", + "font_weight": "700", + "font_size": "8 vmin", + "background_color": "rgba(216,216,216,0)", + "background_x_padding": "33%", + "background_y_padding": "7%", + "background_border_radius": "28%", + "fill_color": "#ffffff", + "stroke_color": "rgba(51,51,51,1)", + "stroke_width": "0.6 vmin", + } + ] +} +text_template_v_2 = { + "type": "composition", + "track": 3, + "elements": [ + { + "type": "text", + "time": 0, + "x": "7.7233%", + "y": "82.9852%", + "width": "84.5534%", + "height": "5.7015%", + "x_anchor": "0%", + "y_anchor": "0%", + "x_alignment": "50%", + "y_alignment": "100%", + "font_family": "Noto Sans", + "font_weight": "700", + "font_size": "6.9999 vmin", + "fill_color": "#ffffff" + } + ] +} +text_template_v_3 = { + "type": "composition", + "track": 3, + "elements": [ + { + "type": "text", + "time": 0, + "x": "0%", + "y": "80%", + "width": "100%", + "height": "15%", + "x_anchor": "0%", + "y_anchor": "0%", + "x_alignment": "50%", + "y_alignment": "50%", + "font_family": "Noto Sans", + "font_weight": "700", + "font_size_maximum": "7 vmin", + "fill_color": "#ffffff", + "animations": [ + { + "time": 0, + "duration": 1, + "easing": "quadratic-out", + "type": "text-wave", + "split": "line", + "overlap": "50%" + } + ] + } + ] +} +text_template_h_1 = { + "type": "composition", + "track": 3, + "elements": [ + { + "type": "text", + "time": 0, + "x": "10%", + "y": "80%", + "width": "80%", + "height": "15%", + "x_anchor": "0%", + "y_anchor": "0%", + "x_alignment": "50%", + "font_family": "Noto Sans", + "font_weight": "700", + "font_size": "5.9998 vmin", + "fill_color": "#ffffff", + "stroke_color": "#333333", + "stroke_width": "0.2 vmin" + } + ] +} + +autotext_template_v_1 = { + "type": "text", + "track": 4, + "time": 0, + "y": "87.9086%", + "width": "100%", + "height": "40%", + "x_alignment": "50%", + "y_alignment": "50%", + "font_family": "Noto Sans", + "font_weight": "700", + "font_size": "8 vmin", + "background_color": "rgba(216,216,216,0)", + "background_x_padding": "33%", + "background_y_padding": "7%", + "background_border_radius": "28%", + "transcript_source": "audio-music", # audio-music과 연동됨 + "transcript_effect": "karaoke", + "fill_color": "#ffffff", + "stroke_color": "rgba(51,51,51,1)", + "stroke_width": "0.6 vmin" +} +autotext_template_h_1 = { + "type": "text", + "track": 4, + "time": 0, + "x": "10%", + "y": "83.2953%", + "width": "80%", + "height": "15%", + "x_anchor": "0%", + "y_anchor": "0%", + "x_alignment": "50%", + "font_family": "Noto Sans", + "font_weight": "700", + "font_size": "5.9998 vmin", + "transcript_source": "audio-music", + "transcript_effect": "karaoke", + "fill_color": "#ffffff", + "stroke_color": "#333333", + "stroke_width": "0.2 vmin" +} +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" + +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, +} + +# 매핑에 없는 업종(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, + project_id: int | None = None, +) -> str: + """orientation과 industry에 따라 Creatomate 템플릿 ID를 선택합니다. + + Args: + orientation: 영상 방향 ("horizontal" 또는 "vertical") + industry: 업종 분류 (stay|restaurant|cafe|attraction 등). + 세로형에서 매핑에 있으면 전용 템플릿을 반환합니다. + project_id: 미매핑 업종(general 등)의 세로형 분배용 키. + project_id % len(VST_LIST)로 결정적으로 선택하므로, 같은 + 프로젝트는 몇 번 호출해도 동일 템플릿을 받아 사전/렌더 단계가 + 일치한다. 없으면 DEFAULT_VERTICAL_TEMPLATE로 폴백한다. + + Returns: + 선택된 템플릿 ID + """ + if orientation == "horizontal": + template_id = DHST0001 + elif orientation == "vertical": + 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}, project_id={project_id}, template_id={template_id}") + return template_id + +async def get_shared_client() -> httpx.AsyncClient: + """공유 HTTP 클라이언트를 반환합니다. 없으면 생성합니다.""" + global _shared_client + if _shared_client is None or _shared_client.is_closed: + _shared_client = httpx.AsyncClient( + timeout=httpx.Timeout( + recovery_settings.CREATOMATE_RENDER_TIMEOUT, + connect=recovery_settings.CREATOMATE_CONNECT_TIMEOUT, + ), + limits=httpx.Limits(max_keepalive_connections=10, max_connections=20), + ) + return _shared_client + + +async def close_shared_client() -> None: + """공유 HTTP 클라이언트를 닫습니다. 앱 종료 시 호출하세요.""" + global _shared_client + if _shared_client is not None and not _shared_client.is_closed: + await _shared_client.aclose() + _shared_client = None + logger.info("[CreatomateService] Shared HTTP client closed") + + +def clear_template_cache() -> None: + """템플릿 캐시를 전체 삭제합니다.""" + global _template_cache + _template_cache.clear() + logger.info("[CreatomateService] Template cache cleared") + + +def _is_cache_valid(cached_at: float) -> bool: + """캐시가 유효한지 확인합니다.""" + return (time.time() - cached_at) < CACHE_TTL_SECONDS + + +class CreatomateService: + """Creatomate API를 통한 영상 생성 서비스 + + 모든 HTTP 호출 메서드는 비동기(async)로 구현되어 있습니다. + """ + + BASE_URL = "https://api.creatomate.com" + + def __init__( + self, + 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|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, project_id=project_id) + self.headers = { + "Content-Type": "application/json", + "Authorization": f"Bearer {self.api_key}", + } + + async def _request( + self, + method: str, + url: str, + timeout: float | None = None, + **kwargs, + ) -> httpx.Response: + """HTTP 요청을 수행합니다. + + Args: + method: HTTP 메서드 ("GET", "POST", etc.) + url: 요청 URL + timeout: 요청 타임아웃 (초). None이면 기본값 사용 + **kwargs: httpx 요청에 전달할 추가 인자 + + Returns: + httpx.Response: 응답 객체 + + Raises: + httpx.HTTPError: 요청 실패 시 + """ + logger.info(f"[Creatomate] {method} {url}") + + # timeout이 None이면 기본 타임아웃 사용 + actual_timeout = timeout if timeout is not None else recovery_settings.CREATOMATE_DEFAULT_TIMEOUT + + client = await get_shared_client() + + if method.upper() == "GET": + response = await client.get( + url, headers=self.headers, timeout=actual_timeout, **kwargs + ) + elif method.upper() == "POST": + response = await client.post( + url, headers=self.headers, timeout=actual_timeout, **kwargs + ) + else: + raise ValueError(f"Unsupported HTTP method: {method}") + + logger.info(f"[Creatomate] Response - Status: {response.status_code}") + return response + + async def get_all_templates_data(self) -> dict: + """모든 템플릿 정보를 조회합니다.""" + url = f"{self.BASE_URL}/v1/templates" + response = await self._request("GET", url) # 기본 타임아웃 사용 + response.raise_for_status() + return response.json() + + async def get_one_template_data( + self, + template_id: str, + use_cache: bool = True, + ) -> dict: + """특정 템플릿 ID로 템플릿 정보를 조회합니다. + + Args: + template_id: 조회할 템플릿 ID + use_cache: 캐시 사용 여부 (기본: True) + + Returns: + 템플릿 데이터 (deep copy) + """ + global _template_cache + + # 캐시 확인 + if use_cache and template_id in _template_cache: + cached = _template_cache[template_id] + if _is_cache_valid(cached["cached_at"]): + logger.debug(f"[CreatomateService] Cache HIT - {template_id}") + return copy.deepcopy(cached["data"]) + else: + # 만료된 캐시 삭제 + del _template_cache[template_id] + logger.debug(f"[CreatomateService] Cache EXPIRED - {template_id}") + + # API 호출 + url = f"{self.BASE_URL}/v1/templates/{template_id}" + response = await self._request("GET", url) # 기본 타임아웃 사용 + response.raise_for_status() + data = response.json() + + # 캐시 저장 + _template_cache[template_id] = { + "data": data, + "cached_at": time.time(), + } + logger.debug(f"[CreatomateService] Cache MISS - {template_id} (cached)") + + return copy.deepcopy(data) + + def parse_template_component_name(self, template_source: list) -> dict: + """템플릿 정보를 파싱하여 리소스 이름을 추출합니다.""" + + def recursive_parse_component(element: dict) -> dict: + if "name" in element: + result_element_name_type = {element["name"]: element["type"]} + else: + result_element_name_type = {} + + if element["type"] == "composition": + minor_component_list = [ + recursive_parse_component(minor) for minor in element["elements"] + ] + # WARNING: Same name component should shroud other component + for minor_component in minor_component_list: + result_element_name_type.update(minor_component) + + return result_element_name_type + + result = {} + for result_element_dict in [ + recursive_parse_component(component) for component in template_source + ]: + result.update(result_element_dict) + + return result + + async def parse_template_name_tag(resource_name : str) -> list: + tag_list = [] + tag_list = resource_name.split("_") + + return tag_list + + + def counting_component( + self, + template : dict, + target_template_type : str + ) -> list: + 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 + return count + + 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 + ) -> tuple[dict, dict]: + """템플릿 슬롯에 이미지를 배정합니다. + + Returns: + (modifications, assigned) 튜플 + - modifications: {슬롯명: image_url, "audio-music": music_url, ...} — modify_element에 전달 + - assigned: {슬롯명: image_tag} — 자막 생성 컨텍스트에 전달 (modify_element에 넣지 않음) + + 배정 전략 (duplicate=False): + - 순수 greedy pop 대신 "난이도 우선 greedy" 적용 + - 사전에 전체 슬롯 점수를 계산해 최고 달성 점수가 낮은(=까다로운) 슬롯을 먼저 배정 + - 이미지 배정 직전 점수를 재계산 (pop 이후 인덱스 변동 대응) + + 배정 전략 (duplicate=True, 이미지 < 슬롯): + - 1단계(최소 커버리지 보장): 이미지별로 가장 잘 맞는 슬롯을 찾아 pool의 모든 이미지를 + 서로 다른 슬롯에 먼저 배정한다 (이미지 수 <= 슬롯 수이므로 전량 배정 가능). + 슬롯을 먼저 소진시키는 것이 아니라 "이미지"를 기준으로 난이도 우선 배정하여, + 특정 이미지 몇 장이 여러 슬롯을 독점하는 것을 방지한다. + - 2단계: 1단계에서 못 채운 나머지 슬롯은 기존처럼 pool 전체에서 최고점 이미지를 + 배정한다 (여기서부터는 중복 재사용 불가피). + + Note: + marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되므로 여기서는 재필터링하지 않는다. + taged_image_list를 그대로 pool로 사용한다. + """ + source_elements = template["source"]["elements"] + template_component_data = self.parse_template_component_name(source_elements) + + pool = taged_image_list + modifications: dict = {} + assigned: dict = {} # {슬롯명: image_tag} — 자막 컨텍스트용 + + # 이미지 슬롯과 텍스트 슬롯 분리 + image_slots = [name for name, t in template_component_data.items() if t == "image"] + text_slots = [(name, t) for name, t in template_component_data.items() if t == "text"] + + if not pool: + logger.warning("[template_matching_taged_image] 태그된 이미지가 없음 — 이미지 슬롯 배정 불가") + elif duplicate: + # 이미지 부족(슬롯 수 > 이미지 수) + # Step 1: 최소 커버리지 보장 — 이미지마다 가장 잘 맞는 슬롯을 찾아 서로 다른 슬롯에 배정 + remaining_slots = list(image_slots) + + 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] + if score > best_score: + best_slot, best_score = slot, score + return best_slot, best_score + + # 사전 점수 계산 (최고 달성 점수가 낮을수록 배정처가 마땅찮은=까다로운 이미지) + prelim_best_score = [ + _best_slot_for_image(image, remaining_slots)[1] for image in pool + ] + ordered_indices = sorted( + range(len(pool)), + key=lambda i: prelim_best_score[i] + ) + + for idx in ordered_indices: + if not remaining_slots: + break + image = pool[idx] + # 슬롯이 이전 배정으로 줄어들었으므로 현재 remaining_slots 기준 재계산 + best_slot, _ = _best_slot_for_image(image, remaining_slots) + if best_slot is None: + continue + modifications[best_slot] = image["image_url"] + assigned[best_slot] = image["image_tag"] + remaining_slots.remove(best_slot) + + # Step 2: 커버리지 배정 후 남은 슬롯은 pool 전체에서 최고점 이미지 배정 (재사용 불가피) + for slot in remaining_slots: + scores = self.calculate_image_slot_score_multi(pool, slot) + if not scores: + continue + best_idx = scores.index(max(scores)) + selected = pool[best_idx] + modifications[slot] = selected["image_url"] + assigned[slot] = selected["image_tag"] + else: + # 이미지 충분(슬롯 수 <= 이미지 수): 난이도 우선 greedy 배정 + # Step 1: 정렬용 사전 점수 계산 (최고 달성 점수가 낮을수록 배정이 어려운 슬롯) + prelim = {slot: self.calculate_image_slot_score_multi(pool, slot) for slot in image_slots} + ordered_slots = sorted( + image_slots, + key=lambda s: max(prelim[s]) if prelim[s] else 0.0 + # 오름차순 → 달성 최대값이 낮은(어려운) 슬롯이 앞으로 + ) + + # Step 2: 난이도 순서로 배정, 배정마다 pool에서 pop + 다음 슬롯은 점수 재계산 + for slot in ordered_slots: + if not pool: + logger.warning(f"[template_matching_taged_image] 이미지 풀 소진 — 남은 슬롯 배정 불가: {slot}") + break + # pop 후 인덱스 변동이 있으므로 현재 pool로 재계산 (사전 점수 재사용 금지) + scores = self.calculate_image_slot_score_multi(pool, slot) + if not scores: + continue + + # 상위 3장 중 점수 비례 확률로 선택 (영상 다양화) + sorted_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True) + top_indices = sorted_indices[:min(3, len(sorted_indices))] + weights = [scores[i] for i in top_indices] + if sum(weights) > 0: + chosen_pool_idx = random.choices(top_indices, weights=weights, k=1)[0] + else: + chosen_pool_idx = random.choice(top_indices) + + selected = pool.pop(chosen_pool_idx) + modifications[slot] = selected["image_url"] + assigned[slot] = selected["image_tag"] + + # 텍스트 슬롯 처리 + for name, _ in text_slots: + if "address_input" in name: + modifications[name] = address + + modifications["audio-music"] = music_url + return modifications, assigned + + # 점수 계산 상수 (모듈 레벨로 빼면 더 좋지만 기존 코드 스타일 따라 클래스 내부에 위치) + _NARR_FLOOR = 0.25 # narrative 최솟값 modulator: base 점수가 완전히 0이 되는 것을 방지 + _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) * space_type_penalty + - base: 태그 카테고리 매칭 합산 (최대 5.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] + slot_tag_dict = self.parse_slot_name_to_tag(slot_name) + if slot_tag_dict is None: + # parse 실패 시 전부 0점 반환 (슬롯 skip) + return [0.0] * len(image_tag_list) + + 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 + ] + + for slot_tag_cate, slot_tag_item in slot_tag_dict.items(): + if slot_tag_cate == "narrative_preference": + slot_tag_narrative = slot_tag_item + continue + + match slot_tag_cate: + case "space_type": + weight = 2.0 + case "subject": + weight = 2.0 + case "camera": + weight = 1.0 + case "motion_recommended": + weight = 0.5 + case _: + raise ValueError(f"[calculate_image_slot_score_multi] unknown slot tag category: {slot_tag_cate}") + + for idx, image_tag in enumerate(image_tag_list): + if slot_tag_item.value in image_tag[slot_tag_cate]: # collect! + base_score_list[idx] += weight + + # narrative 점수 적용: 순수 곱셈 대신 "완화 modulator + 독립 보너스" + # - narr_mod: base를 흐릴 수 있지만 NARR_FLOOR 이하로는 안 떨어짐 + # - NARR_BONUS * narr: base=0이어도 narrative가 높으면 양수 점수 확보 + image_score_list = [] + for idx, image_tag in enumerate(image_tag_list): + raw_narr = image_tag.get("narrative_preference", {}).get(slot_tag_narrative, 0.0) + clamped_narr = min(self._NARR_MAX, max(self._NARR_MIN, float(raw_narr))) + 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 + + def parse_slot_name_to_tag(self, slot_name: str) -> dict[str, StrEnum] | None: + """슬롯 이름을 파싱하여 태그 딕셔너리를 반환합니다. + + 슬롯 이름 형식: {space_type}-{subject}-{camera}-{motion}-{narrative} + 파싱 실패 시 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리). + """ + try: + tag_list = slot_name.split("-") + if len(tag_list) < 5: + logger.warning(f"[parse_slot_name_to_tag] 슬롯명 토큰 부족 ({len(tag_list)}개): '{slot_name}' — 슬롯 skip") + return None + space_type = SpaceType(tag_list[0]) + subject = Subject(tag_list[1]) + camera = Camera(tag_list[2]) + # 모션 동의어(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, + "subject": subject, + "camera": camera, + "motion_recommended": motion, + "narrative_preference": narrative, + } + return tag_dict + except (ValueError, IndexError) as e: + logger.warning(f"[parse_slot_name_to_tag] 슬롯명 파싱 실패: '{slot_name}' — {e} — 슬롯 skip") + return None + + def select_thumbnail_image(self, taged_image_list: list[dict]) -> str | None: + """썸네일 배경 이미지를 독립 선택합니다. + + 일반 씬 배정 풀과 무관하게 narrative_preference의 core+intro 합산 + 최고점 이미지를 고른다 (본편 씬과 중복 허용 — 마케팅 자산이므로 + 항상 최고 품질 사진 사용). + + Args: + taged_image_list: [{"image_url": str, "image_tag": dict}] 형태의 이미지 풀 + + Returns: + 최고점 이미지 URL. 풀이 비어 있으면 None + """ + if not taged_image_list: + return None + + def _score(taged_image: dict) -> float: + narr = taged_image.get("image_tag", {}).get("narrative_preference", {}) + return float(narr.get("core", 0.0)) + float(narr.get("intro", 0.0)) + + best = max(taged_image_list, key=_score) + logger.info( + f"[select_thumbnail_image] selected={best['image_url']}, score={_score(best):.1f}, pool={len(taged_image_list)}" + ) + return best["image_url"] + + def elements_connect_resource_blackbox( + self, + elements: list, + image_url_list: list[str], + music_url: str, + address: str = None + ) -> dict: + """elements 정보와 이미지/가사/음악 리소스를 매핑합니다.""" + template_component_data = self.parse_template_component_name(elements) + + modifications = {} + + for idx, (template_component_name, template_type) in enumerate( + template_component_data.items() + ): + match template_type: + case "image": + modifications[template_component_name] = image_url_list[ + idx % len(image_url_list) + ] + case "text": + if "address_input" in template_component_name: + modifications[template_component_name] = address + + modifications["audio-music"] = music_url + + return modifications + + def modify_element(self, elements: list, modification: dict) -> list: + """elements의 source를 modification에 따라 수정합니다.""" + + def recursive_modify(element: dict) -> None: + if "name" in element: + match element["type"]: + case "image": + element["source"] = modification[element["name"]] + 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"], "") + case "composition": + for minor in element["elements"]: + recursive_modify(minor) + + for minor in elements: + recursive_modify(minor) + + return elements + + async def make_creatomate_call( + self, template_id: str, modifications: dict + ) -> dict: + """Creatomate에 렌더링 요청을 보냅니다 (재시도 로직 포함). + + Args: + template_id: Creatomate 템플릿 ID + modifications: 수정사항 딕셔너리 + + Returns: + Creatomate API 응답 데이터 + + Raises: + CreatomateResponseError: API 오류 또는 재시도 실패 시 + + Note: + response에 요청 정보가 있으니 폴링 필요 + """ + url = f"{self.BASE_URL}/v2/renders" + payload = { + "template_id": template_id, + "modifications": modifications, + } + + last_error: Exception | None = None + + for attempt in range(recovery_settings.CREATOMATE_MAX_RETRIES + 1): + try: + response = await self._request( + "POST", + url, + timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT, + json=payload, + ) + + # 200 OK, 201 Created, 202 Accepted 모두 성공으로 처리 + if response.status_code in (200, 201, 202): + return response.json() + + # 재시도 불가능한 오류 (4xx 클라이언트 오류) + if 400 <= response.status_code < 500: + raise CreatomateResponseError( + f"Client error: {response.status_code}", + original_response={"status": response.status_code, "text": response.text}, + ) + + # 재시도 가능한 오류 (5xx 서버 오류) + last_error = CreatomateResponseError( + f"Server error: {response.status_code}", + original_response={"status": response.status_code, "text": response.text}, + ) + + except httpx.TimeoutException as e: + logger.warning( + f"[Creatomate] Timeout on attempt {attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES + 1}" + ) + last_error = e + + except httpx.HTTPError as e: + logger.warning(f"[Creatomate] HTTP error on attempt {attempt + 1}: {e}") + last_error = e + + except CreatomateResponseError: + raise # CreatomateResponseError는 재시도하지 않고 즉시 전파 + + # 마지막 시도가 아니면 재시도 + if attempt < recovery_settings.CREATOMATE_MAX_RETRIES: + logger.info( + f"[Creatomate] Retrying... ({attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES})" + ) + + # 모든 재시도 실패 + raise CreatomateResponseError( + f"All {recovery_settings.CREATOMATE_MAX_RETRIES + 1} attempts failed", + original_response={"last_error": str(last_error)}, + ) + + async def make_creatomate_custom_call(self, source: dict) -> dict: + """템플릿 없이 Creatomate에 커스텀 렌더링 요청을 보냅니다 (재시도 로직 포함). + + Args: + source: 렌더링 소스 딕셔너리 + + Returns: + Creatomate API 응답 데이터 + + Raises: + CreatomateResponseError: API 오류 또는 재시도 실패 시 + + Note: + response에 요청 정보가 있으니 폴링 필요 + """ + url = f"{self.BASE_URL}/v2/renders" + + source["frame_rate"] = 30 + + last_error: Exception | None = None + + for attempt in range(recovery_settings.CREATOMATE_MAX_RETRIES + 1): + try: + response = await self._request( + "POST", + url, + timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT, + json=source, + ) + + # 200 OK, 201 Created, 202 Accepted 모두 성공으로 처리 + if response.status_code in (200, 201, 202): + return response.json() + + # 재시도 불가능한 오류 (4xx 클라이언트 오류) + if 400 <= response.status_code < 500: + raise CreatomateResponseError( + f"Client error: {response.status_code}", + original_response={"status": response.status_code, "text": response.text}, + ) + + # 재시도 가능한 오류 (5xx 서버 오류) + last_error = CreatomateResponseError( + f"Server error: {response.status_code}", + original_response={"status": response.status_code, "text": response.text}, + ) + + except httpx.TimeoutException as e: + logger.warning( + f"[Creatomate] Timeout on attempt {attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES + 1}" + ) + last_error = e + + except httpx.HTTPError as e: + logger.warning(f"[Creatomate] HTTP error on attempt {attempt + 1}: {e}") + last_error = e + + except CreatomateResponseError: + raise # CreatomateResponseError는 재시도하지 않고 즉시 전파 + + # 마지막 시도가 아니면 재시도 + if attempt < recovery_settings.CREATOMATE_MAX_RETRIES: + logger.info( + f"[Creatomate] Retrying... ({attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES})" + ) + + # 모든 재시도 실패 + raise CreatomateResponseError( + f"All {recovery_settings.CREATOMATE_MAX_RETRIES + 1} attempts failed", + original_response={"last_error": str(last_error)}, + ) + + async def render_thumbnail( + self, + image_url: str, + text_modifications: dict[str, str], + poll_interval: float = 3.0, + max_polls: int = 10, + ) -> str | None: + """썸네일 정지 이미지를 렌더링하고 결과 URL을 반환합니다. + + 영상 렌더와 독립적인 보조 자산 생성이므로, 모든 실패(요청 오류· + 렌더 실패·폴링 타임아웃)는 경고 로그 후 None을 반환하며 예외를 + 전파하지 않는다. 재시도하지 않는다. + + Args: + image_url: 배경 이미지 URL + text_modifications: make_thumbnail_modification() 반환값 + poll_interval: 상태 폴링 간격(초) + max_polls: 최대 폴링 횟수 + + Returns: + 성공 시 썸네일 이미지 URL, 실패 시 None + """ + try: + payload = { + **build_thumbnail_source(image_url, text_modifications), + "output_format": "jpg", + } + response = await self._request( + "POST", + f"{self.BASE_URL}/v2/renders", + timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT, + json=payload, + ) + if response.status_code not in (200, 201, 202): + logger.warning( + f"[render_thumbnail] 렌더 요청 실패 - status: {response.status_code}, body: {response.text}" + ) + return None + + body = response.json() + render = body[0] if isinstance(body, list) and body else body + render_id = render.get("id") if isinstance(render, dict) else None + if not render_id: + logger.warning(f"[render_thumbnail] 렌더 ID 없음 - response: {body}") + return None + + for _ in range(max_polls): + status_result = await self.get_render_status(render_id) + status = status_result.get("status") + if status == "succeeded": + thumbnail_url = status_result.get("url") + logger.info(f"[render_thumbnail] 성공 - url: {thumbnail_url}") + return thumbnail_url + if status == "failed": + logger.warning(f"[render_thumbnail] 렌더 실패 - render_id: {render_id}") + return None + await asyncio.sleep(poll_interval) + + logger.warning( + f"[render_thumbnail] 폴링 타임아웃 ({max_polls}회) - render_id: {render_id}" + ) + return None + + except Exception as e: + logger.warning(f"[render_thumbnail] 예외 발생 - {e}") + return None + + async def get_render_status(self, render_id: str) -> dict: + """렌더링 작업의 상태를 조회합니다. + + Args: + render_id: Creatomate 렌더 ID + + Returns: + 렌더링 상태 정보 + + Note: + 상태 값: + - planned: 예약됨 + - waiting: 대기 중 + - transcribing: 트랜스크립션 중 + - rendering: 렌더링 중 + - succeeded: 성공 + - failed: 실패 + """ + url = f"{self.BASE_URL}/v1/renders/{render_id}" + response = await self._request("GET", url) # 기본 타임아웃 사용 + response.raise_for_status() + return response.json() + + @staticmethod + def _entrance_transition_offset(elem: dict) -> float: + """요소의 entrance transition(transition=true, time==0) duration 합을 반환합니다. + + 순차 배치 시 이 값만큼 이전 요소와 겹치도록 시작점이 앞당겨집니다. + """ + offset = 0.0 + for animation in elem.get("animations", []): + if "reversed" in animation: + continue + if "transition" in animation and animation["transition"] and animation.get("time", 0) == 0: + offset += animation["duration"] + return offset + + def calc_scene_duration(self, template: dict) -> float: + """템플릿의 전체 장면 duration을 계산합니다.""" + total_template_duration = 0.0 + track_maximum_duration = { + SCENE_TRACK : 0, + SUBTITLE_TRACK : 0, + KEYWORD_TRACK : 0 + } + for elem in template["source"]["elements"]: + try: + if elem["track"] not in track_maximum_duration: + continue + if "time" not in elem or elem["time"] == 0: # elem is auto / 만약 마지막 elem이 auto인데 그 앞에 time이 있는 elem 일 시 버그 발생 확률 있음 + track_maximum_duration[elem["track"]] += elem["duration"] - self._entrance_transition_offset(elem) + else: + track_maximum_duration[elem["track"]] = max(track_maximum_duration[elem["track"]], elem["time"] + elem["duration"]) + + except Exception as e: + logger.debug(traceback.format_exc()) + logger.error(f"[calc_scene_duration] Error processing element: {elem}, {e}") + + total_template_duration = max(track_maximum_duration.values()) + + return total_template_duration + + def calc_track_time_ranges(self, template: dict, track: int) -> list[tuple[str, float, float]]: + """최상위 요소 중 지정 track의 (name, start, end) 시간 범위를 계산합니다. + + Creatomate 규칙에 따라 `time`이 없거나 0인 요소는 같은 트랙의 이전 요소 뒤에 + 순차 배치되며, entrance transition duration만큼 이전 요소와 겹치도록 + 시작점을 앞당깁니다(calc_scene_duration의 차감 규칙과 동치). + `time`이 명시된 요소는 해당 시각에서 시작하고, 커서는 그 요소의 끝으로 이동합니다. + """ + ranges: list[tuple[str, float, float]] = [] + cursor = 0.0 + for elem in template["source"]["elements"]: + try: + if elem.get("track") != track: + continue + duration = elem["duration"] + + if "time" not in elem or elem["time"] == 0: # 순차 배치(auto) + start = max(0.0, cursor - self._entrance_transition_offset(elem)) + else: + start = elem["time"] + + end = start + duration + cursor = end + ranges.append((elem.get("name", ""), start, end)) + except Exception as e: + logger.debug(traceback.format_exc()) + logger.error(f"[calc_track_time_ranges] Error processing element: {elem}, {e}") + + return ranges + + @staticmethod + def _scale_element(elem: dict, extend_rate: float) -> None: + """element 하나의 time/duration/animations를 extend_rate만큼 스케일링합니다. + + composition의 자식 element까지 재귀적으로 처리합니다. 재귀하지 않으면 + 부모 컴포지션(씬 슬롯)만 늘어나고 내부 leaf 이미지/애니메이션은 원래 + 길이에서 끝나버려, 슬롯이 끝나기 전 컴포지션의 fill_color(회색)가 + 노출되는 버그가 생긴다 — modify_element/parse_template_component_name과 + 동일하게 composition을 재귀 순회해야 함. + + 오디오 판별은 track 번호가 아닌 type == "audio"로 한다. composition + 내부의 track 번호는 그 컴포지션 안에서만 유효한 로컬 레이어 인덱스라 + 전역 AUDIO_TRACK/KEYWORD_TRACK 상수와 우연히 값이 겹칠 수 있다 + (실제 템플릿에도 다중 레이어 씬의 2번째/3번째 이미지가 track=2, track=4인 + 경우가 존재함). + """ + if elem.get("type") == "audio": + return + + # "time"은 숫자 오프셋 대신 "end"(컴포지션 끝에 고정) 같은 특수 키워드일 수 있음 — + # 이런 값은 렌더 시점에 (이미 스케일링된) duration 기준으로 재계산되므로 그대로 둔다. + if "time" in elem and isinstance(elem["time"], (int, float)): + elem["time"] = elem["time"] * extend_rate + if "duration" in elem and isinstance(elem["duration"], (int, float)): + elem["duration"] = elem["duration"] * extend_rate + + for animation in elem.get("animations", []): + anim_time = animation.get("time", 0) + if isinstance(anim_time, (int, float)) and anim_time != 0: + animation["time"] = anim_time * extend_rate + if isinstance(animation.get("duration"), (int, float)): + animation["duration"] = animation["duration"] * extend_rate + + if elem.get("type") == "composition": + for child in elem.get("elements", []): + CreatomateService._scale_element(child, extend_rate) + + def extend_template_duration(self, template: dict, target_duration: float) -> dict: + """템플릿의 duration을 target_duration으로 확장합니다.""" + # template["duration"] = target_duration + 0.5 # 늘린것보단 짧게 + # target_duration += 1 # 수동으로 직접 변경 및 테스트 필요 : 파란박스 생기는것 + total_template_duration = self.calc_scene_duration(template) + extend_rate = target_duration / total_template_duration + new_template = copy.deepcopy(template) + + for elem in new_template["source"]["elements"]: + try: + if elem["track"] == AUDIO_TRACK : # audio track(최상위 음악 트랙)은 패스 + continue + + self._scale_element(elem, extend_rate) + except Exception as e: + logger.error( + f"[extend_template_duration] Error processing element: {elem}, {e}" + ) + + return new_template + + def lining_lyric(self, text_template: dict, lyric_index: int, lyric_text: str, start_sec: float, end_sec: float, font_family: str = "Noto Sans") -> dict: + duration = end_sec - start_sec + text_scene = copy.deepcopy(text_template) + text_scene["name"] = f"Caption-{lyric_index}" + text_scene["duration"] = duration + text_scene["time"] = start_sec + text_scene["elements"][0]["name"] = f"lyric-{lyric_index}" + text_scene["elements"][0]["text"] = lyric_text + text_scene["elements"][0]["font_family"] = font_family + return text_scene + + def auto_lyric(self, auto_text_template : dict): + text_scene = copy.deepcopy(auto_text_template) + return text_scene + + def get_text_template(self): + match self.orientation: + case "vertical": + return text_template_v_3 + case "horizontal": + return text_template_h_1 + + def get_auto_text_template(self): + match self.orientation: + case "vertical": + return autotext_template_v_1 + case "horizontal": + return autotext_template_h_1 + + 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["track"] == SUBTITLE_TRACK: + subtitle_list.append(elem["name"]) + elif elem["track"] == KEYWORD_TRACK: + keyword_list.append(elem["name"]) + except Exception as e: + logger.error( + f"[extend_template_duration] Error processing element: {elem}, {e}" + ) + + try: + 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 + return pitching_list + + + # 썸네일 헤드라인 글자 수 범위. brand_concept이 셀링포인트 상세 설명에서 오면 + # 문장이 길어져 헤드라인 폰트가 잘게 축소·여러 줄로 wrap되므로 훅 문구 길이로 제한. + # 첫 문장/단어 경계 절단 결과가 MIN 미만이면 더 긴 원문 기준으로 되돌려 보완한다. + THUMB_HEADLINE_MIN_CHARS = 5 + THUMB_HEADLINE_MAX_CHARS = 15 + + @classmethod + def _truncate_thumbnail_headline( + cls, + text: str, + min_chars: int | None = None, + max_chars: int | None = None, + ) -> str: + """썸네일 헤드라인 텍스트를 훅 문구 길이(5~15자)로 축약합니다. + + 1) 첫 문장(마침표/느낌표/물음표 기준)만 취하되, min_chars 미만이면 원문 유지 + 2) max_chars를 넘으면 단어 경계에서 잘라 말줄임표를 붙이되, + 그 결과가 min_chars 미만이면 max_chars에서 하드 컷 + """ + if min_chars is None: + min_chars = cls.THUMB_HEADLINE_MIN_CHARS + if max_chars is None: + max_chars = cls.THUMB_HEADLINE_MAX_CHARS + + text = text.strip() + # 첫 문장만 취함. 구분자 뒤가 공백일 때만 문장 끝으로 간주 ("4.5점" 같은 소수점 오인 방지) + for sep in (".", "!", "?"): + idx = text.find(sep) + if 0 < idx < len(text) - 1 and text[idx + 1] == " ": + first_sentence = text[: idx + (0 if sep == "." else 1)].strip() + if len(first_sentence) >= min_chars: + text = first_sentence + break + + if len(text) <= max_chars: + return text + + cut = text[:max_chars] + if " " in cut: + word_cut = cut[: cut.rfind(" ")].rstrip(" ,.·") + if len(word_cut) >= min_chars: + return word_cut + "…" + return cut.rstrip(" ,.·") + "…" + + def make_thumbnail_modification(self, brand_name : str, region : str, brand_concept : str, category_definition : str, target_keywords : list[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]] + + 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" : self._truncate_thumbnail_headline(brand_concept), + "thumb-badge-category" : category_definition, + } + return mod_dict diff --git a/app/utils/prompts/schemas/image.py b/app/utils/prompts/schemas/image.py index 994cc83..cf0f663 100644 --- a/app/utils/prompts/schemas/image.py +++ b/app/utils/prompts/schemas/image.py @@ -70,6 +70,33 @@ class SpaceType(StrEnum): 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): """이미지 내 주요 피사체 유형. 화면에 무엇이 담겨 있는지를 분류하는 태그.""" @@ -94,6 +121,17 @@ class Subject(StrEnum): # 관광지(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): """이미지의 촬영 기법·구도·조명 특성. 영상 편집 시 컷의 시각적 스타일을 구분하는 태그.""" diff --git a/app/utils/prompts/schemas/marketing.py b/app/utils/prompts/schemas/marketing.py index 2788c14..457cc1e 100644 --- a/app/utils/prompts/schemas/marketing.py +++ b/app/utils/prompts/schemas/marketing.py @@ -44,3 +44,4 @@ class MarketingPromptOutput(BaseModel): target_persona: List[TargetPersona] = Field(..., description="타겟 페르소나") selling_points: List[SellingPoint] = Field(..., description="셀링 포인트") target_keywords: List[str] = Field(..., description="타겟 키워드 리스트") + hook_headline: str = Field(..., description="썸네일 헤드라인용 훅 문구. 브랜드 컨셉/핵심 매력을 5자 이상 15자 이하의 한국어 한 구절로 축약 (마침표·해시태그 없이, 예: '바다 앞 프라이빗 휴식')") # min/max constraint는 openai json schema에서 미작동 보고가 있어 description으로 유도, 소비처에서 길이 안전망 적용 diff --git a/app/utils/thumbnail.py b/app/utils/thumbnail.py new file mode 100644 index 0000000..6e29a2b --- /dev/null +++ b/app/utils/thumbnail.py @@ -0,0 +1,197 @@ +"""세로형 썸네일 정지 이미지 렌더용 컴포지션 및 source 조립. + +CreatomateService에 의존하지 않는 코드 측 자산이다. 실제 렌더 요청·이미지 선택은 +CreatomateService.render_thumbnail / select_thumbnail_image가 담당한다. +""" + +import copy + +# 세로형 썸네일(1080×1920) 정지 이미지 렌더용 컴포지션. +# 운영 템플릿에 의존하지 않는 코드 측 자산 — 디자인 수정 시 이 상수만 변경. +# 텍스트 슬롯 이름은 CreatomateService.make_thumbnail_modification()의 반환 키와 일치해야 한다. +THUMBNAIL_COMPOSITION_V: dict = { + "name": "thumbnail-composition", + "type": "composition", + "track": 1, + "time": 0, + "elements": [ + { + "name": "thumb-background", + "type": "image", + "track": 1, + "time": 0, + "width": "105%", + "height": "105%", + "smart_crop": True, + "color_overlay": "rgba(0,0,0,0.22)", + }, + { + "name": "thumb-badge-category", + "type": "text", + "track": 2, + "time": 0, + "y": "8%", + "width": "75%", + "height": "7%", + "x_alignment": "50%", + "y_alignment": "50%", + "font_family": "Pretendard", + "font_weight": "700", + "font_size_maximum": "4.8 vmin", + "letter_spacing": "12%", + "line_height": "110%", + "background_color": "#FFFFFF", + "background_x_padding": "50%", + "background_y_padding": "65%", + "background_border_radius": "100%", + "fill_color": "#1A1A1A", + }, + { + "name": "thumb-headline-hook_claim-aspirational", + "type": "text", + "track": 3, + "time": 0, + "y": "45%", + "width": "85%", + "height": "22%", + "x_alignment": "50%", + "y_alignment": "50%", + "font_family": "Pretendard", + "font_weight": "900", + "font_size_maximum": "11 vmin", + "letter_spacing": "-2%", + "fill_color": "#FFFFFF", + "stroke_color": "rgba(0,0,0,0.65)", + "stroke_width": "0.8 vmin", + "shadow_color": "rgba(0,0,0,0.55)", + "shadow_blur": "2.5 vmin", + "shadow_y": "0.5 vmin", + }, + { + "name": "thumb-subheadline-selling_point", + "type": "text", + "track": 4, + "time": 0, + "y": "64%", + "width": "88%", + "height": "9%", + "x_alignment": "50%", + "y_alignment": "50%", + "font_family": "Pretendard", + "font_weight": "500", + "font_size_maximum": "6 vmin", + "letter_spacing": "6%", + "line_height": "130%", + "fill_color": "#FFFFFF", + "stroke_color": "rgba(0,0,0,0.55)", + "stroke_width": "0.4 vmin", + }, + { + "name": "thumb-brand-wordmark", + "type": "text", + "track": 5, + "time": 0, + "y": "89%", + "width": "75%", + "height": "9%", + "x_alignment": "50%", + "y_alignment": "50%", + "font_family": "Pretendard", + "font_weight": "900", + "font_size_maximum": "7.5 vmin", + "letter_spacing": "8%", + "line_height": "120%", + "fill_color": "#FFFFFF", + "stroke_color": "rgba(0,0,0,0.6)", + "stroke_width": "0.5 vmin", + "shadow_color": "rgba(0,0,0,0.45)", + "shadow_blur": "1.5 vmin", + "shadow_y": "0.3 vmin", + }, + { + "name": "thumb-hashtag-primary", + "type": "text", + "track": 6, + "time": 0, + "y": "95.5%", + "width": "92%", + "height": "6%", + "x_alignment": "50%", + "y_alignment": "50%", + "font_family": "Pretendard", + "font_weight": "600", + "font_size_maximum": "4.2 vmin", + "letter_spacing": "4%", + "line_height": "120%", + "fill_color": "#FFFFFF", + "stroke_color": "rgba(0,0,0,0.5)", + "stroke_width": "0.3 vmin", + }, + ], +} + + +# 영상 첫 화면 썸네일 오버레이 노출 시간(초). 30fps 기준 3프레임 — SNS 첫 프레임 +# 캡처에는 충분하면서 시청 흐름에는 거의 인지되지 않는 안전 하한선. +THUMBNAIL_INTRO_DURATION = 0.1 + +# 오버레이가 기존 최상위 트랙(1: 씬, 2: 오디오, 3: 자막, 4: 키워드)과 +# 렌더 직전 append되는 가사 캡션(track 3~4) 위에 오도록 여유 있게 잡은 트랙 번호. +THUMBNAIL_INTRO_TRACK = 9 + + +def _fill_thumbnail_slots(composition: dict, image_url: str, text_modifications: dict[str, str]) -> None: + """썸네일 컴포지션의 이미지/텍스트 슬롯을 in-place로 채웁니다.""" + for elem in composition["elements"]: + name = elem.get("name") + if elem["type"] == "image" and name == "thumb-background": + elem["source"] = image_url + elif elem["type"] == "text" and name in text_modifications: + elem["text"] = text_modifications[name] + + +def build_thumbnail_source(image_url: str, text_modifications: dict[str, str]) -> dict: + """썸네일 렌더용 Creatomate source를 조립합니다. + + Args: + image_url: 배경 이미지 URL + text_modifications: make_thumbnail_modification() 반환값 (thumb-* 텍스트 슬롯) + + Returns: + 1080×1920 정지 이미지 렌더 source 딕셔너리 + """ + composition = copy.deepcopy(THUMBNAIL_COMPOSITION_V) + _fill_thumbnail_slots(composition, image_url, text_modifications) + return { + "width": 1080, + "height": 1920, + "elements": [composition], + } + + +def build_thumbnail_intro_element( + image_url: str, + text_modifications: dict[str, str], + duration: float = THUMBNAIL_INTRO_DURATION, +) -> dict: + """영상 첫 화면(SNS 첫 프레임용)에 얹을 썸네일 오버레이 요소를 조립합니다. + + duration 스케일링(extend_template_duration) 이후, 렌더 요청 직전에 + final_template["source"]["elements"]에 append해야 노출 시간이 그대로 유지된다. + 기존 씬·자막·오디오 타이밍은 건드리지 않는 순수 오버레이이다. + + Args: + image_url: 배경 이미지 URL + text_modifications: make_thumbnail_modification() 반환값 (thumb-* 텍스트 슬롯) + duration: 오버레이 노출 시간(초) + + Returns: + time=0, 최상위 트랙에 배치된 composition 요소 딕셔너리 + """ + composition = copy.deepcopy(THUMBNAIL_COMPOSITION_V) + composition["name"] = "thumbnail-intro" + composition["track"] = THUMBNAIL_INTRO_TRACK + composition["time"] = 0 + composition["duration"] = duration + _fill_thumbnail_slots(composition, image_url, text_modifications) + return composition diff --git a/app/video/api/routers/v1/video.py b/app/video/api/routers/v1/video.py index ddb3f8a..181b779 100644 --- a/app/video/api/routers/v1/video.py +++ b/app/video/api/routers/v1/video.py @@ -32,6 +32,9 @@ from app.utils.address_parser import SIDO_CITIES, SIDO_SEARCH_ALIASES from app.lyric.models import Lyric from app.song.models import Song, SongTimestamp from app.utils.creatomate import CreatomateService +# 썸네일 인트로 오버레이 잠정 비활성화 — 재활성화 시 주석 해제 (generate_video 내 블록 참조) +# from app.utils.thumbnail import build_thumbnail_intro_element +# from app.video.services.video import get_image_tags_by_task_id from app.comment.models import Comment from app.database.like_cache import ( @@ -57,6 +60,7 @@ from app.video.schemas.video_schema import ( VideoRenderData, VideoThumbnailItem, ) +from app.video.worker.thumbnail_task import generate_thumbnail_background from app.video.worker.video_task import download_and_upload_video_to_blob @@ -125,6 +129,7 @@ curl -X GET "http://localhost:8000/video/generate/0694b716-dbff-7219-8000-d08cb5 ) async def generate_video( task_id: str, + background_tasks: BackgroundTasks, orientation: Literal["horizontal", "vertical"] = Query( default="vertical", description="영상 방향 (horizontal: 가로형, vertical: 세로형)", @@ -221,10 +226,24 @@ 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"] + # 썸네일 헤드라인(thumb-headline-hook_claim-aspirational)용 문구 추출. + # 1순위: 마케팅 분석 시 LLM이 5~15자로 축약 생성한 hook_headline (신규 데이터) + # 폴백: hook_headline이 없는 구 데이터는 concept 셀링포인트 → core_value → + # 최고점 셀링포인트 순으로 추출 (make_thumbnail_modification에서 15자 절단 안전망 적용) + brand_concept = marketing_intelligence.intel_result.get("hook_headline", "") or "" + selling_points = marketing_intelligence.intel_result["selling_points"] + if not brand_concept: + for sp in selling_points: + if "concept" in sp["english_category"].lower(): + brand_concept = sp["description"] + break + if not brand_concept: + brand_concept = marketing_intelligence.intel_result["market_positioning"].get("core_value", "") + if brand_concept: + logger.info(f"[generate_video] brand_concept 폴백(core_value) 사용 - task_id: {task_id}") + if not brand_concept and selling_points: + brand_concept = max(selling_points, key=lambda sp: sp.get("score", 0))["description"] + logger.info(f"[generate_video] brand_concept 폴백(최고점 셀링포인트) 사용 - task_id: {task_id}") # Lyric 조회 lyric_result = await session.execute( @@ -336,6 +355,19 @@ async def generate_video( ) # 세션이 여기서 자동으로 닫힘 (async with 블록 종료) + # 썸네일 백그라운드 생성 (세로형만, 응답 비차단) — 실패해도 영상 생성에 영향 없음 + background_tasks.add_task( + generate_thumbnail_background, + video_id=video_id, + task_id=task_id, + orientation=orientation, + brand_name=brand_name, + region=region, + brand_concept=brand_concept, + category_definition=category_definition, + target_keywords=target_keywords, + ) + except HTTPException: raise except Exception as e: @@ -360,6 +392,7 @@ async def generate_video( creatomate_service = CreatomateService( orientation=orientation, industry=industry, + project_id=project_id, ) logger.debug( f"[generate_video] Using template_id: {creatomate_service.template_id}, (song duration: {song_duration})" @@ -445,6 +478,24 @@ async def generate_video( ) final_template["source"]["elements"].append(caption) # END - LYRIC AUTO 결정부 + + # 썸네일 인트로 오버레이 (세로형만) — SNS 첫 프레임용 0.1초 정지 화면. + # duration 스케일링·자막 append 이후에 붙여야 노출 시간이 유지되며, + # 실패해도 영상 생성 자체에는 영향을 주지 않는다. + # NOTE: 잠정 비활성화 — 재활성화 시 아래 블록과 상단 import 주석 해제 + # if orientation == "vertical": + # try: + # taged_image_list = await get_image_tags_by_task_id(task_id) + # thumb_image_url = creatomate_service.select_thumbnail_image(taged_image_list) + # if thumb_image_url: + # final_template["source"]["elements"].append( + # build_thumbnail_intro_element(thumb_image_url, thumbnail_modifications) + # ) + # logger.info(f"[generate_video] Thumbnail intro overlay added - task_id: {task_id}") + # else: + # logger.warning(f"[generate_video] 썸네일 인트로 배경 이미지 없음 — skip - task_id: {task_id}") + # except Exception as e: + # logger.warning(f"[generate_video] 썸네일 인트로 추가 실패(무시) - task_id: {task_id}, error: {e}") # logger.debug( # f"[generate_video] final_template: {json.dumps(final_template, indent=2, ensure_ascii=False)}" # ) @@ -977,6 +1028,7 @@ async def get_all_videos( video_id=v.id, store_name=p.store_name, result_movie_url=v.result_movie_url, + thumbnail_url=v.video_thumbnail_url, created_at=v.created_at, like_count=like_count_map.get(v.id) or 0, is_liked_by_me=liked_map.get(v.id, False), diff --git a/app/video/models.py b/app/video/models.py index 888fc7b..598464a 100644 --- a/app/video/models.py +++ b/app/video/models.py @@ -106,6 +106,12 @@ class Video(Base): comment="생성된 영상 URL", ) + video_thumbnail_url: Mapped[Optional[str]] = mapped_column( + String(2048), + nullable=True, + comment="영상 생성 시 백그라운드로 렌더된 썸네일 이미지 URL (실패/미생성 시 NULL)", + ) + is_deleted: Mapped[bool] = mapped_column( Boolean, nullable=False, diff --git a/app/video/schemas/video_schema.py b/app/video/schemas/video_schema.py index f6599cb..ce1411d 100644 --- a/app/video/schemas/video_schema.py +++ b/app/video/schemas/video_schema.py @@ -157,6 +157,7 @@ class VideoListItem(BaseModel): region: Optional[str] = Field(None, description="지역명") task_id: str = Field(..., description="작업 고유 식별자") result_movie_url: Optional[str] = Field(None, description="영상 결과 URL") + thumbnail_url: Optional[str] = Field(None, description="백그라운드 렌더된 썸네일 이미지 URL (세로형만, 실패/미생성 시 null)") created_at: Optional[datetime] = Field(None, description="생성 일시") like_count: int = Field(0, description="좋아요 수") comment_count: int = Field(0, description="댓글 수 (대댓글 포함)") @@ -171,7 +172,8 @@ class VideoThumbnailItem(BaseModel): video_id: int = Field(..., description="영상 고유 ID (상세 페이지 라우팅 키)") store_name: str = Field(..., description="업체명") - result_movie_url: str = Field(..., description="영상 URL — 프론트에서