如何使用Pytube将YouTube视频直接下载到AWS S3存储桶

核心逻辑是跳过本地磁盘写入步骤，将Pytube拉取的视频流直接写入内存缓冲区，再通过boto3 SDK上传到S3，完全不依赖EC2的本地存储，也避免了S3挂载方案的兼容问题。

前置准备 #

• 给你的EC2实例绑定IAM角色，角色需配置对应S3存储桶的s3:PutObject、s3:ListBucket权限，避免在代码中硬写密钥
• 安装依赖：pip install boto3 pytube

修改后的核心代码 #

保留你原有的文件名处理逻辑，只替换下载、校验、上传的部分即可：

import boto3
import io
import re
import unicodedata
from pytube import YouTube

# 保留你原来的文件名处理函数不变
def safe_filename(s: str, max_length: int = 255) -> str:
    ntfs_characters = [chr(i) for i in range(0, 31)]
    characters = [r'"',r"\#",r"\$",r"\%",r"'",r"\*",r"\,",r"\.",r"\/",r"\:",r'"',r"\;",r"\<",r"\>",r"\?",r"\\",r"\^",r"\|",r"\~",r"\\\\",r"\(",r"\)"]
    pattern = "|".join(ntfs_characters + characters)
    regex = re.compile(pattern, re.UNICODE)
    filename = regex.sub("", s)
    return filename[:max_length].rsplit(" ", 0)[0]

def simplify(text):
    try:
        text = unicode(text, 'utf-8')
    except NameError:
        pass
    text = unicodedata.normalize('NFD', text.decode('utf-8')).encode('ascii', 'ignore').decode("utf-8")
    return str(text)

# 初始化S3客户端
s3 = boto3.client('s3', region_name='替换为你的S3桶所在区域，比如ap-southeast-1')
BUCKET_NAME = '替换为你的S3桶名'
S3_PREFIX = '替换为你要存储的路径前缀，比如YT_Interviews_Data/Videos/'

def downloader(lnks, threadlabel):
    failed=[]
    for link in lnks:
        try:
            yt = YouTube(link)
            # 原有文件名处理逻辑
            name= safe_filename(yt.title)
            name= name.strip()
            name= name.encode("utf-8")
            name= simplify(name)
            name= re.sub('[^A-Za-z0-9]+', '', name)
            name= name.replace('[^\w\s]', '')
            file_name = f"{name}.mp4"
            s3_key = f"{S3_PREFIX}{file_name}"
            print(f"处理文件：{file_name}")
            
            # 校验S3中是否已存在该文件，避免重复下载
            try:
                s3.head_object(Bucket=BUCKET_NAME, Key=s3_key)
                print(f"{file_name} 已存在，跳过下载")
                continue
            except s3.exceptions.ClientError as e:
                if e.response['Error']['Code'] != '404':
                    raise
            
            # 获取目标视频流
            stream = yt.streams.filter(progressive=True, file_extension='mp4', type="video").order_by('resolution').desc().first()
            # 将视频流写入内存缓冲区
            buffer = io.BytesIO()
            stream.stream_to_buffer(buffer)
            # 指针移到缓冲区开头，避免上传空文件
            buffer.seek(0)
            
            # 上传到S3
            s3.upload_fileobj(buffer, BUCKET_NAME, s3_key, ExtraArgs={'ContentType': 'video/mp4'})
            print(f"{file_name} 上传成功")
            
        except Exception as e:
            print(f'---------------------- {link} 下载失败，错误：{str(e)} ----------------------')
            failed.append(link)
            continue
    if len(failed)!=0:
        print(f"失败链接：{failed}")
        # 失败日志可直接上传到S3，无需写入本地
        failed_buffer = io.BytesIO('\n'.join(failed).encode('utf-8'))
        s3.upload_fileobj(failed_buffer, BUCKET_NAME, f"{S3_PREFIX}failed_{threadlabel}.txt")

downloader(['https://www.youtube.com/watch?v=Pk4tE6Jfakg'], 'test')

注意事项 #

• 如果需要下载超过2G的大分辨率视频，可以改用S3分片上传接口，避免内存占用过高
• S3挂载方案稳定性很差，s3fs本身是FUSE文件系统，对随机写入、小IO的支持很差，Pytube下载时的写入逻辑很容易触发静默失败
• 不要直接传S3 URI给Pytube的output_path参数，Pytube本身只支持本地文件系统路径，没有适配S3对象存储的逻辑

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