首页 > 文章 > python教程

使用Python为YouTube视频上传添加进度条功能

时间：2025-12-21 15:57:31 133浏览收藏

推广推荐

支持 PC / 移动端，安全直达

对于一个文章开发者来说，牢固扎实的基础是十分重要的，golang学习网就来带大家一点点的掌握基础知识点。今天本篇文章带大家了解《使用Python为YouTube视频上传添加进度条功能》，主要介绍了，希望对大家的知识积累有所帮助，快点收藏起来吧，否则需要时就找不到了！

本教程旨在指导开发者如何在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对象提供了两个核心属性来追踪进度：

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}')

5. 注意事项

终于介绍完啦！小伙伴们，这篇关于《使用Python为YouTube视频上传添加进度条功能》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布文章相关知识，快来关注吧！