全网整合营销服务商

电脑端+手机端+微信端=数据同步管理

免费咨询热线：400-708-3566

使用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函数中，我们需要：

  1. 在上传开始前获取文件的总大小。
  2. 使用manager.counter()创建一个新的进度条实例。
  3. 在while循环中，每次request.next_chunk()返回status时，更新进度条。
  4. 在上传完成或发生错误时，关闭进度条。
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  # js  # json  # go  # instagram  # app  # 字节  # 工具  # ai  # youtube  # stream 


相关文章： 专业网站制作企业网站,如何制作一个企业网站，建设网站的基本步骤有哪些？  高防服务器：AI智能防御DDoS攻击与数据安全保障  网站制作哪家好,cc、.co、.cm哪个域名更适合做网站？  *服务器网站为何频现安全漏洞？  音响网站制作视频教程,隆霸音响官方网站？  如何高效完成自助建站业务培训？  网站制作大概要多少钱一个,做一个平台网站大概多少钱？  常州企业网站制作公司,全国继续教育网怎么登录？  网站制作与设计教程,如何制作一个企业网站，建设网站的基本步骤有哪些？  如何设计高效校园网站？  如何在腾讯云服务器上快速搭建个人网站？  详解免费开源的.NET多类型文件解压缩组件SharpZipLib（.NET组件介绍之七）  如何在IIS中新建站点并解决端口绑定冲突？  如何用手机制作网站和网页,手机移动端的网站能制作成中英双语的吗？  定制建站价位费用解析与套餐推荐全攻略  建站之星手机一键生成：多端自适应+小程序开发快速建站指南  导航网站建站方案与优化指南：一站式高效搭建技巧解析  建站之星后台密码遗忘或太弱？如何重置与强化？  移民网站制作流程,怎么看加拿大移民官网？  如何选择CMS系统实现快速建站与SEO优化？  枣阳网站制作,阳新火车站打的到仙岛湖多少钱？  宝盒自助建站智能生成技巧：SEO优化与关键词设置指南  番禺网站制作公司哪家值得合作,番禺图书馆新馆开放了吗？  Python文件管理规范_工程实践说明【指导】  如何快速查询网址的建站时间与历史轨迹？  外汇网站制作流程,如何在工商银行网站上做外汇买卖？  北京网站制作费用多少,建立一个公司网站的费用.有哪些部分,分别要多少钱？  如何通过免费商城建站系统源码自定义网站主题与功能？  如何通过cPanel快速搭建网站？  高端建站如何打造兼具美学与转化的品牌官网？  济南网站建设制作公司,室内设计网站一般都有哪些功能？  如何在Golang中引入测试模块_Golang测试包导入与使用实践  如何自己制作一个网站链接,如何制作一个企业网站，建设网站的基本步骤有哪些？  郑州企业网站制作公司,郑州招聘网站有哪些？  开源网站制作软件,开源网站什么意思？  广州美橙建站如何快速搭建多端合一网站？  简单实现Android文件上传  Swift中switch语句区间和元组模式匹配  自助网站制作软件,个人如何自助建网站？  威客平台建站流程解析：高效搭建教程与设计优化方案  mc皮肤壁纸制作器,苹果平板怎么设置自己想要的壁纸我的世界？  如何快速搭建安全的FTP站点？  在线流程图制作网站手机版,谁能推荐几个好的CG原画资源网站么？  湖北网站制作公司有哪些,湖北清能集团官网？  视频网站制作教程,怎么样制作优酷网的小视频？  建站之星如何快速更换网站模板？  公众号网站制作网页,微信公众号怎么制作？  如何破解联通资金短缺导致的基站建设难题？  赚钱网站制作软件,建一个网站怎样才能赚钱?是如何盈利的？  建站主机核心功能解析：服务器选择与网站搭建流程指南 

返回目录 在线咨询
上一篇：顶级域名的价格差异为何如此大，如何选择性价比最高的选项？
下一篇：浅谈java的TCP和UDP编程(附实例讲解)

您的项目需求

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

©  云南转妮洋科技有限公司 版权所有   滇ICP备2024034437号 
联系电话： 400-708-3566   E-mail：915688610@qq.com
主营业务：网络营销推广、网站SEO优化、网站定制、营销型网站建设、微商城制作、三级分销系统开发
电话沟通
在线客服