# -*- coding: utf-8 -*- import logging import os import subprocess import tempfile from typing import List, Tuple from app.video_io import get_ffmpeg_cmd logger = logging.getLogger(__name__) def sample_frames(video_path: str, sample_fps: float, duration: float) -> List[Tuple[float, str]]: """按 sample_fps 抽帧,返回 [(time_sec, jpg_path), ...]""" if sample_fps <= 0: sample_fps = 0.5 step = 1.0 / sample_fps out_dir = tempfile.mkdtemp(prefix="snap_frames_") frames: List[Tuple[float, str]] = [] t = 0.0 idx = 0 ffmpeg = get_ffmpeg_cmd("ffmpeg") while t <= duration: out_path = os.path.join(out_dir, f"{idx:06d}.jpg") cmd = [ ffmpeg, "-y", "-ss", f"{t:.3f}", "-i", video_path, "-frames:v", "1", "-q:v", "2", out_path, ] print(f"cmd{cmd}") subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) if os.path.isfile(out_path) and os.path.getsize(out_path) > 0: frames.append((round(t, 3), out_path)) t += step idx += 1 return frames def cleanup_frames(frames: List[Tuple[float, str]]) -> None: if not frames: return out_dir = os.path.dirname(frames[0][1]) import shutil shutil.rmtree(out_dir, ignore_errors=True)