Блог

Я использую API YouTube для удаленной загрузки. Но после некоторого возни с кодом все загружаемые видео становятся «закрытыми (заблокированными)» из-за условий и политик. Я не могу обжаловать это либо из-за того, что «Обжалование этого нарушения недоступно«. Просто чтобы уточнить, я мог загружать раньше и только недавно начал получать эту ошибку.

Код: youtube-загрузчик

 #!/usr/bin/python

import argparse
import http.client
import httplib2
import os
import random
import time
import videoDetails

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

from oauth2client import client # Added
from oauth2client import tools # Added
from oauth2client.file import Storage # Added


# Explicitly tell the underlying HTTP transport library not to retry, since
# we are handling retry logic ourselves.
httplib2.RETRIES = 1

# Maximum number of times to retry before giving up.
MAX_RETRIES = 10

# Always retry when these exceptions are raised.
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)

# Always retry when an apiclient.errors.HttpError with one of these status
# codes is raised.
RETRIABLE_STATUS_CODES = [500, 502, 503, 504]

CLIENT_SECRETS_FILE = 'client_secrets.json'

SCOPES = ['https://www.googleapis.com/auth/youtube.upload']
API_SERVICE_NAME = 'youtube'
API_VERSION = 'v3'

VALID_PRIVACY_STATUSES = ('public', 'private', 'unlisted')



def get_authenticated_service(): # Modified
    credential_path = os.path.join('./', 'credentials.json')
    store = Storage(credential_path)
    credentials = store.get()
    if not credentials or credentials.invalid:
        flow = client.flow_from_clientsecrets(CLIENT_SECRETS_FILE, SCOPES)
        credentials = tools.run_flow(flow, store)
    return build(API_SERVICE_NAME, API_VERSION, credentials=credentials)

def initialize_upload(youtube, options):
  tags = None
  if options.keywords:
    tags = options.keywords.split(',')

  body=dict(
    snippet=dict(
      title=options.getFileName("video").split(".", 1)[0],
      description=options.description,
      tags=tags,
      categoryId=options.category
    ),
    status=dict(
      privacyStatus=options.privacyStatus
    )
  )

  # Call the API's videos.insert method to create and upload the video.
  videoPath = "/UserscaspeOneDriveDocumentsÖvrigtKodningAwsCSGOVideo%s" % (options.getFileName("video"))
  insert_request = youtube.videos().insert(
    part=','.join(body.keys()),
    body=body,
    media_body=MediaFileUpload(videoPath, chunksize=-1, resumable=True)
  )

  resumable_upload(insert_request, options)

# This method implements an exponential backoff strategy to resume a
# failed upload.
def resumable_upload(request, options):
  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 ('The video with the id %s was successfully uploaded!' % response['id'])
          
          # upload thumbnail for Video
          options.insertThumbnail(youtube, 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)

if __name__ == '__main__':
  args = videoDetails.Video()
  youtube = get_authenticated_service()

  try:
    initialize_upload(youtube, args)
  except HttpError as e:
    print ('An HTTP error %d occurred:n%s') % (e.resp.status, e.content)
  

Видеодетали

 import os
from googleapiclient.http import MediaFileUpload

class Video:
    description = "test description"
    category = "22"
    keywords = "test"
    privacyStatus = "public"

    def getFileName(self, type):
        for file in os.listdir("/UserscaspeOneDriveDocumentsÖvrigtKodningAwsCSGOVideo"):
            if type == "video" and file.split(".", 1)[1] != "jpg":
                return file
                break
            elif type == "thumbnail" and file.split(".", 1)[1] != "mp4":
                return file
                break

    def insertThumbnail(self, youtube, videoId):
        thumnailPath = "/UserscaspeOneDriveDocumentsÖvrigtKodningAwsCSGOVideo%s" % (self.getFileName("thumbnail"))

        request = youtube.thumbnails().set(
            videoId=videoId,
            media_body=MediaFileUpload(thumnailPath)
        )
        response = request.execute()
        print(response)
  

Если вы проверите документацию для Video.insert, вы найдете следующее в верхней части страницы. Это новая политика, которая недавно начинает применяться.

Пока ваше приложение не будет проверено, все загружаемые вами видео будут иметь значение private. Сначала вам нужно пройти аудит, после чего вы сможете загружать общедоступные видео.

Обратите внимание, что после проверки вашего приложения это не приведет к автоматической установке всех существующих ранее загруженных видео в общедоступное, вам нужно будет сделать это самостоятельно.