#!/usr/bin/env python3 import requests import sqlite3 from datetime import date as datetime_date import subprocess CHANNEL_NAME = 'teampgp' # Replace with the streamer's username CONN = sqlite3.connect("database.db") CURSOR = CONN.cursor() def create_database(): # Creates table safely using multi-line string CURSOR.execute( """ CREATE TABLE IF NOT EXISTS vods ( id INTEGER PRIMARY KEY, date TEXT NOT NULL, title TEXT NOT NULL, gamename TEXT NOT NULL, downloaded INTEGER NOT NULL, uploaded_yt INTEGER NOT NULL, chat_upload_yt INTEGER NOT NULL ) """ ) CONN.commit() def close_database(): """Commit before we close.""" CONN.commit() CONN.close() def download_vods(): """Find all undownload vods and download them.""" undownload_vods = get_undownloaded_vods() print("Download VODs...") for record_id, record_date, title, game_name, downloaded in undownload_vods: print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title}") output = run_linux_command(f"TwitchDownloaderCLI videodownload --id {record_id} -o save/vod/{record_id}/{record_id}.mp4") output2 = run_linux_command(f"TwitchDownloaderCLI chatdownload --id {record_id} -o save/vod/{record_id}/{record_id}_chat.json -E") if output["success"] is True: print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title} | was successful") mark_as_downloaded(record_id) else: print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title} | was Failed") print("Finished Downloading VODs...") def run_linux_command(command: str): """Executes a Linux command, waits for completion, and returns output.""" try: # shell=True allows running full command strings with pipes/wildcards # text=True returns strings instead of bytes result = subprocess.run( command, shell=True, check=True, capture_output=True, text=True ) return {"success": True, "stdout": result.stdout, "stderr": result.stderr} except subprocess.CalledProcessError as e: # Handles errors if the Linux command returns a non-zero exit code return {"success": False, "stdout": e.stdout, "stderr": e.stderr} def mark_as_downloaded(record_id: int): """Updates the downloaded status to True (1) for a specific record ID.""" # Updates the row matching the specific ID CURSOR.execute( "UPDATE vods SET downloaded = 1 WHERE id = ?", (record_id,) ) CONN.commit() def get_unuploaded_clips(): """Retrieves all clip rows remaining to be chat uploaded.""" CURSOR.execute( "SELECT id, date, title, gamename, clip_by, view_count, downloaded, uploaded_yt, chat_uploaded_yt FROM vods WHERE downloaded = 1 AND chat_uploaded_yt = 0" ) return CURSOR.fetchall() def get_unuploaded_clips(): """Retrieves all clip rows remaining to be uploaded.""" CURSOR.execute( "SELECT id, date, title, gamename, clip_by, view_count, downloaded, uploaded_yt FROM vods WHERE downloaded = 1 AND uploaded_yt = 0" ) return CURSOR.fetchall() def get_undownloaded_vods(): """Retrieves all rows where downloaded status is False (0).""" # Query filters by 0 because SQLite stores booleans as integers CURSOR.execute( "SELECT id, date, title, gamename, downloaded FROM vods WHERE downloaded = 0" ) return CURSOR.fetchall() def insert_record(record_id: int, record_date: datetime_date, title: str, gamename: str): """Inserts a record with ID, date, gamename, and title into a SQLite database.""" # Connects to database file (creates it if missing) # Inserts data using parameterized queries to prevent SQL injection CURSOR.execute( "INSERT OR IGNORE INTO vods (id, date, title, gamename, downloaded, uploaded_yt, chat_uploaded_yt) VALUES (?, ?, ?, ?, ?, ?, ?)", (record_id, str(record_date), title, gamename, False, False, False), ) # Saves changes and closes the connection CONN.commit() def get_vod_ids_simplified(channel_name: str): session = requests.Session() url = "https://gql.twitch.tv/gql" # Case-preserved headers to prevent 400 Bad Request errors session.headers = { "Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Content-Type": "text/plain" } # A simplified query string that hardcodes the ARCHIVE type filter into the request structure query_string = """ query GetChannelVideos($login: String!, $limit: Int!) { user(login: $login) { videos(first: $limit, types: [ARCHIVE]) { edges { node { id title publishedAt game { displayName } } } } } } """ payload = [{ "operationName": "GetChannelVideos", "query": query_string, "variables": { "login": channel_name.lower(), "limit": 50 } }] try: # Prepping ensures Python does not rewrite the Client-ID header case req = requests.Request('POST', url, json=payload) prepped = session.prepare_request(req) response = session.send(prepped) response.raise_for_status() data = response.json() # Pull out the target index array dictionary object result = data[0] if isinstance(data, list) else data if "errors" in result: print(f"Twitch GraphQL Error: {result['errors']}") return [] user_data = result['data']['user'] if not user_data: print(f"Channel '{channel_name}' not found.") return [] edges = user_data['videos']['edges'] vod_ids = [] print(f"--- Latest VODs for {channel_name} ---") for edge in edges: node = edge['node'] # Safe extraction in case a VOD has no category set (Just Chatting, Uncategorized, etc.) game_info = node.get('game') game_name = game_info.get('displayName') if game_info else "Unknown/No Category" print(f"ID: {node['id']} | Date: {node['publishedAt']} | Game: {game_name} | Title: {node['title']}") # Pass game_name to your database logic insert_record(node['id'], node['publishedAt'], node['title'], game_name) vod_ids.append(node['id']) return vod_ids except Exception as e: print(f"An unexpected error occurred: {e}") if 'response' in locals(): print(f"Server Response Text: {response.text}") return [] if __name__ == "__main__": create_database() get_vod_ids_simplified(CHANNEL_NAME) download_vods() close_database()