158 lines
5.5 KiB
Python
158 lines
5.5 KiB
Python
# LIB · 成片合成模块
|
||
# D180 · 2026-07-10
|
||
"""FFmpeg: 多镜头拼接 + 音频合成 + 字幕叠加 → 导出成片"""
|
||
import os, subprocess, json
|
||
from lib.models import get_db
|
||
|
||
FFMPEG = os.path.expanduser("~/.workbuddy/binaries/ffmpeg.exe")
|
||
PROJECT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
def concat_shots(episode_id, output_path=None, add_fade=True):
|
||
"""
|
||
从数据库读取已完成镜头 → FFmpeg 拼接 → 导出成片
|
||
|
||
Args:
|
||
episode_id: 剧集ID
|
||
output_path: 输出路径（默认 outputs/episodes/EP01-final.mp4）
|
||
add_fade: 镜头间加淡入淡出
|
||
"""
|
||
if not os.path.exists(FFMPEG):
|
||
return {"error": f"ffmpeg not found: {FFMPEG}"}
|
||
|
||
db = get_db()
|
||
shots = db.execute("""
|
||
SELECT id, shot_number, output_path, duration_sec FROM shots
|
||
WHERE episode_id = ? AND status = 'done'
|
||
ORDER BY shot_number
|
||
""", (episode_id,)).fetchall()
|
||
|
||
if not shots:
|
||
return {"error": f"No completed shots for {episode_id}"}
|
||
|
||
# 检查所有文件存在
|
||
for s in shots:
|
||
if not s["output_path"] or not os.path.exists(os.path.join(PROJECT, s["output_path"])):
|
||
return {"error": f"Missing: {s['shot_number']} → {s['output_path']}"}
|
||
|
||
if not output_path:
|
||
out_dir = os.path.join(PROJECT, "outputs", "episodes")
|
||
os.makedirs(out_dir, exist_ok=True)
|
||
output_path = os.path.join(out_dir, f"{episode_id}-final.mp4")
|
||
else:
|
||
os.makedirs(os.path.dirname(output_path), exist_ok=True)
|
||
|
||
# 生成 ffmpeg concat 文件列表
|
||
concat_list = os.path.join(PROJECT, "outputs", "temp_concat.txt")
|
||
with open(concat_list, "w", encoding="utf-8") as f:
|
||
for s in shots:
|
||
abs_path = os.path.join(PROJECT, s["output_path"]).replace("\\", "/")
|
||
f.write(f"file '{abs_path}'\n")
|
||
f.write(f"duration {s['duration_sec']}\n")
|
||
# 最后一个文件需要重复一次（ffmpeg concat 约定）
|
||
last = os.path.join(PROJECT, shots[-1]["output_path"]).replace("\\", "/")
|
||
f.write(f"file '{last}'\n")
|
||
|
||
# FFmpeg concat
|
||
cmd = [
|
||
FFMPEG, "-y",
|
||
"-f", "concat",
|
||
"-safe", "0",
|
||
"-i", concat_list,
|
||
"-c:v", "libx264",
|
||
"-preset", "fast",
|
||
"-crf", "23",
|
||
"-c:a", "aac",
|
||
"-pix_fmt", "yuv420p",
|
||
output_path
|
||
]
|
||
|
||
r = subprocess.run(cmd, capture_output=True, text=True)
|
||
os.unlink(concat_list)
|
||
|
||
if r.returncode != 0:
|
||
return {"error": r.stderr[:500]}
|
||
|
||
return {
|
||
"output_path": output_path,
|
||
"shot_count": len(shots),
|
||
"total_duration": sum(s["duration_sec"] for s in shots),
|
||
"shots": [s["shot_number"] for s in shots]
|
||
}
|
||
|
||
def add_audio_and_subtitle(video_path, audio_paths, subtitle_texts, output_path):
|
||
"""
|
||
视频 + 配音 + 字幕 → 合成
|
||
|
||
audio_paths: [{"path": "...", "start": 3.5}, ...] 音频文件和起始时间(秒)
|
||
subtitle_texts: [{"text": "...", "start": 3.5, "end": 8.5}, ...]
|
||
"""
|
||
if not os.path.exists(FFMPEG):
|
||
return {"error": f"ffmpeg not found: {FFMPEG}"}
|
||
|
||
asrc = "" # no original audio
|
||
|
||
# Build audio inputs
|
||
audio_inputs = []
|
||
audio_filters = []
|
||
for i, a in enumerate(audio_paths):
|
||
audio_inputs.extend(["-i", a["path"]])
|
||
delay_ms = int(a["start"] * 1000)
|
||
audio_filters.append(f"[{i+1}:a]adelay={delay_ms}|{delay_ms}[a{i}]")
|
||
|
||
# Build subtitle filter
|
||
sub_parts = []
|
||
for s in subtitle_texts:
|
||
esc = s["text"].replace("'", "\\'").replace(":", "\\:")
|
||
sub_parts.append(f"drawtext=text='{esc}':fontsize=28:fontcolor=white:borderw=2:bordercolor=black:x=(w-text_w)/2:y=h-120:enable='between(t,{s['start']},{s['end']})'")
|
||
sub_filter = ",".join(sub_parts) if sub_parts else "null"
|
||
|
||
# Audio mix
|
||
if audio_inputs:
|
||
n = len(audio_inputs)
|
||
amix = "".join(audio_filters) + f"amix=inputs={n}:duration=first[aout]"
|
||
else:
|
||
amix = ""
|
||
|
||
filter_complex = f"[0:v]{sub_filter}[vout];{amix}" if audio_inputs else f"[0:v]{sub_filter}[vout]"
|
||
|
||
cmd = [FFMPEG, "-y", "-i", video_path] + audio_inputs + [
|
||
"-filter_complex", filter_complex,
|
||
"-map", "[vout]",
|
||
"-map", "[aout]" if audio_inputs else "-an",
|
||
"-c:v", "libx264", "-preset", "fast", "-crf", "23",
|
||
"-c:a", "aac",
|
||
"-pix_fmt", "yuv420p",
|
||
output_path
|
||
]
|
||
|
||
r = subprocess.run(cmd, capture_output=True, text=True)
|
||
return output_path if r.returncode == 0 else {"error": r.stderr[:500]}
|
||
|
||
def render_episode(episode_id):
|
||
"""
|
||
一键成片：数据库 → 拼接所有已完成镜头 → 导出
|
||
"""
|
||
print(f"🎬 合成 {episode_id}...")
|
||
|
||
# Step 1: Concat all done shots
|
||
result = concat_shots(episode_id)
|
||
if "error" in result:
|
||
return result
|
||
|
||
print(f"✅ 拼接完成: {result['shot_count']}镜 {result['total_duration']}s")
|
||
print(f"📁 {result['output_path']}")
|
||
return result
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
if len(sys.argv) < 2:
|
||
print("Usage: python composer.py concat <episode_id>")
|
||
print(" python composer.py render <episode_id>")
|
||
sys.exit(1)
|
||
cmd = sys.argv[1]
|
||
ep = sys.argv[2] if len(sys.argv) > 2 else "DSV-EP01"
|
||
if cmd == "concat":
|
||
print(json.dumps(concat_shots(ep), indent=2, ensure_ascii=False))
|
||
elif cmd == "render":
|
||
print(json.dumps(render_episode(ep), indent=2, ensure_ascii=False))
|