473 lines
17 KiB
Python
473 lines
17 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 typing import Literal
|
|
|
|
import httpx
|
|
|
|
from app.utils.logger import get_logger
|
|
from config import apikey_settings, creatomate_settings
|
|
|
|
# 로거 설정
|
|
logger = get_logger("creatomate")
|
|
|
|
|
|
# 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
|
|
|
|
|
|
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(60.0, connect=10.0),
|
|
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"
|
|
|
|
# 템플릿 설정 (config에서 가져옴)
|
|
TEMPLATE_CONFIG = {
|
|
"horizontal": {
|
|
"template_id": creatomate_settings.TEMPLATE_ID_HORIZONTAL,
|
|
"duration": creatomate_settings.TEMPLATE_DURATION_HORIZONTAL,
|
|
},
|
|
"vertical": {
|
|
"template_id": creatomate_settings.TEMPLATE_ID_VERTICAL,
|
|
"duration": creatomate_settings.TEMPLATE_DURATION_VERTICAL,
|
|
},
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
api_key: str | None = None,
|
|
orientation: OrientationType = "vertical",
|
|
target_duration: float | None = None,
|
|
):
|
|
"""
|
|
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에 따른 템플릿 설정 가져오기
|
|
config = self.TEMPLATE_CONFIG.get(
|
|
orientation, self.TEMPLATE_CONFIG["vertical"]
|
|
)
|
|
self.template_id = config["template_id"]
|
|
self.target_duration = (
|
|
target_duration if target_duration is not None else config["duration"]
|
|
)
|
|
|
|
self.headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
}
|
|
|
|
async def _request(
|
|
self,
|
|
method: str,
|
|
url: str,
|
|
timeout: float = 30.0,
|
|
**kwargs,
|
|
) -> httpx.Response:
|
|
"""HTTP 요청을 수행합니다.
|
|
|
|
Args:
|
|
method: HTTP 메서드 ("GET", "POST", etc.)
|
|
url: 요청 URL
|
|
timeout: 요청 타임아웃 (초)
|
|
**kwargs: httpx 요청에 전달할 추가 인자
|
|
|
|
Returns:
|
|
httpx.Response: 응답 객체
|
|
|
|
Raises:
|
|
httpx.HTTPError: 요청 실패 시
|
|
"""
|
|
logger.info(f"[Creatomate] {method} {url}")
|
|
|
|
client = await get_shared_client()
|
|
|
|
if method.upper() == "GET":
|
|
response = await client.get(
|
|
url, headers=self.headers, timeout=timeout, **kwargs
|
|
)
|
|
elif method.upper() == "POST":
|
|
response = await client.post(
|
|
url, headers=self.headers, timeout=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, timeout=30.0)
|
|
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, timeout=30.0)
|
|
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)
|
|
|
|
# 하위 호환성을 위한 별칭 (deprecated)
|
|
async def get_one_template_data_async(self, template_id: str) -> dict:
|
|
"""특정 템플릿 ID로 템플릿 정보를 조회합니다.
|
|
|
|
Deprecated: get_one_template_data()를 사용하세요.
|
|
"""
|
|
return await self.get_one_template_data(template_id)
|
|
|
|
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 template_connect_resource_blackbox(
|
|
self,
|
|
template_id: str,
|
|
image_url_list: list[str],
|
|
lyric: str,
|
|
music_url: str,
|
|
) -> dict:
|
|
"""템플릿 정보와 이미지/가사/음악 리소스를 매핑합니다.
|
|
|
|
Note:
|
|
- 이미지는 순차적으로 집어넣기
|
|
- 가사는 개행마다 한 텍스트 삽입
|
|
- Template에 audio-music 항목이 있어야 함
|
|
"""
|
|
template_data = await self.get_one_template_data(template_id)
|
|
template_component_data = self.parse_template_component_name(
|
|
template_data["source"]["elements"]
|
|
)
|
|
|
|
lyric = lyric.replace("\r", "")
|
|
lyric_splited = lyric.split("\n")
|
|
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":
|
|
modifications[template_component_name] = lyric_splited[
|
|
idx % len(lyric_splited)
|
|
]
|
|
|
|
modifications["audio-music"] = music_url
|
|
|
|
return modifications
|
|
|
|
def elements_connect_resource_blackbox(
|
|
self,
|
|
elements: list,
|
|
image_url_list: list[str],
|
|
lyric: str,
|
|
music_url: str,
|
|
) -> dict:
|
|
"""elements 정보와 이미지/가사/음악 리소스를 매핑합니다."""
|
|
template_component_data = self.parse_template_component_name(elements)
|
|
|
|
lyric = lyric.replace("\r", "")
|
|
lyric_splited = lyric.split("\n")
|
|
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":
|
|
modifications[template_component_name] = lyric_splited[
|
|
idx % len(lyric_splited)
|
|
]
|
|
|
|
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.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에 렌더링 요청을 보냅니다.
|
|
|
|
Note:
|
|
response에 요청 정보가 있으니 폴링 필요
|
|
"""
|
|
url = f"{self.BASE_URL}/v2/renders"
|
|
data = {
|
|
"template_id": template_id,
|
|
"modifications": modifications,
|
|
}
|
|
response = await self._request("POST", url, timeout=60.0, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def make_creatomate_custom_call(self, source: dict) -> dict:
|
|
"""템플릿 없이 Creatomate에 커스텀 렌더링 요청을 보냅니다.
|
|
|
|
Note:
|
|
response에 요청 정보가 있으니 폴링 필요
|
|
"""
|
|
url = f"{self.BASE_URL}/v2/renders"
|
|
response = await self._request("POST", url, timeout=60.0, json=source)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
# 하위 호환성을 위한 별칭 (deprecated)
|
|
async def make_creatomate_custom_call_async(self, source: dict) -> dict:
|
|
"""템플릿 없이 Creatomate에 커스텀 렌더링 요청을 보냅니다.
|
|
|
|
Deprecated: make_creatomate_custom_call()을 사용하세요.
|
|
"""
|
|
return await self.make_creatomate_custom_call(source)
|
|
|
|
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, timeout=30.0)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
# 하위 호환성을 위한 별칭 (deprecated)
|
|
async def get_render_status_async(self, render_id: str) -> dict:
|
|
"""렌더링 작업의 상태를 조회합니다.
|
|
|
|
Deprecated: get_render_status()를 사용하세요.
|
|
"""
|
|
return await self.get_render_status(render_id)
|
|
|
|
def calc_scene_duration(self, template: dict) -> float:
|
|
"""템플릿의 전체 장면 duration을 계산합니다."""
|
|
total_template_duration = 0.0
|
|
|
|
for elem in template["source"]["elements"]:
|
|
try:
|
|
if elem["type"] == "audio":
|
|
continue
|
|
total_template_duration += elem["duration"]
|
|
if "animations" not in elem:
|
|
continue
|
|
for animation in elem["animations"]:
|
|
assert animation["time"] == 0 # 0이 아닌 경우 확인 필요
|
|
if animation["transition"]:
|
|
total_template_duration -= animation["duration"]
|
|
except Exception as e:
|
|
logger.error(f"[calc_scene_duration] Error processing element: {elem}, {e}")
|
|
|
|
return total_template_duration
|
|
|
|
def extend_template_duration(self, template: dict, target_duration: float) -> dict:
|
|
"""템플릿의 duration을 target_duration으로 확장합니다."""
|
|
target_duration += 0.1 # 수동으로 직접 변경 및 테스트 필요 : 파란박스 생기는것
|
|
template["duration"] = target_duration
|
|
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
|
|
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
|