#!/usr/bin/env python3 import requests import os import subprocess from database import Database import uploader from uploader import CategoryId import youtube_hashtags import transcribe_video CHANNEL_NAME = 'teampgp' # Replace with the streamer's username DB = None 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 transcribe(id: str): """Transcribes the Video File.""" video_file = f"save/{DB.table}/{id}/{id}.mp4" # Check if the video file exists if not os.path.exists(video_file): print(f"Error: File not found: {video_file}") return False # no need to continue if srt transcribe file already exists if os.path.exists(f"save/{DB.table}/{id}/transcribe_{id}.srt"): print(f"video already transcribed:") return True transcribe_video.extract_audio(video_file, f"save/{DB.table}/{id}/temp_{id}_audio.wav") transcribe_video.transcribe_to_srt(f"save/{DB.table}/{id}/temp_{id}_audio.wav", f"save/{DB.table}/{id}/", f"transcribe_{id}") return True def top_hashtags(id: str): """"Hashtags from transcribed SRT file.""" file_srt = f"save/{DB.table}/{id}/transcribe_{id}.srt" # if srt transcribe file not exists if not os.path.exists(file_srt): print(f"Transcribe file not found {file_srt}") return [] tags = youtube_hashtags.get_top_hashtags(file_srt) return tags def download(): """Find all undownload vods and download them.""" undownloaded = DB.get_undownloaded() print(f"Download {DB.table}...") if DB.table == "vods": for record_id, record_date, title, game_name, downloaded, uploaded_yt, chat_upload_yt in undownloaded: 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/vods/{record_id}/{record_id}.mp4 --collision Overwrite") output2 = run_linux_command(f"TwitchDownloaderCLI chatdownload --id {record_id} -o save/vods/{record_id}/{record_id}_chat.json -E --collision Overwrite") if output["success"] is True: print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title} | was successful") DB.mark_as_downloaded(record_id) else: print(f"ID: {record_id} | Download Process failed. {output["stdout"]}. Error: {output["stderr"]}") elif DB.table == "clips": for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt, uploaded_shorts_yt in undownloaded: print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}") # Uses standard clipdownload directive output = run_linux_command(f"TwitchDownloaderCLI clipdownload --id {slug} -o save/clips/{slug}/{slug}.mp4 --collision Overwrite") if output["success"] is True: print(f"Slug: {slug} | Was successfully downloaded.") DB.mark_as_downloaded(slug) else: print(f"Slug: {slug} | Download Process failed. {output["stdout"]}. Error: {output["stderr"]}") print(f"Finished Downloading {DB.table}...") def upload(): """Loops over the downloaded videos entries and uploaded them to youtube.""" print(f"Uploading {DB.table}...") unuploaded = DB.get_unuploaded() twitch_datetime = " Live on Twitch Every Friday and Sunday @7:30 ET https://twitch.tv/teampgp" file_path = "" title = "" description = "" categoryId = CategoryId.GAMING privatcyStatus = 'private' base_tags = ['gaming', 'TeamPGP', 'twitch', 'Level1Techs'] upload_queue = [] if DB.table == "vods": for record_id, record_date, title, game_name, downloaded, uploaded_yt, chats_upload_yt in unuploaded: print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title}") file_path = f"save/vods/{record_id}/{record_id}.mp4" description = f"Game: {game_name}, on {record_date}, #VODS {twitch_datetime}" tags = list(base_tags) tags.extend([f'{game_name}', 'twitch_vods', 'vods']) tags.extend(top_hashtags(record_id)) upload_queue.append([record_id, file_path, title, categoryId, description, privatcyStatus, tags]) elif DB.table == "clips": for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt, shorts_uploaded_yt in unuploaded: print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}") transcribe(slug) file_path = f"save/clips/{slug}/{slug}.mp4" description = f"Game: {game_name}, Cliped By: {clip_by}, on {record_date}, #Shorts #Clips {twitch_datetime}" tags = list(base_tags) tags.extend([f'{game_name}', 'twitch_clips', 'clips', f'{clip_by}', 'shorts']) tags.extend(top_hashtags(slug)) upload_queue.append([slug, file_path, title, categoryId, description, privatcyStatus, tags]) for db_id, file_path, title, categoryId, description, privatcyStatus, tags in upload_queue: output = uploader.upload_video(file_path, title, categoryId, description, privatcyStatus, tags) if output is True: print(f"Title: {title} | Was successfully uploaded.") DB.mark_as_uploaded(db_id) else: print(f"Title: {title} | Download Process failed.") def create_chats(): pass def create_shorts(): pass def get_vod_ids_simplified(channel_name: str): """Queries Twitch's public endpoint directly for trending clips.""" 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" } vods_query_string = """ query GetChannelVideos($login: String!, $limit: Int!) { user(login: $login) { videos(first: $limit, types: [ARCHIVE]) { edges { node { id title publishedAt game { displayName } } } } } } """ 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 } } } } } } """ limit = 50 query_string = "" operation_name = "" if DB.table == "vods": query_string = vods_query_string limit = 50 operation_name = "GetChannelVideos" elif DB.table == "clips": query_string = clips_query_string limit = 40 operation_name = "GetChannelClips" # A simplified query string that hardcodes the ARCHIVE type filter into the request structure payload = [{ "operationName": operation_name, "query": query_string, "variables": { "login": channel_name.lower(), "limit": limit } }] 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 = None if DB.table == "vods": edges = user_data['videos']['edges'] elif DB.table == "clips": edges = user_data['clips']['edges'] video_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" if DB.table == "vods": print(f"ID: {node['id']} | Date: {node['publishedAt']} | Game: {game_name} | Title: {node['title']}") # Pass game_name to your database logic DB.insert_vods_record(node['id'], node['publishedAt'], node['title'], game_name) video_ids.append(node['id']) elif DB.table == "clips": # 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']}") DB.insert_clips_record(node['slug'], node['createdAt'], node['title'], game_name, clip_by, int(node['viewCount'])) video_ids.append(node['slug']) return video_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__": tables = ["vods", "clips"] for table in tables: DB = Database(table) get_vod_ids_simplified(CHANNEL_NAME) download() upload() DB.close_database()