Python YouTube上传脚本中集成实时进度条：使用Enlighten库

在自动化视频上传到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：

pip install enlighten

3. 集成Enlighten进度条到上传逻辑

我们将修改原有的upload_video函数，以引入Enlighten进度条。

3.1 核心修改步骤

1. 获取文件总大小：在开始上传前，使用os.path.getsize(file_path)获取待上传文件的总字节数。
2. 初始化Enlighten管理器：在main函数中一次性创建enlighten.get_manager()实例，并在程序结束时调用manager.stop()，以确保资源正确释放。
3. 创建进度条计数器：在upload_video函数内部，使用manager.counter()创建一个针对当前文件上传的进度条。
4. 更新进度条：在while response is None:循环内部，每次request.next_chunk()返回status时，更新进度条的当前值 (pbar.count = status.resumable_progress)。
5. 关闭进度条：无论上传成功还是失败，在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. 注意事项与总结

1. Enlighten管理器生命周期：enlighten.get_manager()应该在程序启动时调用一次，并在程序结束时（无论正常结束还是异常退出）通过manager.stop()来释放资源。在main函数中使用try...finally块是管理manager生命周期的最佳实践。
2. 进度条描述：desc参数可以用来显示当前正在上传的文件名或标题，title[:40]用于截断过长的标题，以适应控制台显示。
3. 单位和总数：确保unit参数设置为'bytes'，total参数设置为文件的总字节数，这样Enlighten就能正确地显示MB/GB等单位。
4. 错误处理：在try...except块中，无论上传成功还是失败，都应调用pbar.close()来确保进度条被正确移除，避免在控制台留下残余。
5. 运行环境：Enlighten在大多数终端环境下都能良好工作，包括通过.bat文件运行Python脚本。如果遇到显示问题，请检查终端是否支持ANSI转义序列。

通过以上修改，您的YouTube上传脚本将能提供清晰、专业的实时上传进度反馈，极大地提升用户体验。这不仅让用户了解上传状态，也方便调试和监控自动化流程。

# python  # go  # instagram  # 字节  # ai  # youtube  # stream  # google  # twitter 

相关文章： 北京营销型网站制作公司,可以用python做一个营销推广网站吗？  如何彻底卸载建站之星软件？  临沂网站制作公司有哪些,临沂第四中学官网？  怎么制作一个起泡网,水泡粪全漏粪育肥舍冬季氨气超过25ppm，可以有哪些措施降低舍内氨气水平？  C#怎么创建控制台应用 C# Console App项目创建方法  清单制作人网站有哪些,近日“兴风作浪的姑奶奶”引起很多人的关注这是什么事情？  如何通过建站之星自助学习解决操作问题？  湖州网站制作公司有哪些,浙江中蓝新能源公司官网？  网站制作企业,网站的banner和导航栏是指什么？  ,在苏州找工作，上哪个网站比较好？  再谈Python中的字符串与字符编码（推荐）  手机网站制作平台,手机靓号代理商怎么制作属于自己的手机靓号网站？  视频网站制作教程,怎么样制作优酷网的小视频？  网站制作免费,什么网站能看正片电影？  建站主机数据库如何配置才能提升网站性能？  内部网站制作流程,如何建立公司内部网站？  建站之星如何修改网站生成路径？  制作网站的软件下载免费,今日头条开宝箱老是需要下载怎么回事？  电商网站制作价格怎么算,网上拍卖流程以及规则？  如何批量查询域名的建站时间记录？  如何选择PHP开源工具快速搭建网站？  建站之星代理平台如何选择最佳方案？  如何自定义建站之星模板颜色并下载新样式？  建站VPS选购需注意哪些关键参数？  建站之星在线客服如何快速接入解答？  免费的流程图制作网站有哪些,2025年教师初级职称申报网上流程？  制作企业网站建设方案,怎样建设一个公司网站？  ,购物网站怎么盈利呢？  如何在腾讯云免费申请建站？  建站主机如何选？性能与价格怎样平衡？  深圳网站制作设计招聘,关于服装设计的流行趋势，哪里的资料比较全面？  定制建站流程步骤详解：一站式方案设计与开发指南  存储型VPS适合搭建中小型网站吗？  深圳网站制作案例,网页的相关名词有哪些？  建站之星手机一键生成：多端自适应+小程序开发快速建站指南  如何确保西部建站助手FTP传输的安全性？  如何在Ubuntu系统下快速搭建WordPress个人网站？  无锡营销型网站制作公司,无锡网选车牌流程？  如何在腾讯云服务器快速搭建个人网站？  济南网站制作的价格,历城一职专官方网站？  实例解析Array和String方法  ui设计制作网站有哪些,手机UI设计网址吗？  如何快速生成高效建站系统源代码？  上海网站制作开发公司,上海买房比较好的网站有哪些？  如何在云主机上快速搭建多站点网站？  智能起名网站制作软件有哪些,制作logo的软件？  如何高效配置IIS服务器搭建网站？  如何快速查询网站的真实建站时间？  mc皮肤壁纸制作器,苹果平板怎么设置自己想要的壁纸我的世界？  太平洋网站制作公司,网络用语太平洋是什么意思？