python_scripts/twitch_download_thumbnails.py

import os
import json
import asyncio
import requests
from twitchAPI.twitch import Twitch
from twitchAPI.helper import first

import database

# 1. Fill in your credentials from the Twitch Developer Console
SECRETS = None
TWITCH = None
USER = None
CHANNEL_NAME = "teampgp"

async def get_twitch():
    global SECRETS, TWITCH, 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 download_live_thumbnail(twitch_client, streamer_username: str, w: int = 1920, h: int = 1080):
    """Fetches and saves the live stream thumbnail for an active broadcast."""
    print(f"🔎 Checking live status for: {streamer_username}...")
    
    # Query the live streams endpoint
    stream_generator = twitch_client.get_streams(user_logins=[streamer_username])
    stream_data = await first(stream_generator)
    
    if not stream_data:
        print(f"❌ User '{streamer_username}' is offline. Live thumbnails require an active stream.")
        return

    # Twitch API live streams use the {width} and {height} format
    raw_url = stream_data.thumbnail_url
    clean_url = raw_url.replace('{width}', str(w)).replace('{height}', str(h))
    
    filename = f"live_{streamer_username}_{w}x{h}.jpg"
    save_image(clean_url, filename)

async def download_vod_thumbnail(twitch_client, vod_id: str, w: int = 1920, h: int = 1080):
    """Fetches and saves a thumbnail from a past broadcast VOD ID."""
    print(f"🔎 Searching for VOD ID: {vod_id}...")
    
    # Query the videos endpoint
    video_generator = twitch_client.get_videos(vod_id)
    video_data = await first(video_generator)

    filename = f"download/videos/{vod_id}/{vod_id}_{w}x{h}.jpg"
    if os.path.exists(filename):
        return
    
    if not video_data:
        print(f"❌ VOD ID {vod_id} could not be found.")
        return

    # Twitch VOD endpoints typically format string tokens as %{width} and %{height}
    raw_url = video_data.thumbnail_url
    if not raw_url:
        print("❌ This VOD does not have an available thumbnail.")
        return
        
    clean_url = raw_url.replace('%{width}', str(w)).replace('%{height}', str(h))
    
    #filename = f"vod_{vod_id}_{w}x{h}.jpg"
    save_image(clean_url, filename)

async def download_clip_thumbnail(clip_id: str, url: str):
    print(f"🔎 Searching for Clip ID: {clip_id}...")
           
    filename = f"download/clips/{clip_id}/{clip_id}.jpg"
    if os.path.exists(filename):
        return
           
    save_image(url, filename, False)

def save_image(url: str, filename: str, stream: bool = True):
    """Helper function to stream image bytes directly to a file."""
    try:
        response = requests.get(url, stream)
        if response.status_code == 200:
            with open(filename, 'wb') as file:
                if stream is True:
                    for chunk in response.iter_content(1024):
                        file.write(chunk)
                else:
                    file.write(response.content)
            print(f"✅ Success! Saved as: {filename}")
        else:
            print(f"❌ Download failed. HTTP Status: {response.status_code}")
    except Exception as e:
        print(f"❌ An error occurred during file writing: {e}")

async def main():
    # Initialize connection & automatically authorize the App token
    await get_twitch()
    twitch = TWITCH # Twitch(APP_ID, APP_SECRET)
    
    # --- OPTION A: Download Live Thumbnail ---
    # Target user must be streaming live right now
    target_streamer = CHANNEL_NAME
    #await download_live_thumbnail(twitch, target_streamer, 1920, 1080)
    
    # --- OPTION B: Download Past Broadcast VOD Thumbnail ---
    # Extract the ID sequence from your target video link
    # target_vod = "2145678901"
    db = database.Database()
    vods = db.get_vods()
    for vod in vods:
        (
            id,
            title,
            created_at,
            view_count,
            duration,
            url,
            thumbnail_url,
            game_id,
            game_name,
            stream_id,
            creator_name,
            clip_is,
            downloaded,
            uploaded_yt,
            uploaded_yt_chats,
            uploaded_yt_shorts,
        ) = vod

        await download_vod_thumbnail(twitch, id, 1920, 1080)

    clips = db.get_clips()
    for clip in clips:
        (
            id,
            title,
            created_at,
            view_count,
            duration,
            url,
            thumbnail_url,
            game_id,
            game_name,
            stream_id,
            creator_name,
            clip_is,
            downloaded,
            uploaded_yt,
            uploaded_yt_chats,
            uploaded_yt_shorts,
        ) = clip
    
        await download_clip_thumbnail(id, thumbnail_url)

if __name__ == '__main__':
    asyncio.run(main())