如何用Python实现YouTube转音频直接存储至GCS而非本地磁盘?
直接将YouTube音频流上传至GCS(无需本地存储)
不用将音频文件写入本地磁盘,可通过pytube获取音频流的字节数据,直接上传到Google Cloud Storage,实现无本地存储的转换上传流程。
1. 安装依赖
确保已安装所需库:
pip install pytube tldextract google-cloud-storage
2. 修改代码实现直接上传
替换原有的Download函数,改用内存字节流完成上传:
import tldextract from pytube import YouTube import logging from google.cloud import storage import io def Download(video_list): # 初始化GCS客户端(需提前配置认证:设置GOOGLE_APPLICATION_CREDENTIALS环境变量指向服务账号密钥) storage_client = storage.Client() gcs_bucket_name = "你的GCS桶名称" # 替换为实际桶名 bucket = storage_client.bucket(gcs_bucket_name) for url in video_list: logging.info('Url: %s' % url) youtube = YouTube(url.strip()) # 获取指定itag的音频流(itag 139为m4a格式,可根据需求调整) stream = youtube.streams.get_by_itag('139') # 将音频流读取到内存字节容器,不写入本地磁盘 audio_stream = io.BytesIO(stream.stream()) # 自定义GCS文件名:替换标题中的非法字符,也可使用视频ID确保唯一性 gcs_file_name = f"{youtube.title.replace('/', '_')}.mp3" blob = bucket.blob(gcs_file_name) # 上传字节流到GCS,指定音频类型 blob.upload_from_file(audio_stream, content_type="audio/mpeg") logging.info(f"音频已上传至GCS:gs://{gcs_bucket_name}/{gcs_file_name}") video_list = [] run = True while run == True: link = str(input("Enter a YouTube URL or press D to download the video(s): ")) domain_name = tldextract.extract(link).domain if link.lower() == 'd': Download(video_list) run = False elif domain_name == 'youtube': video_list.append(link) else: print("Invalid youtube URL.")
关键说明
stream.stream():pytube提供的方法,直接返回音频的原始数据流对象,跳过本地文件写入步骤。io.BytesIO:将数据流包装为内存中的字节容器,兼容GCS客户端的upload_from_file方法。- 文件名处理:替换视频标题中的
/等非法字符,避免GCS路径错误;也可使用youtube.video_id作为文件名,彻底避免重名问题。 - GCS认证:本地运行时需通过
GOOGLE_APPLICATION_CREDENTIALS环境变量指定服务账号密钥文件;在Google Cloud托管环境(如Cloud Functions、GCE)中运行时,会自动获取默认认证。
内容的提问来源于stack exchange,提问作者UtkMal
相关产品推荐
相关产品推荐

