868 lines
32 KiB
Python
868 lines
32 KiB
Python
"""
|
|
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 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 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 = [DVST0001,DVST0002,DVST0003, DVST0001T]
|
|
|
|
SCENE_TRACK = 1
|
|
AUDIO_TRACK = 2
|
|
SUBTITLE_TRACK = 3
|
|
KEYWORD_TRACK = 4
|
|
|
|
def select_template(orientation:OrientationType):
|
|
if orientation == "horizontal":
|
|
return DHST0001
|
|
elif orientation == "vertical":
|
|
return DVST0001T
|
|
else:
|
|
raise
|
|
|
|
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_name" : str , "image_tag" : dict}]
|
|
music_url: str,
|
|
address : str,
|
|
duplicate : bool = False
|
|
) -> list:
|
|
source_elements = template["source"]["elements"]
|
|
template_component_data = self.parse_template_component_name(source_elements)
|
|
|
|
modifications = {}
|
|
|
|
for slot_idx, (template_component_name, template_type) in enumerate(template_component_data.items()):
|
|
match template_type:
|
|
case "image":
|
|
image_score_list = self.calculate_image_slot_score_multi(taged_image_list, template_component_name)
|
|
maximum_idx = image_score_list.index(max(image_score_list))
|
|
if duplicate:
|
|
selected = taged_image_list[maximum_idx]
|
|
else:
|
|
selected = taged_image_list.pop(maximum_idx)
|
|
image_name = selected["image_url"]
|
|
modifications[template_component_name] =image_name
|
|
pass
|
|
case "text":
|
|
if "address_input" in template_component_name:
|
|
modifications[template_component_name] = address
|
|
|
|
modifications["audio-music"] = music_url
|
|
return modifications
|
|
|
|
def calculate_image_slot_score_multi(self, taged_image_list : list[dict], slot_name : str):
|
|
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)
|
|
image_score_list = [0] * len(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
|
|
case "subject" :
|
|
weight = 2
|
|
case "camera":
|
|
weight = 1
|
|
case "motion_recommended" :
|
|
weight = 0.5
|
|
case _:
|
|
raise
|
|
|
|
for idx, image_tag in enumerate(image_tag_list):
|
|
if slot_tag_item.value in image_tag[slot_tag_cate]: #collect!
|
|
image_score_list[idx] += weight
|
|
|
|
for idx, image_tag in enumerate(image_tag_list):
|
|
image_narrative_score = image_tag["narrative_preference"][slot_tag_narrative]
|
|
image_score_list[idx] = image_score_list[idx] * image_narrative_score
|
|
|
|
return image_score_list
|
|
|
|
def parse_slot_name_to_tag(self, slot_name : str) -> dict[str, StrEnum]:
|
|
tag_list = slot_name.split("-")
|
|
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
|
|
|
|
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"
|
|
|
|
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()
|
|
|
|
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"]
|
|
|
|
if "animations" not in elem:
|
|
continue
|
|
for animation in elem["animations"]:
|
|
assert animation["time"] == 0 # 0이 아닌 경우 확인 필요
|
|
if "transition" in animation and animation["transition"]:
|
|
track_maximum_duration[elem["track"]] -= animation["duration"]
|
|
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 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["type"] == "audio":
|
|
# continue
|
|
if elem["track"] == AUDIO_TRACK : # audio track은 패스
|
|
continue
|
|
|
|
if "time" in elem:
|
|
elem["time"] = elem["time"] * extend_rate
|
|
if "duration" in elem:
|
|
elem["duration"] = elem["duration"] * extend_rate
|
|
|
|
if "animations" not in elem:
|
|
continue
|
|
for animation in elem["animations"]:
|
|
assert animation["time"] == 0 # 0이 아닌 경우 확인 필요
|
|
animation["duration"] = animation["duration"] * 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 |