From 5f06c3c59de635099179652677bbe52670628219 Mon Sep 17 00:00:00 2001 From: bluebamus Date: Wed, 24 Dec 2025 17:57:50 +0900 Subject: [PATCH] =?UTF-8?q?creatomate=20=EC=BD=94=EB=93=9C=EB=A5=BC=20util?= =?UTF-8?q?s=EB=A1=9C=20=EC=9D=B4=EB=8F=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/utils/creatomate.py | 254 ++++++++++++++++++++++++++ app/video/services/base.py | 24 --- app/video/worker/video_tesk.py | 0 poc/crawling/creatomate/creatomate.py | 172 +++++++++-------- poc/crawling/creatomate/test.py | 30 ++- 5 files changed, 364 insertions(+), 116 deletions(-) create mode 100644 app/utils/creatomate.py delete mode 100644 app/video/services/base.py create mode 100644 app/video/worker/video_tesk.py diff --git a/app/utils/creatomate.py b/app/utils/creatomate.py new file mode 100644 index 0000000..13ee715 --- /dev/null +++ b/app/utils/creatomate.py @@ -0,0 +1,254 @@ +""" +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 diff --git a/app/video/services/base.py b/app/video/services/base.py deleted file mode 100644 index 2a0b0a9..0000000 --- a/app/video/services/base.py +++ /dev/null @@ -1,24 +0,0 @@ -from uuid import UUID -from sqlalchemy.ext.asyncio import AsyncSession -from sqlmodel import SQLModel - - -class BaseService: - def __init__(self, model, session: AsyncSession): - self.model = model - self.session = session - - async def _get(self, id: UUID): - return await self.session.get(self.model, id) - - async def _add(self, entity): - self.session.add(entity) - await self.session.commit() - await self.session.refresh(entity) - return entity - - async def _update(self, entity): - return await self._add(entity) - - async def _delete(self, entity): - await self.session.delete(entity) \ No newline at end of file diff --git a/app/video/worker/video_tesk.py b/app/video/worker/video_tesk.py new file mode 100644 index 0000000..e69de29 diff --git a/poc/crawling/creatomate/creatomate.py b/poc/crawling/creatomate/creatomate.py index 2ec1e58..f48f11b 100644 --- a/poc/crawling/creatomate/creatomate.py +++ b/poc/crawling/creatomate/creatomate.py @@ -1,53 +1,58 @@ -import requests import copy +import requests + CREATOMATE_API_KEY = "df9e4382d7e84fe790bf8a2168152be195d5a3568524ceb66ed989a2dea809f7d3065d6803b2e3dd9d02b5e5ec1c9823" -#ACCOUNT_URL = "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/" +# ACCOUNT_URL = "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/" # Creatomate 템플릿 정보 전부 가져오기 -class Creatomate(): - base_url : str = "https://api.creatomate.com" + +class Creatomate: + base_url: str = "https://api.creatomate.com" def __init__(self, api_key): self.api_key = api_key def get_all_templates_data(self) -> dict: - url = Creatomate.base_url + f"/v1/templates" + url = Creatomate.base_url + "/v1/templates" headers = { "Content-Type": "application/json", - "Authorization": f"Bearer {self.api_key}" + "Authorization": f"Bearer {self.api_key}", } response = requests.get(url, headers=headers) return response.json() - # Creatomate 템플릿 ID 로부터 해당 템플릿 정보 가져오기 - def get_one_template_data(self, template_id:str) -> dict: + def get_one_template_data(self, template_id: str) -> dict: url = Creatomate.base_url + f"/v1/templates/{template_id}" headers = { "Content-Type": "application/json", - "Authorization": f"Bearer {self.api_key}" + "Authorization": f"Bearer {self.api_key}", } response = requests.get(url, headers=headers) return response.json() # 템플릿 정보 파싱하여 리소스 이름 추출하기 - def parse_template_component_name(self, template_source : dict) -> dict: - def recursive_parse_component(element:dict) -> dict: - if 'name' in element: - result_element_name_type = {element['name'] : element['type']} + def parse_template_component_name(self, template_source: dict) -> 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']] + if element["type"] == "composition": + minor_component_list = [ + recursive_parse_component(minor) for minor in element["elements"] + ] for minor_component in minor_component_list: ## WARNING : Same name component should shroud other component. be aware 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]: + for result_element_dict in [ + recursive_parse_component(component) for component in template_source + ]: result.update(result_element_dict) return result @@ -56,55 +61,73 @@ class Creatomate(): # 이미지는 순차적으로 집어넣기 # 가사는 개행마다 한 텍스트 삽입 # Template에 audio-music 항목이 있어야 함. (추가된 템플릿 Cafe뿐임) - def template_connect_resource_blackbox(self, template_id:str, image_url_list:list[str], lyric:str, music_url:str) -> dict: + def template_connect_resource_blackbox( + self, template_id: str, image_url_list: list[str], lyric: str, music_url: str + ) -> dict: template_data = self.get_one_template_data(template_id) - template_component_data = self.parse_template_component_name(template_data['source']['elements']) + template_component_data = self.parse_template_component_name( + template_data["source"]["elements"] + ) lyric.replace("\r", "") lyric_splited = lyric.split("\n") modifications = {} - for idx, (template_component_name, template_type) in enumerate(template_component_data.items()): + 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)] + 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: + def elements_connect_resource_blackbox( + self, elements: list, image_url_list: list[str], lyric: str, music_url: str + ) -> dict: template_component_data = self.parse_template_component_name(elements) lyric.replace("\r", "") lyric_splited = lyric.split("\n") modifications = {} - for idx, (template_component_name, template_type) in enumerate(template_component_data.items()): + 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)] + 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): - def recursive_modify(element:dict) -> dict: - 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']: + def modify_element(self, elements: list, modification: dict): + def recursive_modify(element: dict) -> dict: + 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: @@ -114,16 +137,13 @@ class Creatomate(): # Creatomate에 생성 요청 # response에 요청 정보 있으니 풀링 필요 - def make_creatomate_call(self, template_id:str, modifications:dict): - url = Creatomate.base_url + f"/v2/renders" - - data = { - "template_id": template_id, - "modifications": modifications - } + def make_creatomate_call(self, template_id: str, modifications: dict): + url = Creatomate.base_url + "/v2/renders" + + data = {"template_id": template_id, "modifications": modifications} headers = { "Content-Type": "application/json", - "Authorization": f"Bearer {self.api_key}" + "Authorization": f"Bearer {self.api_key}", } response = requests.post(url, json=data, headers=headers) @@ -131,49 +151,49 @@ class Creatomate(): # Creatomate에 생성 요청 without template # response에 요청 정보 있으니 풀링 필요 - def make_creatomate_custom_call(self, source:str): - url = Creatomate.base_url + f"/v2/renders" + def make_creatomate_custom_call(self, source: str): + url = Creatomate.base_url + "/v2/renders" data = source headers = { "Content-Type": "application/json", - "Authorization": f"Bearer {self.api_key}" + "Authorization": f"Bearer {self.api_key}", } response = requests.post(url, json=data, headers=headers) return response - def calc_scene_duration(self, template : dict): + def calc_scene_duration(self, template: dict): total_template_duration = 0 - for elem in template['source']['elements']: + for elem in template["source"]["elements"]: try: - if elem['type'] == 'audio': + 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'] + 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: print(elem) return total_template_duration - def extend_template_duration(self, template : dict, target_duration : float): - template['duration'] = target_duration + def extend_template_duration(self, template: dict, target_duration: float): + 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']: + for elem in new_template["source"]["elements"]: try: - if elem['type'] == 'audio': + 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 + 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: print(elem) return new_template diff --git a/poc/crawling/creatomate/test.py b/poc/crawling/creatomate/test.py index b391acc..63590b0 100644 --- a/poc/crawling/creatomate/test.py +++ b/poc/crawling/creatomate/test.py @@ -1,18 +1,21 @@ import creatomate + CREATOMATE_API_KEY = "df9e4382d7e84fe790bf8a2168152be195d5a3568524ceb66ed989a2dea809f7d3065d6803b2e3dd9d02b5e5ec1c9823" shortform_4_template_id = "e8c7b43f-de4b-4ba3-b8eb-5df688569193" -target_duration = 90.0 # s +target_duration = 90.0 # s creato = creatomate.Creatomate(CREATOMATE_API_KEY) template = creato.get_one_template_data(shortform_4_template_id) -uploaded_image_url_list = ["https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818306_000_385523a5_99f2e8a8.jpg", - "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818306_001_d4cf6ec9_b81a1fdc.jpg", - "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818307_002_e4a0b276_680c5020.jpg", - "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818307_003_657f8c26_9f2c7168.jpg", - "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818307_004_9500e39d_24b9dad0.jpg", - "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818308_005_c3536641_9d490ccf.jpg"] +uploaded_image_url_list = [ + "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818306_000_385523a5_99f2e8a8.jpg", + "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818306_001_d4cf6ec9_b81a1fdc.jpg", + "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818307_002_e4a0b276_680c5020.jpg", + "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818307_003_657f8c26_9f2c7168.jpg", + "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818307_004_9500e39d_24b9dad0.jpg", + "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/crawling-images/crawler4_img_1755818308_005_c3536641_9d490ccf.jpg", +] lyric = """ 진짜 맛있는 추어탕의 향연 @@ -42,16 +45,11 @@ lyric = """ song_url = "https://ado2mediastoragepublic.blob.core.windows.net/ado2-media-public-access/ado2-media-original/dev-user-idx/dev-task-idx/stay.mp3" modifications = creato.elements_connect_resource_blackbox( - template['source']['elements'], - uploaded_image_url_list, - lyric, - song_url + template["source"]["elements"], uploaded_image_url_list, lyric, song_url ) -new_elements = creato.modify_element(template['source']['elements'], modifications) -template['source']['elements'] = new_elements +new_elements = creato.modify_element(template["source"]["elements"], modifications) +template["source"]["elements"] = new_elements last_template = creato.extend_template_duration(template, target_duration) -creato.make_creatomate_custom_call(last_template['source']) - - +creato.make_creatomate_custom_call(last_template["source"])