Python视频工具案例如何封装视频处理

本文目录导读：

1. 基础视频处理工具类封装
2. GUI视频处理工具

我来为你详细介绍视频处理工具封装的方法,包含完整的案例代码。

基础视频处理工具类封装

1 使用MoviePy的封装示例

import moviepy.editor as mp
from moviepy.video.fx import resize, rotate
import numpy as np
from typing import Optional, Tuple, List
import os
class VideoProcessor:
    """视频处理工具类"""
    def __init__(self, video_path: str):
        """
        初始化视频处理器
        Args:
            video_path: 视频文件路径
        """
        self.video_path = video_path
        self.video = None
        self._load_video()
    def _load_video(self):
        """加载视频文件"""
        if not os.path.exists(self.video_path):
            raise FileNotFoundError(f"视频文件不存在: {self.video_path}")
        self.video = mp.VideoFileClip(self.video_path)
    def cut_video(self, start_time: float, end_time: float, 
                  output_path: str) -> bool:
        """
        截取视频片段
        Args:
            start_time: 开始时间（秒）
            end_time: 结束时间（秒）
            output_path: 输出路径
        Returns:
            bool: 是否成功
        """
        try:
            clip = self.video.subclip(start_time, end_time)
            clip.write_videofile(output_path, codec="libx264")
            clip.close()
            return True
        except Exception as e:
            print(f"截取失败: {e}")
            return False
    def resize_video(self, width: int, height: int, 
                     output_path: str) -> bool:
        """
        调整视频尺寸
        Args:
            width: 目标宽度
            height: 目标高度
            output_path: 输出路径
        """
        try:
            resized = self.video.resize((width, height))
            resized.write_videofile(output_path, codec="libx264")
            resized.close()
            return True
        except Exception as e:
            print(f"调整尺寸失败: {e}")
            return False
    def add_text(self, text: str, position: Tuple[int, int], 
                 duration: float, output_path: str,
                 font_size: int = 24, color: str = "white") -> bool:
        """
        添加文本字幕
        Args:
            text: 文本内容
            position: 位置 (x, y)
            duration: 持续时间（秒）
            output_path: 输出路径
            font_size: 字体大小
            color: 字体颜色
        """
        try:
            txt_clip = mp.TextClip(text, fontsize=font_size, 
                                   color=color, font="Arial")
            txt_clip = txt_clip.set_position(position).set_duration(duration)
            video_with_text = mp.CompositeVideoClip([self.video, txt_clip])
            video_with_text.write_videofile(output_path, codec="libx264")
            txt_clip.close()
            video_with_text.close()
            return True
        except Exception as e:
            print(f"添加文字失败: {e}")
            return False
    def merge_videos(self, video_list: List[str], 
                     output_path: str) -> bool:
        """
        合并多个视频
        Args:
            video_list: 视频文件路径列表
            output_path: 输出路径
        """
        try:
            clips = [mp.VideoFileClip(v) for v in video_list]
            final_clip = mp.concatenate_videoclips(clips)
            final_clip.write_videofile(output_path, codec="libx264")
            for clip in clips:
                clip.close()
            final_clip.close()
            return True
        except Exception as e:
            print(f"合并视频失败: {e}")
            return False
    def extract_audio(self, output_path: str, 
                      audio_format: str = "mp3") -> bool:
        """
        提取音频
        Args:
            output_path: 输出路径
            audio_format: 音频格式
        """
        try:
            audio = self.video.audio
            audio.write_audiofile(output_path, codec=audio_format)
            audio.close()
            return True
        except Exception as e:
            print(f"提取音频失败: {e}")
            return False
    def add_audio(self, audio_path: str, output_path: str,
                  volume: float = 1.0) -> bool:
        """
        添加音频
        Args:
            audio_path: 音频文件路径
            output_path: 输出路径
            volume: 音量 (0.0-1.0)
        """
        try:
            audio_clip = mp.AudioFileClip(audio_path).volumex(volume)
            video_with_audio = self.video.set_audio(audio_clip)
            video_with_audio.write_videofile(output_path, codec="libx264")
            audio_clip.close()
            video_with_audio.close()
            return True
        except Exception as e:
            print(f"添加音频失败: {e}")
            return False
    def get_video_info(self) -> dict:
        """
        获取视频信息
        Returns:
            dict: 视频信息字典
        """
        return {
            "duration": self.video.duration,
            "fps": self.video.fps,
            "width": self.video.w,
            "height": self.video.h,
            "size": self.video.size,
            "rotation": self.video.rotation
        }
    def close(self):
        """关闭视频文件"""
        if self.video:
            self.video.close()
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

2 使用OpenCV的封装示例

