96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
"""
|
|
视频合成服务
|
|
"""
|
|
import os
|
|
import subprocess
|
|
import json
|
|
from pathlib import Path
|
|
from loguru import logger
|
|
from typing import Optional
|
|
|
|
class VideoService:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def _run_ffmpeg(self, cmd: list) -> bool:
|
|
cmd_str = ' '.join(f'"{c}"' if ' ' in c or '\\' in c else c for c in cmd)
|
|
logger.debug(f"FFmpeg CMD: {cmd_str}")
|
|
try:
|
|
# Synchronous call for BackgroundTasks compatibility
|
|
result = subprocess.run(
|
|
cmd_str,
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding='utf-8',
|
|
)
|
|
if result.returncode != 0:
|
|
logger.error(f"FFmpeg Error: {result.stderr}")
|
|
return False
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"FFmpeg Exception: {e}")
|
|
return False
|
|
|
|
def _get_duration(self, file_path: str) -> float:
|
|
# Synchronous call for BackgroundTasks compatibility
|
|
cmd = f'ffprobe -v error -show_entries format=duration -of default=noprint_wrappers=1:nokey=1 "{file_path}"'
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
return float(result.stdout.strip())
|
|
except Exception:
|
|
return 0.0
|
|
|
|
async def compose(
|
|
self,
|
|
video_path: str,
|
|
audio_path: str,
|
|
output_path: str,
|
|
subtitle_path: Optional[str] = None
|
|
) -> str:
|
|
"""合成视频"""
|
|
# Ensure output dir
|
|
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
video_duration = self._get_duration(video_path)
|
|
audio_duration = self._get_duration(audio_path)
|
|
|
|
# Audio loop if needed
|
|
loop_count = 1
|
|
if audio_duration > video_duration and video_duration > 0:
|
|
loop_count = int(audio_duration / video_duration) + 1
|
|
|
|
cmd = ["ffmpeg", "-y"]
|
|
|
|
# Input video (stream_loop must be before -i)
|
|
if loop_count > 1:
|
|
cmd.extend(["-stream_loop", str(loop_count)])
|
|
cmd.extend(["-i", video_path])
|
|
|
|
# Input audio
|
|
cmd.extend(["-i", audio_path])
|
|
|
|
# Filter complex
|
|
filter_complex = []
|
|
|
|
# Subtitles (skip for now to mimic previous state or implement basic)
|
|
# Previous state: subtitles disabled due to font issues
|
|
# if subtitle_path: ...
|
|
|
|
# Audio map
|
|
cmd.extend(["-c:v", "libx264", "-c:a", "aac", "-shortest"])
|
|
# Use audio from input 1
|
|
cmd.extend(["-map", "0:v", "-map", "1:a"])
|
|
|
|
cmd.append(output_path)
|
|
|
|
if self._run_ffmpeg(cmd):
|
|
return output_path
|
|
else:
|
|
raise RuntimeError("FFmpeg composition failed")
|