【问题标题】:How to implement refres token to youtube API如何将刷新令牌实现到 youtube API
【发布时间】:2022-08-16 23:16:11
【问题描述】:

我想使用 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()

    标签: python youtube-api google-oauth google-api-python-client refresh-token


    【解决方案1】:

    诀窍是让客户端库为您完成大部分工作。

    在下面的代码中你会发现我们第一次创建了 token.json 文件

    该文件将包含

    1. 来自用户的访问令牌
    2. 刷新令牌
    3. 用于创建它的客户端 ID
    4. 用于创建它的客户端
    5. 它授权的范围。

      令牌.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'):
              creds = Credentials.from_authorized_user_file('token.json', SCOPES)
          # 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:\YouTube\dev\credentials.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()
      

      笔记

      1. 访问令牌有效期为一小时,代码将根据需要对其进行刷新。
      2. 目前正在测试的应用程序的刷新令牌有效期为 7 天。此时您需要删除令牌 json 文件并再次授权应用程序。
      3. 一旦您的应用程序投入生产,刷新令牌将不再过期

    【讨论】:

      猜你喜欢
      • 2021-07-11
      • 1970-01-01
      • 2021-06-13
      • 1970-01-01
      • 2012-04-10
      • 2017-08-11
      • 2020-11-09
      • 2021-11-17
      • 1970-01-01
      相关资源
      最近更新 更多