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

pip install 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}hd{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 ! ⏬
"
    f"VOD HD d'un stream Twitch de Ben_OnAir uploadée automatiquement

"
    "═════════════════════

"
    "► Je stream ici : https://www.twitch.tv/ben_onair
"
    "► Ma Chaîne Youtube : https://www.youtube.com/@BenOnAir
"
    "► Mon Twitter : https://twitter.com/Ben_OnAir
"
    "► Mon Insta : https://www.instagram.com/ben_onair/
"
    "► 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"
 -> 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: {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}')