1166 lines
48 KiB
Python
1166 lines
48 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 random
|
|
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 app.utils.common import normalize_location
|
|
|
|
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 = "fe11aeab-ff29-4bc8-9f75-c695c7e243e6"
|
|
DVRT0001 = "3c4b74c4-82d7-4742-9b05-4b999fef4cbd"
|
|
DVCF0001 = "ebb532ec-e49a-4a7e-b0e2-a8132f1bf1f7"
|
|
DVAT0001 = "14d4fa22-49b8-49ca-a233-ac0556436f60"
|
|
DVSL0001 = "20c526a7-b184-46f3-9138-9dfaff2fa342"
|
|
DVCL0001 = "b57dca80-167d-430a-8905-30f4674ff72b"
|
|
DVFT0001 = "03b6d521-4864-4e69-8e6c-1deab9f0ce6e"
|
|
DVAC0001 = "4db8f5a3-4fd1-42cf-86b3-d997f0713d99"
|
|
|
|
DHST0001 = "660be601-080a-43ea-bf0f-adcf4596fa98"
|
|
|
|
HST_LIST = [DHST0001]
|
|
VST_LIST = [DVST0001,DVRT0001,DVCF0001,DVAT0001,DVSL0001,DVCL0001,DVFT0001,DVAC0001]
|
|
|
|
# industry → 세로형 템플릿 매핑 (신규 세로형 템플릿 추가 시 여기에만 등록)
|
|
VERTICAL_INDUSTRY_TEMPLATE_MAP: dict[str, str] = {
|
|
"stay": DVST0001,
|
|
"restaurant": DVRT0001,
|
|
"cafe": DVCF0001,
|
|
"attraction": DVAT0001,
|
|
"salon": DVSL0001,
|
|
"clinic": DVCL0001,
|
|
"fitness": DVFT0001,
|
|
"academy": DVAC0001,
|
|
}
|
|
|
|
# 매핑에 없는 업종(general)의 폴백 템플릿
|
|
DEFAULT_VERTICAL_TEMPLATE = DVST0001
|
|
|
|
# 템플릿 슬롯명의 모션 동의어 → MotionRecommended 표준값 정규화 맵.
|
|
# 동의어를 enum에 추가하면 LLM 태그 후보가 갈라져 매칭 정확도가 떨어지므로,
|
|
# enum은 표준 어휘만 유지하고 슬롯명 파싱 경계에서 흡수한다.
|
|
MOTION_TOKEN_NORMALIZATION: dict[str, MotionRecommended] = {
|
|
"zoom_in": MotionRecommended.slow_zoom_in,
|
|
"zoom_out": MotionRecommended.slow_zoom_out,
|
|
"pan_left": MotionRecommended.slow_pan,
|
|
"pan_right": MotionRecommended.slow_pan,
|
|
"tilt_up": MotionRecommended.slow_pan,
|
|
"tilt_down": MotionRecommended.slow_pan,
|
|
"push_diag": MotionRecommended.dolly,
|
|
"montage": MotionRecommended.static,
|
|
}
|
|
|
|
SCENE_TRACK = 1
|
|
AUDIO_TRACK = 2
|
|
SUBTITLE_TRACK = 3
|
|
KEYWORD_TRACK = 4
|
|
|
|
def select_template(
|
|
orientation: OrientationType,
|
|
industry: str | None = None,
|
|
project_id: int | None = None,
|
|
) -> str:
|
|
"""orientation과 industry에 따라 Creatomate 템플릿 ID를 선택합니다.
|
|
|
|
Args:
|
|
orientation: 영상 방향 ("horizontal" 또는 "vertical")
|
|
industry: 업종 분류 (stay|restaurant|cafe|attraction 등).
|
|
세로형에서 매핑에 있으면 전용 템플릿을 반환합니다.
|
|
project_id: 미매핑 업종(general 등)의 세로형 분배용 키.
|
|
project_id % len(VST_LIST)로 결정적으로 선택하므로, 같은
|
|
프로젝트는 몇 번 호출해도 동일 템플릿을 받아 사전/렌더 단계가
|
|
일치한다. 없으면 DEFAULT_VERTICAL_TEMPLATE로 폴백한다.
|
|
|
|
Returns:
|
|
선택된 템플릿 ID
|
|
"""
|
|
if orientation == "horizontal":
|
|
template_id = DHST0001
|
|
elif orientation == "vertical":
|
|
mapped = VERTICAL_INDUSTRY_TEMPLATE_MAP.get(industry)
|
|
if mapped is not None:
|
|
template_id = mapped
|
|
elif project_id is not None:
|
|
# 미매핑 업종: project_id 나머지로 VST_LIST 결정적 분배 (공유 상태 없음)
|
|
template_id = VST_LIST[project_id % len(VST_LIST)]
|
|
else:
|
|
template_id = DEFAULT_VERTICAL_TEMPLATE
|
|
else:
|
|
raise
|
|
logger.info(f"[select_template] orientation={orientation}, industry={industry}, project_id={project_id}, template_id={template_id}")
|
|
return template_id
|
|
|
|
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",
|
|
industry: str | None = None,
|
|
project_id: int | None = None,
|
|
):
|
|
"""
|
|
Args:
|
|
api_key: Creatomate API 키 (Bearer token으로 사용)
|
|
None일 경우 config에서 자동으로 가져옴
|
|
orientation: 영상 방향 ("horizontal" 또는 "vertical", 기본값: "vertical")
|
|
industry: 업종 분류 (stay|restaurant|cafe|attraction 등). 세로형 템플릿 선택에 사용되며,
|
|
매핑에 있으면 전용 템플릿을 사용
|
|
project_id: 미매핑 업종(general 등)의 세로형 분배용 키. 매핑에 없을 때
|
|
project_id % len(VST_LIST)로 결정적 선택
|
|
"""
|
|
self.api_key = api_key or apikey_settings.CREATOMATE_API_KEY
|
|
self.orientation = orientation
|
|
|
|
# orientation·industry에 따른 템플릿 설정 가져오기
|
|
self.template_id = select_template(orientation, industry=industry, project_id=project_id)
|
|
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_url": str, "image_tag": dict}] — 이미 마케팅 적합성 필터를 통과한 이미지만 전달할 것
|
|
music_url: str,
|
|
address: str,
|
|
duplicate: bool = False
|
|
) -> tuple[dict, dict]:
|
|
"""템플릿 슬롯에 이미지를 배정합니다.
|
|
|
|
Returns:
|
|
(modifications, assigned) 튜플
|
|
- modifications: {슬롯명: image_url, "audio-music": music_url, ...} — modify_element에 전달
|
|
- assigned: {슬롯명: image_tag} — 자막 생성 컨텍스트에 전달 (modify_element에 넣지 않음)
|
|
|
|
배정 전략 (duplicate=False):
|
|
- 순수 greedy pop 대신 "난이도 우선 greedy" 적용
|
|
- 사전에 전체 슬롯 점수를 계산해 최고 달성 점수가 낮은(=까다로운) 슬롯을 먼저 배정
|
|
- 이미지 배정 직전 점수를 재계산 (pop 이후 인덱스 변동 대응)
|
|
|
|
배정 전략 (duplicate=True, 이미지 < 슬롯):
|
|
- 1단계(최소 커버리지 보장): 이미지별로 가장 잘 맞는 슬롯을 찾아 pool의 모든 이미지를
|
|
서로 다른 슬롯에 먼저 배정한다 (이미지 수 <= 슬롯 수이므로 전량 배정 가능).
|
|
슬롯을 먼저 소진시키는 것이 아니라 "이미지"를 기준으로 난이도 우선 배정하여,
|
|
특정 이미지 몇 장이 여러 슬롯을 독점하는 것을 방지한다.
|
|
- 2단계: 1단계에서 못 채운 나머지 슬롯은 기존처럼 pool 전체에서 최고점 이미지를
|
|
배정한다 (여기서부터는 중복 재사용 불가피).
|
|
|
|
Note:
|
|
marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되므로 여기서는 재필터링하지 않는다.
|
|
taged_image_list를 그대로 pool로 사용한다.
|
|
"""
|
|
source_elements = template["source"]["elements"]
|
|
template_component_data = self.parse_template_component_name(source_elements)
|
|
|
|
pool = taged_image_list
|
|
modifications: dict = {}
|
|
assigned: dict = {} # {슬롯명: image_tag} — 자막 컨텍스트용
|
|
|
|
# 이미지 슬롯과 텍스트 슬롯 분리
|
|
image_slots = [name for name, t in template_component_data.items() if t == "image"]
|
|
text_slots = [(name, t) for name, t in template_component_data.items() if t == "text"]
|
|
|
|
if not pool:
|
|
logger.warning("[template_matching_taged_image] 태그된 이미지가 없음 — 이미지 슬롯 배정 불가")
|
|
elif duplicate:
|
|
# 이미지 부족(슬롯 수 > 이미지 수)
|
|
# Step 1: 최소 커버리지 보장 — 이미지마다 가장 잘 맞는 슬롯을 찾아 서로 다른 슬롯에 배정
|
|
remaining_slots = list(image_slots)
|
|
|
|
def _best_slot_for_image(image: dict, slots: list) -> tuple[str | None, float]:
|
|
best_slot, best_score = None, -1.0
|
|
for slot in slots:
|
|
score = self.calculate_image_slot_score_multi([image], slot)[0]
|
|
if score > best_score:
|
|
best_slot, best_score = slot, score
|
|
return best_slot, best_score
|
|
|
|
# 사전 점수 계산 (최고 달성 점수가 낮을수록 배정처가 마땅찮은=까다로운 이미지)
|
|
prelim_best_score = [
|
|
_best_slot_for_image(image, remaining_slots)[1] for image in pool
|
|
]
|
|
ordered_indices = sorted(
|
|
range(len(pool)),
|
|
key=lambda i: prelim_best_score[i]
|
|
)
|
|
|
|
for idx in ordered_indices:
|
|
if not remaining_slots:
|
|
break
|
|
image = pool[idx]
|
|
# 슬롯이 이전 배정으로 줄어들었으므로 현재 remaining_slots 기준 재계산
|
|
best_slot, _ = _best_slot_for_image(image, remaining_slots)
|
|
if best_slot is None:
|
|
continue
|
|
modifications[best_slot] = image["image_url"]
|
|
assigned[best_slot] = image["image_tag"]
|
|
remaining_slots.remove(best_slot)
|
|
|
|
# Step 2: 커버리지 배정 후 남은 슬롯은 pool 전체에서 최고점 이미지 배정 (재사용 불가피)
|
|
for slot in remaining_slots:
|
|
scores = self.calculate_image_slot_score_multi(pool, slot)
|
|
if not scores:
|
|
continue
|
|
best_idx = scores.index(max(scores))
|
|
selected = pool[best_idx]
|
|
modifications[slot] = selected["image_url"]
|
|
assigned[slot] = selected["image_tag"]
|
|
else:
|
|
# 이미지 충분(슬롯 수 <= 이미지 수): 난이도 우선 greedy 배정
|
|
# Step 1: 정렬용 사전 점수 계산 (최고 달성 점수가 낮을수록 배정이 어려운 슬롯)
|
|
prelim = {slot: self.calculate_image_slot_score_multi(pool, slot) for slot in image_slots}
|
|
ordered_slots = sorted(
|
|
image_slots,
|
|
key=lambda s: max(prelim[s]) if prelim[s] else 0.0
|
|
# 오름차순 → 달성 최대값이 낮은(어려운) 슬롯이 앞으로
|
|
)
|
|
|
|
# Step 2: 난이도 순서로 배정, 배정마다 pool에서 pop + 다음 슬롯은 점수 재계산
|
|
for slot in ordered_slots:
|
|
if not pool:
|
|
logger.warning(f"[template_matching_taged_image] 이미지 풀 소진 — 남은 슬롯 배정 불가: {slot}")
|
|
break
|
|
# pop 후 인덱스 변동이 있으므로 현재 pool로 재계산 (사전 점수 재사용 금지)
|
|
scores = self.calculate_image_slot_score_multi(pool, slot)
|
|
if not scores:
|
|
continue
|
|
|
|
# 상위 3장 중 점수 비례 확률로 선택 (영상 다양화)
|
|
sorted_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)
|
|
top_indices = sorted_indices[:min(3, len(sorted_indices))]
|
|
weights = [scores[i] for i in top_indices]
|
|
if sum(weights) > 0:
|
|
chosen_pool_idx = random.choices(top_indices, weights=weights, k=1)[0]
|
|
else:
|
|
chosen_pool_idx = random.choice(top_indices)
|
|
|
|
selected = pool.pop(chosen_pool_idx)
|
|
modifications[slot] = selected["image_url"]
|
|
assigned[slot] = selected["image_tag"]
|
|
|
|
# 텍스트 슬롯 처리
|
|
for name, _ in text_slots:
|
|
if "address_input" in name:
|
|
modifications[name] = address
|
|
|
|
modifications["audio-music"] = music_url
|
|
return modifications, assigned
|
|
|
|
# 점수 계산 상수 (모듈 레벨로 빼면 더 좋지만 기존 코드 스타일 따라 클래스 내부에 위치)
|
|
_NARR_FLOOR = 0.25 # narrative 최솟값 modulator: base 점수가 완전히 0이 되는 것을 방지
|
|
_NARR_BONUS = 1.5 # narrative 독립 보너스: base=0이어도 narrative 신호가 점수에 남음
|
|
_NARR_MIN = 0.0 # narrative_preference 값 clamp 하한 (무경계 스키마 방어)
|
|
_NARR_MAX = 100.0 # narrative_preference 값 clamp 상한. NarrativePreference 스키마(0.0~100.0)와 스케일 일치
|
|
_SPACE_TYPE_MISMATCH_PENALTY = 0.1 # 슬롯이 요구하는 space_type과 다른 이미지에 적용하는 페널티 배수 (하드 필터에 가깝게)
|
|
|
|
def calculate_image_slot_score_multi(self, taged_image_list : list[dict], slot_name : str):
|
|
"""이미지 슬롯에 대한 각 이미지의 매칭 점수를 계산합니다.
|
|
|
|
점수 = (base * narr_mod + NARR_BONUS * narr) * space_type_penalty
|
|
- base: 태그 카테고리 매칭 합산 (최대 5.5)
|
|
- narr_mod: [NARR_FLOOR, 1.0] 구간으로 완화된 narrative 계수
|
|
- NARR_BONUS * narr: narrative 독립 보너스 (base=0이어도 신호 유지)
|
|
- space_type_penalty: 슬롯이 요구하는 space_type과 이미지가 불일치하면
|
|
SPACE_TYPE_MISMATCH_PENALTY(0.1)를 곱해 사실상 탈락시킴. narrative_preference
|
|
점수가 아무리 높아도(예: bathroom 이미지가 intro narr=100) 슬롯이 지정한
|
|
space_type(예: exterior_front)과 다르면 우선 배정되지 않도록 하기 위함 —
|
|
같은 space_type 이미지가 pool에 하나도 없을 때만 상대적으로 선택됨(fallback 유지).
|
|
이 구조로 "base=0, narrative가 높음"과 "완전히 틀린 이미지(0,0)"를 구별할 수 있음.
|
|
"""
|
|
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)
|
|
if slot_tag_dict is None:
|
|
# parse 실패 시 전부 0점 반환 (슬롯 skip)
|
|
return [0.0] * len(image_tag_list)
|
|
|
|
base_score_list = [0.0] * len(image_tag_list)
|
|
# slot_tag_narrative = NarrativePhase.accent # 기본값 (narrative 토큰 없는 슬롯 대비)
|
|
slot_space_type = slot_tag_dict.get("space_type")
|
|
space_type_match_list = [
|
|
slot_space_type is not None and slot_space_type.value in image_tag.get("space_type", [])
|
|
for image_tag in 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.0
|
|
case "subject":
|
|
weight = 2.0
|
|
case "camera":
|
|
weight = 1.0
|
|
case "motion_recommended":
|
|
weight = 0.5
|
|
case _:
|
|
raise ValueError(f"[calculate_image_slot_score_multi] unknown slot tag category: {slot_tag_cate}")
|
|
|
|
for idx, image_tag in enumerate(image_tag_list):
|
|
if slot_tag_item.value in image_tag[slot_tag_cate]: # collect!
|
|
base_score_list[idx] += weight
|
|
|
|
# narrative 점수 적용: 순수 곱셈 대신 "완화 modulator + 독립 보너스"
|
|
# - narr_mod: base를 흐릴 수 있지만 NARR_FLOOR 이하로는 안 떨어짐
|
|
# - NARR_BONUS * narr: base=0이어도 narrative가 높으면 양수 점수 확보
|
|
image_score_list = []
|
|
for idx, image_tag in enumerate(image_tag_list):
|
|
raw_narr = image_tag.get("narrative_preference", {}).get(slot_tag_narrative, 0.0)
|
|
clamped_narr = min(self._NARR_MAX, max(self._NARR_MIN, float(raw_narr)))
|
|
narr = clamped_narr / self._NARR_MAX # 0.0~1.0로 정규화
|
|
narr_mod = self._NARR_FLOOR + (1.0 - self._NARR_FLOOR) * narr
|
|
score = base_score_list[idx] * narr_mod + self._NARR_BONUS * narr
|
|
if not space_type_match_list[idx]:
|
|
score *= self._SPACE_TYPE_MISMATCH_PENALTY
|
|
image_score_list.append(score)
|
|
|
|
return image_score_list
|
|
|
|
def parse_slot_name_to_tag(self, slot_name: str) -> dict[str, StrEnum] | None:
|
|
"""슬롯 이름을 파싱하여 태그 딕셔너리를 반환합니다.
|
|
|
|
슬롯 이름 형식: {space_type}-{subject}-{camera}-{motion}-{narrative}
|
|
파싱 실패 시 None을 반환합니다 (호출자가 해당 슬롯을 skip+log 처리).
|
|
"""
|
|
try:
|
|
tag_list = slot_name.split("-")
|
|
if len(tag_list) < 5:
|
|
logger.warning(f"[parse_slot_name_to_tag] 슬롯명 토큰 부족 ({len(tag_list)}개): '{slot_name}' — 슬롯 skip")
|
|
return None
|
|
space_type = SpaceType(tag_list[0])
|
|
subject = Subject(tag_list[1])
|
|
camera = Camera(tag_list[2])
|
|
# 모션 동의어(zoom_in, pan_left 등)는 표준 enum 값으로 정규화 후 변환
|
|
motion = MOTION_TOKEN_NORMALIZATION.get(tag_list[3]) or 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
|
|
except (ValueError, IndexError) as e:
|
|
logger.warning(f"[parse_slot_name_to_tag] 슬롯명 파싱 실패: '{slot_name}' — {e} — 슬롯 skip")
|
|
return None
|
|
|
|
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"
|
|
|
|
source["frame_rate"] = 30
|
|
|
|
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()
|
|
|
|
@staticmethod
|
|
def _entrance_transition_offset(elem: dict) -> float:
|
|
"""요소의 entrance transition(transition=true, time==0) duration 합을 반환합니다.
|
|
|
|
순차 배치 시 이 값만큼 이전 요소와 겹치도록 시작점이 앞당겨집니다.
|
|
"""
|
|
offset = 0.0
|
|
for animation in elem.get("animations", []):
|
|
if "reversed" in animation:
|
|
continue
|
|
if "transition" in animation and animation["transition"] and animation.get("time", 0) == 0:
|
|
offset += animation["duration"]
|
|
return offset
|
|
|
|
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"] - self._entrance_transition_offset(elem)
|
|
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 calc_track_time_ranges(self, template: dict, track: int) -> list[tuple[str, float, float]]:
|
|
"""최상위 요소 중 지정 track의 (name, start, end) 시간 범위를 계산합니다.
|
|
|
|
Creatomate 규칙에 따라 `time`이 없거나 0인 요소는 같은 트랙의 이전 요소 뒤에
|
|
순차 배치되며, entrance transition duration만큼 이전 요소와 겹치도록
|
|
시작점을 앞당깁니다(calc_scene_duration의 차감 규칙과 동치).
|
|
`time`이 명시된 요소는 해당 시각에서 시작하고, 커서는 그 요소의 끝으로 이동합니다.
|
|
"""
|
|
ranges: list[tuple[str, float, float]] = []
|
|
cursor = 0.0
|
|
for elem in template["source"]["elements"]:
|
|
try:
|
|
if elem.get("track") != track:
|
|
continue
|
|
duration = elem["duration"]
|
|
|
|
if "time" not in elem or elem["time"] == 0: # 순차 배치(auto)
|
|
start = max(0.0, cursor - self._entrance_transition_offset(elem))
|
|
else:
|
|
start = elem["time"]
|
|
|
|
end = start + duration
|
|
cursor = end
|
|
ranges.append((elem.get("name", ""), start, end))
|
|
except Exception as e:
|
|
logger.debug(traceback.format_exc())
|
|
logger.error(f"[calc_track_time_ranges] Error processing element: {elem}, {e}")
|
|
|
|
return ranges
|
|
|
|
@staticmethod
|
|
def _scale_element(elem: dict, extend_rate: float) -> None:
|
|
"""element 하나의 time/duration/animations를 extend_rate만큼 스케일링합니다.
|
|
|
|
composition의 자식 element까지 재귀적으로 처리합니다. 재귀하지 않으면
|
|
부모 컴포지션(씬 슬롯)만 늘어나고 내부 leaf 이미지/애니메이션은 원래
|
|
길이에서 끝나버려, 슬롯이 끝나기 전 컴포지션의 fill_color(회색)가
|
|
노출되는 버그가 생긴다 — modify_element/parse_template_component_name과
|
|
동일하게 composition을 재귀 순회해야 함.
|
|
|
|
오디오 판별은 track 번호가 아닌 type == "audio"로 한다. composition
|
|
내부의 track 번호는 그 컴포지션 안에서만 유효한 로컬 레이어 인덱스라
|
|
전역 AUDIO_TRACK/KEYWORD_TRACK 상수와 우연히 값이 겹칠 수 있다
|
|
(실제 템플릿에도 다중 레이어 씬의 2번째/3번째 이미지가 track=2, track=4인
|
|
경우가 존재함).
|
|
"""
|
|
if elem.get("type") == "audio":
|
|
return
|
|
|
|
# "time"은 숫자 오프셋 대신 "end"(컴포지션 끝에 고정) 같은 특수 키워드일 수 있음 —
|
|
# 이런 값은 렌더 시점에 (이미 스케일링된) duration 기준으로 재계산되므로 그대로 둔다.
|
|
if "time" in elem and isinstance(elem["time"], (int, float)):
|
|
elem["time"] = elem["time"] * extend_rate
|
|
if "duration" in elem and isinstance(elem["duration"], (int, float)):
|
|
elem["duration"] = elem["duration"] * extend_rate
|
|
|
|
for animation in elem.get("animations", []):
|
|
anim_time = animation.get("time", 0)
|
|
if isinstance(anim_time, (int, float)) and anim_time != 0:
|
|
animation["time"] = anim_time * extend_rate
|
|
if isinstance(animation.get("duration"), (int, float)):
|
|
animation["duration"] = animation["duration"] * extend_rate
|
|
|
|
if elem.get("type") == "composition":
|
|
for child in elem.get("elements", []):
|
|
CreatomateService._scale_element(child, extend_rate)
|
|
|
|
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["track"] == AUDIO_TRACK : # audio track(최상위 음악 트랙)은 패스
|
|
continue
|
|
|
|
self._scale_element(elem, 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
|
|
|
|
|
|
def make_thumbnail_modification(self, brand_name : str, region : str, category_definition : str, target_keywords : list[str]):
|
|
|
|
len_keywords = len(target_keywords) if len(target_keywords) < 3 else 3
|
|
|
|
hashtaged_target_keywords = [f"#{tk}" for tk in target_keywords[:len_keywords]]
|
|
|
|
mod_dict = {
|
|
"thumb-hashtag-primary" : ' '.join(hashtaged_target_keywords),
|
|
"thumb-headline-brand_name-factual" : brand_name,
|
|
"thumb-subheadline-local_info-factual" : normalize_location(region),
|
|
"thumb-badge-category" : category_definition,
|
|
}
|
|
return mod_dict
|