307 lines
9.5 KiB
Python
Executable File
307 lines
9.5 KiB
Python
Executable File
"""FFmpeg 工具函数 — Worker 层.
|
||
|
||
业务相关的滤镜构建、视频探测、视频标准化等能力放在这里；
|
||
底层原语（run_ffmpeg / 二进制路径 / 默认超时）已下沉到 packages/shared/ffmpeg_utils.py，
|
||
本模块 re-export 保持向后兼容。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import subprocess # nosec B404
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
# 底层原语从 shared 层导入，application 层和 worker 层共用同一份实现
|
||
from shared.ffmpeg_utils import ( # noqa: F401
|
||
DEFAULT_FFMPEG_TIMEOUT,
|
||
FFMPEG_BIN,
|
||
FFPROBE_BIN,
|
||
run_ffmpeg,
|
||
)
|
||
|
||
# xfade 转场纯逻辑已抽离到 domain 层，这里 re-export 保持向后兼容
|
||
from packages.domain.xfade_builder import DEFAULT_TRANSITION_DURATION as _default_transition_duration_base # noqa: F401
|
||
from packages.domain.xfade_builder import SUPPORTED_TRANSITIONS # noqa: F401
|
||
from packages.domain.xfade_builder import XFADE_TRANSITION_MAP # noqa: F401
|
||
from packages.domain.xfade_builder import XFade_TRANSITION_NAMES # noqa: F401
|
||
from packages.domain.xfade_builder import build_xfade_filter_chain as _build_xfade_filter_chain_base
|
||
from packages.domain.xfade_builder import chain_filters as _chain_filters_base
|
||
from packages.domain.xfade_builder import resolve_xfade_transition as _resolve_xfade_transition_base
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── 常量（Worker 层业务相关） ────────────────────────────────────────────────
|
||
|
||
DEFAULT_OUTPUT_WIDTH = 1280
|
||
DEFAULT_OUTPUT_HEIGHT = 720
|
||
DEFAULT_FPS = 25
|
||
|
||
# 向后兼容：DEFAULT_TRANSITION_DURATION 从 domain 层导出
|
||
DEFAULT_TRANSITION_DURATION = _default_transition_duration_base
|
||
|
||
|
||
# 向后兼容：薄包装函数
|
||
def chain_filters(filters: list[str], output_label: str, *, input_label: str = "0:v") -> str:
|
||
return _chain_filters_base(filters, output_label, input_label=input_label)
|
||
|
||
|
||
def resolve_xfade_transition(transition_name: Any) -> str:
|
||
return _resolve_xfade_transition_base(transition_name)
|
||
|
||
|
||
def build_xfade_filter_chain(
|
||
clip_durations: list[float],
|
||
clip_video_labels: list[str],
|
||
transitions: list[str],
|
||
*,
|
||
transition_duration: float = DEFAULT_TRANSITION_DURATION,
|
||
output_label: str = "outv",
|
||
) -> tuple[str, float]:
|
||
return _build_xfade_filter_chain_base(
|
||
clip_durations,
|
||
clip_video_labels,
|
||
transitions,
|
||
transition_duration=transition_duration,
|
||
output_label=output_label,
|
||
)
|
||
|
||
|
||
# ── FFprobe 探测 ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
def run_ffprobe(
|
||
command: list[str],
|
||
*,
|
||
capture_output: bool = True,
|
||
timeout: int = 30,
|
||
) -> tuple[str, str]:
|
||
"""执行 FFprobe 命令。
|
||
|
||
Args:
|
||
command: 完整的 ffprobe 命令列表（含 "ffprobe" 本身）
|
||
capture_output: 是否捕获 stdout/stderr
|
||
timeout: 超时时间（秒），默认 30s；None 表示不设超时
|
||
|
||
Returns:
|
||
(stdout, stderr) 元组
|
||
|
||
Raises:
|
||
subprocess.CalledProcessError: 命令执行失败时抛出
|
||
subprocess.TimeoutExpired: 超时未完成时抛出
|
||
"""
|
||
try:
|
||
result = subprocess.run( # nosec B603
|
||
command,
|
||
check=True,
|
||
stdout=subprocess.PIPE if capture_output else None,
|
||
stderr=subprocess.PIPE if capture_output else None,
|
||
text=True,
|
||
timeout=timeout,
|
||
)
|
||
return (result.stdout or "", result.stderr or "")
|
||
except subprocess.TimeoutExpired:
|
||
logger.error(
|
||
"FFprobe 命令超时 (%ds): command=%s",
|
||
timeout or -1,
|
||
" ".join(str(c) for c in command[:20]),
|
||
)
|
||
raise
|
||
except subprocess.CalledProcessError as e:
|
||
stderr_text = (e.stderr or "").strip()
|
||
logger.error(
|
||
"FFprobe 命令失败: exit_code=%d command=%s\nstderr:\n%s",
|
||
e.returncode,
|
||
" ".join(str(c) for c in command[:20]),
|
||
stderr_text[:5000],
|
||
)
|
||
raise
|
||
|
||
|
||
def probe_has_audio(local_path: str | Path) -> bool:
|
||
"""探测文件是否包含音频流。
|
||
|
||
Args:
|
||
local_path: 本地文件路径
|
||
|
||
Returns:
|
||
True 表示有音频流（或探测失败保守返回），False 表示确认无音频流
|
||
"""
|
||
try:
|
||
result = subprocess.run( # nosec B603
|
||
[
|
||
FFPROBE_BIN,
|
||
"-v",
|
||
"error",
|
||
"-select_streams",
|
||
"a:0",
|
||
"-show_entries",
|
||
"stream=codec_type",
|
||
"-of",
|
||
"default=noprint_wrappers=1:nokey=1",
|
||
str(local_path),
|
||
],
|
||
check=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
timeout=10,
|
||
)
|
||
return result.stdout.strip() == "audio"
|
||
except Exception:
|
||
# 探测失败保守返回 True，让 FFmpeg 自己处理（避免误删音频）
|
||
return True
|
||
|
||
|
||
def probe_duration(local_path: str | Path) -> float:
|
||
"""用 ffprobe 获取视频时长（秒）。
|
||
|
||
失败时返回默认值 5.0 秒。
|
||
"""
|
||
try:
|
||
result = subprocess.run( # nosec B603
|
||
[
|
||
FFPROBE_BIN,
|
||
"-v",
|
||
"error",
|
||
"-show_entries",
|
||
"format=duration",
|
||
"-of",
|
||
"default=noprint_wrappers=1:nokey=1",
|
||
str(local_path),
|
||
],
|
||
check=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
)
|
||
return round(float(result.stdout.strip()), 3)
|
||
except Exception:
|
||
return 5.0
|
||
|
||
|
||
def probe_video_info(video_path: str) -> dict[str, Any]:
|
||
"""获取视频信息（宽、高、时长、fps、编码、像素格式）。
|
||
|
||
Returns:
|
||
{
|
||
"width": int, "height": int, "duration": float, "fps": float,
|
||
"video_codec": str, "audio_codec": str, "pix_fmt": str,
|
||
"has_audio": bool,
|
||
}
|
||
失败时返回默认值。
|
||
"""
|
||
try:
|
||
result = subprocess.run( # nosec B603
|
||
[
|
||
FFPROBE_BIN,
|
||
"-v",
|
||
"error",
|
||
"-show_entries",
|
||
"stream=width,height,r_frame_rate,duration,codec_name,codec_type,pix_fmt",
|
||
"-show_entries",
|
||
"format=duration",
|
||
"-of",
|
||
"json",
|
||
video_path,
|
||
],
|
||
check=True,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
timeout=15,
|
||
)
|
||
|
||
import json
|
||
|
||
info = json.loads(result.stdout)
|
||
streams = info.get("streams", [])
|
||
fmt = info.get("format", {})
|
||
|
||
video_stream = next((s for s in streams if s.get("codec_type") == "video"), {})
|
||
audio_stream = next((s for s in streams if s.get("codec_type") == "audio"), {})
|
||
|
||
width = int(video_stream.get("width", DEFAULT_OUTPUT_WIDTH))
|
||
height = int(video_stream.get("height", DEFAULT_OUTPUT_HEIGHT))
|
||
video_codec = video_stream.get("codec_name", "") or ""
|
||
pix_fmt = video_stream.get("pix_fmt", "") or ""
|
||
|
||
# 解析帧率
|
||
fps_str = video_stream.get("r_frame_rate", "25/1")
|
||
if "/" in fps_str:
|
||
num, den = fps_str.split("/")
|
||
fps = float(num) / float(den) if float(den) > 0 else DEFAULT_FPS
|
||
else:
|
||
fps = float(fps_str) if fps_str else DEFAULT_FPS
|
||
|
||
# 时长
|
||
duration = float(fmt.get("duration", 0)) or float(video_stream.get("duration", 0))
|
||
|
||
has_audio = bool(audio_stream)
|
||
audio_codec = audio_stream.get("codec_name", "") or ""
|
||
|
||
return {
|
||
"width": width,
|
||
"height": height,
|
||
"duration": duration,
|
||
"fps": round(fps, 2),
|
||
"video_codec": video_codec,
|
||
"audio_codec": audio_codec,
|
||
"pix_fmt": pix_fmt,
|
||
"has_audio": has_audio,
|
||
}
|
||
except Exception as e:
|
||
logger.warning("获取视频信息失败: %s, error: %s", video_path, e)
|
||
return {
|
||
"width": DEFAULT_OUTPUT_WIDTH,
|
||
"height": DEFAULT_OUTPUT_HEIGHT,
|
||
"duration": 0.0,
|
||
"fps": DEFAULT_FPS,
|
||
"video_codec": "",
|
||
"audio_codec": "",
|
||
"pix_fmt": "",
|
||
"has_audio": True,
|
||
}
|
||
|
||
|
||
def normalize_video(
|
||
input_path: str,
|
||
output_path: str,
|
||
*,
|
||
width: int = DEFAULT_OUTPUT_WIDTH,
|
||
height: int = DEFAULT_OUTPUT_HEIGHT,
|
||
fps: int = DEFAULT_FPS,
|
||
) -> dict[str, Any]:
|
||
"""标准化视频（缩放 + 恒定帧率）。
|
||
|
||
使用 scale + pad 保持宽高比，黑边填充到目标分辨率。
|
||
|
||
Returns:
|
||
{"width": int, "height": int, "path": str}
|
||
"""
|
||
command = [
|
||
FFMPEG_BIN,
|
||
"-y",
|
||
"-i",
|
||
input_path,
|
||
"-vf",
|
||
f"scale={width}:{height}:force_original_aspect_ratio=decrease,"
|
||
f"pad={width}:{height}:(ow-iw)/2:(oh-ih)/2:black,"
|
||
f"fps={fps}",
|
||
"-c:v",
|
||
"libx264",
|
||
"-crf",
|
||
"23",
|
||
"-preset",
|
||
"medium",
|
||
"-c:a",
|
||
"aac",
|
||
"-b:a",
|
||
"128k",
|
||
"-movflags",
|
||
"+faststart",
|
||
output_path,
|
||
]
|
||
run_ffmpeg(command)
|
||
return {"width": width, "height": height, "path": output_path}
|