Python YouTube视频上传进度条实现教程

本教程旨在指导开发者如何在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，因为它具有以下优点：

非侵入式打印： 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视频上传脚本添加了实时进度条。这不仅提供了直观的上传反馈，极大地提升了用户体验，也使得自动化脚本在处理耗时任务时更加专业和可靠。遵循上述步骤和最佳实践，您可以轻松地将此功能应用到自己的项目中。

# python # js # json # go # instagram # app # 字节 # 回调函数 # ai # youtube # stream

返回目录在线咨询

上一篇：ASP网站制作入门：从基础到实战全解析
下一篇：详解nodeJS之路径PATH模块

您的项目需求

*请认真填写需求信息，我们会在24小时内与您取得联系。

全网整合营销服务商