求助：为YouTube自动上传VOD的Python脚本添加进度条

要实现上传进度可视化，我们可以利用Google API的可恢复上传返回的状态信息，结合进度条库（如tqdm）或自定义文本进度条来展示已上传/剩余MB、上传进度。

步骤1：安装依赖（可选，用tqdm的话） #

pip install tqdm

修改后的完整代码 #

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
from tqdm import tqdm  # 新增导入

WEBHOOK_URL = "xxxx"


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调用失败: {response.status_code}, {response.text}")
   

def upload_video(youtube, file_path):
    print(f" -> 检测到文件: {file_path}")
    title = format_title(os.path.basename(file_path)).replace("_", "|").replace(".mkv", "")
    print(f" * 上传中 : {title}...")
    send_discord_webhook(f'- 上传中 : {title}')
    
    # 获取文件总大小，用于进度计算
    file_size = os.path.getsize(file_path)
    MB = 1024 * 1024
    total_mb = file_size / MB
    
    # 从文件加载预设标签
    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/"
    )
    
    # 设置可恢复上传请求
    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
    # 初始化进度条
    pbar = tqdm(total=file_size, unit='B', unit_scale=True, unit_divisor=MB, desc=f"上传 {title}")
    try:
        while response is None:
            status, response = request.next_chunk()
            if status:
                # 更新进度条
                uploaded_bytes = status.resumable_progress
                pbar.update(uploaded_bytes - pbar.n)
                # 计算并显示已上传/剩余MB
                uploaded_mb = uploaded_bytes / MB
                remaining_mb = total_mb - uploaded_mb
                pbar.set_postfix({"已上传": f"{uploaded_mb:.2f}MB", "剩余": f"{remaining_mb:.2f}MB"})
        
        # 上传完成后关闭进度条
        pbar.close()
        if 'id' in response:
            video_url = f"https://www.youtube.com/watch?v={response['id']}"
            print(f" -> 上传完成 {file_path}，视频ID {response['id']} - {video_url}")
            send_discord_webhook(f'- 上传完成 : {title} | {video_url}')
            webbrowser.open(video_url, new=2)
        else:
            print(f"上传失败 {file_path}，未获取到YouTube视频ID。")
            send_discord_webhook(f'- 上传失败 : {title}')
    except googleapiclient.errors.HttpError as e:
        pbar.close()
        print(f"上传时发生HTTP错误 {e.resp.status}: {e.content}")
        send_discord_webhook(f'- 上传失败，HTTP错误 {e.resp.status}: {title}')
    except Exception as e:
        pbar.close()
        print(f"上传错误: {e}")
        send_discord_webhook(f'- 上传错误: {title} - {str(e)}')


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))]
    for file_path in files:
        upload_video(youtube, file_path)
        # 上传完成后移动本地文件到指定目录
        print(f" * 移动本地文件: {file_path}")
        shutil.move(file_path, "M:\\VOD UPLOADED")
        print(f" -> 已移动本地文件: {file_path}")

try:
    main("M:\\VOD TO UPLOAD")
except googleapiclient.errors.HttpError:
    print(f" -> 配额超限!")
    send_discord_webhook(f'- 配额超限! 上传暂停。')
except Exception as e:
    print(f" -> 发生错误 : {e}")
    send_discord_webhook(f'-> 发生错误 : {e}')

关键修改说明 #

1. 导入tqdm库：生成动态进度条，支持单位自动转换，提升终端显示体验。
2. 获取文件总大小：通过os.path.getsize()获取字节数，转换为MB用于进度计算。
3. 进度条初始化：设置总大小、显示单位、描述文本，让进度展示更直观。
4. 进度更新逻辑：在next_chunk()返回status时，用status.resumable_progress获取已上传字节数，更新进度条并实时计算显示已上传/剩余MB。
5. 异常处理优化：异常发生时关闭进度条，避免终端输出混乱。

无tqdm替代方案（自定义文本进度条） #

如果不想依赖第三方库，可以替换进度条部分为以下自定义函数：

def print_progress(uploaded_bytes, total_bytes):
    percent = (uploaded_bytes / total_bytes) * 100
    bar_length = 50
    filled_length = int(bar_length * uploaded_bytes // total_bytes)
    bar = '=' * filled_length + '-' * (bar_length - filled_length)
    MB = 1024 * 1024
    uploaded_mb = uploaded_bytes / MB
    remaining_mb = (total_bytes - uploaded_bytes) / MB
    print(f"\r[{bar}] {percent:.1f}% | 已上传: {uploaded_mb:.2f}MB | 剩余: {remaining_mb:.2f}MB", end='', flush=True)

然后在upload_video函数中，去掉tqdm相关代码，替换为：

# 初始化进度显示
print_progress(0, file_size)
try:
    while response is None:
        status, response = request.next_chunk()
        if status:
            uploaded_bytes = status.resumable_progress
            print_progress(uploaded_bytes, file_size)
    # 上传完成后换行
    print()
    # ...后续上传完成逻辑

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