239 lines
6.7 KiB
Python
239 lines
6.7 KiB
Python
import logging
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from app.config import settings
|
|
|
|
import asyncio
|
|
|
|
logger = logging.getLogger("videogen")
|
|
|
|
|
|
class VideoCoverError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _clean_cover_format(value: str | None) -> str:
|
|
ext = (value or "jpg").strip().lower().lstrip(".")
|
|
return ext or "jpg"
|
|
|
|
|
|
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)
|
|
|
|
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 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。
|
|
- output_path 建议使用 .jpg 或 .webp。
|
|
"""
|
|
video_file = Path(video_path)
|
|
output_file = Path(output_path)
|
|
|
|
if not video_file.exists():
|
|
raise VideoCoverError(f"视频文件不存在: {video_file}")
|
|
|
|
output_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
ffmpeg_bin = get_ffmpeg_bin()
|
|
scale_filter = f"scale={width}:-2"
|
|
|
|
cmd = [
|
|
ffmpeg_bin,
|
|
"-y",
|
|
"-nostdin",
|
|
"-hide_banner",
|
|
"-loglevel",
|
|
"error",
|
|
"-ss",
|
|
seek_time,
|
|
"-i",
|
|
str(video_file),
|
|
"-frames:v",
|
|
"1",
|
|
"-vf",
|
|
scale_filter,
|
|
"-q:v",
|
|
"3",
|
|
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 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(settings.VIDEO_COVER_FORMAT)
|
|
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 settings.VIDEO_COVER_SEEK_TIME
|
|
second_seek = fallback_seek_time or settings.VIDEO_COVER_FALLBACK_SEEK_TIME
|
|
cover_width = width or settings.VIDEO_COVER_WIDTH
|
|
cover_timeout = timeout or settings.VIDEO_COVER_TIMEOUT_SECONDS
|
|
|
|
try:
|
|
return generate_video_cover(
|
|
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(
|
|
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]:
|
|
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,
|
|
)
|