423 lines
11 KiB
Python
423 lines
11 KiB
Python
import asyncio
|
||
import logging
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
from app.config import settings
|
||
|
||
logger = logging.getLogger("videogen")
|
||
|
||
|
||
class VideoCoverError(RuntimeError):
|
||
pass
|
||
|
||
|
||
def _clean_cover_format(value: str | None) -> str:
|
||
"""
|
||
清理封面格式。
|
||
|
||
只建议使用：
|
||
- jpg / jpeg
|
||
- png
|
||
- webp
|
||
|
||
默认 jpg，避免配置为空导致路径异常。
|
||
"""
|
||
ext = (value or "jpg").strip().lower().lstrip(".")
|
||
if not ext:
|
||
return "jpg"
|
||
|
||
allowed_exts = {"jpg", "jpeg", "png", "webp"}
|
||
if ext not in allowed_exts:
|
||
logger.warning("不支持的视频封面格式: %s，已回退为 jpg", ext)
|
||
return "jpg"
|
||
|
||
return ext
|
||
|
||
|
||
def _is_valid_file(path: str | None) -> bool:
|
||
if not path:
|
||
return False
|
||
|
||
try:
|
||
return os.path.isfile(path) and os.path.getsize(path) > 0
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _safe_remove(path: str | None) -> None:
|
||
if not path:
|
||
return
|
||
|
||
try:
|
||
if os.path.exists(path):
|
||
os.remove(path)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _make_part_path(final_path: str) -> str:
|
||
"""
|
||
生成原子写入临时文件路径。
|
||
|
||
重点：
|
||
- ffmpeg 会根据输出文件最后的扩展名判断格式。
|
||
- 不能生成 xxx.png.uuid.part，因为最后扩展名是 .part。
|
||
- 必须生成 xxx.uuid.part.png，让最后扩展名仍然是 .png / .jpg / .webp。
|
||
|
||
示例：
|
||
final_path:
|
||
storage/generate/covers/2026/06/05/0019.png
|
||
|
||
part_path:
|
||
storage/generate/covers/2026/06/05/0019.abcd1234.part.png
|
||
"""
|
||
path = Path(final_path)
|
||
|
||
suffix = path.suffix
|
||
if not suffix:
|
||
suffix = ".jpg"
|
||
|
||
return str(path.with_name(f"{path.stem}.{uuid.uuid4().hex}.part{suffix}"))
|
||
|
||
|
||
def get_ffmpeg_bin() -> str:
|
||
"""
|
||
获取 ffmpeg 可执行文件路径。
|
||
|
||
优先级：
|
||
1. 环境变量 FFMPEG_BIN
|
||
2. settings.FFMPEG_BIN
|
||
3. 系统 PATH 中的 ffmpeg / ffmpeg.exe
|
||
"""
|
||
env_bin = settings.FFMPEG_BIN
|
||
if env_bin:
|
||
ffmpeg_path = Path(env_bin)
|
||
if ffmpeg_path.exists():
|
||
return str(ffmpeg_path)
|
||
|
||
setting_bin = getattr(settings, "FFMPEG_BIN", None)
|
||
if setting_bin:
|
||
ffmpeg_path = Path(setting_bin)
|
||
if ffmpeg_path.exists():
|
||
return str(ffmpeg_path)
|
||
|
||
found = shutil.which("ffmpeg")
|
||
if found:
|
||
return found
|
||
|
||
found_exe = shutil.which("ffmpeg.exe")
|
||
if found_exe:
|
||
return found_exe
|
||
|
||
raise VideoCoverError("未找到 ffmpeg，请安装 ffmpeg 或配置环境变量 FFMPEG_BIN")
|
||
|
||
|
||
def _build_quality_args(output_path: str) -> list[str]:
|
||
"""
|
||
根据封面格式生成对应质量参数。
|
||
|
||
说明：
|
||
- jpg/jpeg：使用 -q:v，数值越小质量越高，3 比较适中。
|
||
- png：不使用 -q:v，避免无效参数造成干扰。
|
||
- webp：使用 -q:v 80，体积和质量比较均衡。
|
||
"""
|
||
ext = Path(output_path).suffix.lower().lstrip(".")
|
||
|
||
if ext in {"jpg", "jpeg"}:
|
||
return ["-q:v", "3"]
|
||
|
||
if ext == "webp":
|
||
return ["-q:v", "80"]
|
||
|
||
return []
|
||
|
||
|
||
def generate_video_cover(
|
||
video_path: str,
|
||
output_path: str,
|
||
seek_time: str = "00:00:01",
|
||
width: int = 720,
|
||
timeout: int = 15,
|
||
) -> str:
|
||
"""
|
||
从视频中截取封面图。
|
||
|
||
说明：
|
||
- 默认截第 1 秒，避免首帧黑屏。
|
||
- Windows / Linux 都可用。
|
||
- 不使用 shell=True，避免路径空格、命令注入问题。
|
||
- 使用 -nostdin + timeout，避免 ffmpeg 卡死 worker。
|
||
- 使用 -f image2 明确告诉 ffmpeg 输出单张图片。
|
||
- output_path 的最后后缀必须是 .jpg / .jpeg / .png / .webp。
|
||
"""
|
||
video_file = Path(video_path)
|
||
output_file = Path(output_path)
|
||
|
||
if not video_file.exists():
|
||
raise VideoCoverError(f"视频文件不存在: {video_file}")
|
||
|
||
if not video_file.is_file():
|
||
raise VideoCoverError(f"视频路径不是文件: {video_file}")
|
||
|
||
output_ext = output_file.suffix.lower().lstrip(".")
|
||
if output_ext not in {"jpg", "jpeg", "png", "webp"}:
|
||
raise VideoCoverError(
|
||
f"封面输出文件扩展名不支持: {output_file}，"
|
||
"请使用 .jpg / .jpeg / .png / .webp"
|
||
)
|
||
|
||
output_file.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
ffmpeg_bin = get_ffmpeg_bin()
|
||
|
||
safe_width = int(width or 720)
|
||
if safe_width <= 0:
|
||
safe_width = 720
|
||
|
||
scale_filter = f"scale={safe_width}:-2"
|
||
quality_args = _build_quality_args(str(output_file))
|
||
|
||
cmd = [
|
||
ffmpeg_bin,
|
||
"-y",
|
||
"-nostdin",
|
||
"-hide_banner",
|
||
"-loglevel",
|
||
"error",
|
||
|
||
# 放在 -i 前，速度更快。
|
||
"-ss",
|
||
seek_time,
|
||
|
||
"-i",
|
||
str(video_file),
|
||
|
||
# 只截一帧。
|
||
"-frames:v",
|
||
"1",
|
||
|
||
# 等比缩放，宽度固定，高度自动修正为偶数。
|
||
"-vf",
|
||
scale_filter,
|
||
|
||
# 明确输出单张图片，避免 ffmpeg 对临时文件格式判断异常。
|
||
"-f",
|
||
"image2",
|
||
|
||
*quality_args,
|
||
|
||
str(output_file),
|
||
]
|
||
|
||
try:
|
||
result = subprocess.run(
|
||
cmd,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.PIPE,
|
||
text=True,
|
||
encoding="utf-8",
|
||
errors="replace",
|
||
timeout=timeout,
|
||
)
|
||
except subprocess.TimeoutExpired as exc:
|
||
raise VideoCoverError(f"ffmpeg 截图超时: {video_file}") from exc
|
||
except OSError as exc:
|
||
raise VideoCoverError(f"ffmpeg 执行失败: {exc}") from exc
|
||
|
||
if result.returncode != 0:
|
||
err = (result.stderr or "").strip()
|
||
raise VideoCoverError(f"ffmpeg 截图失败: {err[-1000:]}")
|
||
|
||
if not output_file.exists() or output_file.stat().st_size <= 0:
|
||
raise VideoCoverError(f"封面生成失败，输出文件为空: {output_file}")
|
||
|
||
return str(output_file)
|
||
|
||
|
||
def generate_video_cover_atomically(
|
||
video_path: str,
|
||
output_path: str,
|
||
seek_time: str = "00:00:01",
|
||
width: int = 720,
|
||
timeout: int = 15,
|
||
) -> str:
|
||
"""
|
||
原子方式生成视频封面。
|
||
|
||
流程：
|
||
1. 如果最终封面已存在且有效，直接返回。
|
||
2. 先生成到临时文件。
|
||
3. 临时文件有效后 os.replace 到最终路径。
|
||
4. 任意异常清理临时文件。
|
||
|
||
关键：
|
||
临时文件最后后缀必须保留图片格式，例如：
|
||
- 正确：xxx.uuid.part.png
|
||
- 错误：xxx.png.uuid.part
|
||
"""
|
||
if _is_valid_file(output_path):
|
||
return output_path
|
||
|
||
output_file = Path(output_path)
|
||
output_file.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
part_path = _make_part_path(output_path)
|
||
|
||
try:
|
||
generate_video_cover(
|
||
video_path=video_path,
|
||
output_path=part_path,
|
||
seek_time=seek_time,
|
||
width=width,
|
||
timeout=timeout,
|
||
)
|
||
|
||
if not _is_valid_file(part_path):
|
||
raise VideoCoverError(f"封面临时文件为空: {part_path}")
|
||
|
||
os.replace(part_path, output_path)
|
||
|
||
if not _is_valid_file(output_path):
|
||
raise VideoCoverError(f"封面最终文件为空: {output_path}")
|
||
|
||
return output_path
|
||
|
||
except Exception:
|
||
_safe_remove(part_path)
|
||
raise
|
||
|
||
|
||
def build_video_cover_path_and_url(record_id: str, date_dir: str) -> tuple[str, str]:
|
||
"""
|
||
根据记录ID和日期目录生成本地封面文件路径与对外URL。
|
||
|
||
注意：
|
||
- 这里不做签名，接口返回时统一通过 build_resource_signed_url 处理。
|
||
- URL 固定使用 /generate/covers，便于 OpenResty/Nginx 统一做资源验签。
|
||
"""
|
||
ext = _clean_cover_format(getattr(settings, "VIDEO_COVER_FORMAT", "jpg"))
|
||
|
||
cover_dir = os.path.join(settings.STORAGE_VIDEO_COVER_LOCAL_PATH, date_dir)
|
||
cover_path = os.path.join(cover_dir, f"{record_id}.{ext}")
|
||
cover_url = f"/generate/covers/{date_dir}/{record_id}.{ext}"
|
||
|
||
return cover_path, cover_url
|
||
|
||
|
||
def try_generate_video_cover(
|
||
*,
|
||
video_path: str,
|
||
output_path: str,
|
||
seek_time: str | None = None,
|
||
fallback_seek_time: str | None = None,
|
||
width: int | None = None,
|
||
timeout: int | None = None,
|
||
log_prefix: str = "视频封面生成",
|
||
) -> str | None:
|
||
"""
|
||
best-effort 封面生成。
|
||
|
||
规则：
|
||
- 任何异常都捕获并记录日志。
|
||
- 主 seek_time 失败后，使用 fallback_seek_time 再尝试一次。
|
||
- 最终失败返回 None，不影响视频任务 completed。
|
||
"""
|
||
first_seek = seek_time or getattr(settings, "VIDEO_COVER_SEEK_TIME", "00:00:01")
|
||
second_seek = fallback_seek_time or getattr(
|
||
settings,
|
||
"VIDEO_COVER_FALLBACK_SEEK_TIME",
|
||
"00:00:03",
|
||
)
|
||
cover_width = width or getattr(settings, "VIDEO_COVER_WIDTH", 720)
|
||
cover_timeout = timeout or getattr(settings, "VIDEO_COVER_TIMEOUT_SECONDS", 15)
|
||
|
||
try:
|
||
return generate_video_cover_atomically(
|
||
video_path=video_path,
|
||
output_path=output_path,
|
||
seek_time=first_seek,
|
||
width=cover_width,
|
||
timeout=cover_timeout,
|
||
)
|
||
except Exception as first_exc:
|
||
if second_seek and second_seek != first_seek:
|
||
try:
|
||
return generate_video_cover_atomically(
|
||
video_path=video_path,
|
||
output_path=output_path,
|
||
seek_time=second_seek,
|
||
width=cover_width,
|
||
timeout=cover_timeout,
|
||
)
|
||
except Exception as second_exc:
|
||
logger.warning(
|
||
"%s失败，已忽略，不影响视频生成成功。video_path=%s, output_path=%s, first_error=%s, fallback_error=%s",
|
||
log_prefix,
|
||
video_path,
|
||
output_path,
|
||
first_exc,
|
||
second_exc,
|
||
exc_info=True,
|
||
)
|
||
return None
|
||
|
||
logger.warning(
|
||
"%s失败，已忽略，不影响视频生成成功。video_path=%s, output_path=%s, error=%s",
|
||
log_prefix,
|
||
video_path,
|
||
output_path,
|
||
first_exc,
|
||
exc_info=True,
|
||
)
|
||
return None
|
||
|
||
|
||
def create_video_cover_for_local_video(
|
||
*,
|
||
record_id: str,
|
||
video_path: str,
|
||
date_dir: str,
|
||
log_prefix: str = "视频封面生成",
|
||
) -> tuple[str | None, str | None]:
|
||
"""
|
||
给本地视频生成封面。
|
||
|
||
返回：
|
||
- cover_url: 成功时为 /generate/covers/...，失败为 None
|
||
- cover_path: 成功时为本地文件路径，失败为 None
|
||
"""
|
||
cover_path, cover_url = build_video_cover_path_and_url(record_id, date_dir)
|
||
|
||
generated_path = try_generate_video_cover(
|
||
video_path=video_path,
|
||
output_path=cover_path,
|
||
log_prefix=log_prefix,
|
||
)
|
||
|
||
if not generated_path:
|
||
return None, None
|
||
|
||
return cover_url, generated_path
|
||
|
||
|
||
async def async_create_video_cover_for_local_video(
|
||
*,
|
||
record_id: str,
|
||
video_path: str,
|
||
date_dir: str,
|
||
log_prefix: str = "视频封面生成",
|
||
) -> tuple[str | None, str | None]:
|
||
"""
|
||
异步封装，避免在 FastAPI event loop 中直接执行 ffmpeg 阻塞。
|
||
"""
|
||
return await asyncio.to_thread(
|
||
create_video_cover_for_local_video,
|
||
record_id=record_id,
|
||
video_path=video_path,
|
||
date_dir=date_dir,
|
||
log_prefix=log_prefix,
|
||
) |