You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何通过OAuth授权多个渠道并使用YouTube Analytics API批量获取多渠道分析报告?

如何通过OAuth授权多个渠道并使用YouTube Analytics API批量获取多渠道分析报告?

嗨,我太懂你现在的处境了——单个频道用YouTube Analytics API查数据顺风顺水,但要批量搞多个频道的时候,OAuth授权这块就摸不着头脑了对吧?别慌,我来给你拆解清楚两种常见场景的解决办法,再给你改改代码就能用!

先搞懂核心逻辑:OAuth授权和YouTube频道的绑定关系

首先得明确:YouTube的OAuth授权是和Google账号绑定的,不是单个频道。也就是说:

  • 如果你的所有目标频道都归同一个Google账号所有,那只需要授权一次这个账号,就能访问该账号下的所有频道数据;
  • 如果频道分属不同的Google账号,那你得给每个账号单独授权,并且管理好各自的凭证文件,调用API时切换不同的凭证就行。

场景1:所有目标频道属于同一个Google账号

这种情况最省心,步骤如下:

  1. 先获取该Google账号下的所有频道ID
    你需要先调用YouTube Data API(要确保你的项目已经启用这个API哦),获取当前授权账号下的所有YouTube频道ID。
  2. 循环调用YouTube Analytics API查询每个频道的数据
    拿到所有频道ID后,直接循环遍历,把每个ID代入你原来的查询代码里就行。

给你改好的代码示例:

from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from datetime import date
import json

# 配置信息
SCOPES = ["https://www.googleapis.com/auth/yt-analytics.readonly", "https://www.googleapis.com/auth/youtube.readonly"]
API_SERVICE_NAME_ANALYTICS = "youtubeAnalytics"
API_VERSION_ANALYTICS = "v2"
API_SERVICE_NAME_YOUTUBE = "youtube"
API_VERSION_YOUTUBE = "v3"
CLIENT_SECRETS_FILE = "client_secret_2.json"

def get_services():
    # 一次性授权,同时获取Analytics和Data API的服务
    flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRETS_FILE, SCOPES)
    credentials = flow.run_local_server()
    analytics_service = build(API_SERVICE_NAME_ANALYTICS, API_VERSION_ANALYTICS, credentials=credentials)
    youtube_service = build(API_SERVICE_NAME_YOUTUBE, API_VERSION_YOUTUBE, credentials=credentials)
    return analytics_service, youtube_service

def get_all_channel_ids(youtube_service):
    # 获取当前账号下的所有频道ID
    response = youtube_service.channels().list(
        part="id",
        mine=True
    ).execute()
    return [item["id"] for item in response.get("items", [])]

def fetch_channel_analytics(analytics_service, channel_id, today):
    # 单个频道的数据分析查询(沿用你原来的逻辑)
    result = analytics_service.reports().query(
        ids="channel==" + channel_id,
        startDate='2023-11-01',
        endDate=today,
        metrics='views,likes,subscribersGained,estimatedMinutesWatched,averageViewDuration,shares',
        dimensions='day,creatorContentType',
        sort='day'
    ).execute()
    return channel_id, result

if __name__ == "__main__":
    today = date.today().isoformat()
    # 获取服务
    analytics_service, youtube_service = get_services()
    # 获取所有频道ID
    channel_ids = get_all_channel_ids(youtube_service)
    # 批量查询每个频道的数据
    for channel_id in channel_ids:
        print(f"正在查询频道 {channel_id} 的数据...")
        channel_id, data = fetch_channel_analytics(analytics_service, channel_id, today)
        # 这里可以把数据保存到文件或者数据库,比如:
        with open(f"{channel_id}_analytics.json", "w") as f:
            json.dump(data, f)
        print(f"频道 {channel_id} 的数据查询完成!")

场景2:频道分属不同的Google账号

这种情况需要给每个Google账号单独授权,并保存各自的凭证文件,调用时切换不同的凭证。

  1. 为每个Google账号生成并保存凭证
    你可以修改get_service函数,让它能指定凭证保存路径,第一次授权后把凭证存下来,后续直接加载不需要重复授权。
  2. 循环加载不同账号的凭证,调用API查询

示例代码(简化版):

from googleapiclient.discovery import build
from google_auth_oauthlib.flow import InstalledAppFlow
from google.oauth2.credentials import Credentials
import json
from datetime import date

SCOPES = ["https://www.googleapis.com/auth/yt-analytics.readonly"]
API_SERVICE_NAME = "youtubeAnalytics"
API_VERSION = "v2"
CLIENT_SECRETS_FILE = "client_secret_2.json"
# 存储不同账号的凭证文件路径,你可以根据实际情况添加更多
ACCOUNT_CREDENTIALS = {
    "账号A": "credentials_accountA.json",
    "账号B": "credentials_accountB.json"
}

def get_or_create_credentials(account_name):
    cred_path = ACCOUNT_CREDENTIALS[account_name]
    try:
        # 尝试加载已有的凭证
        credentials = Credentials.from_authorized_user_file(cred_path, SCOPES)
    except FileNotFoundError:
        # 没有凭证则重新授权
        flow = InstalledAppFlow.from_client_secrets_file(CLIENT_SECRETS_FILE, SCOPES)
        credentials = flow.run_local_server()
        # 保存凭证到文件,下次直接用
        with open(cred_path, "w") as f:
            f.write(credentials.to_json())
    return credentials

def get_service(credentials):
    return build(API_SERVICE_NAME, API_VERSION, credentials=credentials)

def fetch_analytics_data(service, channel_id, today):
    result = service.reports().query(
        ids="channel==" + channel_id,
        startDate='2023-11-01',
        endDate=today,
        metrics='views,likes,subscribersGained,estimatedMinutesWatched,averageViewDuration,shares',
        dimensions='day,creatorContentType',
        sort='day'
    ).execute()
    return result

if __name__ == "__main__":
    today = date.today().isoformat()
    # 循环处理每个账号
    for account_name, _ in ACCOUNT_CREDENTIALS.items():
        print(f"正在处理账号 {account_name}...")
        credentials = get_or_create_credentials(account_name)
        service = get_service(credentials)
        # 这里可以复用场景1里的get_all_channel_ids函数,动态获取该账号下的频道ID
        # 这里先假设你已经拿到了对应账号的频道列表
        channel_ids = ["你的频道ID1", "你的频道ID2"]
        for channel_id in channel_ids:
            data = fetch_analytics_data(service, channel_id, today)
            print(f"账号 {account_name} 下频道 {channel_id} 的数据查询完成!")

几个要注意的小细节

  • 不管哪种场景,都要确保你的Google Cloud项目已经启用了YouTube Analytics API和(如果需要获取频道列表的话)YouTube Data API v3
  • 批量调用API时要注意Google的API配额限制,别一次性调用太频繁导致被限流;
  • 授权时要确保登录的Google账号对目标频道有YouTube Analytics的访问权限(比如是频道的所有者或管理者)。

备注:内容来源于stack exchange,提问作者Sarah Sales

火山引擎 最新活动