基于boto3实现帧处理后视频分片上传至AWS S3的问题

嘿，我来帮你解决这个问题！你现在的核心痛点是如何把处理后的视频帧转换成标准的MP4字节流，并且按AWS要求的5MB分片上传，同时还要避免写入本地磁盘。我先梳理下你现有代码的小问题，再给出完整的解决方案。

先修正你的MultipartUpload类的小问题 #

你的原类里有两个明显的bug：

1. open方法里直接用了s3而不是self.s3，会触发NameError
2. upload方法需要手动传入part_number，不如在类内部自动维护分片编号更方便

修正后的类如下：

import boto3

class MultipartUpload:
    def __init__(self, bucket, key):
        self.bucket = bucket
        self.key = key
        self.s3 = boto3.client("s3")
        self.parts = []
        self.part_number = 1  # 内部维护分片编号

    def open(self):
        response = self.s3.create_multipart_upload(
            ACL="public-read",
            Bucket=self.bucket,
            ContentType="video/mp4",
            Key=self.key
        )
        self.upload_id = response["UploadId"]

    def upload(self, bytes_data):
        if not bytes_data:
            return
        response = self.s3.upload_part(
            Bucket=self.bucket,
            Key=self.key,
            Body=bytes_data,
            PartNumber=self.part_number,
            UploadId=self.upload_id
        )
        self.parts.append({"ETag": response["ETag"], "PartNumber": self.part_number})
        self.part_number += 1  # 分片编号自动递增

    def close(self):
        response = self.s3.complete_multipart_upload(
            Bucket=self.bucket,
            Key=self.key,
            UploadId=self.upload_id,
            MultipartUpload={"Parts": self.parts}
        )
        print("分片上传完成：", response)

核心解决方案：将处理后的帧编码为MP4字节流并分片上传 #

直接处理单帧无法生成标准的MP4文件，必须通过编码器把连续帧转换成视频格式的字节流。这里推荐用ffmpeg-python（需要先安装：pip install ffmpeg-python），它能在内存中完成编码，完全不需要写入本地磁盘，还能生成兼容的MP4流。

完整的处理+上传代码：

import cv2
from io import BytesIO
import ffmpeg

def main():
    AWS_BUCKET_NAME = "你的S3桶名"
    AWS_KEY = "上传后的视频文件名.mp4"
    video_path = "本地视频路径.mp4"

    # 初始化分片上传
    multipart_upload = MultipartUpload(AWS_BUCKET_NAME, AWS_KEY)
    multipart_upload.open()

    # 打开本地视频
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        print("打开视频失败！")
        return

    # 获取原视频的关键参数（保证输出视频和原视频匹配）
    fps = cap.get(cv2.CAP_PROP_FPS)
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))

    # 设置FFmpeg编码管道：从标准输入读原始帧，输出MP4字节流到标准输出
    process = (
        ffmpeg
        .input('pipe:', format='rawvideo', pix_fmt='bgr24', s=f'{width}x{height}', r=fps)
        .output('pipe:', format='mp4', vcodec='libx264', pix_fmt='yuv420p')
        .run_async(pipe_stdin=True, pipe_stdout=True)
    )

    chunk_size = 5 * 1024 * 1024  # AWS要求的最小分片大小5MB
    current_chunk = b''

    while cap.isOpened():
        ret, frame = cap.read()
        if not ret:
            break

        # --- 这里写你的帧处理逻辑 ---
        processed_frame = frame  # 示例：直接使用原帧，替换成你的处理代码
        # --- 帧处理结束 ---

        # 将处理后的帧写入FFmpeg编码管道
        process.stdin.write(processed_frame.tobytes())

        # 读取FFmpeg输出的MP4字节，累计到分片缓冲区
        while True:
            # 每次读1MB，避免一次性占用过多内存
            data = process.stdout.read(1024*1024)
            if not data:
                break
            current_chunk += data

            # 当缓冲区达到5MB时，上传该分片
            if len(current_chunk) >= chunk_size:
                multipart_upload.upload(current_chunk)
                current_chunk = b''

    # 关闭FFmpeg输入管道，等待编码完成
    process.stdin.close()
    process.wait()

    # 读取剩余的编码数据
    remaining_data = process.stdout.read()
    if remaining_data:
        current_chunk += remaining_data

    # 上传最后一个分片
    if len(current_chunk) > 0:
        multipart_upload.upload(current_chunk)

    # 完成分片上传
    multipart_upload.close()

    # 释放资源
    cap.release()

if __name__ == "__main__":
    main()

方案说明 #

1. FFmpeg编码管道：通过管道将处理后的帧传递给FFmpeg，实时编码成标准MP4字节流，完全在内存中操作，不需要本地文件。
2. 分片逻辑：累计编码后的字节数据，达到5MB就上传一个分片，最后处理剩余的不足5MB的部分（AWS允许最后一个分片小于5MB）。
3. 参数匹配：复用原视频的分辨率、帧率，保证输出视频的兼容性。

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