twitch获取直播主上传视频
import requests
import json
# Twitch API 凭证
CLIENT_ID = 'XXXXXX'
CLIENT_SECRET = 'XXXXXX'
# 获取OAuth Token
def get_oauth_token():
url = 'https://id.twitch.tv/oauth2/token'
params = {
'client_id': CLIENT_ID,
'client_secret': CLIENT_SECRET,
'grant_type': 'client_credentials'
}
response = requests.post(url, params=params)
if response.status_code == 200:
token = response.json().get('access_token')
print(f"获取到的 OAuth Token: {token}")
return token
else:
print(f"获取 OAuth Token 失败,状态码: {response.status_code}, 内容: {response.text}")
return None
# 直播主用户名
usernames = ['blastpremier', 'eslcsb', 'eslcs_gg', 'eslcsb']
user_id_map = {}
def get_user_id(username, headers):
url = f"https://api.twitch.tv/helix/users?login={username}"
response = requests.get(url, headers=headers)
if response.status_code != 200:
print(f"请求失败,状态码: {response.status_code}, 内容: {response.text}")
return None
data = response.json()
if 'data' in data and data['data']:
return data['data'][0]['id']
else:
print(f"无法获取用户ID,返回数据: {data}")
return None
def get_all_videos(user_id, headers):
url = f"https://api.twitch.tv/helix/videos?user_id={user_id}&first=100"
videos = []
while url:
response = requests.get(url, headers=headers)
if response.status_code != 200:
print(f"获取视频失败,状态码: {response.status_code}, 内容: {response.text}")
break
data = response.json()
videos.extend(data.get('data', []))
# 检查是否有下一页
pagination = data.get('pagination', {})
cursor = pagination.get('cursor')
if cursor:
url = f"https://api.twitch.tv/helix/videos?user_id={user_id}&first=100&after={cursor}"
else:
url = None
return videos
def main():
# 获取 OAuth Token
oauth_token = get_oauth_token()
if not oauth_token:
print("无法获取 OAuth Token,程序结束。")
return
headers = {
'Client-ID': CLIENT_ID,
'Authorization': f'Bearer {oauth_token}'
}
all_videos = {}
for username in usernames:
if username not in user_id_map:
user_id = get_user_id(username, headers)
if user_id:
user_id_map[username] = user_id
else:
print(f"未找到用户:{username}")
continue
user_id = user_id_map[username]
videos = get_all_videos(user_id, headers)
if videos:
print(f"{username} 的第一个视频的完整字段: ")
print(json.dumps(videos[0], indent=4, ensure_ascii=False))
all_videos[username] = videos
print(f"{username} 上传的视频数量:{len(videos)}")
with open('twitch_videos.json', 'w', encoding='utf-8') as f:
json.dump(all_videos, f, ensure_ascii=False, indent=4)
if __name__ == "__main__":
main()