255 lines
9.0 KiB
Python
255 lines
9.0 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 = creatomate.get_all_templates_data()
|
|
|
|
# 특정 템플릿 조회
|
|
template = creatomate.get_one_template_data(template_id)
|
|
|
|
# 영상 렌더링 요청
|
|
response = creatomate.make_creatomate_call(template_id, modifications)
|
|
```
|
|
"""
|
|
|
|
import copy
|
|
|
|
import httpx
|
|
|
|
from config import apikey_settings
|
|
|
|
|
|
class CreatomateService:
|
|
"""Creatomate API를 통한 영상 생성 서비스"""
|
|
|
|
BASE_URL = "https://api.creatomate.com"
|
|
|
|
def __init__(self, api_key: str | None = None):
|
|
"""
|
|
Args:
|
|
api_key: Creatomate API 키 (Bearer token으로 사용)
|
|
None일 경우 config에서 자동으로 가져옴
|
|
"""
|
|
self.api_key = api_key or apikey_settings.CREATOMATE_API_KEY
|
|
self.headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
}
|
|
|
|
def get_all_templates_data(self) -> dict:
|
|
"""모든 템플릿 정보를 조회합니다."""
|
|
url = f"{self.BASE_URL}/v1/templates"
|
|
response = httpx.get(url, headers=self.headers, timeout=30.0)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_one_template_data(self, template_id: str) -> dict:
|
|
"""특정 템플릿 ID로 템플릿 정보를 조회합니다."""
|
|
url = f"{self.BASE_URL}/v1/templates/{template_id}"
|
|
response = httpx.get(url, headers=self.headers, timeout=30.0)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
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
|
|
|
|
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 = 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
|
|
|
|
def make_creatomate_call(self, template_id: str, modifications: dict):
|
|
"""Creatomate에 렌더링 요청을 보냅니다.
|
|
|
|
Note:
|
|
response에 요청 정보가 있으니 폴링 필요
|
|
"""
|
|
url = f"{self.BASE_URL}/v2/renders"
|
|
data = {
|
|
"template_id": template_id,
|
|
"modifications": modifications,
|
|
}
|
|
response = httpx.post(url, json=data, headers=self.headers, timeout=60.0)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def make_creatomate_custom_call(self, source: dict):
|
|
"""템플릿 없이 Creatomate에 커스텀 렌더링 요청을 보냅니다.
|
|
|
|
Note:
|
|
response에 요청 정보가 있으니 폴링 필요
|
|
"""
|
|
url = f"{self.BASE_URL}/v2/renders"
|
|
response = httpx.post(url, json=source, headers=self.headers, timeout=60.0)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
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
|