o2o-castad-backend/generator/media.py

467 lines
22 KiB
Python

# -*- coding: utf-8 -*-
"""
Best3 저수준 도구함 (TTS · 비전 · ffmpeg · BGM · 폰트)
======================================================
one_short / render 가 공통으로 쓰는 유틸 모음. 단독 실행용 아님.
· 비전: analyze_photos(사진/클립프레임 라벨·점수), extract_frame
· 음성: gen_tts(Gemini TTS), pcm_to_wav, wav_seconds, process_audio(무음제거·속도)
· ffmpeg: _ffmpeg, _run, pick_encoder(QSV/libx264), _AUDIO_ARGS
· 자막: _wrap_lines(픽셀폭 줄바꿈)
· BGM: resolve_bgm(폴더 랜덤 1곡), mix_bgm
설치: pip install google-genai pillow imageio-ffmpeg
"""
import json
import re
import shutil
import subprocess
import time
import wave
from pathlib import Path
from google.genai import types
HERE = Path(__file__).parent
W, H = 1080, 1920 # 세로 쇼츠
FPS = 30
VISION_MODEL = "gemini-2.5-flash" # 사진/클립 비전 분석용
TTS_MODEL = "gemini-2.5-flash-preview-tts"
TTS_VOICE = "Aoede" # 산뜻하고 가벼운 보이스. 대안: Charon(차분) / Sulafat(따뜻) / Laomedeia(경쾌)
TTS_STYLE = ("다음 문장을 밝고 경쾌하게, 적당히 빠른 속도로 활기차고 신나는 톤으로 읽어줘. "
"지루하지 않게 텐션을 올리되, 과한 병맛·오버는 아니고 세련되게:")
LINK_CTA = "소개한 곳 링크는 고정댓글에서 확인하세요" # info 카드 링크 안내
# 한글 자막용 폰트 — 주아체(fonts/ 동봉), 없으면 맑은고딕 볼드. 제목/정보용은 송명체.
FONT = str(HERE / "fonts" / "Jua-Regular.ttf")
FONT_FALLBACK = r"C:\Windows\Fonts\malgunbd.ttf"
TITLE_FONT = str(HERE / "fonts" / "SongMyung-Regular.ttf")
_AUDIO_ARGS = ["-c:a", "aac", "-b:a", "160k", "-ar", "44100", "-ac", "2"]
# 수동 입력(링크 없이 직접 업로드)용 — 로컬 폴더의 이미지/영상 파일 목록
_LOCAL_IMG_EXTS = (".jpg", ".jpeg", ".png", ".webp", ".bmp")
_LOCAL_VID_EXTS = (".mp4", ".mov", ".webm", ".m4v")
def list_media(folder, kind="image"):
"""로컬 폴더의 이미지(kind='image') 또는 영상(kind='video') 파일을 정렬해 [Path...] 반환.
링크 없이 직접 사진/클립을 올릴 때 사용. 폴더가 없으면 빈 리스트."""
exts = _LOCAL_VID_EXTS if kind == "video" else _LOCAL_IMG_EXTS
d = Path(folder) if folder else None
if not d or not d.is_dir():
return []
return sorted(p for p in d.iterdir() if p.suffix.lower() in exts)
# ---------------------------------------------------------------- ffmpeg
def _ffmpeg() -> str:
import imageio_ffmpeg
return imageio_ffmpeg.get_ffmpeg_exe()
def _run(cmd):
r = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace")
if r.returncode != 0:
raise RuntimeError("ffmpeg 실패:\n" + (r.stderr or "")[-1500:])
return r
def pick_encoder(exe) -> list:
"""Intel QSV 하드웨어 인코딩이 실제로 동작하면 그걸, 아니면 libx264 ultrafast."""
test = subprocess.run(
[exe, "-hide_banner", "-f", "lavfi", "-i", "color=c=black:s=128x128:d=1",
"-c:v", "h264_qsv", "-f", "null", "-"],
capture_output=True, text=True, errors="replace")
if test.returncode == 0:
print(" 인코더: h264_qsv (Intel 하드웨어)")
return ["-c:v", "h264_qsv", "-global_quality", "24", "-preset", "fast"]
print(" 인코더: libx264 (superfast)")
return ["-c:v", "libx264", "-preset", "superfast", "-crf", "26"]
# ---------------------------------------------------------------- 비전
def analyze_photos(client, photo_paths, model=None):
"""사진/클립프레임을 비전으로 한 번에 분석.
반환 rows: 입력 순서대로 [{label, char_fit, has_text}].
label = 보이는 핵심 한국어 한 줄(나레이션 매칭용)
char_fit = 컷이 매력적인 정도 1~10(클립·사진 순위용)
has_text = 글자/자막/캡션이 박혀 보이면 True
실패 시 빈 라벨/0/False."""
n = len(photo_paths)
parts = [types.Part.from_text(text=(
"여러 이미지(가게 사진 또는 영상 캡처)다. 각 이미지를 분석해라.\n"
"- label: 보이는 핵심을 짧은 한국어 한 줄로(예: '대게 한 상', '바다 노을 뷰', '야외 수영장', "
"'아늑한 객실', '외관/간판'). 메뉴판이면 '메뉴판'.\n"
"- char_fit: 광고 컷으로 매력적인 정도 1~10 "
"(음식·음료·수영장·아늑한 실내=높게 / 메뉴판·로고·밋밋한 외관·사람 위주=낮게).\n"
"- has_text: 이미지에 '글자/자막/캡션/큰 로고 텍스트'가 박혀 보이면 true, 깨끗하면 false.\n"
"각 이미지 앞 [이미지 N] 의 N 을 index 로 써라."))]
for i, p in enumerate(photo_paths):
parts.append(types.Part.from_text(text=f"[이미지 {i}]"))
parts.append(types.Part.from_bytes(data=Path(p).read_bytes(), mime_type="image/jpeg"))
schema = {"type": "object", "properties": {"photos": {"type": "array", "items": {
"type": "object", "properties": {
"index": {"type": "integer"}, "label": {"type": "string"},
"char_fit": {"type": "integer"}, "has_text": {"type": "boolean"}},
"required": ["index", "label", "char_fit", "has_text"]}}}, "required": ["photos"]}
resp = client.models.generate_content(
model=model or VISION_MODEL, contents=parts,
config=types.GenerateContentConfig(
temperature=0.2, response_mime_type="application/json", response_schema=schema))
by = {r["index"]: r for r in json.loads(resp.text).get("photos", [])
if 0 <= r.get("index", -1) < n}
return [{"label": (by.get(i, {}).get("label") or "").strip(),
"char_fit": by.get(i, {}).get("char_fit", 0),
"has_text": bool(by.get(i, {}).get("has_text", False))} for i in range(n)]
def extract_frame(video_path, out_jpg, at=1.0):
"""영상에서 프레임 1장 추출(비전 분석/썸네일용)."""
_run([_ffmpeg(), "-y", "-ss", str(at), "-i", str(video_path),
"-frames:v", "1", str(out_jpg)])
return out_jpg
# ---------------------------------------------------------------- 음성
def _extract_audio(resp):
"""TTS 응답에서 (pcm, rate) 추출. 비었으면 None 반환(재시도 신호)."""
cands = getattr(resp, "candidates", None) or []
for c in cands:
content = getattr(c, "content", None)
parts = getattr(content, "parts", None) or [] if content else []
for p in parts:
inline = getattr(p, "inline_data", None)
if inline and inline.data:
m = re.search(r"rate=(\d+)", inline.mime_type or "")
return inline.data, (int(m.group(1)) if m else 24000)
return None
def gen_tts(client, text, voice, retries=4) -> bytes:
"""Gemini TTS. 프리뷰 모델이 가끔 빈 응답(content=None)을 주므로 재시도한다."""
last = ""
for attempt in range(1, retries + 1):
try:
resp = client.models.generate_content(
model=TTS_MODEL,
contents=f"{TTS_STYLE}\n\n{text}",
config=types.GenerateContentConfig(
response_modalities=["AUDIO"],
speech_config=types.SpeechConfig(
voice_config=types.VoiceConfig(
prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=voice)
)
),
),
)
got = _extract_audio(resp)
if got:
return got
# 비었으면 왜 비었는지(finish_reason 등) 기록하고 재시도
fr = ""
try:
fr = str(getattr(resp.candidates[0], "finish_reason", "") or "")
except Exception:
fr = "no-candidates"
last = f"빈 응답(finish_reason={fr or '?'})"
except Exception as e:
last = f"{type(e).__name__}: {e}"
if attempt < retries:
wait = 2 * attempt # 2s, 4s, 6s … 점증 백오프
print(f"\n ↻ TTS 재시도 {attempt}/{retries-1} ({last}) — {wait}s 후",
end="", flush=True)
time.sleep(wait)
raise RuntimeError(f"TTS 실패(재시도 {retries}회): {last}\n 문장: {text[:40]}")
def pcm_to_wav(pcm, path, rate):
with wave.open(str(path), "wb") as w:
w.setnchannels(1); w.setsampwidth(2); w.setframerate(rate)
w.writeframes(pcm)
def wav_seconds(path: Path) -> float:
with wave.open(str(path), "rb") as w:
return w.getnframes() / w.getframerate()
def _atempo_chain(speed: float) -> list:
"""atempo 는 0.5~2.0 범위만 받으므로 큰 배율은 곱으로 쪼개 체이닝."""
chain, s = [], speed
while s > 2.0:
chain.append("atempo=2.0"); s /= 2.0
while s < 0.5:
chain.append("atempo=0.5"); s *= 2.0
chain.append(f"atempo={s:.4f}")
return chain
def process_audio(exe, src_wav, dst_wav, speed=1.0, trim_silence=True,
threshold="-38dB", keep=0.06):
"""컷편집용 음성 가공: 앞뒤 침묵 제거 + 말 속도 올리기.
- trim_silence: 앞/뒤 침묵을 잘라 컷이 탁탁 넘어가게(가운데 호흡은 유지).
silenceremove 로 앞을 자르고, areverse 로 뒤집어 다시 앞(=원래 뒤)을 자른 뒤 복원.
keep: 잘라낸 끝에 남길 여유(초). 작을수록 마디가 바짝 붙는다.
- speed: 1.0=원본, 1.3 이면 30% 빠르게(음정 유지). 길이도 그만큼 줄어든다.
가공 결과가 비거나 실패하면 원본을 그대로 복사해 안전하게 폴백.
"""
af = []
if trim_silence:
one = (f"silenceremove=start_periods=1:start_silence={keep}:"
f"start_threshold={threshold}:detection=peak")
af += [one, "areverse", one, "areverse"]
if speed and abs(speed - 1.0) > 1e-3:
af += _atempo_chain(speed)
if not af:
shutil.copyfile(src_wav, dst_wav); return
try:
_run([exe, "-y", "-i", str(src_wav), "-af", ",".join(af),
"-ar", "24000", "-ac", "1", str(dst_wav)])
if wav_seconds(dst_wav) < 0.15: # 과하게 잘렸으면 폴백
raise RuntimeError("가공 후 음성이 너무 짧음")
except (RuntimeError, OSError):
shutil.copyfile(src_wav, dst_wav)
# ---------------------------------------------------------------- 자막 유틸
def _wrap_lines(draw, text, font, max_w):
"""글자 단위로 픽셀 폭(max_w)에 맞춰 줄바꿈."""
lines, cur = [], ""
for ch in text:
if ch == "\n":
lines.append(cur); cur = ""; continue
if draw.textlength(cur + ch, font=font) <= max_w:
cur += ch
else:
lines.append(cur); cur = ch
if cur:
lines.append(cur)
return lines or [""]
# ---------------------------------------------------------------- 카메라 모션
#
# 엔진 4종(animation / samgukji / greekroman / odyssey)이 각자 복사본을 갖고 있던
# `_scene_motion` 을 여기로 올렸다. 완전히 같은 코드였고(md5 일치), 효과를 늘릴 때마다
# 4곳을 손으로 맞춰야 해 어긋나기 쉬웠다.
#
# 표현식은 ffmpeg `zoompan` 의 z/x/y 만 반환한다 — d/s/fps 는 호출부가 붙인다.
# 회전·틸트 필터는 쓰지 않는다: 렌더가 이미 떨림 방지로 3배 업스케일 중이라
# 필터를 더 얹으면 메모리와 시간이 그만큼 붙는다. zoompan 하나로 낼 수 있는
# 범위 안에서만 다양화했다.
#: 스토리보드가 고를 수 있는 카메라 종류. gen.py 의 SB_SCHEMA enum 과 같은 값이어야 한다.
CAMERA_KINDS = (
"reveal", # 살짝 당긴 상태에서 전체로 — 첫 컷(훅) 기본값
"punch_in", # 앞부분에서 확 들어가 고정 — 반전·폭로
"slow_in", # 완만한 줌인 — 몰입이 쌓이는 서술
"slow_out", # 완만한 줌아웃 — 상황이 드러나는 서술
"pan_left", # 우→좌
"pan_right", # 좌→우
"tilt_up", # 아래→위 — 올려다보는 느낌(권위·거대함)
"tilt_down", # 위→아래 — 내려다보는 느낌(추락·좌절)
"shake", # 미세 흔들림 — 충격·당황
"hold", # 완전 정지 — 정적인 대사, 앞뒤 대비용
)
# ── 움직임이 '계단처럼' 보이지 않게 하는 규칙 ────────────────────────────
#
# zoompan 은 크롭 사각형을 **입력 정수 픽셀**로 반올림한다. 3배 업스케일이면
# 출력 기준 약 0.33px 격자이고, 이 격자 오차(측정값 jerk 0.13~0.16px)는
# 무엇을 해도 거의 줄지 않는다. 실측(2026-08-10):
#
# 업스케일 3배 → 6배 → 8배 : jerk 0.127 → 0.094 → 0.085 (33% 개선)
# 그런데 렌더 시간은 : 5.6초 → 22.5초 → 29.3초 (5배)
#
# 즉 업스케일은 나쁜 거래다. 바닥이 고정이라면 **프레임당 이동량을 그 위로
# 올리는 것**이 답이다. 실제로 부드러웠던 팬은 1.06px/프레임이었고,
# 계단이 보이던 줌은 0.25~0.38px/프레임이었다.
#
# 컷 길이별 불균일도(= jerk / 프레임당 이동), 확대폭 10% 고정 시
# 2초 22% · 4초 34% · 6초 63% ← 컷이 길수록 급격히 나빠진다
# 확대폭을 35% 로 올리면 6초 컷도 16% 로 떨어진다.
#
# 원인은 **확대·이동 폭이 컷 길이와 무관하게 고정**이었다는 것이다. 컷이 길수록
# 같은 거리를 더 잘게 나눠 가므로 한 걸음이 격자 아래로 내려간다.
# → 아래 두 상수로 '한 걸음'을 목표치에 맞춰 폭을 역산한다.
#
# 가감속(smoothstep)은 넣지 않았다. 양 끝에서 속도가 0 이 되어 그 구간이
# 격자 아래로 떨어진다 — 실측에서 정지 프레임이 0개에서 7개로 늘었다.
# 등속이 이 격자 위에서는 가장 부드럽다.
#: 프레임당 목표 이동량(출력 px). 부드러웠던 팬의 실측치(1.06px)를 기준으로 잡았다.
TARGET_PX_PER_FRAME = 1.0
#: 확대폭 한계. 너무 크면 원본 화질이 드러나고 화면이 과하게 요동친다.
_ZOOM_MIN, _ZOOM_MAX = 0.08, 0.30
#: 팬·틸트 이동 거리(화면 대비 비율) 한계.
_PAN_MIN, _PAN_MAX = 0.06, 0.25
#: 출력 가로 폭. 이동량 계산의 기준이며 W 와 같다.
_OUT_W = 1080
#: 확대율 1% 당 화면에서 움직이는 픽셀 수(실측 회귀).
#: 확대폭 a, 프레임 f 일 때 프레임당 이동 = a * 450 / f 로 맞았다.
_ZOOM_PX_GAIN = 450.0
def _zoom_amount(frames: int) -> float:
"""컷 길이에 맞춘 확대폭. 한 걸음이 목표치(≈1px)가 되도록 역산한다."""
f = max(frames, 1)
want = f * TARGET_PX_PER_FRAME / _ZOOM_PX_GAIN
return max(_ZOOM_MIN, min(_ZOOM_MAX, want))
def _pan_zoom(frames: int) -> tuple:
"""팬·틸트용 (확대율, 이동비율). 확대율은 이동할 여백을 만들기 위한 것이다.
이동 거리 = 화면폭 * ratio 이므로, 프레임당 목표 이동에서 ratio 를 역산한다.
"""
f = max(frames, 1)
ratio = f * TARGET_PX_PER_FRAME / _OUT_W
ratio = max(_PAN_MIN, min(_PAN_MAX, ratio))
return 1.0 / (1.0 - ratio), ratio
#: 역할을 알 수 없는 컷에 순환시킬 기본 목록.
#: reveal/punch_in/shake 는 강한 효과라 여기서 뺐다 — 아무 데나 붙으면 산만해진다.
_NEUTRAL_KINDS = ("slow_in", "pan_right", "slow_out", "tilt_up", "pan_left", "tilt_down")
_CX = "iw/2-(iw/zoom/2)" # 가로 중앙 고정
_CY = "ih/2-(ih/zoom/2)" # 세로 중앙 고정
def motion_seed(text: str) -> int:
"""스토리보드 텍스트에서 카메라 순환 시작점을 뽑는다.
난수를 쓰지 않는 이유: 같은 콘텐츠를 다시 렌더하면 같은 영상이 나와야
비교·디버깅이 된다. 반면 콘텐츠가 다르면 시작점이 달라지므로
**영상마다 카메라 순서가 달라진다** — 결정론과 다양성을 함께 얻는다.
"""
import hashlib
digest = hashlib.md5((text or "").encode("utf-8", "ignore")).hexdigest()
return int(digest[:8], 16)
def _camera_expr(kind: str, frames: int) -> str:
"""카메라 종류 → zoompan 표현식.
확대·이동 폭은 상수가 아니라 **컷 길이에서 역산**한다(위 주석 참조).
긴 컷일수록 폭을 키워야 한 걸음이 양자화 격자 위에 남는다.
"""
f = max(frames, 1)
a = _zoom_amount(f) # 이 컷에 맞는 확대폭
pz, _ = _pan_zoom(f) # 팬·틸트용 확대율(여백 확보)
if kind == "reveal":
# 당긴 상태에서 전체로. 확대폭을 조금 더 크게 잡아 '드러나는' 느낌을 준다
ra = min(_ZOOM_MAX, a * 1.4)
return f"zoompan=z='max({1 + ra:.4f}-{ra:.4f}*on/{f},1.0)':x='{_CX}':y='{_CY}'"
if kind == "punch_in":
# 앞 35% 구간에서 들어가고 나머지는 정지. 정지 구간은 움직임이 아예 없어
# 격자 문제가 생기지 않고, 돌진 구간은 걸음이 크므로 부드럽다
rush = max(1, int(f * 0.35))
pa = min(_ZOOM_MAX, a * 1.4)
return f"zoompan=z='min(1.0+{pa:.4f}*on/{rush},{1 + pa:.4f})':x='{_CX}':y='{_CY}'"
if kind == "slow_in":
return f"zoompan=z='1.0+{a:.4f}*on/{f}':x='{_CX}':y='{_CY}'"
if kind == "slow_out":
return f"zoompan=z='{1 + a:.4f}-{a:.4f}*on/{f}':x='{_CX}':y='{_CY}'"
if kind == "pan_right":
return f"zoompan=z='{pz:.4f}':x='(iw-iw/zoom)*on/{f}':y='{_CY}'"
if kind == "pan_left":
# `(1-on/f)` 로 쓰면 1 에 가까운 값을 큰 수에 곱하게 되어 반올림이 어긋난다.
# 실측에서 pan_right 4.1% 대 pan_left 22.3% 로 방향만 다른데 5배 차이가 났다.
# 증가항을 그대로 두고 '빼는' 형태로 쓰면 두 방향의 격자 패턴이 같아진다.
return f"zoompan=z='{pz:.4f}':x='(iw-iw/zoom)-(iw-iw/zoom)*on/{f}':y='{_CY}'"
if kind == "tilt_down":
return f"zoompan=z='{pz:.4f}':x='{_CX}':y='(ih-ih/zoom)*on/{f}'"
if kind == "tilt_up":
return f"zoompan=z='{pz:.4f}':x='{_CX}':y='(ih-ih/zoom)-(ih-ih/zoom)*on/{f}'"
if kind == "shake":
# 흔들림은 의도된 진동이라 걸음이 크다(프레임당 수 px) — 격자 영향 없음
return ("zoompan=z='1.06':"
f"x='{_CX}+8*sin(on*1.6)':"
f"y='{_CY}+6*cos(on*2.2)'")
# hold — **완전 정지**. 예전에는 3px 사인 드리프트를 줬으나, 그 폭이 출력
# 기준 1px 라 격자에 걸려 프레임의 1/3 이 멈췄다 튀었다(실측 20/59).
# 미세하게 움직이느니 아예 안 움직이는 편이 훨씬 깔끔하다.
return f"zoompan=z='1.04':x='{_CX}':y='{_CY}'"
def scene_motion(scene, idx: int, n: int, frames: int, seed: int = 0) -> str:
"""장면 하나의 zoompan 표현식.
카메라를 정하는 순서는 셋이고, 앞에서 정해지면 뒤는 보지 않는다:
1. **스토리보드가 지정한 값** (`scene["camera"]`).
LLM 이 장면 내용을 알고 고른 것이라 가장 정확하다.
2. **위치 휴리스틱** — 첫 컷은 reveal, 마지막 컷은 시선을 모으는 slow_in.
스토리보드가 camera 를 안 줬을 때의 안전망이다.
3. **seed 순환** — 중립 효과를 `(idx + seed)` 로 돌린다. idx 가 1씩
늘어나므로 **인접 컷이 같은 효과가 되는 일이 없고**, seed 가 콘텐츠마다
달라 영상끼리도 순서가 겹치지 않는다.
`scene` 이 dict 가 아니거나 모르는 값이면 조용히 3단계로 떨어진다 —
카메라 때문에 렌더 전체가 실패하면 안 된다.
"""
if isinstance(scene, dict):
want = str(scene.get("camera") or "").strip().lower()
if want in CAMERA_KINDS:
return _camera_expr(want, frames)
if idx == 1:
return _camera_expr("reveal", frames)
if n >= 3 and idx == n:
return _camera_expr("slow_in", frames)
kind = _NEUTRAL_KINDS[(idx + seed) % len(_NEUTRAL_KINDS)]
return _camera_expr(kind, frames)
# ---------------------------------------------------------------- BGM
AUDIO_EXTS = (".mp3", ".wav", ".m4a", ".aac", ".ogg", ".flac", ".opus")
def resolve_bgm(bgm_arg):
"""--bgm 값 → 실제 음악 파일 경로(없으면 None).
해석 순서(분위기별로 bgm/ 폴더에 나눠 담는 걸 지원):
1) 파일이면 그대로
2) 폴더면 그 안 음악 중 무작위 1곡
3) 이름만 주면 HERE/<값> → HERE/bgm/<값> 순으로 파일/폴더 탐색
"""
if not bgm_arg or str(bgm_arg).strip().lower() in ("none", "off", "no"):
return None
import random
p = Path(bgm_arg)
cands = [p] if p.is_absolute() else [HERE / p, HERE / "bgm" / p]
for c in cands:
if c.is_file():
return c
if c.is_dir():
files = [f for f in sorted(c.iterdir()) if f.suffix.lower() in AUDIO_EXTS]
if files:
return random.choice(files)
return None
def mix_bgm(exe, video_path, bgm_path, out_path, volume=0.3, fade=1.0):
"""완성 영상(나레이션 음성 포함) 위에 BGM 한 트랙을 작은 볼륨으로 깔아 최종본 저장.
- BGM 이 영상보다 짧으면 무한 루프(-stream_loop -1), 길면 영상 길이에 맞춰 잘림.
- 나레이션[0:a] + BGM[1:a]*volume 를 amix(duration=first)로 섞어 영상 길이에 맞춘다.
- 영상은 재인코딩하지 않는다(-c:v copy) → 빠름. 시작에 짧은 페이드인.
"""
fc = (f"[1:a]volume={volume},afade=t=in:st=0:d={fade}[bg];"
f"[0:a][bg]amix=inputs=2:duration=first:dropout_transition=0[a]")
_run([exe, "-y", "-i", str(video_path), "-stream_loop", "-1", "-i", str(bgm_path),
"-filter_complex", fc, "-map", "0:v", "-map", "[a]",
"-c:v", "copy", *_AUDIO_ARGS, "-shortest", "-movflags", "+faststart",
str(out_path)])