352 lines
13 KiB
Python
352 lines
13 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)
|
|
```
|
|
"""
|
|
|
|
import copy
|
|
from typing import Literal
|
|
|
|
import httpx
|
|
|
|
from config import apikey_settings, creatomate_settings
|
|
|
|
|
|
# Orientation 타입 정의
|
|
OrientationType = Literal["horizontal", "vertical"]
|
|
|
|
|
|
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 get_all_templates_data(self) -> dict:
|
|
"""모든 템플릿 정보를 조회합니다."""
|
|
url = f"{self.BASE_URL}/v1/templates"
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url, headers=self.headers, timeout=30.0)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def get_one_template_data(self, template_id: str) -> dict:
|
|
"""특정 템플릿 ID로 템플릿 정보를 조회합니다."""
|
|
url = f"{self.BASE_URL}/v1/templates/{template_id}"
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url, headers=self.headers, timeout=30.0)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
# 하위 호환성을 위한 별칭 (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,
|
|
}
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
url, json=data, headers=self.headers, timeout=60.0
|
|
)
|
|
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"
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
url, json=source, headers=self.headers, timeout=60.0
|
|
)
|
|
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}"
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url, headers=self.headers, 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:
|
|
print(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으로 확장합니다."""
|
|
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:
|
|
print(
|
|
f"[extend_template_duration] Error processing element: {elem}, {e}"
|
|
)
|
|
|
|
return new_template
|