本教程旨在指导开发者如何在python youtube视频上传脚本中集成实时进度条。我们将利用`googleapiclient.http.mediauploadprogress`对象获取上传进度信息，并结合`enlighten`库实现一个清晰、不干扰现有输出的进度显示，从而提升用户体验并提供直观的上传反馈。
在自动化视频上传至YouTube的场景中，尤其当处理大文件时，缺乏实时进度反馈会使得用户难以判断上传状态，甚至误以为程序卡死。googleapiclient库提供了实现这一功能所需的基础，结合合适的第三方进度条库，我们可以轻松地为上传过程添加直观的进度显示。
理解 googleapiclient 的上传进度机制
googleapiclient在执行可恢复（resumable）上传时，request.next_chunk()方法会返回一个status对象，该对象是googleapiclient.http.MediaUploadProgress的实例。这个status对象包含了两个关键属性，对于实现进度条至关重要：
- status.resumable_progress: 表示当前已上传的字节数。
- status.total_size: 表示待上传文件的总字节数。
通过这两个属性，我们可以计算出上传的百分比，并据此更新进度条。
选择合适的进度条库：Enlighten
市面上有多种Python进度条库，如tqdm、progressbar2等。本教程推荐使用Enlighten，因为它具有以下优点：
立即学习“Python免费学习笔记（深入）”；
- 非侵入式打印： Enlighten设计上能够很好地处理标准输出（print语句），即使在进度条活跃时，print输出也不会打乱进度条的显示，这对于保持现有日志输出的整洁性非常有利。
- 易于集成： API设计简洁，易于快速上手。
- 功能丰富： 支持多种进度条样式、速率显示等。
实现步骤
1. 安装 Enlighten
首先，确保您的环境中安装了Enlighten库：
pip install enlighten
2. 修改 upload_video 函数
我们将对原有的upload_video函数进行改造，引入Enlighten来显示上传进度。
import os
import webbrowser
import re
import google_auth_oauthlib.flow
import googleapiclient.discovery
import googleapiclient.errors
import googleapiclient.http # 导入 MediaUploadProgress 所需
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): # 传入 enlighten 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=file_size, desc=title 用于显示进度条的总量和描述
# unit='B', unit_scale=True, unit_decimals=2 用于显示字节单位，并自动缩放（KB, MB, GB）
pbar = manager.counter(total=file_size, desc=f"Uploading '{title}'", unit='B', unit_scale=True, unit_decimals=2)
# 定义一个回调函数，用于更新进度条
# 这个函数将在 MediaFileUpload 每次上传一个 chunk 后被调用
def progress_callback(current_progress, total_size):
pbar.update(current_progress - pbar.count) # 更新进度条，pbar.count是当前进度条已显示的值
# current_progress 是 MediaUploadProgress.resumable_progress
# 这样可以确保每次更新都是增量
# Setup request for resumable upload
media_body = googleapiclient.http.MediaFileUpload(
file_path,
resumable=True,
chunksize=-1, # -1 表示由库自动决定 chunk size
callback=lambda request_id, response: progress_callback(request_id, response) # 将回调函数传递给 MediaFileUpload
)
request = youtube.videos().insert(
part="snippet,status",
body={
"snippet": {
"categoryId": "20",
"description": description,
"title": title,
"tags": tags
},
"status": {
"privacyStatus": "private"
}
},
media_body=media_body
)
response = None
while response is None:
try:
# request.next_chunk() 返回 status 和 response
# status 是 MediaUploadProgress 对象
status, response = request.next_chunk()
if status: # 确保 status 不是 None
# 在这里，progress_callback 已经通过 media_body 的 callback 参数被调用了
# 但是，如果 MediaFileUpload 的 callback 不够灵活，也可以在这里手动更新
# pbar.update(status.resumable_progress - pbar.count) # 确保是增量更新
pass # 因为我们已经通过 callback 机制更新了，所以这里可以留空
if response is not None:
if 'id' in response:
video_url = f"https://www.youtube.com/watch?v={response['id']}"
print(f"\n -> 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"\nFailed 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"\nAn HTTP error {e.resp.status} occurred while uploading {file_path}: {e.content}")
pbar.close() # 发生错误时关闭进度条
break
except Exception as e:
print(f"\nAn unexpected error occurred: {e}")
pbar.close() # 发生错误时关闭进度条
break
def main(directory_path):
youtube = authenticate_youtube()
# 初始化 Enlighten 管理器
# 所有的进度条都将由这个管理器控制，确保它们不会互相干扰
manager = enlighten.get_manager()
files = [os.path.join(directory_path, f) for f in os.listdir(directory_path) if os.path.isfile(os.path.join(directory_path, f))]
for file_path in files:
upload_video(youtube, file_path, manager) # 传递 manager 给上传函数
print(f" * Moving local file: {file_path}")
shutil.move(file_path, "M:\\VOD UPLOADED")
print(f" -> Moved local file: {file_path}")
manager.stop() # 所有上传完成后停止 Enlighten 管理器
try:
main("M:\\VOD TO UPLOAD")
except googleapiclient.errors.HttpError:
print(f" -> Quota Exceeded!")
except Exception as e:
print(f" -> An error occured : {e}")
send_discord_webhook(f'-> An error occured : {e}')
代码修改说明：
-
导入 enlighten 和 googleapiclient.http：
import enlighten import googleapiclient.http
-
在 main 函数中初始化 enlighten 管理器：
manager = enlighten.get_manager() # ... manager.stop() # 在所有操作完成后停止管理器
enlighten.get_manager() 创建一个管理器实例，所有由该管理器创建的进度条都将协同工作，避免输出混乱。
-
修改 upload_video 函数签名：
def upload_video(youtube, file_path, manager):
现在 upload_video 函数接收 manager 参数。
-
获取文件大小并初始化进度条：
file_size = os.path.getsize(file_path) pbar = manager.counter(total=file_size, desc=f"Uploading '{title}'", unit='B', unit_scale=True, unit_decimals=2)- os.path.getsize(file_path) 用于获取文件的总字节数，作为进度条的total值。
- manager.counter(...) 创建一个新的进度条。desc参数用于显示进度条的描述文本，unit、unit_scale、unit_decimals则用于美化字节单位的显示。
-
定义 progress_callback 函数并传递给 MediaFileUpload：googleapiclient.http.MediaFileUpload 接受一个 callback 参数。这个回调函数会在每次成功上传一个数据块后被调用，并接收两个参数：request_id (通常是已上传的字节数) 和 response (这里通常是 None，或者一个 MediaUploadProgress 对象，具体取决于库版本和实现细节)。
def progress_callback(current_progress, total_size): pbar.update(current_progress - pbar.count) media_body = googleapiclient.http.MediaFileUpload( file_path, resumable=True, chunksize=-1, callback=lambda request_id, response: progress_callback(request_id, response) )- pbar.update(current_progress - pbar.count)：enlighten的update方法期望接收的是增量值，而不是绝对值。pbar.count是进度条当前已显示的累计值，current_progress是MediaUploadProgress.resumable_progress的绝对值。因此，我们计算两者之差作为本次更新的增量。
-
在上传完成或失败时关闭进度条：
pbar.close()
无论上传成功与否，都应调用pbar.close()来清除进度条，保持终端界面的整洁。
调整 print 语句： 在进度条显示期间，为了避免新的print输出覆盖进度条，可以在print语句前添加换行符\n，确保输出显示在进度条下方。
注意事项与最佳实践
- 错误处理： 确保在任何异常（如网络中断、API错误）发生时，进度条都能被正确关闭，避免遗留在终端。
- 文件大小： 对于非常小的文件，进度条可能会一闪而过，效果不明显。可以考虑为小文件不显示进度条，或设置一个最小显示时间。
- chunksize 参数： MediaFileUpload的chunksize参数控制每次上传的数据块大小。chunksize=-1是推荐值，让库自动优化。如果设置过小，可能会导致回调过于频繁，增加开销；过大则可能导致进度条更新不够平滑。
- 日志与进度条的协调： Enlighten在这方面做得很好，但如果使用其他库，可能需要更复杂的逻辑来确保日志输出不会破坏进度条。
- 用户体验： 除了进度条，还可以显示上传速度、剩余时间等信息，进一步提升用户体验。Enlighten的counter对象支持这些显示。
总结
通过集成googleapiclient.http.MediaUploadProgress和Enlighten库，我们成功地为Python YouTube视频上传脚本添加了实时进度条。这不仅提供了直观的上传反馈，极大地提升了用户体验，也使得自动化脚本在处理耗时任务时更加专业和可靠。遵循上述步骤和最佳实践，您可以轻松地将此功能应用到自己的项目中。
