YouTube API - 尝试获取频道成员

Question

我目前正在尝试编写一个 Python 脚本来使用 YouTube API 获取我的 YouTube 频道成员。我创建了一个 OAuth 客户端并完成了设置所有令牌的必要步骤（我认为）。此 OAuth 客户端在我使用它来获取我的频道订阅和其他信息的列表时工作，但是当我尝试获取我的成员时，我收到 403 错误。我相信这是因为 Members Documention 声明我必须填写特定的 form 才能访问此 API 路由。我在表格中填写了我认为 Google 需要的所有信息。但是我没有收到任何关于此批准状态的更新，也没有收到任何指示，如果我用正确的信息填写了此表格。

有没有人有使用 YouTube API 获取频道成员的经验，到目前为止我是否正确执行了此操作？如果我这样做正确，应用程序需要多长时间才能获准访问成员 API 路由？

对于代码，我只是使用Python Web Client Quickstart Guide。我也尝试过使用示波器但没有成功......

import os
import flask
import requests

import google.oauth2.credentials
import google_auth_oauthlib.flow
import googleapiclient.discovery

CLIENT_SECRETS_FILE = "client_secret.json"

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

app = flask.Flask(__name__)
app.secret_key = '5f24ef3cf326d03'


@app.route('/')
def index():
  return print_index_table()


@app.route('/test')
def test_api_request():
  if 'credentials' not in flask.session:
    return flask.redirect('authorize')

  credentials = google.oauth2.credentials.Credentials(
      **flask.session['credentials'])

  youtube = googleapiclient.discovery.build(
      API_SERVICE_NAME, API_VERSION, credentials=credentials)

  # THIS WORKS
  # subs = youtube.subscriptions().list(
  #     part="snippet,contentDetails",
  #     mine=True
  # ).execute()

  # THIS DOESNT WORK
  subs = youtube.members().list(
      part="snippet",
      maxResults=50
  ).execute()

  flask.session['credentials'] = credentials_to_dict(credentials)

  return flask.jsonify(**subs)


@app.route('/authorize')
def authorize():
  flow = google_auth_oauthlib.flow.Flow.from_client_secrets_file(
      CLIENT_SECRETS_FILE, scopes=SCOPES)

  flow.redirect_uri = flask.url_for('oauth2callback', _external=True)

  authorization_url, state = flow.authorization_url(
      access_type='offline',
      include_granted_scopes='true')

  flask.session['state'] = state

  return flask.redirect(authorization_url)


@app.route('/oauth2callback')
def oauth2callback():
  state = flask.session['state']

  flow = google_auth_oauthlib.flow.Flow.from_client_secrets_file(
      CLIENT_SECRETS_FILE, scopes=SCOPES, state=state)
  flow.redirect_uri = flask.url_for('oauth2callback', _external=True)

  authorization_response = flask.request.url
  flow.fetch_token(authorization_response=authorization_response)

  credentials = flow.credentials
  flask.session['credentials'] = credentials_to_dict(credentials)

  return flask.redirect(flask.url_for('test_api_request'))


@app.route('/revoke')
def revoke():
  if 'credentials' not in flask.session:
    return ('You need to <a href="/authorize">authorize</a> before ' +
            'testing the code to revoke credentials.')

  credentials = google.oauth2.credentials.Credentials(
      **flask.session['credentials'])

  revoke = requests.post('https://oauth2.googleapis.com/revoke',
                         params={'token': credentials.token},
                         headers={'content-type': 'application/x-www-form-urlencoded'})

  status_code = getattr(revoke, 'status_code')
  if status_code == 200:
    return('Credentials successfully revoked.' + print_index_table())
  else:
    return('An error occurred.' + print_index_table())


@app.route('/clear')
def clear_credentials():
  if 'credentials' in flask.session:
    del flask.session['credentials']
  return ('Credentials have been cleared.<br><br>' +
          print_index_table())


def credentials_to_dict(credentials):
  return {'token': credentials.token,
          'refresh_token': credentials.refresh_token,
          'token_uri': credentials.token_uri,
          'client_id': credentials.client_id,
          'client_secret': credentials.client_secret,
          'scopes': credentials.scopes}


