如何加快Moviepy导出长视频的速度？优化方案咨询

问题背景 #

使用Moviepy 2.1.2（搭配PIL 9.5.0）将音频与图片合成带缩放淡出效果的2.5小时720p长视频，当前导出耗时1.5-2小时。已尝试调整write_videofile的preset和threads参数，禁用缩放效果时导出速度约23it/s，启用时约24it/s，需进一步优化导出速度。

现有代码 #

主生成函数：

# import
import math
import numpy

from PIL import Image

from moviepy import AudioFileClip, ImageClip, CompositeVideoClip
from moviepy.audio.fx import MultiplyVolume
from moviepy.video.fx import FadeOut


def generate_video(video_folder_path, audio_file_path, image_file_path, temp_path):

    # define zoom effect function
    def zoom_effect(clip, zoom_ratio=1.2):
        def effect(get_frame, t):
            img = Image.fromarray(get_frame(t))
            base_size = img.size

            new_size = [
                math.ceil(img.size[0] * zoom_ratio * 1 / (1 - (t / clip.duration) * (1 - zoom_ratio))),
                math.ceil(img.size[1] * zoom_ratio * 1 / (1 - (t / clip.duration) * (1 - zoom_ratio)))
            ]

            img = img.resize(new_size, Image.LANCZOS)

            x = math.ceil((new_size[0] - base_size[0]) / 2)
            y = math.ceil((new_size[1] - base_size[1]) / 2)

            img = img.crop([
                x, y, new_size[0] - x, new_size[1] - y
            ]).resize(base_size, Image.LANCZOS)

            result = numpy.array(img)
            img.close()

            return result

        return clip.transform(effect)

    # define export settings
    audio_volume = 1
    aspect_ratio = 16 / 9
    z_ratio = 1.18
    fade_out_duration = 1.15
    export_fps = 24

    # create clips
    audio_clip = AudioFileClip(audio_file_path).with_effects([MultiplyVolume(audio_volume)])
    image_clip = ImageClip(image_file_path, duration=audio_clip.duration).with_position(('center', 'center'))

    # set dimensions
    height = image_clip.size[1]
    width = int(height * aspect_ratio)

    # apply zoom effect
    image_clip = zoom_effect(image_clip, z_ratio)

    # compose
    video_clip = CompositeVideoClip([image_clip], size=(width, height)).with_effects([FadeOut(fade_out_duration)])

    # set audio
    video_clip.audio = audio_clip

    # write
    video_clip.write_videofile(
        video_folder_path + '/main_video.mp4', fps=export_fps,
        temp_audiofile_path=temp_path, preset='ultrafast', threads=18)

批量执行循环：

for video_folder in sorted([f for f in os.scandir('path/to/directory')
                            if os.path.isdir(f)], key=lambda x: id_to_num(x.name)):
    generate_video(video_folder.path)

具体优化方案 #

1. 重构缩放效果，减少计算与格式转换开销 #

当前自定义zoom_effect的多次格式转换、重复计算和慢插值是性能核心瓶颈，可从以下几点优化：

• 替换插值算法：将Image.LANCZOS改为Image.BICUBIC或Image.BILINEAR，视觉差异可接受的前提下，大幅提升缩放速度；
• 预计算固定参数：提前缓存视频时长、基础尺寸等固定值，避免每帧重复计算；
• 减少格式转换：直接用numpy/OpenCV处理帧，跳过PIL的数组与图像互转步骤；

优化后的缩放效果示例：

import cv2

def zoom_effect(clip, zoom_ratio=1.2):
    duration = clip.duration
    base_w, base_h = clip.size
    
    def scale_factor(t):
        return 1 - (t / duration) * (1 - zoom_ratio)
    
    def effect(get_frame, t):
        frame = get_frame(t)
        sf = scale_factor(t)
        new_w = math.ceil(base_w * zoom_ratio / sf)
        new_h = math.ceil(base_h * zoom_ratio / sf)
        
        # 用OpenCV线性插值缩放，比PIL LANCZOS更快
        resized_frame = cv2.resize(frame, (new_w, new_h), interpolation=cv2.INTER_LINEAR)
        # 计算中心裁剪区域
        x_start = (new_w - base_w) // 2
        y_start = (new_h - base_h) // 2
        cropped_frame = resized_frame[y_start:y_start+base_h, x_start:x_start+base_w]
        
        return cropped_frame
    
    return clip.transform(effect)

2. 启用硬件加速编码 #

Moviepy默认使用CPU编码，改用硬件编码器可将导出速度提升数倍：

• NVIDIA显卡：设置codec='h264_nvenc'；
• AMD显卡：设置codec='h264_amf'；
• 调整threads为CPU核心数（如threads=os.cpu_count()），避免线程过多导致调度开销。

修改后的导出代码：

import os

video_clip.write_videofile(
    f"{video_folder_path}/main_video.mp4", 
    fps=export_fps,
    temp_audiofile_path=temp_path, 
    preset='ultrafast', 
    threads=os.cpu_count(),
    codec='h264_nvenc'  # 根据显卡类型替换
)

3. 预处理图片，减少每帧计算 #

• 提前调整图片尺寸：将原始图片预转为目标视频分辨率（720p），避免每帧渲染时重复调整：

# 创建ImageClip前预处理图片
with Image.open(image_file_path) as img:
    target_size = (width, height)
    img = img.resize(target_size, Image.BILINEAR)
    preprocessed_img_path = f"{temp_path}/preprocessed_img.jpg"
    img.save(preprocessed_img_path)

# 使用预处理后的图片创建剪辑
image_clip = ImageClip(preprocessed_img_path, duration=audio_clip.duration).with_position(('center', 'center'))

• 移除冗余复合剪辑：仅单剪辑时直接设置尺寸和效果，无需嵌套CompositeVideoClip：

# 替换原CompositeVideoClip代码
video_clip = image_clip.set_size((width, height)).with_effects([FadeOut(fade_out_duration)])

4. 批量处理并行化 #

将串行处理改为并行处理，充分利用CPU资源：

from concurrent.futures import ProcessPoolExecutor

def process_single_folder(video_folder):
    # 补充generate_video所需的参数逻辑，如audio_file_path、image_file_path的获取
    generate_video(video_folder.path, ...)

if __name__ == "__main__":
    folders = sorted([f for f in os.scandir('path/to/directory') if os.path.isdir(f)], 
                     key=lambda x: id_to_num(x.name))
    # 按CPU核心数的一半设置并行数，避免资源耗尽
    with ProcessPoolExecutor(max_workers=os.cpu_count()//2) as executor:
        executor.map(process_single_folder, folders)

5. 优化IO性能 #

• 将临时文件目录temp_path和输出目录都设置在SSD上，机械硬盘的IO速度会严重拖慢导出进度；
• 导出过程中避免同时进行其他磁盘密集型操作。

内容的提问来源于stack exchange，提问作者Dev