292 lines
13 KiB
Python
292 lines
13 KiB
Python
"""
|
|
Creative Assets Background Tasks
|
|
|
|
영상 생성 전 사전 준비(이미지 배정 + 자막 생성)를 수행하는 백그라운드 태스크를 정의합니다.
|
|
"""
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.database.session import BackgroundSessionLocal
|
|
from app.home.models import Project, MarketingIntel
|
|
from app.utils.subtitles import SubtitleContentsGenerator
|
|
from app.utils.creatomate import (
|
|
CreatomateService,
|
|
SCENE_TRACK,
|
|
SUBTITLE_TRACK,
|
|
KEYWORD_TRACK,
|
|
)
|
|
from app.utils.logger import get_logger
|
|
from app.video.services.video import get_image_tags_by_task_id
|
|
from app.utils.prompts.schemas.image import NARRATIVE_PHASE_LABEL
|
|
|
|
# 로거 설정
|
|
logger = get_logger("video")
|
|
|
|
|
|
async def generate_subtitle_background(
|
|
orientation: str,
|
|
task_id: str,
|
|
max_retries: int = 3,
|
|
) -> None:
|
|
"""자막 + 이미지 배정을 사전에 수행하는 백그라운드 태스크.
|
|
|
|
내부적으로 `generate_creative_assets_background`를 호출한다.
|
|
"""
|
|
await generate_creative_assets_background(
|
|
orientation=orientation,
|
|
task_id=task_id,
|
|
max_retries=max_retries,
|
|
)
|
|
|
|
|
|
async def generate_creative_assets_background(
|
|
orientation: str,
|
|
task_id: str,
|
|
max_retries: int = 3,
|
|
) -> None:
|
|
"""이미지 배정 + 자막 생성을 사전에 수행하는 백그라운드 태스크.
|
|
|
|
실행 순서:
|
|
1. 템플릿 조회 (읽기 전용, 렌더 없음)
|
|
2. 이미지 매칭 — get_image_tags_by_task_id → template_matching_taged_image
|
|
→ MarketingIntel.image_match 저장
|
|
3. 자막 생성 — 배정된 이미지 컨텍스트(슬롯별 공간/단계 정보)를 pitching 리스트에 포함
|
|
→ MarketingIntel.subtitle 저장
|
|
"""
|
|
logger.info(f"[generate_creative_assets_background] START - task_id: {task_id}, orientation: {orientation}")
|
|
|
|
for attempt in range(1, max_retries + 1):
|
|
try:
|
|
# ── 프로젝트 / 마케팅 인텔 로드 ──────────────────────────────────
|
|
# 템플릿 선택에 project.industry가 필요하므로 서비스 생성보다 먼저 로드
|
|
async with BackgroundSessionLocal() as session:
|
|
project_result = await session.execute(
|
|
select(Project)
|
|
.where(Project.task_id == task_id)
|
|
.order_by(Project.created_at.desc())
|
|
.limit(1)
|
|
)
|
|
project = project_result.scalar_one_or_none()
|
|
marketing_result = await session.execute(
|
|
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
|
)
|
|
marketing_intelligence = marketing_result.scalar_one_or_none()
|
|
|
|
creatomate_service = CreatomateService(orientation=orientation, industry=project.industry, project_id=project.id)
|
|
template = await creatomate_service.get_one_template_data(creatomate_service.template_id)
|
|
|
|
store_address = project.detail_region_info
|
|
customer_name = project.store_name
|
|
language = project.language or "Korean"
|
|
logger.info(
|
|
f"[generate_creative_assets_background] customer: {customer_name}, "
|
|
f"address: {store_address}, language: {language} - task_id: {task_id}"
|
|
)
|
|
|
|
# ── Step 1: 이미지 매칭 ─────────────────────────────────────────
|
|
# marketing_acceptable 필터링은 크롤링 단계에서 이미 완료되었으므로 재필터링하지 않는다.
|
|
taged_image_list = await get_image_tags_by_task_id(task_id)
|
|
min_image_num = creatomate_service.counting_component(
|
|
template=template,
|
|
target_template_type="image",
|
|
)
|
|
duplicate = len(taged_image_list) < min_image_num
|
|
logger.info(
|
|
f"[generate_creative_assets_background] image matching — "
|
|
f"pool={len(taged_image_list)}, slots={min_image_num}, "
|
|
f"duplicate={duplicate} - task_id: {task_id}"
|
|
)
|
|
|
|
image_modifications, assigned_image_tags = creatomate_service.template_matching_taged_image(
|
|
template=template,
|
|
taged_image_list=taged_image_list,
|
|
music_url="", # 음악 URL은 영상 생성 시점에 결정 — 사전 단계에서는 빈 값
|
|
address=store_address,
|
|
duplicate=duplicate,
|
|
)
|
|
# audio-music은 영상 생성 시점에 덮어씌워지므로 image_match에서 제거
|
|
image_match = {k: v for k, v in image_modifications.items() if k != "audio-music"}
|
|
|
|
logger.info(f"[generate_creative_assets_background] image_match: {list(image_match.keys())} - task_id: {task_id}")
|
|
|
|
# image_match 저장
|
|
async with BackgroundSessionLocal() as session:
|
|
marketing_result = await session.execute(
|
|
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
|
)
|
|
mi = marketing_result.scalar_one_or_none()
|
|
mi.image_match = image_match
|
|
await session.commit()
|
|
|
|
logger.info(f"[generate_creative_assets_background] image_match saved - task_id: {task_id}")
|
|
|
|
# ── Step 2: 자막 생성 (이미지 컨텍스트 포함) ──────────────────────
|
|
# 슬롯별 배정된 이미지의 공간/내러티브 정보를 pitching 레이블에 포함하여
|
|
# LLM이 "이 자막이 나올 때 화면에 보이는 이미지"를 인지하고 자막을 작성할 수 있게 함.
|
|
pitchings_raw = creatomate_service.extract_text_format_from_template(template)
|
|
pitchings_with_context = _build_pitching_list_with_image_context(
|
|
pitching_label_list=pitchings_raw,
|
|
assigned_image_tags=assigned_image_tags,
|
|
template=template,
|
|
creatomate_service=creatomate_service,
|
|
)
|
|
|
|
subtitle_generator = SubtitleContentsGenerator()
|
|
generated_subtitles = await subtitle_generator.generate_subtitle_contents(
|
|
marketing_intelligence=marketing_intelligence.intel_result,
|
|
pitching_label_list=pitchings_with_context,
|
|
customer_name=customer_name,
|
|
detail_region_info=store_address,
|
|
language=language,
|
|
industry=project.industry,
|
|
)
|
|
pitching_output_list = generated_subtitles.pitching_results
|
|
|
|
# LLM이 pitching_tag에 " | 화면: ..." 컨텍스트 접미를 그대로 되돌려줄 수 있으므로 제거
|
|
subtitle_modifications = {
|
|
pitching_output.pitching_tag.split(IMAGE_CONTEXT_SEP)[0].strip(): pitching_output.pitching_data
|
|
for pitching_output in pitching_output_list
|
|
}
|
|
logger.info(f"[generate_creative_assets_background] subtitle_modifications: {subtitle_modifications}")
|
|
|
|
# subtitle 저장
|
|
async with BackgroundSessionLocal() as session:
|
|
marketing_result = await session.execute(
|
|
select(MarketingIntel).where(MarketingIntel.id == project.marketing_intelligence)
|
|
)
|
|
mi = marketing_result.scalar_one_or_none()
|
|
mi.subtitle = subtitle_modifications
|
|
await session.commit()
|
|
|
|
logger.info(
|
|
f"[generate_creative_assets_background] DONE - task_id: {task_id} "
|
|
f"(attempt {attempt}/{max_retries})"
|
|
)
|
|
return
|
|
|
|
except Exception as e:
|
|
logger.error(
|
|
f"[generate_creative_assets_background] FAILED (attempt {attempt}/{max_retries}) "
|
|
f"- task_id: {task_id}, error: {e}",
|
|
exc_info=True,
|
|
)
|
|
if attempt < max_retries:
|
|
logger.info(
|
|
f"[generate_creative_assets_background] 재시도 중... "
|
|
f"({attempt + 1}/{max_retries}) - task_id: {task_id}"
|
|
)
|
|
|
|
logger.error(f"[generate_creative_assets_background] 모든 재시도 실패 - task_id: {task_id}")
|
|
|
|
|
|
# 이미지 컨텍스트 접미 구분자 — 생성(_context_for_label)과 제거(subtitle_modifications) 양쪽에서 사용
|
|
IMAGE_CONTEXT_SEP = " | "
|
|
|
|
|
|
def _merge_unique(tags: list[dict], key: str) -> list:
|
|
"""태그들의 key 리스트를 순서 유지하며 중복 제거해 병합합니다."""
|
|
return list(dict.fromkeys(item for tag in tags for item in tag.get(key, [])))
|
|
|
|
|
|
def _aggregate_tags_by_composition(
|
|
template: dict,
|
|
assigned_image_tags: dict[str, dict],
|
|
creatomate_service: CreatomateService,
|
|
) -> dict[str, dict]:
|
|
"""SCENE_TRACK 최상위 컴포지션별로 배정된 이미지 태그를 집계합니다.
|
|
|
|
컴포지션 하위 leaf image가 여러 개면(다중 이미지 씬) space_type/subject를
|
|
순서 유지 중복 제거로 병합합니다. 배정된 태그가 없는 컴포지션(CTA 등)은
|
|
결과에 포함하지 않습니다.
|
|
"""
|
|
comp_tags: dict[str, dict] = {}
|
|
for elem in template["source"]["elements"]:
|
|
if elem.get("track") != SCENE_TRACK or not elem.get("name"):
|
|
continue
|
|
name_to_type = creatomate_service.parse_template_component_name([elem])
|
|
leaf_tags = [
|
|
assigned_image_tags[name]
|
|
for name, elem_type in name_to_type.items()
|
|
if elem_type == "image" and name in assigned_image_tags
|
|
]
|
|
if not leaf_tags:
|
|
continue
|
|
comp_tags[elem["name"]] = {
|
|
"space_type": _merge_unique(leaf_tags, "space_type"),
|
|
"subject": _merge_unique(leaf_tags, "subject"),
|
|
}
|
|
return comp_tags
|
|
|
|
|
|
def _context_for_label(
|
|
text_range: tuple[float, float],
|
|
scene_ranges: list[tuple[str, float, float]],
|
|
comp_tags: dict[str, dict],
|
|
) -> str:
|
|
"""텍스트 재생 구간과 시간 겹침이 최대인 씬의 이미지 컨텍스트 접미를 만듭니다.
|
|
|
|
태그가 배정된 컴포지션 중 겹침이 최대인 씬을 선택하며(동률 시 이른 시작 우선),
|
|
겹치는 씬이 없으면 빈 문자열을 반환합니다.
|
|
"""
|
|
text_start, text_end = text_range
|
|
candidates = [
|
|
(min(text_end, comp_end) - max(text_start, comp_start), -comp_start, comp_name)
|
|
for comp_name, comp_start, comp_end in scene_ranges
|
|
if comp_name in comp_tags
|
|
]
|
|
candidates = [c for c in candidates if c[0] > 0]
|
|
if not candidates:
|
|
return ""
|
|
|
|
best_comp = max(candidates)[2]
|
|
tag = comp_tags[best_comp]
|
|
space_types = tag.get("space_type", [])
|
|
subjects = tag.get("subject", [])
|
|
space_str = "/".join(space_types[:2]) if space_types else "?"
|
|
subject_str = subjects[0] if subjects else "?"
|
|
# 컴포지션명 형식: {space_type}-{subject}-{camera}-{motion}-{narrative_phase}
|
|
comp_tokens = best_comp.split("-")
|
|
phase = comp_tokens[4] if len(comp_tokens) >= 5 else ""
|
|
phase_str = NARRATIVE_PHASE_LABEL.get(phase, phase)
|
|
return f"{IMAGE_CONTEXT_SEP}화면: {space_str} / {subject_str} / {phase_str} 단계"
|
|
|
|
|
|
def _build_pitching_list_with_image_context(
|
|
pitching_label_list: list[str],
|
|
assigned_image_tags: dict[str, dict],
|
|
template: dict,
|
|
creatomate_service: CreatomateService,
|
|
) -> list[str]:
|
|
"""pitching 레이블 리스트에 배정 이미지 컨텍스트를 포함하여 반환합니다.
|
|
|
|
텍스트 레이어(SUBTITLE/KEYWORD_TRACK)와 이미지 컴포지션(SCENE_TRACK)의
|
|
시간 범위를 계산해, 각 텍스트가 재생되는 동안 화면에 가장 오래 보이는
|
|
(시간 겹침이 최대인) 컴포지션의 이미지 태그를 컨텍스트로 붙입니다.
|
|
narrative_phase 토큰 매칭과 달리 phase 이름이 어긋나거나(자막 outro vs
|
|
이미지 accent) 같은 phase에 씬이 여러 개여도 올바른 짝을 찾습니다.
|
|
|
|
겹치는 씬이 없거나 해당 씬에 배정된 이미지가 없으면(CTA 등) 컨텍스트를
|
|
생략합니다. 렌더 시 extend_template_duration은 전 트랙 균등 배율이므로
|
|
원본 템플릿 기준 겹침 관계는 스케일 후에도 유지됩니다.
|
|
|
|
출력 예시:
|
|
"subtitle-welcome-emotion_cue-empathic-001 | 화면: lobby / furniture / 진입(welcome) 단계"
|
|
"""
|
|
comp_tags = _aggregate_tags_by_composition(template, assigned_image_tags, creatomate_service)
|
|
if not comp_tags:
|
|
return pitching_label_list
|
|
|
|
scene_ranges = creatomate_service.calc_track_time_ranges(template, SCENE_TRACK)
|
|
text_ranges: dict[str, tuple[float, float]] = {}
|
|
for track in (SUBTITLE_TRACK, KEYWORD_TRACK):
|
|
for name, start, end in creatomate_service.calc_track_time_ranges(template, track):
|
|
text_ranges[name] = (start, end)
|
|
|
|
result = []
|
|
for label in pitching_label_list:
|
|
text_range = text_ranges.get(label)
|
|
context_str = _context_for_label(text_range, scene_ranges, comp_tags) if text_range else ""
|
|
result.append(label + context_str)
|
|
|
|
return result
|