我想用youtube API上传一个视频，我总是从谷歌收到一条消息到终端检查这个链接以验证你的应用程序，我读到我可以使用刷新令牌功能，我在互联网上找到了一些，但它不起作用。我不知道为什么，但我已经完成了之前的所有步骤，所以对于每一次帮助，我都会很高兴。感谢
功能：
import pickle
import os
from google_auth_oauthlib.flow import Flow, InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload, MediaIoBaseDownload
from google.auth.transport.requests import Request
import datetime
def Create_Service(client_secret_file, api_name, api_version, *scopes):
print(client_secret_file, api_name, api_version, scopes, sep='-')
CLIENT_SECRET_FILE = client_secret_file
API_SERVICE_NAME = api_name
API_VERSION = api_version
SCOPES = [scope for scope in scopes[0]]
print(SCOPES)
cred = None
pickle_file = f'token_{API_SERVICE_NAME}_{API_VERSION}.pickle'
# print(pickle_file)
if os.path.exists(pickle_file):
with open(pickle_file, 'rb') as token:
cred = pickle.load(token)
if not cred or not cred.valid:
if cred and cred.expired and cred.refresh_token:
cred.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRET_FILE, SCOPES)
cred = flow.run_console()
with open(pickle_file, 'wb') as token:
pickle.dump(cred, token)
try:
service = build(API_SERVICE_NAME, API_VERSION, credentials=cred)
print(API_SERVICE_NAME, 'service created successfully')
return service
except Exception as e:
print('Unable to connect.')
print(e)
return None
def convert_to_RFC_datetime(year=1900, month=1, day=1, hour=0, minute=0):
dt = datetime.datetime(year, month, day, hour, minute, 0).isoformat() + 'Z'
return dt
这是我的代码：
import argparse
import http.client
import httplib2
import os
import random
import time
import datetime
import google.oauth2.credentials
import google_auth_oauthlib.flow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload
from google_auth_oauthlib.flow import InstalledAppFlow
httplib2.RETRIES = 1
MAX_RETRIES = 10
RETRIABLE_EXCEPTIONS = (httplib2.HttpLib2Error, IOError, http.client.NotConnected,
http.client.IncompleteRead, http.client.ImproperConnectionState,
http.client.CannotSendRequest, http.client.CannotSendHeader,
http.client.ResponseNotReady, http.client.BadStatusLine)
RETRIABLE_STATUS_CODES = [500, 502, 503, 504]
CLIENT_SECRETS_FILE = 'youtube_client.json'
SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
API_SERVICE_NAME = 'youtube'
API_VERSION = 'v3'
VALID_PRIVACY_STATUSES = ('public', 'private', 'unlisted')
upload_date_time = datetime.datetime(2022, 12, 25, 12, 30, 0).isoformat() + '.000Z'
request_body = {
'snippet': {
'categoryI': 10,
'title': 'best music on the youtube | happy mood mix | AMP',
'description': "test",
'tags': ['Travel', 'video test', 'Travel Tips']
},
'status': {
'privacyStatus': 'private',
'publishAt': upload_date_time,
'selfDeclaredMadeForKids': False,
},
'notifySubscribers': False
}
def get_authenticated_service():
flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRETS_FILE, SCOPES)
credentials = flow.run_console()
return build(API_SERVICE_NAME, API_VERSION, credentials = credentials)
def initialize_upload(youtube,body,file):
insert_request = youtube.videos().insert(
part='snippet,status',
body=body,
media_body=MediaFileUpload(file, chunksize=-1, resumable=True))
response = resumable_upload(insert_request)
return response
def resumable_upload(request):
response = None
error = None
retry = 0
while response is None:
try:
print('Uploading file...')
status, response = request.next_chunk()
if response is not None:
if 'id' in response:
print('Video id "%s" was successfully uploaded.' % response['id'])
else:
exit('The upload failed with an unexpected response: %s' % response)
except HttpError as e:
if e.resp.status in RETRIABLE_STATUS_CODES:
error = 'A retriable HTTP error %d occurred:n%s' % (e.resp.status,
e.content)
else:
raise
except RETRIABLE_EXCEPTIONS as e:
error = 'A retriable error occurred: %s' % e
if error is not None:
print(error)
retry += 1
if retry > MAX_RETRIES:
exit('No longer attempting to retry.')
max_sleep = 2 ** retry
sleep_seconds = random.random() * max_sleep
print('Sleeping %f seconds and then retrying...' % sleep_seconds)
time.sleep(sleep_seconds)
return response['id']
if __name__ == '__main__':
youtube = get_authenticated_service()
try:
response = initialize_upload(youtube,request_body,"output.mp4" )
except HttpError as e:
print('An HTTP error %d occurred:n%s' % (e.resp.status, e.content))
youtube.thumbnails().set(
videoId=response.get('id'),
media_body=MediaFileUpload('thumbnail.png')
).execute()
诀窍是让客户端库为您完成大部分工作。
在下面的代码中，你会发现第一次创建it-us-token.json文件时是
此文件将包含
- 用户的访问令牌
- 刷新令牌
- 用于创建它的客户端id
- 用于创建它的客户端
- 它授权的范围
token.json
{
"token": "[redacted]",
"refresh_token": "[redacted]",
"token_uri": "https://oauth2.googleapis.com/token",
"client_id": "[redacted]",
"client_secret": "[redacted]",
"scopes": [
"https://www.googleapis.com/auth/youtube"
],
"expiry": "2022-08-15T13:05:53.162649Z"
}
每次运行时，它都会检查该文件是否存在。如果确实存在，则会尝试使用from_authorized_user_file加载凭据。如果它不存在，则将请求用户访问。
代码
# To install the Google client library for Python, run the following command:
# pip install --upgrade google-api-python-client google-auth-httplib2 google-auth-oauthlib
from __future__ import print_function
import os.path
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
# If modifying these scopes, delete the file token.json.
SCOPES = ['https://www.googleapis.com/auth/youtube']
def main():
"""Shows basic usage of the YouTube v3 API.
Prints the names and ids of the first 10 files the user has access to.
"""
creds = None
# The file token.json stores the user's access and refresh tokens, and is
# created automatically when the authorization flow completes for the first
# time.
if os.path.exists('token.json'):
try:
creds = Credentials.from_authorized_user_file('token.json', SCOPES)
creds.refresh(Request())
except google.auth.exceptions.RefreshError as error:
# if refresh token fails, reset creds to none.
creds = None
print(f'Refresh token expired requesting authorization again: {error}')
# If there are no (valid) credentials available, let the user log in.
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
'C:YouTubedevcredentials.json', SCOPES)
creds = flow.run_local_server(port=0)
# Save the credentials for the next run
with open('token.json', 'w') as token:
token.write(creds.to_json())
try:
service = build('youtube', 'v3', credentials=creds)
# Call the YouTube v3 API
results = service.videos().list(
part='snippet',
chart='mostPopular',
fields="nextPageToken, items").execute()
items = results.get('items', [])
if not items:
print('No videos found.')
return
print('Videos:')
for item in items:
print(u'{0} ({1})'.format(item['id'], item['snippet']['title']))
except HttpError as error:
# TODO(developer) - Handle errors from drive API.
print(f'An error occurred: {error}')
if __name__ == '__main__':
main()
注释
- 访问令牌有效期为一小时，代码将根据需要进行刷新
- 目前正在测试中的应用程序的刷新令牌将在七天内有效。此时，您将需要删除令牌json文件并再次授权应用程序
- 一旦您的应用程序投入生产，刷新令牌将不再过期