如何通过OAuth授权多个渠道并使用YouTube Analytics API批量获取多渠道分析报告?
如何通过OAuth授权多个渠道并使用YouTube Analytics API批量获取多渠道分析报告?
嗨,我太懂你现在的处境了——单个频道用YouTube Analytics API查数据顺风顺水,但要批量搞多个频道的时候,OAuth授权这块就摸不着头脑了对吧?别慌,我来给你拆解清楚两种常见场景的解决办法,再给你改改代码就能用!
先搞懂核心逻辑:OAuth授权和YouTube频道的绑定关系
首先得明确:YouTube的OAuth授权是和Google账号绑定的,不是单个频道。也就是说:
- 如果你的所有目标频道都归同一个Google账号所有,那只需要授权一次这个账号,就能访问该账号下的所有频道数据;
- 如果频道分属不同的Google账号,那你得给每个账号单独授权,并且管理好各自的凭证文件,调用API时切换不同的凭证就行。
场景1:所有目标频道属于同一个Google账号
这种情况最省心,步骤如下:
- 先获取该Google账号下的所有频道ID
你需要先调用YouTube Data API(要确保你的项目已经启用这个API哦),获取当前授权账号下的所有YouTube频道ID。 - 循环调用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账号单独授权,并保存各自的凭证文件,调用时切换不同的凭证。
- 为每个Google账号生成并保存凭证
你可以修改get_service函数,让它能指定凭证保存路径,第一次授权后把凭证存下来,后续直接加载不需要重复授权。 - 循环加载不同账号的凭证,调用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