import cv2
import numpy as np
from pathlib import Path
import tempfile
import subprocess
class OpenCVVideoProcessor:
    """使用OpenCV的视频处理工具类"""
    def __init__(self, video_path: str):
        self.video_path = video_path
        self.cap = None
        self.fps = 0
        self.total_frames = 0
        self.width = 0
        self.height = 0
        self._load_video()
    def _load_video(self):
        """加载视频"""
        self.cap = cv2.VideoCapture(self.video_path)
        if not self.cap.isOpened():
            raise ValueError(f"无法打开视频文件: {self.video_path}")
        self.fps = int(self.cap.get(cv2.CAP_PROP_FPS))
        self.total_frames = int(self.cap.get(cv2.CAP_PROP_FRAME_COUNT))
        self.width = int(self.cap.get(cv2.CAP_PROP_FRAME_WIDTH))
        self.height = int(self.cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    def apply_filter(self, filter_type: str = "gray", 
                     output_path: str = None) -> bool:
        """
        应用滤镜
        Args:
            filter_type: 滤镜类型 (gray, blur, edge, invert)
            output_path: 输出路径
        """
        if output_path is None:
            output_path = f"filtered_{Path(self.video_path).name}"
        fourcc = cv2.VideoWriter_fourcc(*'mp4v')
        out = cv2.VideoWriter(output_path, fourcc, self.fps, 
                             (self.width, self.height))
        while True:
            ret, frame = self.cap.read()
            if not ret:
                break
            if filter_type == "gray":
                frame = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
                frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)
            elif filter_type == "blur":
                frame = cv2.GaussianBlur(frame, (15, 15), 0)
            elif filter_type == "edge":
                gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
                frame = cv2.Canny(gray, 100, 200)
                frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)
            elif filter_type == "invert":
                frame = cv2.bitwise_not(frame)
            out.write(frame)
        out.release()
        self.cap.release()
        return True
    def extract_frames(self, frame_interval: int = 1, 
                       output_dir: str = "frames") -> List[str]:
        """
        提取帧
        Args:
            frame_interval: 帧间隔
            output_dir: 输出目录
        Returns:
            List[str]: 保存的帧文件路径列表
        """
        Path(output_dir).mkdir(parents=True, exist_ok=True)
        frame_files = []
        frame_count = 0
        while True:
            ret, frame = self.cap.read()
            if not ret:
                break
            if frame_count % frame_interval == 0:
                frame_path = f"{output_dir}/frame_{frame_count:06d}.jpg"
                cv2.imwrite(frame_path, frame)
                frame_files.append(frame_path)
            frame_count += 1
        self.cap.release()
        return frame_files
    def __del__(self):
        """析构函数"""
        if self.cap:
            self.cap.release()

3 使用FFmpeg的高级封装

import subprocess
import json
from pathlib import Path
from typing import Optional
class FFmpegVideoProcessor:
    """使用FFmpeg的高级视频处理"""
    def __init__(self, ffmpeg_path: str = "ffmpeg"):
        self.ffmpeg_path = ffmpeg_path
    def compress_video(self, input_path: str, output_path: str,
                       target_size_mb: int = 10) -> bool:
        """
        压缩视频到目标大小
        Args:
            input_path: 输入路径
            output_path: 输出路径
            target_size_mb: 目标大小（MB）
        """
        # 获取视频信息
        probe_cmd = [
            self.ffmpeg_path,
            "-i", input_path,
            "-f", "null",
            "-"
        ]
        # 计算比特率
        duration = self._get_duration(input_path)
        target_bitrate = (target_size_mb * 8 * 1024 * 1024) / duration
        cmd = [
            self.ffmpeg_path,
            "-i", input_path,
            "-b:v", f"{int(target_bitrate)}",
            "-bufsize", f"{int(target_bitrate * 2)}",
            "-maxrate", f"{int(target_bitrate * 1.5)}",
            "-y",
            output_path
        ]
        try:
            subprocess.run(cmd, check=True, capture_output=True)
            return True
        except subprocess.CalledProcessError as e:
            print(f"压缩失败: {e.stderr.decode()}")
            return False
    def add_subtitles(self, input_path: str, subtitle_path: str,
                      output_path: str) -> bool:
        """
        添加字幕
        Args:
            input_path: 输入路径
            subtitle_path: 字幕文件路径
            output_path: 输出路径
        """
        cmd = [
            self.ffmpeg_path,
            "-i", input_path,
            "-vf", f"subtitles={subtitle_path}",
            "-y",
            output_path
        ]
        try:
            subprocess.run(cmd, check=True)
            return True
        except subprocess.CalledProcessError as e:
            print(f"添加字幕失败: {e}")
            return False
    def change_speed(self, input_path: str, output_path: str,
                     speed_factor: float = 1.5) -> bool:
        """
        改变视频速度
        Args:
            input_path: 输入路径
            output_path: 输出路径
            speed_factor: 速度因子 (1.0=正常)
        """
        # 计算相应的音频和视频滤镜
        video_filter = f"setpts={1/speed_factor}*PTS"
        audio_filter = f"atempo={min(speed_factor, 2.0)}"
        # 如果速度因子大于2.0，需要多次应用音频滤镜
        if speed_factor > 2.0:
            audio_filter = f"atempo={speed_factor:.2f}[a1];{audio_filter}"
        cmd = [
            self.ffmpeg_path,
            "-i", input_path,
            "-vf", video_filter,
            "-af", audio_filter,
            "-y",
            output_path
        ]
        try:
            subprocess.run(cmd, check=True)
            return True
        except subprocess.CalledProcessError as e:
            print(f"改变速度失败: {e}")
            return False
    def _get_duration(self, video_path: str) -> float:
        """获取视频时长（秒）"""
        cmd = [
            self.ffmpeg_path,
            "-i", video_path,
            "-f", "null",
            "-"
        ]
        result = subprocess.run(cmd, capture_output=True, text=True)
        # 从输出中提取时长
        import re
        duration_match = re.search(r"Duration: (\d+):(\d+):(\d+\.\d+)", 
                                  result.stderr)
        if duration_match:
            hours = int(duration_match.group(1))
            minutes = int(duration_match.group(2))
            seconds = float(duration_match.group(3))
            return hours * 3600 + minutes * 60 + seconds
        return 0.0

