Made some chanages and now Can't remmebr what I did
This commit is contained in:
+165
@@ -0,0 +1,165 @@
|
||||
import asyncio
|
||||
import json
|
||||
import requests
|
||||
from twitchAPI.twitch import Twitch
|
||||
from twitchAPI.helper import first
|
||||
# Import the explicit VideoType Enum to prevent the AttributeError
|
||||
from twitchAPI.type import VideoType
|
||||
|
||||
SECRETS = None
|
||||
TWITCH = None
|
||||
USER = None
|
||||
GAME_CACHE = {} # Local cache dictionary to store game_id -> game_name mapping
|
||||
|
||||
CHANNEL_NAME = "teampgp"
|
||||
|
||||
async def get_twitch():
|
||||
global SECRETS
|
||||
global TWITCH
|
||||
global USER
|
||||
if SECRETS is None:
|
||||
with open('twitch_secrets.json', 'r') as f:
|
||||
SECRETS = json.load(f)
|
||||
if TWITCH is None:
|
||||
TWITCH = await Twitch(SECRETS['client_id'], SECRETS['client_secret'])
|
||||
if USER is None:
|
||||
USER = await first(TWITCH.get_users(logins=[CHANNEL_NAME]))
|
||||
if not USER:
|
||||
print("User not found.")
|
||||
|
||||
async def get_game_name_by_id(game_id: str) -> str:
|
||||
"""Helper function to fetch game names and cache them locally."""
|
||||
if not game_id:
|
||||
return "Unknown / No Category"
|
||||
if game_id in GAME_CACHE:
|
||||
return GAME_CACHE[game_id]
|
||||
try:
|
||||
game_generator = TWITCH.get_games(game_ids=[game_id])
|
||||
game = await first(game_generator)
|
||||
if game:
|
||||
GAME_CACHE[game_id] = game.name
|
||||
return game.name
|
||||
except Exception:
|
||||
pass
|
||||
return "Unknown Game"
|
||||
|
||||
async def get_vod_game_name(vod_id: str) -> str:
|
||||
game_id = 0
|
||||
game_name = "Unknown Game"
|
||||
url = "https://gql.twitch.tv/gql"
|
||||
|
||||
headers = {
|
||||
"Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
|
||||
payload = [{
|
||||
"operationName": "VideoMetadata",
|
||||
"variables": {
|
||||
"channelLogin": "",
|
||||
"videoID": vod_id
|
||||
},
|
||||
"extensions": {
|
||||
"persistedQuery": {
|
||||
"version": 1,
|
||||
"sha256Hash": "45111672eea2e507f8ba44d101a61862f9c56b11dee09a15634cb75cb9b9084d"
|
||||
}
|
||||
}
|
||||
}]
|
||||
|
||||
response = requests.post(url, headers=headers, json=payload)
|
||||
data = response.json()
|
||||
|
||||
# Parsing the Game ID out of the response array
|
||||
video_info = data[0]['data']['video']
|
||||
if video_info and video_info.get('game'):
|
||||
game_id = video_info['game']['id']
|
||||
game_name = video_info['game']['displayName']
|
||||
print(f"Game: {game_name} (ID: {game_id})")
|
||||
else:
|
||||
print("No game information found for this VOD.")
|
||||
|
||||
return game_id, game_name
|
||||
|
||||
async def get_streamer_vods():
|
||||
await get_twitch()
|
||||
|
||||
print(f"Starting VOD extraction for {USER.display_name}...")
|
||||
vod_generator = TWITCH.get_videos(user_id=USER.id, first=100, video_type=VideoType.ALL)
|
||||
|
||||
all_vods = []
|
||||
|
||||
async for v in vod_generator:
|
||||
# FIXED: Resolving game category using the automatic stream markers
|
||||
game_id, game_name = await get_vod_game_name(v.id)
|
||||
|
||||
#print(f"{v}")
|
||||
|
||||
vod_data = {
|
||||
"id": v.id,
|
||||
"title": v.title,
|
||||
"created_at": str(v.published_at),
|
||||
"view_count": v.view_count,
|
||||
"duration": v.duration,
|
||||
"url": v.url,
|
||||
"thumbnail_url": v.thumbnail_url,
|
||||
"game_id": game_id,
|
||||
"game_name": game_name,
|
||||
"stream_id": v.stream_id,
|
||||
"creator_name": CHANNEL_NAME
|
||||
}
|
||||
all_vods.append(vod_data)
|
||||
print(f"Collected VOD: {v.title} | Category: {game_name} ({v.duration})")
|
||||
|
||||
print(f"\nFinished extracting VODs. Total gathered: {len(all_vods)}")
|
||||
return all_vods
|
||||
|
||||
async def get_streamer_clips():
|
||||
await get_twitch()
|
||||
|
||||
print(f"Starting clip extraction for {USER.display_name}...")
|
||||
|
||||
clip_generator = TWITCH.get_clips(broadcaster_id=USER.id, first=100)
|
||||
|
||||
all_clips = []
|
||||
|
||||
async for c in clip_generator:
|
||||
# Clips DO have game_id attributes natively supported
|
||||
game_name = await get_game_name_by_id(c.game_id)
|
||||
|
||||
clip_data = {
|
||||
"id": c.id,
|
||||
"title": c.title,
|
||||
"created_at": str(c.created_at),
|
||||
"view_count": c.view_count,
|
||||
"duration": c.duration,
|
||||
"url": c.url,
|
||||
"thumbnail_url": c.thumbnail_url,
|
||||
"game_id": c.game_id,
|
||||
"game_name": game_name,
|
||||
"stream_id": "0",
|
||||
"creator_name": c.creator_name,
|
||||
}
|
||||
all_clips.append(clip_data)
|
||||
print(f"Collected clip: {c.title} | Category: {game_name} ({c.view_count} views)")
|
||||
|
||||
print(f"\nFinished extracting clips. Total gathered: {len(all_clips)}")
|
||||
return all_clips
|
||||
|
||||
async def main():
|
||||
print("--- Script Started ---")
|
||||
|
||||
# 1. Pull clips (with categories)
|
||||
#clips_list = await get_streamer_clips()
|
||||
#print(f"\nSuccessfully received a list of {len(clips_list)} clips in main().")
|
||||
#if clips_list:
|
||||
# print(f"Top clip: '{clips_list[0]['title']}' (Game: {clips_list[0]['game_name']})")
|
||||
|
||||
# 2. Pull VODs (without categories)
|
||||
vods_list = await get_streamer_vods()
|
||||
#print(f"\nSuccessfully received a list of {len(vods_list)} VODs in main().")
|
||||
if vods_list:
|
||||
print(f"Recent VOD: '{vods_list[0]['title']}'")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user