本教程旨在指导开发者如何在Python YouTube视频上传脚本中集成实时进度条功能。通过深入理解googleapiclient.http.MediaUploadProgress对象，结合如Enlighten等第三方库，实现精确显示已上传字节、总文件大小及上传百分比，从而显著提升脚本的用户体验和监控能力，尤其适用于自动化视频上传场景。
Python YouTube视频上传进度条实现教程
在自动化YouTube视频上传流程中，实时了解上传进度对于用户体验和脚本监控至关重要。本教程将详细介绍如何利用Google API Python客户端提供的功能，并结合第三方进度条库，为您的Python上传脚本添加一个动态且专业的进度条。
1. 理解YouTube上传机制与进度反馈
YouTube API通过googleapiclient.http.MediaFileUpload支持可恢复上传（resumable upload）。这意味着即使网络中断，上传也可以从中断处继续。在上传过程中，request.next_chunk()方法会返回一个status对象和一个response对象。
- status: 当上传仍在进行时，status是一个MediaUploadProgress对象，它包含了当前上传进度的关键信息。
- response: 当文件上传完成时，response会包含YouTube返回的视频信息，而status将为None。
MediaUploadProgress对象提供了两个核心属性来追踪进度：
立即学习“Python免费学习笔记（深入）”；
- resumable_progress: 表示当前已成功上传的字节数。
- total_size: 表示文件的总字节数。
利用这两个属性，我们可以计算出上传的百分比，并将其展示在进度条中。
2. 选择合适的进度条库
为了在命令行界面美观且高效地显示进度条，推荐使用专门的进度条库。Enlighten是一个优秀的Python进度条库，它支持多行进度条、自动刷新，并且不会干扰正常的print输出，非常适合集成到现有脚本中。
安装 Enlighten： 您可以使用pip轻松安装Enlighten：
pip install enlighten
3. 集成进度条到YouTube上传脚本
以下是将进度条功能集成到现有upload_video函数中的步骤和示例代码。
3.1 导入所需库
首先，确保在脚本顶部导入enlighten库，并添加os用于获取文件大小。
import os import enlighten # 导入enlighten库 # ... 其他现有导入 ... import googleapiclient.discovery import googleapiclient.errors import googleapiclient.http # 确保导入此模块以访问MediaFileUpload
3.2 修改 main 函数以管理 Enlighten Manager
为了更好地管理多个视频上传时的进度条，建议在main函数中初始化和停止enlighten的Manager，并将其传递给upload_video函数。
# ... authenticate_youtube, save_token, load_token, format_title, send_discord_webhook 等函数保持不变 ...
def main(directory_path):
youtube = authenticate_youtube()
files = [os.path.join(directory_path, f) for f in os.listdir(directory_path) if os.path.isfile(os.path.join(directory_path, f))]
# 初始化 Enlighten Manager
manager = enlighten.get_manager()
try:
for file_path in files:
# 将 manager 传递给 upload_video 函数
upload_video(youtube, file_path, manager)
# 上传完成后移动本地文件
print(f" * Moving local file: {file_path}")
shutil.move(file_path, "M:\\VOD UPLOADED")
print(f" -> Moved local file: {file_path}")
finally:
# 确保在所有上传完成后停止 Manager
manager.stop()
# ... try-except 块保持不变 ...3.3 修改 upload_video 函数以显示进度条
在upload_video函数中，我们需要：
- 在上传开始前获取文件的总大小。
- 使用manager.counter()创建一个新的进度条实例。
- 在while循环中，每次request.next_chunk()返回status时，更新进度条。
- 在上传完成或发生错误时，关闭进度条。
def upload_video(youtube, file_path, manager): # 接收 manager 参数
print(f" -> Detected file: {file_path}")
title = format_title(os.path.basename(file_path)).replace("_", "|").replace(".mkv", "")
print(f" * Uploading : {title}...")
send_discord_webhook(f'- Uploading : {title}')
tags_file_path = "tags.txt"
with open(tags_file_path, 'r') as tags_file:
tags = tags_file.read().splitlines()
description = (
"⏬ Déroule-moi ! ⏬\n"
f"VOD HD d'un stream Twitch de Ben_OnAir uploadée automatiquement\n\n"
"═════════════════════\n\n"
"► Je stream ici : https://www.twitch.tv/ben_onair\n"
"► Ma Chaîne Youtube : https://www.youtube.com/@BenOnAir\n"
"► Mon Twitter : https://twitter.com/Ben_OnAir\n"
"► Mon Insta : https://www.instagram.com/ben_onair/\n"
"► Mon Book : https://ben-book.fr/"
)
# 获取文件总大小
file_size = os.path.getsize(file_path)
# 初始化 Enlighten 进度条
# total: 文件总大小，unit: 单位，desc: 进度条描述
pbar = manager.counter(total=file_size, unit='bytes', desc=f"Uploading {title}")
request = youtube.videos().insert(
part="snippet,status",
body={
"snippet": {
"categoryId": "20",
"description": description,
"title": title,
"tags": tags
},
"status": {
"privacyStatus": "private"
}
},
media_body=googleapiclient.http.MediaFileUpload(file_path, resumable=True, chunksize=-1)
)
response = None
while response is None:
try:
status, response = request.next_chunk()
if status: # 检查 status 是否不为空，表示上传仍在进行
# 更新进度条：pbar.update() 接受的是增量值
# status.resumable_progress 是当前已上传的总字节数
# pbar.count 是进度条当前显示的总字节数
# 所以更新的增量是两者之差
pbar.update(status.resumable_progress - pbar.count)
if response is not None:
if 'id' in response:
video_url = f"https://www.youtube.com/watch?v={response['id']}"
print(f" -> Uploaded {file_path} with video ID {response['id']} - {video_url}")
send_discord_webhook(f'- Uploaded : {title} | {video_url}')
webbrowser.open(video_url, new=2)
else:
print(f"Failed to upload {file_path}. No video ID returned from YouTube.")
send_discord_webhook(f'- Failed uploading : {title}')
pbar.close() # 上传完成，关闭进度条
except googleapiclient.errors.HttpError as e:
print(f"An HTTP error {e.resp.status} occurred while uploading {file_path}: {e.content}")
pbar.close() # 发生错误，关闭进度条
break
except Exception as e: # 捕获其他可能的异常
print(f"An unexpected error occurred during upload: {e}")
pbar.close() # 发生错误，关闭进度条
break
4. 完整代码示例
将上述修改整合到您的原始脚本中，一个带有进度条的YouTube上传工具就完成了。
import os
import webbrowser
import re
import google_auth_oauthlib.flow
import googleapiclient.discovery
import googleapiclient.errors
import googleapiclient.http # 确保导入此模块
import pickle
import shutil
import time
import requests
import enlighten # 导入enlighten库
WEBHOOK_URL = "xxxx" # 请替换为您的Discord Webhook URL
def authenticate_youtube():
os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "1"
api_service_name = "youtube"
api_version = "v3"
client_secrets_file = "client_secrets.json"
token_filename = "youtube_token.pickle"
credentials = load_token(token_filename)
if not credentials or not credentials.valid:
if credentials and credentials.expired and credentials.refresh_token:
credentials.refresh(googleapiclient.errors.Request())
else:
flow = google_auth_oauthlib.flow.InstalledAppFlow.from_client_secrets_file(client_secrets_file, ["https://www.googleapis.com/auth/youtube.upload"])
credentials = flow.run_local_server(port=0)
save_token(credentials, token_filename)
youtube = googleapiclient.discovery.build(api_service_name, api_version, credentials=credentials)
return youtube
def save_token(credentials, token_file):
with open(token_file, 'wb') as token:
pickle.dump(credentials, token)
def load_token(token_file):
if os.path.exists(token_file):
with open(token_file, 'rb') as token:
return pickle.load(token)
return None
def format_title(filename):
match = re.search(r"Stream - (\d{4})_(\d{2})_(\d{2}) - (\d{2}h\d{2})", filename)
if match:
year, month, day, time = match.groups()
sanitized_title = f"VOD | Stream du {day} {month} {year}"
else:
match = re.search(r"(\d{4})-(\d{2})-(\d{2})_(\d{2})-(\d{2})-(\d{2})", filename)
if match:
year, month, day, hour, minute, second = match.groups()
sanitized_title = f"VOD | Stream du {day} {month} {year}"
else:
sanitized_title = filename
return re.sub(r'[^\w\-_\. ]', '_', sanitized_title)
def send_discord_webhook(message):
data = {"content": message}
response = requests.post(WEBHOOK_URL, json=data)
if response.status_code != 204:
print(f"Webhook call failed: {response.status_code}, {response.text}")
def upload_video(youtube, file_path, manager): # 接收 manager 参数
print(f" -> Detected file: {file_path}")
title = format_title(os.path.basename(file_path)).replace("_", "|").replace(".mkv", "")
print(f" * Uploading : {title}...")
send_discord_webhook(f'- Uploading : {title}')
tags_file_path = "tags.txt"
with open(tags_file_path, 'r') as tags_file:
tags = tags_file.read().splitlines()
description = (
"⏬ Déroule-moi ! ⏬\n"
f"VOD HD d'un stream Twitch de Ben_OnAir uploadée automatiquement\n\n"
"═════════════════════\n\n"
"► Je stream ici : https://www.twitch.tv/ben_onair\n"
"► Ma Chaîne Youtube : https://www.youtube.com/@BenOnAir\n"
"► Mon Twitter : https://twitter.com/Ben_OnAir\n"
"► Mon Insta : https://www.instagram.com/ben_onair/\n"
"► Mon Book : https://ben-book.fr/"
)
# 获取文件总大小
file_size = os.path.getsize(file_path)
# 初始化 Enlighten 进度条
pbar = manager.counter(total=file_size, unit='bytes', desc=f"Uploading {title}")
request = youtube.videos().insert(
part="snippet,status",
body={
"snippet": {
"categoryId": "20",
"description": description,
"title": title,
"tags": tags
},
"status": {
"privacyStatus": "private"
}
},
media_body=googleapiclient.http.MediaFileUpload(file_path, resumable=True, chunksize=-1)
)
response = None
while response is None:
try:
status, response = request.next_chunk()
if status: # 检查 status 是否不为空，表示上传仍在进行
pbar.update(status.resumable_progress - pbar.count)
if response is not None:
if 'id' in response:
video_url = f"https://www.youtube.com/watch?v={response['id']}"
print(f" -> Uploaded {file_path} with video ID {response['id']} - {video_url}")
send_discord_webhook(f'- Uploaded : {title} | {video_url}')
webbrowser.open(video_url, new=2)
else:
print(f"Failed to upload {file_path}. No video ID returned from YouTube.")
send_discord_webhook(f'- Failed uploading : {title}')
pbar.close() # 上传完成，关闭进度条
except googleapiclient.errors.HttpError as e:
print(f"An HTTP error {e.resp.status} occurred while uploading {file_path}: {e.content}")
pbar.close() # 发生错误，关闭进度条
break
except Exception as e: # 捕获其他可能的异常
print(f"An unexpected error occurred during upload: {e}")
pbar.close() # 发生错误，关闭进度条
break
def main(directory_path):
youtube = authenticate_youtube()
files = [os.path.join(directory_path, f) for f in os.listdir(directory_path) if os.path.isfile(os.path.join(directory_path, f))]
manager = enlighten.get_manager() # 初始化 Enlighten Manager
try:
for file_path in files:
upload_video(youtube, file_path, manager) # 将 manager 传递给 upload_video
print(f" * Moving local file: {file_path}")
shutil.move(file_path, "M:\\VOD UPLOADED") # 替换为您的目标目录
print(f" -> Moved local file: {file_path}")
finally:
manager.stop() # 确保在所有上传完成后停止 Manager
try:
main("M:\\VOD TO UPLOAD") # 替换为您的视频源目录
except googleapiclient.errors.HttpError as e:
if e.resp.status == 403 and "quotaExceeded" in e.content.decode():
print(f" -> Quota Exceeded! Error: {e.content.decode()}")
send_discord_webhook(f'-> Quota Exceeded! Error: {e.content.decode()}')
else:
print(f" -> An HTTP error occurred: {e}")
send_discord_webhook(f'-> An HTTP error occurred: {e}')
except Exception as e:
print(f" -> An error occured : {e}")
send_discord_webhook(f'-> An error occured : {e}')
