#!/usr/bin/env python3 import requests import sqlite3 import subprocess from datetime import datetime import uploader from uploader import CategoryId CHANNEL_NAME = 'teampgp' # Replace with the streamer's username # Stores data locally CONN = sqlite3.connect("clips_database.db") CURSOR = CONN.cursor() def create_database(): """Creates a table structured explicitly for Twitch clip properties.""" CURSOR.execute( """ CREATE TABLE IF NOT EXISTS clips ( slug TEXT PRIMARY KEY, date TEXT NOT NULL, title TEXT NOT NULL, gamename TEXT NOT NULL, clip_by TEXT NOT NULL, view_count INTEGER NOT NULL, downloaded INTEGER NOT NULL, uploaded_yt INTEGER NOT NULL ) """ ) CONN.commit() def close_database(): """Commits queries before ending connection context.""" CONN.commit() CONN.close() def download_clips(): """Loops over undownloaded metadata entries to write files down locally.""" undownloaded_clips = get_undownloaded_clips() print("Downloading Clips...") for slug, record_date, title, game_name, clip_by, views, downloaded in undownloaded_clips: print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}") # Ensures destination folder structures exist before executing CLI tool run_linux_command(f"mkdir -p save/clips/{slug}") # Uses standard clipdownload directive output = run_linux_command(f"TwitchDownloaderCLI clipdownload --id {slug} -o save/clips/{slug}/{slug}.mp4") if output["success"] is True: print(f"Slug: {slug} | Was successfully downloaded.") mark_as_downloaded(slug) else: print(f"Slug: {slug} | Download Process failed.") print("Finished Downloading Clips...") def upload_clips(): """Loops over the downloaded videos entries and uploaded them to youtube.""" print("Uploading Clips...") unpuloaded_clips = get_unuploaded_clips() for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in unpuloaded_clips: print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}") file_path = f"save/clips/{slug}/{slug}.mp4" title = title description = f"Game: {game_name}, Cliped By: {clip_by}, on {record_date}" categoryId = CategoryId.SHORTS privatcyStatus = 'private' tags = ['#SHORT', '#GAMING', '#TeamPGP', f'#{game_name}', f'#{clip_by}'] if (uploader.upload_video(file_path, title, description, categoryId, privatcyStatus, tags) is True): print(f"Slug: {slug} | Was successfully uploaded.") mark_as_uploaded(slug) else: print(f"Slug: {slug} | Upload Process failed.") def run_linux_command(command: str): """Executes a Linux command, waits for completion, and returns output.""" try: 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: return {"success": False, "stdout": e.stdout, "stderr": e.stderr} def mark_as_uploaded(slug: str): """Flags a specific clip row record to uploaded (1).""" CURSOR.execute( "UPDATE clips SET uploaded_yt = 1 WHERE slug = ?", (slug,) ) CONN.commit() def mark_as_downloaded(slug: str): """Flags a specific clip row record to downloaded (1).""" CURSOR.execute( "UPDATE clips SET downloaded = 1 WHERE slug = ?", (slug,) ) CONN.commit() def get_unuploaded_clips(): """Retrieves all clip rows remaining to be uploaded.""" CURSOR.execute( "SELECT slug, date, title, gamename, clip_by, view_count, downloaded, uploaded_yt FROM clips WHERE downloaded = 1 AND uploaded_yt = 0" ) return CURSOR.fetchall() def get_undownloaded_clips(): """Retrieves all clip rows remaining to be captured.""" CURSOR.execute( "SELECT slug, date, title, gamename, clip_by, view_count, downloaded FROM clips WHERE downloaded = 0" ) return CURSOR.fetchall() def insert_record(slug: str, record_date_str: str, title: str, gamename: str, clip_by: str, views: int): """Cleans up ISO-8601 strings into unified date structures for the database.""" try: clean_date = datetime.strptime(record_date_str, "%Y-%m-%dT%H:%M:%SZ").date() except ValueError: clean_date = record_date_str CURSOR.execute( "INSERT OR IGNORE INTO clips (slug, date, title, gamename, clip_by, view_count, downloaded, uploaded_yt) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (slug, str(clean_date), title, gamename, clip_by, views, False, False), ) CONN.commit() def get_channel_clips(channel_name: str): """Queries Twitch's public endpoint directly for trending clips.""" session = requests.Session() url = "https://gql.twitch.tv/gql" 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" } # Twitch GQL schema for channel discovery clips query_string = """ query GetChannelClips($login: String!, $limit: Int!) { user(login: $login) { clips(first: $limit, criteria: { period: ALL_TIME }) { edges { node { slug title createdAt viewCount game { displayName } curator { login } } } } } } """ payload = [{ "operationName": "GetChannelClips", "query": query_string, "variables": { "login": channel_name.lower(), "limit": 40 } }] try: req = requests.Request('POST', url, json=payload) prepped = session.prepare_request(req) response = session.send(prepped) response.raise_for_status() data = response.json() 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['clips']['edges'] slugs = [] print(f"--- Top Clips for {channel_name} ---") for edge in edges: node = edge['node'] game_info = node.get('game') game_name = game_info.get('displayName') if game_info else "Unknown/No Category" slug_id = node['slug'] # Safe extraction in case the curator account was deleted/missing curator_info = node.get('curator') clip_by = curator_info.get('login') if curator_info else "Unknown Creator" print(f"Slug: {node['slug']} | Date: {node['createdAt']} | Game: {game_name} | By: {clip_by} | Views: {node['viewCount']} | Title: {node['title']}") insert_record(node['slug'], node['createdAt'], node['title'], game_name, clip_by, int(node['viewCount'])) slugs.append(slug_id) return slugs except Exception as e: print(f"An unexpected error occurred: {e}") return [] if __name__ == "__main__": create_database() get_channel_clips(CHANNEL_NAME) download_clips() upload_clips() close_database()