Spaces:
Sleeping
Sleeping
File size: 2,090 Bytes
eb8c5e1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 |
# core/assembler.py
"""
Assembler combines video + music and optionally overlays images/text,
and produces final assembled output. Uses ffmpeg for merging audio/video.
Returns a dict with final path and metadata.
"""
import subprocess
import time
from pathlib import Path
from typing import Dict
from config import OUTPUT_DIR
OUTPUT_DIR = Path(OUTPUT_DIR)
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
def assemble_video(video_result: Dict, music_result: Dict, out_name: str = None) -> Dict:
"""
video_result: dict returned from video_generator
music_result: dict returned from music_generator
Returns dict { "final_path":..., "duration":..., "meta":... }
"""
t0 = time.time()
video_path = Path(video_result.get("video_path"))
music_path = Path(music_result.get("music_path"))
out_name = out_name or f"final_{int(time.time()*1000)}.mp4"
out_path = OUTPUT_DIR / out_name
# If ffmpeg is available, merge audio and video
if video_path.exists() and music_path.exists():
cmd = [
"ffmpeg", "-y",
"-i", str(video_path),
"-i", str(music_path),
"-c:v", "copy", # copy video stream
"-c:a", "aac",
"-shortest",
str(out_path)
]
try:
subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
duration = video_result.get("duration", 0.0)
except Exception as e:
print(f"[assembler] ffmpeg merge failed: {e}. Creating placeholder final output.")
with open(out_path, "wb") as f:
f.write(b"")
duration = 0.0
else:
# If audio or video missing, create a placeholder
with open(out_path, "wb") as f:
f.write(b"")
duration = 0.0
meta = {
"video_src": str(video_path),
"music_src": str(music_path),
"assembled_at": time.time() - t0
}
return {"final_path": str(out_path), "duration": duration, "meta": meta}
|