1368 lines
59 KiB
Python
1368 lines
59 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
|
||
|
||
# 썸네일 컴포지션의 이미지 슬롯명 접미사 (8개 업종 템플릿 전수 확인, 예:
|
||
# "exterior_front-architecture_detail-wide_angle-slow_zoom_in-intro-core-9999").
|
||
# 일반 씬 이미지는 "-0001"~"-0030"류 4자리 순번을 쓰는 반면 썸네일만 "-9999"라
|
||
# 슬롯명 접미사만으로 안전하게 식별 가능(가로 템플릿처럼 썸네일 컴포지션이
|
||
# 없는 템플릿도 있음).
|
||
THUMBNAIL_SLOT_MARKER = "-9999"
|
||
|
||
|
||
def is_fixed_slot_name(name: str) -> bool:
|
||
"""이름이 '-fixed'로 끝나는 요소인지 판별합니다.
|
||
|
||
템플릿 명명 규칙상 마지막 토큰이 'fixed'인 요소(예: 'brand-cta-bi_logo-
|
||
factual-fixed', 'brand-cta-contact_phone-factual-fixed')는 회사 연락처
|
||
문구·로고 등 매 영상마다 절대 바뀌면 안 되는 고정 콘텐츠다. 이런 요소는
|
||
이미지 배정/자막 생성 대상에서 제외하고 템플릿에 이미 설정된 값을 그대로
|
||
보존해야 한다.
|
||
"""
|
||
if not name:
|
||
return False
|
||
return name.rsplit("-", 1)[-1] == "fixed"
|
||
|
||
# 언어별 대체 폰트 (Google Fonts — Creatomate 기본 지원).
|
||
# 템플릿 기본 폰트(GyeonggiTitleOTF/Pretendard)는 한글·라틴 전용이라 여기 있는
|
||
# 언어는 렌더 직전 전체 텍스트 폰트를 교체한다. 매핑에 없는 언어(Korean,
|
||
# English 등)는 템플릿 원본 폰트를 유지한다. 중국어는 간체(SC) 기준.
|
||
LANGUAGE_FONT_MAP = {
|
||
"Japanese": "Noto Sans JP",
|
||
"Chinese": "Noto Sans SC",
|
||
"Thai": "Noto Sans Thai",
|
||
"Vietnamese": "Noto Sans",
|
||
}
|
||
|
||
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:
|
||
"""target_template_type 요소 개수를 셉니다.
|
||
|
||
target_template_type이 "image"인 경우, 고정 자산('-fixed' 명명) 및
|
||
5토큰 명명 규칙을 따르지 않는 요소(고정 워터마크 등)는 콘텐츠 슬롯이
|
||
아니므로 카운트에서 제외합니다 (template_matching_taged_image의
|
||
image_slots 필터링과 동일 기준).
|
||
"""
|
||
source_elements = template["source"]["elements"]
|
||
template_component_data = self.parse_template_component_name(source_elements)
|
||
count = 0
|
||
|
||
for name, template_type in template_component_data.items():
|
||
if template_type != target_template_type:
|
||
continue
|
||
if target_template_type == "image" and (is_fixed_slot_name(name) or self.parse_slot_name_to_tag(name) is None):
|
||
continue
|
||
count += 1
|
||
return count
|
||
|
||
def _slot_scores_with_fitness(
|
||
self,
|
||
pool_subset: list[dict],
|
||
slot: str,
|
||
thumbnail_fitness_map: dict | None,
|
||
) -> list[float]:
|
||
"""슬롯 점수에 썸네일 픽셀 적합도를 결합합니다.
|
||
|
||
썸네일 슬롯(-9999)에 한해 태그 점수에 적합도를 반영한다:
|
||
- reject(저해상도/극단 가로형) → 0점 (하드 배제)
|
||
- 정상 → 태그 점수 × 적합도 배율(0.25~1.0)
|
||
- 적합도 데이터 없음(헤더 확보/파싱 실패) → 태그 점수 그대로(중립).
|
||
다운로드 실패를 배제로 오판해 풀 전체가 날아가는 것을 막기 위함.
|
||
일반 씬 슬롯은 태그 점수를 그대로 반환한다.
|
||
"""
|
||
scores = self.calculate_image_slot_score_multi(pool_subset, slot)
|
||
if not thumbnail_fitness_map or not slot.endswith(THUMBNAIL_SLOT_MARKER):
|
||
return scores
|
||
|
||
adjusted = []
|
||
for item, score in zip(pool_subset, scores):
|
||
fitness = thumbnail_fitness_map.get(item.get("image_url"))
|
||
if fitness is None:
|
||
adjusted.append(score)
|
||
elif fitness["reject"]:
|
||
adjusted.append(0.0)
|
||
else:
|
||
adjusted.append(score * fitness["score"])
|
||
return adjusted
|
||
|
||
def rank_thumbnail_candidates(
|
||
self,
|
||
template: dict,
|
||
taged_image_list: list,
|
||
thumbnail_fitness_map: dict | None = None,
|
||
top_n: int = 3,
|
||
) -> dict[str, list[dict]]:
|
||
"""썸네일 슬롯별로 상위 후보(태그×픽셀 점수 내림차순)를 반환합니다.
|
||
|
||
비전 LLM 최종 선택(Phase 2)에 넘길 후보 압축용. 읽기 전용 — pool을
|
||
변형하지 않으며 배정도 하지 않는다. 점수 0 이하(픽셀 reject 포함)는
|
||
제외한다.
|
||
|
||
Returns:
|
||
{슬롯명: [pool item, ...]} — 슬롯당 최대 top_n개, 점수 내림차순.
|
||
후보가 없는 슬롯은 키 자체를 포함하지 않는다.
|
||
"""
|
||
component = self.parse_template_component_name(template["source"]["elements"])
|
||
thumbnail_slots = [
|
||
name for name, t in component.items()
|
||
if t == "image" and name.endswith(THUMBNAIL_SLOT_MARKER)
|
||
and not is_fixed_slot_name(name) and self.parse_slot_name_to_tag(name) is not None
|
||
]
|
||
result: dict[str, list[dict]] = {}
|
||
for slot in thumbnail_slots:
|
||
scores = self._slot_scores_with_fitness(taged_image_list, slot, thumbnail_fitness_map)
|
||
ranked = sorted(
|
||
range(len(scores)), key=lambda i: scores[i], reverse=True
|
||
)
|
||
candidates = [taged_image_list[i] for i in ranked if scores[i] > 0][:top_n]
|
||
if candidates:
|
||
result[slot] = candidates
|
||
return result
|
||
|
||
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,
|
||
thumbnail_fitness_map: dict | None = None, # {image_url: {"score", "reject"}} — 썸네일 슬롯 픽셀 적합도
|
||
thumbnail_choice: dict | None = None, # {슬롯명: image_url} — 비전 LLM 최종 선택(있으면 결정론 최고점 대신 사용)
|
||
) -> 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} — 자막 컨텍스트용
|
||
|
||
# 이미지 슬롯과 텍스트 슬롯 분리
|
||
# 고정 자산(이름이 '-fixed'로 끝나는 로고 등) 및 5토큰 명명 규칙을 따르지
|
||
# 않는 image 요소는 콘텐츠 슬롯이 아니므로 배정 대상에서 제외 — 그렇지
|
||
# 않으면 파싱 실패로 0점 처리되어 "가장 까다로운 슬롯"으로 취급되고
|
||
# 무작위 이미지로 덮어써진다.
|
||
image_slots = [
|
||
name for name, t in template_component_data.items()
|
||
if t == "image" and not is_fixed_slot_name(name) and self.parse_slot_name_to_tag(name) is not None
|
||
]
|
||
text_slots = [(name, t) for name, t in template_component_data.items() if t == "text"]
|
||
|
||
# ── 썸네일 슬롯(-9999) 우선 배정 ─────────────────────────────────
|
||
# 썸네일은 영상에서 가장 노출이 큰 표면이므로 씬 슬롯과 다르게 취급한다:
|
||
# 1) 씬 슬롯이 pool에서 좋은 컷을 pop해 가기 전에 먼저 배정 (같은 태그
|
||
# 계열 씬 슬롯이 여러 개면 썸네일 차례에 적합 컷이 소진되는 문제 방지)
|
||
# 2) "상위 3장 가중 랜덤" 없이 결정론적 최고점 선택 — 랜덤 다양화가
|
||
# 저점수 컷(예: space_type 불일치 ×0.1 페널티 컷)을 확률적으로
|
||
# 썸네일에 앉히는 사고 방지
|
||
# duplicate(이미지 부족) 시에는 pop하지 않아 씬 커버리지를 깎지 않는다.
|
||
# thumbnail_choice(비전 LLM 최종 선택)가 해당 슬롯에 있으면 그 이미지를,
|
||
# 없으면(선택 실패/미제공) 결정론적 최고점 컷을 사용한다 — 폴백 안전.
|
||
thumbnail_choice = thumbnail_choice or {}
|
||
thumbnail_slots = [s for s in image_slots if s.endswith(THUMBNAIL_SLOT_MARKER)]
|
||
image_slots = [s for s in image_slots if not s.endswith(THUMBNAIL_SLOT_MARKER)]
|
||
for slot in thumbnail_slots:
|
||
if not pool:
|
||
logger.warning(f"[template_matching_taged_image] 이미지 풀 없음 — 썸네일 슬롯 배정 불가: {slot}")
|
||
break
|
||
scores = self._slot_scores_with_fitness(pool, slot, thumbnail_fitness_map)
|
||
chosen_url = thumbnail_choice.get(slot)
|
||
chosen_idx = next(
|
||
(i for i, item in enumerate(pool) if item["image_url"] == chosen_url), None
|
||
) if chosen_url else None
|
||
if chosen_idx is not None:
|
||
sel_idx, sel_source = chosen_idx, "vision"
|
||
else:
|
||
sel_idx, sel_source = max(range(len(scores)), key=lambda i: scores[i]), "결정론"
|
||
selected = pool[sel_idx] if duplicate else pool.pop(sel_idx)
|
||
modifications[slot] = selected["image_url"]
|
||
assigned[slot] = selected["image_tag"]
|
||
logger.info(
|
||
f"[template_matching_taged_image] 썸네일 배정({sel_source}) — slot: {slot}, "
|
||
f"score: {scores[sel_idx]:.3f}, url: {selected['image_url']}"
|
||
)
|
||
|
||
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._slot_scores_with_fitness([image], slot, thumbnail_fitness_map)[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._slot_scores_with_fitness(pool, slot, thumbnail_fitness_map)
|
||
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._slot_scores_with_fitness(pool, slot, thumbnail_fitness_map) 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._slot_scores_with_fitness(pool, slot, thumbnail_fitness_map)
|
||
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에 따라 수정합니다.
|
||
|
||
modification에 없는 image/text 요소(예: '-fixed' 명명의 고정 로고·
|
||
연락처 문구처럼 배정/자막 생성 대상에서 제외된 고정 콘텐츠)는 템플릿에
|
||
이미 설정된 source/text를 그대로 보존한다 (KeyError·빈 문자열로
|
||
렌더가 깨지지 않도록).
|
||
"""
|
||
|
||
def recursive_modify(element: dict) -> None:
|
||
if "name" in element:
|
||
match element["type"]:
|
||
case "image":
|
||
element["source"] = modification.get(element["name"], element.get("source"))
|
||
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"], element.get("text", ""))
|
||
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 apply_language_font(self, template: dict, language: str) -> dict:
|
||
"""출력 언어에 맞춰 템플릿 내 모든 텍스트 요소의 폰트를 교체합니다.
|
||
|
||
템플릿 기본 폰트(GyeonggiTitleOTF/Pretendard)는 한글·라틴 전용이라
|
||
CJK/태국어 글리프가 없다. LANGUAGE_FONT_MAP에 있는 언어는 전체 텍스트
|
||
폰트를 해당 언어 지원 폰트(Google Fonts)로 교체하고, 매핑에 없는
|
||
언어(Korean, English 등)는 템플릿 원본 폰트를 유지한다.
|
||
"""
|
||
font_family = LANGUAGE_FONT_MAP.get(language)
|
||
if not font_family:
|
||
return template
|
||
|
||
def recursive_apply(element: dict) -> None:
|
||
if element.get("type") == "text":
|
||
element["font_family"] = font_family
|
||
for minor in element.get("elements") or []:
|
||
recursive_apply(minor)
|
||
|
||
for elem in template["source"]["elements"]:
|
||
recursive_apply(elem)
|
||
|
||
logger.info(
|
||
f"[apply_language_font] 텍스트 폰트 교체 완료 - language: {language}, font: {font_family}"
|
||
)
|
||
return template
|
||
|
||
def extract_text_format_from_template(self, template:dict):
|
||
keyword_list = []
|
||
subtitle_list = []
|
||
for elem in template["source"]["elements"]:
|
||
try: #최상위 내 텍스트만 검사
|
||
if elem["type"] == "text" and not is_fixed_slot_name(elem["name"]):
|
||
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.")
|
||
|
||
# 썸네일 텍스트(thumb-*)는 전용 트랙이 아닌 thumbnail-composition 내부에
|
||
# 중첩되어 있어 위 최상위 트랙 스캔에 잡히지 않는다. 다국어 자막 생성에
|
||
# 함께 포함되도록 재귀 파싱으로 수집한다.
|
||
component_data = self.parse_template_component_name(
|
||
template["source"]["elements"]
|
||
)
|
||
thumbnail_list = [
|
||
name
|
||
for name, elem_type in component_data.items()
|
||
if elem_type == "text" and name.startswith("thumb-")
|
||
]
|
||
|
||
pitching_list = keyword_list + subtitle_list + thumbnail_list
|
||
return pitching_list
|
||
|
||
|
||
def make_thumbnail_modification(self, brand_name : str, region : str, category_definition : str, target_keywords : list[str], detail_region_info : 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]]
|
||
|
||
# 지역 표기: 상세주소의 공백 기준 앞 두 마디 사용
|
||
# (예: "전북 군산시 절골길 18" → "전북 군산시").
|
||
# 상세주소가 없으면 기존 region 정규화 표기로 폴백.
|
||
if detail_region_info and detail_region_info.strip():
|
||
region_display = " ".join(detail_region_info.split()[:2])
|
||
else:
|
||
region_display = normalize_location(region)
|
||
|
||
mod_dict = {
|
||
"thumb-hashtag-primary" : ' '.join(hashtaged_target_keywords),
|
||
"thumb-headline-brand_name-factual" : brand_name,
|
||
"thumb-subheadline-local_info-factual" : region_display,
|
||
"thumb-badge-category" : category_definition,
|
||
}
|
||
return mod_dict
|