python_scripts/package/twitch/video_info.py

#!/usr/bin/env python3
import asyncio
import httpx  # Switched from requests to prevent async loop freezing

from twitchAPI.helper import first
from twitchAPI.type import VideoType

import package.twitch.config as config
from package.config import DB as DB
from package.database.database import TwitchVideo

GAME_CACHE = {} # Local cache dictionary to store game_id -> game_name mapping

async def get_game_id_by_name(game_name: str) -> str:
    if config.SECRETS is None:
        await config.get_twitch()    
    # Query the 'get_games' endpoint using the names parameter
    # The API expects a list, even for a single game name
    async for game in config.TWITCH.get_games(names=[game_name]):
        print(f"Game Name: {game.name}")
        print(f"Game ID:   {game.id}")
        print(f"IGDB ID:   {game.igdb_id}")
        return game.id

    print(f"No game found with the exact name: '{game_name}'")
    return None

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 = config.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) -> tuple:
    """Asynchronously query Twitch GQL endpoint for VOD game metadata."""
    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": str(vod_id)
        },
        "extensions": {
            "persistedQuery": {
                "version": 1,
                "sha256Hash": "45111672eea2e507f8ba44d101a61862f9c56b11dee09a15634cb75cb9b9084d"
            }
        }
    }]

    # 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

async def get_streamer_vods() -> list:
    await config.get_twitch()
    
    print(f"Starting VOD extraction for {config.USER.display_name}...")
    vod_generator = config.TWITCH.get_videos(user_id=config.USER.id, first=100, video_type=VideoType.ALL)
 
    all_vods = []

    async for v in vod_generator:
        # 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 = game_id
        except ValueError:
            clean_game_id = 0

        vod_data = TwitchVideo()
        vod_data.id = v.id
        vod_data.title = v.title
        vod_data.created_at = str(v.published_at)
        vod_data.view_count = int(v.view_count) if v.view_count else 0
        vod_data.duration = v.duration
        vod_data.url = v.url
        vod_data.thumbnail_url = v.thumbnail_url
        vod_data.game_id = clean_game_id
        vod_data.game_name = game_name
        vod_data.stream_id = str(v.stream_id) if v.stream_id else "0"
        vod_data.creator_name = config.CHANNEL_NAME
        vod_data.clip_is = False
  
        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() -> list:
    await config.get_twitch()

    print(f"Starting clip extraction for {config.USER.display_name}...")
    clip_generator = config.TWITCH.get_clips(broadcaster_id=config.USER.id, first=100)
    all_clips = []

    async for c in clip_generator:
        game_name = await get_game_name_by_id(c.game_id)

        clip_data = TwitchVideo()
        clip_data.id = c.id
        clip_data.title = c.title
        clip_data.created_at = str(c.created_at)
        clip_data.view_count = int(c.view_count) if c.view_count else 0
        clip_data.duration = c.duration
        clip_data.url = c.url
        clip_data.thumbnail_url = c.thumbnail_url
        clip_data.game_id = c.game_id
        clip_data.game_name = game_name
        clip_data.stream_id = "0"
        clip_data.creator_name = c.creator_name
        clip_data.clip_is = True

        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 video_info():
    print("--- Script Started ---")
    
    vods_list = await get_streamer_vods()
    
    for v in vods_list:
        DB.insert(v)

    if vods_list:
        print("====================================================")
        print(f"Recent VOD: '{vods_list[0].title}'")
        print("====================================================")

    clips_list = await get_streamer_clips()
        
    for v in clips_list:
        DB.insert(v)

    if clips_list:
        print("====================================================")
        print(f"Recent VOD: '{clips_list[0].title}'")
        print("====================================================")

if __name__ == "__main__":
    asyncio.run(video_info())