我一直在编写一个脚本，该脚本可以帮助我自动执行将 VOD 从本地驱动器上传到 YouTube 的过程。 目前只有开始上传、上传完成和上传失败时有反馈。
我想添加一个进度条来了解我的上传进度，其中包含一些数据，例如 Mo 已上传 / Mo 还需上传...
这是我的代码，我尝试了不同的方法来添加进度条，但没有任何效果。 我正在使用 bat 文件运行代码。
代码显然一点也不干净，我使用了 GPT 的一些帮助作为块管理器，这是一项正在进行的工作，需要一些改进，但它效果很好！
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
WEBHOOK_URL = "xxxx"
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 the original format
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 the new format
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
# Sanitize the 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):
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/"
)
# 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"
}
},
media_body=googleapiclient.http.MediaFileUpload(file_path, resumable=True, chunksize=-1)
)
response = None
while response is None:
try:
status, response = request.next_chunk()
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.")
end_discord_webhook(f'- Failed uploading : {title}')
except googleapiclient.errors.HttpError as e:
print(f"An HTTP error {e.resp.status} occurred while uploading {file_path}: {e.content}")
break
# You can handle specific HTTP error codes here if needed
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))]
for file_path in files:
upload_video(youtube, file_path)
# 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}")
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}')```
如果您只是对如何在终端中显示进度条感兴趣，您可以通过打印由 ASCII 字符（例如
[######............]\r 跳转到行首来实现此目的因此，您必须决定
bar_lengthdone_length = (progress_done / progress_total) * bar_length
print(
"\r["
+ "#" * done_length
+ "." * (bar_length - done_length)
+ "]"
)
然后您需要获取您的
progress_doneprogress_totalprogress_totalprogress_done如果您的问题是如何获得这些值，我想我无法真正帮助您。
看起来您正在使用 while 循环来等待上传完成。这不提供任何有关视频已上传量的信息，它只是等待响应。
您可以检查 Google API 是否可以为您提供该信息，或者您使用的某些库是否能够检查您的文件已发送了多少。
也许您应该提供一些更具体的信息，了解如何在代码中发送请求以及哪个库用于什么目的。另请考虑查看文档。
希望我能帮忙！