diff --git a/twitch_video_info.py b/twitch_video_info.py index bd0cfa7..d7533db 100644 --- a/twitch_video_info.py +++ b/twitch_video_info.py @@ -1,10 +1,9 @@ import asyncio import json -import requests +import httpx # Switched from requests to prevent async loop freezing import database 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 @@ -15,9 +14,7 @@ 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 + global SECRETS, TWITCH, USER if SECRETS is None: with open('twitch_secrets.json', 'r') as f: SECRETS = json.load(f) @@ -44,8 +41,9 @@ async def get_game_name_by_id(game_id: str) -> str: pass return "Unknown Game" -async def get_vod_game_name(vod_id: str) -> str: - game_id = 0 +async def get_vod_game_name(vod_id: str): + """Asynchronously query Twitch GQL endpoint for VOD game metadata.""" + game_id = "0" game_name = "Unknown Game" url = "https://gql.twitch.tv/gql" @@ -58,7 +56,7 @@ async def get_vod_game_name(vod_id: str) -> str: "operationName": "VideoMetadata", "variables": { "channelLogin": "", - "videoID": vod_id + "videoID": str(vod_id) }, "extensions": { "persistedQuery": { @@ -68,17 +66,21 @@ async def get_vod_game_name(vod_id: str) -> str: } }] - 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.") + # Using httpx async client to prevent blocking the asyncio event loop + async with httpx.AsyncClient() as client: + try: + response = await client.post(url, headers=headers, json=payload) + if response.status_code == 200: + data = response.json() + video_info = data[0].get('data', {}).get('video') + if video_info and video_info.get('game'): + game_id = str(video_info['game']['id']) + game_name = video_info['game']['displayName'] + print(f"GQL Found: {game_name} (ID: {game_id})") + else: + print(f"No game information found in GQL for VOD {vod_id}.") + except Exception as e: + print(f"Error fetching GQL metadata for VOD {vod_id}: {e}") return game_id, game_name @@ -91,18 +93,24 @@ async def get_streamer_vods(): all_vods = [] async for v in vod_generator: - # FIXED: Resolving game category using the automatic stream markers + # Resolving game category safely without freezing the event loop game_id, game_name = await get_vod_game_name(v.id) + + # Safely convert game_id to integer if possible, otherwise default to 0 + try: + clean_game_id = int(game_id) + except ValueError: + clean_game_id = 0 vod_data = { "id": v.id, "title": v.title, "created_at": str(v.published_at), - "view_count": int(v.view_count), + "view_count": int(v.view_count) if v.view_count else 0, "duration": v.duration, "url": v.url, "thumbnail_url": v.thumbnail_url, - "game_id": int(game_id), + "game_id": clean_game_id, "game_name": game_name, "stream_id": str(v.stream_id) if v.stream_id else "0", "creator_name": CHANNEL_NAME, @@ -118,24 +126,26 @@ 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) + + try: + clean_game_id = int(c.game_id) + except (ValueError, TypeError): + clean_game_id = 0 clip_data = { "id": c.id, "title": c.title, "created_at": str(c.created_at), - "view_count": int(c.view_count), + "view_count": int(c.view_count) if c.view_count else 0, "duration": c.duration, "url": c.url, "thumbnail_url": c.thumbnail_url, - "game_id": int(c.game_id), + "game_id": clean_game_id, "game_name": game_name, "stream_id": "0", "creator_name": c.creator_name, @@ -150,22 +160,30 @@ async def get_streamer_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']})") - db = database.Database() - # 2. Pull VODs (without categories) vods_list = await get_streamer_vods() + for v in vods_list: - db.insert_video_record(v['id'], v['title'], v['created_at'], v['view_count'], v['duration'], v['url'], v['thumbnail_url'], - v['game_id'], v['game_name'], v['stream_id'], v['creator_name'], v['clip_is']) + db.insert_video_record( + v['id'], v['title'], v['created_at'], v['view_count'], v['duration'], + v['url'], v['thumbnail_url'], v['game_id'], v['game_name'], + v['stream_id'], v['creator_name'], v['clip_is'] + ) - #print(f"\nSuccessfully received a list of {len(vods_list)} VODs in main().") if vods_list: print(f"Recent VOD: '{vods_list[0]['title']}'") + clips_list = await get_streamer_clips() + + for v in clips_list: + db.insert_video_record( + v['id'], v['title'], v['created_at'], v['view_count'], v['duration'], + v['url'], v['thumbnail_url'], v['game_id'], v['game_name'], + v['stream_id'], v['creator_name'], v['clip_is'] + ) + + if clips_list: + print(f"Recent VOD: '{clips_list[0]['title']}'") + if __name__ == "__main__": asyncio.run(main())