""" 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 = "75161273-0422-4771-adeb-816bd7263fb0" DVST0002 = "c68cf750-bc40-485a-a2c5-3f9fe301e386" DVST0003 = "e1fb5b00-1f02-4f63-99fa-7524b433ba47" DHST0001 = "660be601-080a-43ea-bf0f-adcf4596fa98" DHST0002 = "3f194cc7-464e-4581-9db2-179d42d3e40f" DHST0003 = "f45df555-2956-4a13-9004-ead047070b3d" DVST0001T = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6" HST_LIST = [DHST0001,DHST0002,DHST0003] VST_LIST = [DVST0001T] SCENE_TRACK = 1 AUDIO_TRACK = 2 SUBTITLE_TRACK = 3 KEYWORD_TRACK = 4 def select_template(orientation:OrientationType): if orientation == "horizontal": template_id = DHST0001 elif orientation == "vertical": template_id = random.choice(VST_LIST) else: raise logger.info(f"[select_template] orientation={orientation}, 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" ): """ Args: api_key: Creatomate API 키 (Bearer token으로 사용) None일 경우 config에서 자동으로 가져옴 orientation: 영상 방향 ("horizontal" 또는 "vertical", 기본값: "vertical") target_duration: 목표 영상 길이 (초) None일 경우 orientation에 해당하는 기본값 사용 """ self.api_key = api_key or apikey_settings.CREATOMATE_API_KEY self.orientation = orientation # orientation에 따른 템플릿 설정 가져오기 self.template_id = select_template(orientation) 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]) motion = 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