update

twitch_download_thumbnails.py

@@ -0,0 +1,162 @@
+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())