GUI视频处理工具

import tkinter as tk
from tkinter import ttk, filedialog, messagebox
from pathlib import Path
import threading
class VideoProcessorGUI:
    """视频处理GUI工具"""
    def __init__(self):
        self.root = tk.Tk()
        self.root.title("视频处理工具")
        self.root.geometry("600x400")
        self.processor = None
        self.input_path = ""
        self.output_path = ""
        self._setup_ui()
    def _setup_ui(self):
        """设置UI"""
        # 文件选择
        file_frame = ttk.LabelFrame(self.root, text="文件选择", padding=10)
        file_frame.pack(fill="x", padx=10, pady=5)
        ttk.Button(file_frame, text="选择视频", 
                  command=self._select_file).pack(side="left", padx=5)
        self.file_label = ttk.Label(file_frame, text="未选择文件")
        self.file_label.pack(side="left", padx=10)
        # 操作选择
        operation_frame = ttk.LabelFrame(self.root, text="操作", padding=10)
        operation_frame.pack(fill="x", padx=10, pady=5)
        self.operation_var = tk.StringVar(value="cut")
        operations = [
            ("截取视频", "cut"),
            ("调整大小", "resize"),
            ("添加文字", "text"),
            ("提取音频", "audio"),
            ("添加滤镜", "filter")
        ]
        for text, value in operations:
            ttk.Radiobutton(operation_frame, text=text, 
                           variable=self.operation_var,
                           value=value).pack(side="left", padx=10)
        # 参数设置
        param_frame = ttk.LabelFrame(self.root, text="参数设置", padding=10)
        param_frame.pack(fill="both", expand=True, padx=10, pady=5)
        # 不同的操作显示不同的参数
        self.param_frame = param_frame
        self._show_cut_params()
        # 操作按钮
        button_frame = ttk.Frame(self.root, padding=10)
        button_frame.pack(fill="x", padx=10, pady=5)
        ttk.Button(button_frame, text="选择输出路径", 
                  command=self._select_output).pack(side="left", padx=5)
        ttk.Button(button_frame, text="开始处理", 
                  command=self._start_processing).pack(side="left", padx=5)
        ttk.Button(button_frame, text="退出", 
                  command=self.root.quit).pack(side="right", padx=5)
        # 进度条
        self.progress = ttk.Progressbar(self.root, mode="indeterminate")
        self.progress.pack(fill="x", padx=10, pady=5)
    def _show_cut_params(self):
        """显示截取参数"""
        for widget in self.param_frame.winfo_children():
            widget.destroy()
        ttk.Label(self.param_frame, text="开始时间（秒）:").grid(row=0, column=0, padx=5, pady=5)
        self.start_time = ttk.Entry(self.param_frame)
        self.start_time.grid(row=0, column=1, padx=5, pady=5)
        self.start_time.insert(0, "0")
        ttk.Label(self.param_frame, text="结束时间（秒）:").grid(row=1, column=0, padx=5, pady=5)
        self.end_time = ttk.Entry(self.param_frame)
        self.end_time.grid(row=1, column=1, padx=5, pady=5)
        self.end_time.insert(0, "10")
    def _select_file(self):
        """选择文件"""
        file_path = filedialog.askopenfilename(
            title="选择视频文件",
            filetypes=[("视频文件", "*.mp4 *.avi *.mov *.mkv")]
        )
        if file_path:
            self.input_path = file_path
            self.file_label.config(text=Path(file_path).name)
    def _select_output(self):
        """选择输出路径"""
        file_path = filedialog.asksaveasfilename(
            title="选择输出路径",
            defaultextension=".mp4",
            filetypes=[("MP4文件", "*.mp4"), ("所有文件", "*.*")]
        )
        if file_path:
            self.output_path = file_path
    def _start_processing(self):
        """开始处理"""
        if not self.input_path or not self.output_path:
            messagebox.showerror("错误", "请选择输入和输出文件")
            return
        # 在新线程中处理，避免阻塞GUI
        thread = threading.Thread(target=self._process_video)
        thread.start()
    def _process_video(self):
        """处理视频"""
        self.progress.start()
        try:
            # 根据选择的操