在自动化视频上传到youtube的场景中，尤其当处理大文件时，缺乏实时进度反馈会使得用户难以判断上传状态，甚至误认为程序卡死。本教程将详细介绍如何在现有的python youtube上传脚本中，通过集成进度条来提供清晰的上传进度可视化。
1. 理解YouTube API的上传进度机制
Google API Python客户端库在执行可恢复上传（resumable upload）时，提供了一个关键机制来获取上传进度。youtube.videos().insert()方法配合googleapiclient.http.MediaFileUpload(..., resumable=True)，在调用request.next_chunk()时，会返回两个值：status和response。
- status：当上传仍在进行中时，这是一个googleapiclient.http.MediaUploadProgress对象。它包含两个重要属性：
- status.resumable_progress：当前已上传的字节数。
- status.total_size：文件的总字节数。
- response：当上传完成时，此变量将包含YouTube API的最终响应（例如，上传视频的ID）；否则，它将为None。
利用status.resumable_progress和status.total_size，我们就可以计算出上传的百分比和剩余量，从而驱动进度条的显示。
2. 选择合适的进度条库
为了在控制台输出中优雅地显示进度条，同时不干扰其他print语句，推荐使用Enlighten库。Enlighten具有以下优点：
- 非侵入性：它能智能地管理控制台输出，确保进度条不会被其他打印内容覆盖或打乱。
- 易于使用：提供简洁的API来创建和更新进度条。
- 功能丰富：支持多种显示格式、单位、颜色等。
首先，确保你的环境中安装了Enlighten：
立即学习“Python免费学习笔记（深入）”；
pip install enlighten
3. 集成Enlighten进度条到上传逻辑
我们将修改原有的upload_video函数，以引入Enlighten进度条。
3.1 核心修改步骤
- 获取文件总大小：在开始上传前，使用os.path.getsize(file_path)获取待上传文件的总字节数。
- 初始化Enlighten管理器：在main函数中一次性创建enlighten.get_manager()实例，并在程序结束时调用manager.stop()，以确保资源正确释放。
- 创建进度条计数器：在upload_video函数内部，使用manager.counter()创建一个针对当前文件上传的进度条。
- 更新进度条：在while response is None:循环内部，每次request.next_chunk()返回status时，更新进度条的当前值 (pbar.count = status.resumable_progress)。
- 关闭进度条：无论上传成功还是失败，在upload_video函数结束前（或在相应的分支中），调用pbar.close()来关闭进度条。
3.2 示例代码：修改 upload_video 和 main 函数
import os
import webbrowser
import re
import google_auth_oauthlib.flow
import googleapiclient.discovery
import googleapiclient.errors
import pickle
import shutil
import time
import requests
import enlighten # 导入enlighten库
WEBHOOK_URL = "xxxx" # 请替换为您的Webhook URL
# ... (authenticate_youtube, save_token, load_token, format_title, send_discord_webhook 函数保持不变) ...
def upload_video(youtube, file_path, manager):
"""
上传视频到YouTube，并显示进度条。
:param youtube: YouTube API服务对象。
:param file_path: 待上传视频文件的完整路径。
:param manager: Enlighten管理器实例。
"""
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}')
# Load predefined tags from the file
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进度条
# desc: 进度条前缀描述
# unit: 单位，例如'bytes'
# total: 总大小
# min_display_duration: 最小显示时长，防止进度条闪烁
pbar = manager.counter(total=file_size, unit='bytes', desc=f"Uploading {title[:40]}...", min_display_duration=0.5)
# Setup request for resumable upload
request = youtube.videos().insert(
part="snippet,status",
body={
"snippet": {
"categoryId": "20",
"description": description,
"title": title,
"tags": tags
},
"status": {
"privacyStatus": "private"
}
},
# MediaFileUpload的chunksize=-1对于resumable=True通常是默认行为，可以省略
media_body=googleapiclient.http.MediaFileUpload(file_path, resumable=True)
)
response = None
while response is None:
try:
status, response = request.next_chunk()
if status: # 检查status是否为MediaUploadProgress对象
# 更新进度条的当前值
pbar.count = status.resumable_progress
# pbar会自动处理显示，无需额外print
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}') # 修正原代码中的end_discord_webhook
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
# You can handle specific HTTP error codes here if needed
def main(directory_path):
"""
主函数，处理指定目录下的所有视频文件上传。
:param directory_path: 包含待上传视频文件的目录路径。
"""
youtube = authenticate_youtube()
# 初始化Enlighten管理器，最好在主循环外一次性创建
manager = enlighten.get_manager()
try:
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:
# 将manager实例传递给upload_video函数
upload_video(youtube, file_path, manager)
# Move the local video file to a different directory after upload
print(f" * Moving local file: {file_path}")
shutil.move(file_path, "M:\\VOD UPLOADED")
print(f" -> Moved local file: {file_path}")
finally:
# 确保在程序结束时关闭Enlighten管理器，即使发生异常
manager.stop()
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}')
4. 注意事项与总结
- Enlighten管理器生命周期：enlighten.get_manager()应该在程序启动时调用一次，并在程序结束时（无论正常结束还是异常退出）通过manager.stop()来释放资源。在main函数中使用try...finally块是管理manager生命周期的最佳实践。
- 进度条描述：desc参数可以用来显示当前正在上传的文件名或标题，title[:40]用于截断过长的标题，以适应控制台显示。
- 单位和总数：确保unit参数设置为'bytes'，total参数设置为文件的总字节数，这样Enlighten就能正确地显示MB/GB等单位。
- 错误处理：在try...except块中，无论上传成功还是失败，都应调用pbar.close()来确保进度条被正确移除，避免在控制台留下残余。
- 运行环境：Enlighten在大多数终端环境下都能良好工作，包括通过.bat文件运行Python脚本。如果遇到显示问题，请检查终端是否支持ANSI转义序列。
通过以上修改，您的YouTube上传脚本将能提供清晰、专业的实时上传进度反馈，极大地提升用户体验。这不仅让用户了解上传状态，也方便调试和监控自动化流程。
