feat(v3): PR 5 — caption templates (3 styles) + MotionEngine (7 patterns)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -20,6 +20,105 @@ logger = logging.getLogger(__name__)
|
|||||||
|
|
||||||
BASE_DIR = Path(__file__).parent.parent.parent
|
BASE_DIR = Path(__file__).parent.parent.parent
|
||||||
|
|
||||||
|
CAPTION_TEMPLATES = {
|
||||||
|
'hormozi': {
|
||||||
|
'font_size': 64,
|
||||||
|
'highlight_color': '#FFD700',
|
||||||
|
'animation': 'pop_in',
|
||||||
|
'position': 'center',
|
||||||
|
'outline_width': 4,
|
||||||
|
'auto_emoji': False,
|
||||||
|
},
|
||||||
|
'tiktok_viral': {
|
||||||
|
'font_size': 56,
|
||||||
|
'highlight_color': '#FF6B6B',
|
||||||
|
'animation': 'bounce',
|
||||||
|
'auto_emoji': True,
|
||||||
|
'position': 'center_bottom',
|
||||||
|
},
|
||||||
|
'brand_4thpath': {
|
||||||
|
'font_size': 52,
|
||||||
|
'highlight_color': '#00D4FF',
|
||||||
|
'animation': 'typewriter',
|
||||||
|
'position': 'center',
|
||||||
|
'overlay_gradient': True,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
# Corner → caption template mapping
|
||||||
|
CORNER_CAPTION_MAP = {
|
||||||
|
'쉬운세상': 'hormozi',
|
||||||
|
'숨은보물': 'tiktok_viral',
|
||||||
|
'바이브리포트': 'hormozi',
|
||||||
|
'팩트체크': 'brand_4thpath',
|
||||||
|
'한컷': 'tiktok_viral',
|
||||||
|
'웹소설': 'brand_4thpath',
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def smart_line_break(text: str, max_chars: int = 18) -> list[str]:
|
||||||
|
"""
|
||||||
|
Break Korean text at semantic boundaries, not mid-word.
|
||||||
|
Never break before 조사 (particles) or 어미 (endings).
|
||||||
|
|
||||||
|
Returns list of line strings.
|
||||||
|
"""
|
||||||
|
# Common Korean particles/endings that should not start a new line
|
||||||
|
PARTICLES = ['은', '는', '이', '가', '을', '를', '의', '에', '에서', '으로', '로',
|
||||||
|
'과', '와', '도', '만', '까지', '부터', '보다', '처럼', '같이',
|
||||||
|
'한테', '에게', '이라', '라고', '이고', '이며', '고', '며', '면',
|
||||||
|
'이면', '이나', '나', '든지', '거나', '지만', '이지만', '지만',
|
||||||
|
'니까', '으니까', '이니까', '서', '아서', '어서', '며', '고']
|
||||||
|
|
||||||
|
if len(text) <= max_chars:
|
||||||
|
return [text] if text else []
|
||||||
|
|
||||||
|
lines = []
|
||||||
|
remaining = text
|
||||||
|
|
||||||
|
while len(remaining) > max_chars:
|
||||||
|
# Find best break point near max_chars
|
||||||
|
break_at = max_chars
|
||||||
|
|
||||||
|
# Look for space or punctuation near the limit
|
||||||
|
for i in range(max_chars, max(0, max_chars - 6), -1):
|
||||||
|
if i >= len(remaining):
|
||||||
|
continue
|
||||||
|
char = remaining[i]
|
||||||
|
prev_char = remaining[i-1] if i > 0 else ''
|
||||||
|
next_char = remaining[i+1] if i+1 < len(remaining) else ''
|
||||||
|
|
||||||
|
# Break at space
|
||||||
|
if char == ' ':
|
||||||
|
# Check if next word starts with a particle
|
||||||
|
next_word = remaining[i+1:i+4]
|
||||||
|
is_particle_start = any(next_word.startswith(p) for p in PARTICLES)
|
||||||
|
if not is_particle_start:
|
||||||
|
break_at = i
|
||||||
|
break
|
||||||
|
|
||||||
|
# Break after punctuation
|
||||||
|
if prev_char in ('。', ',', ',', '.', '!', '?', '~'):
|
||||||
|
break_at = i
|
||||||
|
break
|
||||||
|
|
||||||
|
lines.append(remaining[:break_at].strip())
|
||||||
|
remaining = remaining[break_at:].strip()
|
||||||
|
|
||||||
|
if remaining:
|
||||||
|
lines.append(remaining)
|
||||||
|
|
||||||
|
return [l for l in lines if l]
|
||||||
|
|
||||||
|
|
||||||
|
def get_template_for_corner(corner: str) -> dict:
|
||||||
|
"""
|
||||||
|
Get caption template config for a given content corner.
|
||||||
|
Falls back to 'hormozi' template if corner not in map.
|
||||||
|
"""
|
||||||
|
template_name = CORNER_CAPTION_MAP.get(corner, 'hormozi')
|
||||||
|
return CAPTION_TEMPLATES.get(template_name, CAPTION_TEMPLATES['hormozi'])
|
||||||
|
|
||||||
|
|
||||||
def _load_config() -> dict:
|
def _load_config() -> dict:
|
||||||
cfg_path = BASE_DIR / 'config' / 'shorts_config.json'
|
cfg_path = BASE_DIR / 'config' / 'shorts_config.json'
|
||||||
@@ -200,6 +299,7 @@ def render_captions(
|
|||||||
timestamp: str,
|
timestamp: str,
|
||||||
wav_duration: float = 0.0,
|
wav_duration: float = 0.0,
|
||||||
cfg: Optional[dict] = None,
|
cfg: Optional[dict] = None,
|
||||||
|
corner: str = '',
|
||||||
) -> Path:
|
) -> Path:
|
||||||
"""
|
"""
|
||||||
스크립트 + 단어별 타임스탬프 → ASS 자막 파일 생성.
|
스크립트 + 단어별 타임스탬프 → ASS 자막 파일 생성.
|
||||||
@@ -211,6 +311,7 @@ def render_captions(
|
|||||||
timestamp: 파일명 prefix
|
timestamp: 파일명 prefix
|
||||||
wav_duration: TTS 오디오 총 길이 (균등 분할 폴백용)
|
wav_duration: TTS 오디오 총 길이 (균등 분할 폴백용)
|
||||||
cfg: shorts_config.json dict
|
cfg: shorts_config.json dict
|
||||||
|
corner: content corner name (e.g. '쉬운세상') for template selection
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
ass_path
|
ass_path
|
||||||
@@ -222,6 +323,20 @@ def render_captions(
|
|||||||
ass_path = output_dir / f'{timestamp}.ass'
|
ass_path = output_dir / f'{timestamp}.ass'
|
||||||
|
|
||||||
cap_cfg = cfg.get('caption', {})
|
cap_cfg = cfg.get('caption', {})
|
||||||
|
|
||||||
|
# Apply corner-specific template overrides if corner is provided
|
||||||
|
if corner:
|
||||||
|
template = get_template_for_corner(corner)
|
||||||
|
# Override cfg caption section with template values
|
||||||
|
cap_cfg = dict(cap_cfg) # make a shallow copy to avoid mutating original
|
||||||
|
if 'font_size' in template:
|
||||||
|
cap_cfg['font_size'] = template['font_size']
|
||||||
|
if 'highlight_color' in template:
|
||||||
|
cap_cfg['highlight_color'] = template['highlight_color']
|
||||||
|
if 'outline_width' in template:
|
||||||
|
cap_cfg['outline_width'] = template['outline_width']
|
||||||
|
logger.info(f'[캡션] 코너 "{corner}" → 템플릿 적용: {template}')
|
||||||
|
|
||||||
max_chars = cap_cfg.get('max_chars_per_line_ko', 18)
|
max_chars = cap_cfg.get('max_chars_per_line_ko', 18)
|
||||||
highlight_color = cap_cfg.get('highlight_color', '#FFD700')
|
highlight_color = cap_cfg.get('highlight_color', '#FFD700')
|
||||||
default_color = cap_cfg.get('default_color', '#FFFFFF')
|
default_color = cap_cfg.get('default_color', '#FFFFFF')
|
||||||
@@ -235,8 +350,10 @@ def render_captions(
|
|||||||
wav_duration = 20.0
|
wav_duration = 20.0
|
||||||
timestamps = _build_uniform_timestamps(script, wav_duration)
|
timestamps = _build_uniform_timestamps(script, wav_duration)
|
||||||
|
|
||||||
# ASS 헤더
|
# ASS 헤더 (rebuild cfg with updated cap_cfg so header reflects template overrides)
|
||||||
header = _ass_header(cfg)
|
effective_cfg = dict(cfg)
|
||||||
|
effective_cfg['caption'] = cap_cfg
|
||||||
|
header = _ass_header(effective_cfg)
|
||||||
events = []
|
events = []
|
||||||
|
|
||||||
# 훅 이벤트 (첫 1.5초 중앙 표시)
|
# 훅 이벤트 (첫 1.5초 중앙 표시)
|
||||||
|
|||||||
@@ -0,0 +1,196 @@
|
|||||||
|
"""
|
||||||
|
bots/shorts/motion_engine.py
|
||||||
|
Motion pattern engine for video clips.
|
||||||
|
|
||||||
|
Applies one of 7 motion patterns to still images using FFmpeg.
|
||||||
|
Ensures no 2 consecutive clips use the same pattern.
|
||||||
|
|
||||||
|
Patterns:
|
||||||
|
1. ken_burns_in — slow zoom in
|
||||||
|
2. ken_burns_out — slow zoom out
|
||||||
|
3. pan_left — pan from right to left
|
||||||
|
4. pan_right — pan from left to right
|
||||||
|
5. parallax — layered depth effect (approximated)
|
||||||
|
6. rotate_slow — very slow rotation
|
||||||
|
7. glitch_reveal — glitch-style reveal
|
||||||
|
"""
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import random
|
||||||
|
import subprocess
|
||||||
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
PATTERNS = [
|
||||||
|
'ken_burns_in',
|
||||||
|
'ken_burns_out',
|
||||||
|
'pan_left',
|
||||||
|
'pan_right',
|
||||||
|
'parallax',
|
||||||
|
'rotate_slow',
|
||||||
|
'glitch_reveal',
|
||||||
|
]
|
||||||
|
|
||||||
|
# FFmpeg filter_complex expressions for each pattern
|
||||||
|
# Input: scale to 1120x1990 (slightly larger than 1080x1920 for motion room)
|
||||||
|
PATTERN_FILTERS = {
|
||||||
|
'ken_burns_in': (
|
||||||
|
"scale=1120:1990,"
|
||||||
|
"zoompan=z='min(zoom+0.0008,1.08)':x='iw/2-(iw/zoom/2)':y='ih/2-(ih/zoom/2)'"
|
||||||
|
":d={dur_frames}:s=1080x1920:fps=30"
|
||||||
|
),
|
||||||
|
'ken_burns_out': (
|
||||||
|
"scale=1120:1990,"
|
||||||
|
"zoompan=z='if(lte(zoom,1.0),1.08,max(zoom-0.0008,1.0))'"
|
||||||
|
":x='iw/2-(iw/zoom/2)':y='ih/2-(ih/zoom/2)'"
|
||||||
|
":d={dur_frames}:s=1080x1920:fps=30"
|
||||||
|
),
|
||||||
|
'pan_left': (
|
||||||
|
"scale=1200:1920,"
|
||||||
|
"crop=1080:1920:'min(iw-ow, (iw-ow)*t/{duration})':0"
|
||||||
|
),
|
||||||
|
'pan_right': (
|
||||||
|
"scale=1200:1920,"
|
||||||
|
"crop=1080:1920:'(iw-ow)*(1-t/{duration})':0"
|
||||||
|
),
|
||||||
|
'parallax': (
|
||||||
|
# Approximate parallax: zoom + horizontal pan
|
||||||
|
"scale=1200:1990,"
|
||||||
|
"zoompan=z='1.05':x='iw/2-(iw/zoom/2)+50*sin(2*PI*t/{duration})'"
|
||||||
|
":y='ih/2-(ih/zoom/2)':d={dur_frames}:s=1080x1920:fps=30"
|
||||||
|
),
|
||||||
|
'rotate_slow': (
|
||||||
|
"scale=1200:1200,"
|
||||||
|
"rotate='0.02*t':c=black:ow=1080:oh=1920"
|
||||||
|
),
|
||||||
|
'glitch_reveal': (
|
||||||
|
# Fade in with slight chromatic aberration approximation
|
||||||
|
"scale=1080:1920,"
|
||||||
|
"fade=t=in:st=0:d=0.3,"
|
||||||
|
"hue=h='if(lt(t,0.3),10*sin(30*t),0)'"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class MotionEngine:
|
||||||
|
"""
|
||||||
|
Applies motion patterns to still images.
|
||||||
|
Auto-selects patterns to avoid repeating the last 2 used.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self):
|
||||||
|
self._recent: list[str] = [] # last patterns used (max 2)
|
||||||
|
self._ffmpeg = os.environ.get('FFMPEG_PATH', 'ffmpeg')
|
||||||
|
|
||||||
|
def apply(self, image_path: str, duration: float, output_path: Optional[str] = None) -> str:
|
||||||
|
"""
|
||||||
|
Apply a motion pattern to a still image.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
image_path: Path to input image (PNG/JPG, 1080x1920 recommended)
|
||||||
|
duration: Duration of output video in seconds
|
||||||
|
output_path: Output MP4 path. If None, creates temp file.
|
||||||
|
|
||||||
|
Returns: Path to motion-applied video clip (MP4)
|
||||||
|
"""
|
||||||
|
if output_path is None:
|
||||||
|
# Create a temp file that persists (caller is responsible for cleanup)
|
||||||
|
tmp = tempfile.NamedTemporaryFile(suffix='.mp4', delete=False)
|
||||||
|
output_path = tmp.name
|
||||||
|
tmp.close()
|
||||||
|
|
||||||
|
pattern = self._next_pattern()
|
||||||
|
success = self._ffmpeg_motion(image_path, duration, pattern, output_path)
|
||||||
|
|
||||||
|
if not success:
|
||||||
|
logger.warning(f'[모션] {pattern} 패턴 실패 — ken_burns_in으로 폴백')
|
||||||
|
success = self._ffmpeg_motion(image_path, duration, 'ken_burns_in', output_path)
|
||||||
|
|
||||||
|
if success:
|
||||||
|
logger.info(f'[모션] 패턴 적용: {pattern} ({duration:.1f}초)')
|
||||||
|
return output_path
|
||||||
|
else:
|
||||||
|
logger.error(f'[모션] 모든 패턴 실패: {image_path}')
|
||||||
|
return ''
|
||||||
|
|
||||||
|
def _next_pattern(self) -> str:
|
||||||
|
"""Select next pattern, avoiding last 2 used."""
|
||||||
|
available = [p for p in PATTERNS if p not in self._recent[-2:]]
|
||||||
|
if not available:
|
||||||
|
available = PATTERNS
|
||||||
|
choice = random.choice(available)
|
||||||
|
self._recent.append(choice)
|
||||||
|
if len(self._recent) > 4: # Keep small buffer
|
||||||
|
self._recent = self._recent[-4:]
|
||||||
|
return choice
|
||||||
|
|
||||||
|
def _ffmpeg_motion(self, image_path: str, duration: float,
|
||||||
|
pattern: str, output_path: str) -> bool:
|
||||||
|
"""Apply a specific motion pattern using FFmpeg."""
|
||||||
|
dur_frames = int(duration * 30)
|
||||||
|
|
||||||
|
vf_template = PATTERN_FILTERS.get(pattern, PATTERN_FILTERS['ken_burns_in'])
|
||||||
|
vf = vf_template.format(
|
||||||
|
duration=f'{duration:.3f}',
|
||||||
|
dur_frames=dur_frames,
|
||||||
|
)
|
||||||
|
|
||||||
|
cmd = [
|
||||||
|
self._ffmpeg, '-y',
|
||||||
|
'-loop', '1',
|
||||||
|
'-i', str(image_path),
|
||||||
|
'-t', f'{duration:.3f}',
|
||||||
|
'-vf', vf,
|
||||||
|
'-c:v', 'libx264',
|
||||||
|
'-crf', '20',
|
||||||
|
'-preset', 'fast',
|
||||||
|
'-pix_fmt', 'yuv420p',
|
||||||
|
'-an',
|
||||||
|
'-r', '30',
|
||||||
|
str(output_path),
|
||||||
|
]
|
||||||
|
|
||||||
|
try:
|
||||||
|
result = subprocess.run(
|
||||||
|
cmd,
|
||||||
|
capture_output=True,
|
||||||
|
timeout=120,
|
||||||
|
)
|
||||||
|
if result.returncode != 0:
|
||||||
|
logger.warning(f'[모션] FFmpeg 오류 ({pattern}): {result.stderr.decode(errors="ignore")[-200:]}')
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
except subprocess.TimeoutExpired:
|
||||||
|
logger.warning(f'[모션] 타임아웃 ({pattern})')
|
||||||
|
return False
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning(f'[모션] 예외 ({pattern}): {e}')
|
||||||
|
return False
|
||||||
|
|
||||||
|
def get_recent(self) -> list[str]:
|
||||||
|
"""Return recently used patterns."""
|
||||||
|
return list(self._recent)
|
||||||
|
|
||||||
|
|
||||||
|
# ── Standalone test ──────────────────────────────────────────────
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
import sys
|
||||||
|
if '--test' in sys.argv:
|
||||||
|
engine = MotionEngine()
|
||||||
|
print("=== Motion Engine Test ===")
|
||||||
|
print(f"Available patterns: {PATTERNS}")
|
||||||
|
print(f"Pattern sequence (10 picks):")
|
||||||
|
for i in range(10):
|
||||||
|
p = engine._next_pattern()
|
||||||
|
print(f" {i+1}. {p}")
|
||||||
|
print(f"No pattern repeated consecutively: ", end='')
|
||||||
|
recent_list = engine.get_recent()
|
||||||
|
no_consec = all(
|
||||||
|
recent_list[i] != recent_list[i+1]
|
||||||
|
for i in range(len(recent_list)-1)
|
||||||
|
)
|
||||||
|
print("PASS" if no_consec else "FAIL")
|
||||||
Reference in New Issue
Block a user