def print_index_table():
  return ('<table>' +
          '<tr><td><a href="/test">Test an API request</a></td>' +
          '<td>Submit an API request and see a formatted JSON response. ' +
          '    Go through the authorization flow if there are no stored ' +
          '    credentials for the user.</td></tr>' +
          '<tr><td><a href="/authorize">Test the auth flow directly</a></td>' +
          '<td>Go directly to the authorization flow. If there are stored ' +
          '    credentials, you still might not be prompted to reauthorize ' +
          '    the application.</td></tr>' +
          '<tr><td><a href="/revoke">Revoke current credentials</a></td>' +
          '<td>Revoke the access token associated with the current user ' +
          '    session. After revoking credentials, if you go to the test ' +
          '    page, you should see an <code>invalid_grant</code> error.' +
          '</td></tr>' +
          '<tr><td><a href="/clear">Clear Flask session credentials</a></td>' +
          '<td>Clear the access token currently stored in the user session. ' +
          '    After clearing the token, if you <a href="/test">test the ' +
          '    API request</a> again, you should go back to the auth flow.' +
          '</td></tr></table>')


if __name__ == '__main__':
  # When running locally, disable OAuthlib's HTTPs verification.
  # ACTION ITEM for developers:
  #     When running in production *do not* leave this option enabled.
  os.environ['OAUTHLIB_INSECURE_TRANSPORT'] = '1'

  app.run('localhost', 8080, debug=True)

具体来说，这些是当前有效的部分（我被替换的频道）和不有效的部分（我的频道成员）...

  # THIS WORKS
  # subs = youtube.subscriptions().list(
  #     part="snippet,contentDetails",
  #     mine=True
  # ).execute()

  # THIS DOESNT WORK
  subs = youtube.members().list(
      part="snippet",
      maxResults=50
  ).execute()

Answer 1

import requests
from bs4 import BeautifulSoup
import pypyodbc    
connection = pypyodbc.connect('Driver={SQL Server};Server=ipserver;Database=DBContext;uid=yourid;pwd=passwd')   
import datetime
sql_select_Query = "  select  URLChannel,idstr from [dbo].[Channel] group by [IDStr], [URLChannel]"
cursors2 = connection.cursor()
cursors2.execute(sql_select_Query)
# get all records
records = cursors2.fetchall()
for row in records:
    youtube = build('youtube', 'v3',
                developerKey='yourdeveloperkey')
 
    ch_request = youtube.channels().list(
        part='statistics,snippet',
        id=row[1])
 
    print("url:   ", row[0]) 
    # Channel Information
    ch_response = ch_request.execute()
    urlchannel = row[0]
    idStr = row[1]
    sub = 0
    checkcondition = ch_response['items'][0]['statistics']['hiddenSubscriberCount']
    if checkcondition == False:
        sub = ch_response['items'][0]['statistics']['subscriberCount']
    vid = ch_response['items'][0]['statistics']['videoCount']
    views = ch_response['items'][0]['statistics']['viewCount']
    snippet = ch_response["items"][0]["snippet"]
    statistics = ch_response["items"][0]["statistics"]
    #channel_country = snippet["country"]
    channel_country = ''
    channel_description = snippet["description"]
    channel_creation_date = snippet["publishedAt"]
    channel_title = snippet["title"]
    cursor = connection.cursor()    
    SQLCommand = ("INSERT INTO [dbo].[Channel]([Channel_title],[Channel_country],[Channel_creation_date],[Channel_description],[subscriberCount],[videoCount],[viewCount],[URLChannel],[IDStr],[UpdateTime],[CreateTime],[UpdateUser],[CreateUser]) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)")    
    Values = [channel_title, channel_country, channel_creation_date, channel_description, sub,vid,views,urlchannel,idStr,datetime.datetime.now(),datetime.datetime.now(),"usercreate","userupdate"]   
    #Processing Query    
    cursor.execute(SQLCommand,Values)     
    #Commiting any pending transaction to the database.    
    connection.commit()    
    #closing connection    
    print("Data Successfully Inserted")