如何为YouTube API实现Refresh Token以避免重复验证

我在使用YouTube API上传视频时，每次都会收到谷歌的终端提示，要求打开链接验证应用。了解到可以通过Refresh Token功能解决该问题，我在网上找到了相关实现代码，但无法正常运行。我已完成所有前置步骤，希望能得到相关帮助，谢谢。

找到的Refresh Token实现函数 #

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'

    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()

修正后的完整可运行代码 #

将两个代码整合并修复错误后，完整可运行代码如下：

import pickle
import os
import http.client
import httplib2
import random
import time
import datetime
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from googleapiclient.http import MediaFileUpload
from google.auth.transport.requests import Request

# 全局配置
httplib2.RETRIES = 1
MAX_RETRIES = 10
CLIENT_SECRETS_FILE = 'youtube_client.json'
SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
API_SERVICE_NAME = 'youtube'
API_VERSION = 'v3'

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]

def Create_Service(client_secret_file, api_name, api_version, *scopes):
    CLIENT_SECRET_FILE = client_secret_file
    API_SERVICE_NAME = api_name
    API_VERSION = api_version
    SCOPES = [scope for scope in scopes[0]]

    cred = None
    pickle_file = f'token_{API_SERVICE_NAME}_{API_VERSION}.pickle'

    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(f'{API_SERVICE_NAME} service created successfully')
        return service
    except Exception as e:
        print('Unable to connect.')
        print(e)
        return None

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']

def initialize_upload(youtube, body, file):
  insert_request = youtube.videos().insert(
    part='snippet,status',
    body=body,
    media_body=MediaFileUpload(file, chunksize=-1, resumable=True))
  return resumable_upload(insert_request)

if __name__ == '__main__':
    # 创建认证服务
    youtube = Create_Service(CLIENT_SECRETS_FILE, API_SERVICE_NAME, API_VERSION, SCOPES)
    if not youtube:
        exit('Failed to create YouTube service')

    # 上传配置（修复了categoryI的拼写错误）
    upload_date_time = datetime.datetime(2022, 12, 25, 12, 30, 0).isoformat() + '.000Z'
    request_body = {
        'snippet': {
            'categoryId': 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
    }

    try:
        video_id = initialize_upload(youtube, request_body, "output.mp4")
        # 上传缩略图
        youtube.thumbnails().set(
            videoId=video_id,
            media_body=MediaFileUpload('thumbnail.png')
        ).execute()
        print('Thumbnail uploaded successfully')
    except HttpError as e:
        print('An HTTP error %d occurred:\n%s' % (e.resp.status, e.content))

关键修改说明 #

• 整合认证逻辑：用Create_Service替代原有的get_authenticated_service，实现token的持久化存储（生成token_youtube_v3.pickle文件），首次验证后后续无需重复验证
• 修复拼写错误：将request_body中的categoryI改为正确的categoryId，避免API请求报错
• 清理冗余代码：移除重复导入的模块，简化代码结构
• 修正字符串转义：去掉不必要的转义符，确保代码语法正确

使用步骤 #

1. 确保已安装依赖包：pip install google-auth-oauthlib google-api-python-client httplib2
2. 将youtube_client.json放在代码同目录下
3. 首次运行时，按照终端提示打开链接完成验证，之后会自动生成pickle文件，后续运行无需再验证

内容的提问来源于stack exchange，提问作者absoluutly_begginer