From 81e76b25d0917705fc953fa8ce070d8f2cc2eb6b Mon Sep 17 00:00:00 2001 From: SumGuyV5 Date: Thu, 23 Jul 2026 04:55:36 +0000 Subject: [PATCH] Made some chanages and now Can't remmebr what I did --- .gitignore | 2 + auth_twitch.py | 74 ++++++++ database.py | 50 +++--- linux.py | 17 ++ requirements.txt | 1 - transcribe_video.py | 10 +- twitch_down_uploader.py | 239 +++++++++++++++----------- twitch_download_clips.py | 1 - twitch_download_vod.py | 358 ++++++++++----------------------------- twitch_vod.py | 165 ++++++++++++++++++ youtube_short.py | 11 +- 11 files changed, 513 insertions(+), 415 deletions(-) create mode 100644 auth_twitch.py create mode 100644 linux.py create mode 100644 twitch_vod.py diff --git a/.gitignore b/.gitignore index 57cfa2c..5786515 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,5 @@ database.db save __pycache__ +twitch_secrets.json +.vscode/settings.json diff --git a/auth_twitch.py b/auth_twitch.py new file mode 100644 index 0000000..bd2a555 --- /dev/null +++ b/auth_twitch.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +import asyncio +import json +import os +import webbrowser +from twitchAPI.twitch import Twitch +from twitchAPI.oauth import UserAuthenticator +from twitchAPI.type import AuthScope + +SECRETS_FILE = "twitch_secrets.json" + +def load_credentials(): + """Loads existing Client ID and Secret from your JSON file.""" + if not os.path.exists(SECRETS_FILE): + raise FileNotFoundError(f"Could not find {SECRETS_FILE} in this directory.") + with open(SECRETS_FILE, "r") as f: + data = json.load(f) + return data.get("client_id"), data.get("client_secret") + +def save_token_to_json(token): + """Saves the generated token into twitch_secrets.json under 'manual_token'.""" + with open(SECRETS_FILE, "r") as f: + data = json.load(f) + + # Inject the new token + data["manual_token"] = token + + with open(SECRETS_FILE, "w") as f: + json.dump(data, f, indent=4) + print(f"\n[SUCCESS] Token saved inside '{SECRETS_FILE}' under 'manual_token'!") + +async def main(): + try: + client_id, client_secret = load_credentials() + if not client_id or not client_secret: + print("[ERROR] Please add your client_id and client_secret to the JSON file first.") + return + + print("Initializing local connection loop...") + # Initialize official Twitch connection interface + twitch = await Twitch(client_id, client_secret) + + # Scopes: We leave this empty [] since VOD collection only requires basic public clearance + scopes = [] + + # Create an authenticator that automatically sets up http://localhost:17563 + auth = UserAuthenticator(twitch, scopes, url="http://localhost:17563") + + # Request authentication URL + auth_url = auth.return_auth_url() + print(f"\nIf your browser does not open automatically, copy and paste this URL into your browser:\n{auth_url}\n") + + # Open your system default browser to let you manually click "Authorize" + webbrowser.open(auth_url) + + print("Waiting for you to click 'Authorize' in your web browser...") + # The script halts here, running a local background server until you click authorize + token, refresh_token = await auth.authenticate() + + print(f"\nSuccessfully generated Token: {token}") + + # Save it right back into your configuration file + save_token_to_json(token) + + # Gracefully shut down the library connection + await twitch.close() + + except Exception as e: + print(f"\n[ERROR] An error occurred: {e}") + print("Double-check that http://localhost:17563 is added to your Twitch Dev Console.") + +if __name__ == "__main__": + # Run the asynchronous loop + asyncio.run(main()) \ No newline at end of file diff --git a/database.py b/database.py index 9ef6bae..f28ef14 100644 --- a/database.py +++ b/database.py @@ -1,7 +1,6 @@ #!/usr/bin/env python3 import sqlite3 from datetime import datetime -from datetime import date as datetime_date from pathlib import Path class Database: @@ -15,7 +14,7 @@ class Database: self.CONN = sqlite3.connect("database.db") self.CURSOR = self.CONN.cursor() - if self.table == "vods": + if self.table == "videos": self.columns = "id, date, title, gamename, downloaded, uploaded_yt, chats_upload_yt" elif self.table == "clips": self.columns = "slug, date, title, gamename, clip_by, view_count, downloaded, uploaded_yt, shorts_upload_yt" @@ -24,10 +23,14 @@ class Database: self.create_database() def __del__(self): - self.close_database() + # Destructors are unpredictable in Python; explicitly close when done instead + try: + self.close_database() + except: + pass def create_database(self): - """Creates a table structured explicitly for Twitch clip properties.""" + """Creates a table structured explicitly for Twitch clips properties.""" self.CURSOR.execute( """ CREATE TABLE IF NOT EXISTS clips ( @@ -44,10 +47,10 @@ class Database: """ ) - """Creates a table structured explicitly for Twitch vods properties.""" + """Creates a table structured explicitly for Twitch videos properties.""" self.CURSOR.execute( """ - CREATE TABLE IF NOT EXISTS vods ( + CREATE TABLE IF NOT EXISTS videos ( id INTEGER PRIMARY KEY, date TEXT NOT NULL, title TEXT NOT NULL, @@ -63,8 +66,9 @@ class Database: def close_database(self): """Commit before we close.""" - self.CONN.commit() - self.CONN.close() + if self.CONN: + self.CONN.commit() + self.CONN.close() def __mark_as(self, record_id: int, set_sql: str): id = "id" @@ -117,32 +121,32 @@ class Database: self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 0") return self.CURSOR.fetchall() - def insert_vods_record(self, record_id: int, record_date_str: datetime_date, title: str, gamename: str): - """Inserts a record with ID, date, gamename, and title into a SQLite database.""" + def insert_videos_record(self, record_id: int, record_date_str: str, title: str, gamename: str): + """Inserts a record with ID, full datetime string, gamename, and title.""" try: - clean_date = datetime.strptime(record_date_str, "%Y-%m-%dT%H:%M:%SZ").date() - except ValueError: - clean_date = record_date_str - # Connects to database file (creates it if missing) + # Parses into a Python datetime object, then drops timezone info to create a clean string + dt_obj = datetime.strptime(record_date_str, "%Y-%m-%dT%H:%M:%SZ") + clean_datetime = dt_obj.strftime("%Y-%m-%d %H:%M:%S") + except (ValueError, TypeError): + clean_datetime = record_date_str - # Inserts data using parameterized queries to prevent SQL injection self.CURSOR.execute( f"INSERT OR IGNORE INTO {self.table} ({self.columns}) VALUES (?, ?, ?, ?, ?, ?, ?)", - (record_id, str(clean_date), title, gamename, False, False, False), + (record_id, str(clean_datetime), title, gamename, False, False, False), ) - - # Saves changes and closes the connection self.CONN.commit() def insert_clips_record(self, 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.""" + """Cleans up ISO-8601 strings into full datetime 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 + # Parses into a Python datetime object, then drops timezone info to create a clean string + dt_obj = datetime.strptime(record_date_str, "%Y-%m-%dT%H:%M:%SZ") + clean_datetime = dt_obj.strftime("%Y-%m-%d %H:%M:%S") + except (ValueError, TypeError): + clean_datetime = record_date_str self.CURSOR.execute( f"INSERT OR IGNORE INTO {self.table} ({self.columns}) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - (slug, str(clean_date), title, gamename, clip_by, views, False, False, False), + (slug, str(clean_datetime), title, gamename, clip_by, views, False, False, False), ) self.CONN.commit() \ No newline at end of file diff --git a/linux.py b/linux.py new file mode 100644 index 0000000..108778a --- /dev/null +++ b/linux.py @@ -0,0 +1,17 @@ +#!/usr/bin/env python3 +import subprocess + +def run_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} \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 8491f41..d04ffe0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,3 @@ -apt-listchanges==4.8 attrs==26.1.0 beautifulsoup4==4.14.3 certifi==2026.2.25 diff --git a/transcribe_video.py b/transcribe_video.py index 0b61db2..7223195 100644 --- a/transcribe_video.py +++ b/transcribe_video.py @@ -2,6 +2,7 @@ import os import subprocess import whisper +import linux from moviepy import VideoFileClip from whisper.utils import get_writer @@ -15,15 +16,8 @@ def extract_audio(video_path: str, audio_temp_path: str): # -map_chapters -1 removes chapter layouts that break the parser. # -sn strips text/subtitle streams that crash MoviePy. # -c copy copies video and audio instantly without quality loss. - cleanup_cmd = [ - "ffmpeg", "-y", "-i", video_path, - "-map_chapters", "-1", "-sn", - "-c", "copy", sanitized_video_path - ] + linux.run_command(f"ffmpeg -y -i {video_path} -map_chapters -1 -sn -c copy {sanitized_video_path}") - # Run the sanitization process silently - subprocess.run(cleanup_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - print("Extracting uncompressed WAV audio...") try: # Load the sanitized file instead of the raw Twitch clip diff --git a/twitch_down_uploader.py b/twitch_down_uploader.py index a0e5b5a..3d22596 100644 --- a/twitch_down_uploader.py +++ b/twitch_down_uploader.py @@ -1,7 +1,8 @@ #!/usr/bin/env python3 import requests import os -import subprocess +import linux +import time from database import Database @@ -14,21 +15,6 @@ 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" @@ -61,16 +47,17 @@ def top_hashtags(id: str): return tags def download(): - """Find all undownload vods and download them.""" + """Find all undownload videos and download them.""" undownloaded = DB.get_undownloaded() print(f"Download {DB.table}...") - if DB.table == "vods": + if DB.table == "videos": 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") + output = linux.run_command(f"TwitchDownloaderCLI videodownload --id {record_id} -o save/videos/{record_id}/{record_id}.mp4 --collision Overwrite") + output2 = linux.run_command(f"TwitchDownloaderCLI chatdownload --id {record_id} -o save/videos/{record_id}/{record_id}_chat.json -E --collision Overwrite") + time.sleep(1) 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) @@ -82,7 +69,8 @@ def download(): 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") + output = linux.run_command(f"TwitchDownloaderCLI clipdownload --id {slug} -o save/clips/{slug}/{slug}.mp4 --collision Overwrite") + time.sleep(1) if output["success"] is True: print(f"Slug: {slug} | Was successfully downloaded.") @@ -109,11 +97,11 @@ def upload(): upload_queue = [] - if DB.table == "vods": + if DB.table == "videos": 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" + file_path = f"save/videos/{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']) @@ -161,10 +149,14 @@ def get_vod_ids_simplified(channel_name: str): "Content-Type": "text/plain" } - vods_query_string = """ - query GetChannelVideos($login: String!, $limit: Int!) { + videos_query_string = """ + query GetChannelVideos($login: String!, $limit: Int!, $after: Cursor) { user(login: $login) { - videos(first: $limit, types: [ARCHIVE]) { + videos(first: $limit, types: [ARCHIVE], after: $after) { + pageInfo { + hasNextPage + endCursor + } edges { node { id @@ -181,10 +173,15 @@ def get_vod_ids_simplified(channel_name: str): """ clips_query_string = """ - query GetChannelClips($login: String!, $limit: Int!) { + query GetChannelClips($login: String!, $limit: Int!, $after: Cursor) { user(login: $login) { - clips(first: $limit, criteria: { period: ALL_TIME }) { + clips(first: $limit, criteria: { period: ALL_TIME }, after: $after) { + pageInfo { + hasNextPage + endCursor + } edges { + cursor node { slug title @@ -202,93 +199,129 @@ def get_vod_ids_simplified(channel_name: str): } } """ + + video_ids = [] + has_next_page = True + cursor = None limit = 50 query_string = "" operation_name = "" - if DB.table == "vods": - query_string = vods_query_string - limit = 50 + if DB.table == "videos": + query_string = videos_query_string + limit = 100 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 [] + while has_next_page: + time.sleep(1) + # 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, + "after": cursor + } + }] + + try: + # Prepping ensures Python does not rewrite the Client-ID header case + req = requests.Request('POST', url, json=payload) + #req = requests.Request('POST', url, data=json.dumps(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.get('data', {}).get('user', {}) + if not user_data: + print(f"Channel '{channel_name}' not found.") + return [] + + edges = user_data.get(DB.table, {}).get('edges', []) + + # --- BREAK CONDITION 1: Stop if Twitch returns no more data items --- + if not edges or len(edges) == 0: + print("No more items returned by the server. Ending pagination loop.") + break + + last_edge_cursor = None + print(f"--- Processing {DB.table} for {channel_name} ---") + for edge in edges: + last_edge_cursor = edge.get("cursor") + node = edge.get('node', {}) + if not node: + continue + + game_info = node.get('game') + game_name = game_info.get('displayName') if game_info else "Unknown/No Category" + + # FIXED: Switched fields to use safe .get() metrics to completely prevent KeyErrors + node_id = node.get('id') + node_title = node.get('title', 'No Title') + + if DB.table == "videos": + published_at = node.get('publishedAt') + print(f"ID: {node_id} | Date: {published_at} | Game: {game_name} | Title: {node_title}") + + DB.insert_videos_record(node_id, published_at, node_title, game_name) + if node_id: + video_ids.append(node_id) + + elif DB.table == "clips": + slug = node.get('slug') + created_at = node.get('createdAt') + view_count = node.get('viewCount', 0) + + curator_info = node.get('curator') + clip_by = curator_info.get('login') if curator_info else "Unknown Creator" + + print(f"Slug: {slug} | Date: {created_at} | Game: {game_name} | By: {clip_by} | Views: {view_count} | Title: {node_title}") + + DB.insert_clips_record(slug, created_at, node_title, game_name, clip_by, int(view_count)) + if slug: + video_ids.append(slug) + + # --- CORRECTED PAGINATION ENGINE FOR BOTH TABLES --- + page_info = user_data.get(DB.table, {}).get('pageInfo', {}) + has_next_page = page_info.get("hasNextPage", False) + + next_cursor = page_info.get("endCursor") or last_edge_cursor + + if not next_cursor or next_cursor == cursor: + print("Cursor did not advance or is null. Safely terminating loop.") + break + + cursor = next_cursor + + except Exception as e: + print(f"An unexpected error occurred: {e}") + if 'response' in locals(): + print(f"Server Response Text: {response.text}") + return [] + + return video_ids if __name__ == "__main__": - tables = ["vods", "clips"] + tables = ["clips"] for table in tables: DB = Database(table) + get_vod_ids_simplified(CHANNEL_NAME) - download() - upload() - DB.close_database() \ No newline at end of file + #download() + #upload() + #DB.close_database() \ No newline at end of file diff --git a/twitch_download_clips.py b/twitch_download_clips.py index 008694c..aa1d25a 100644 --- a/twitch_download_clips.py +++ b/twitch_download_clips.py @@ -1,6 +1,5 @@ #!/usr/bin/env python3 import requests -import sqlite3 import subprocess from datetime import datetime diff --git a/twitch_download_vod.py b/twitch_download_vod.py index 2420865..6bd6e89 100644 --- a/twitch_download_vod.py +++ b/twitch_download_vod.py @@ -1,286 +1,102 @@ #!/usr/bin/env python3 +import json +import os import requests -import sqlite3 -import subprocess +# The target Twitch streamer username +TWITCH_USERNAME = "SumGuyV5" -from database import Database +def load_secrets(filepath="twitch_secrets.json"): + """Loads client credentials and potential manual token from JSON file.""" + if not os.path.exists(filepath): + raise FileNotFoundError(f"Missing credential file: '{filepath}'") + + with open(filepath, "r") as file: + secrets = json.load(file) + + if "client_id" not in secrets or "client_secret" not in secrets: + raise KeyError("JSON file must contain 'client_id' and 'client_secret'.") + + return secrets["client_id"], secrets["client_secret"], secrets.get("manual_token") -import uploader -from uploader import CategoryId +def get_app_access_token(client_id, client_secret): + """Generates an App Access Token using the correct Twitch ID server.""" + auth_url = "https://twitch.tv" # FIXED: Correct auth endpoint + payload = { + "client_id": client_id, + "client_secret": client_secret, + "grant_type": "client_credentials" + } + headers = {"Content-Type": "application/x-www-form-urlencoded"} + + response = requests.post(auth_url, data=payload, headers=headers) + response.raise_for_status() + return response.json()["access_token"] -CHANNEL_NAME = 'teampgp' # Replace with the streamer's username -DB = Database("vods") +def get_user_id(username, headers): + """Retrieves the unique numerical Twitch User ID from Helix.""" + url = f"https://twitch.tv{username}" # FIXED: Endpoint & parameter + response = requests.get(url, headers=headers) + response.raise_for_status() + data = response.json().get("data") + if data and len(data) > 0: + return data[0]["id"] # FIXED: Helix data array returns user dictionaries + else: + raise ValueError(f"Twitch user '{username}' not found.") -def run_linux_command(command: str): - """Executes a Linux command, waits for completion, and returns output.""" +def get_channel_vods(user_id, headers, limit=10): + """Fetches past broadcasts (VODs) using valid Helix syntax.""" + # FIXED: Restructured URL to use correct endpoint and standard query parameters + url = f"https://twitch.tv{user_id}&type=archive&first={limit}" + response = requests.get(url, headers=headers) + response.raise_for_status() + return response.json().get("data", []) + +def main(): 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(slug: str): - """Transcribes the Video File.""" - video_file = f"save/clips/{slug}/{slug}.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/clips/{slug}/transcribe_{slug}.srt"): - print(f"video already transcribed:") - return True - - import transcribe_video - transcribe_video.extract_audio(video_file, f"save/clips/{slug}/temp_{slug}_audio.wav") - transcribe_video.transcribe_to_srt(f"save/clips/{slug}/temp_{slug}_audio.wav", f"save/clips/{slug}/", f"transcribe_{slug}") - - return True - -def top_hashtags(slug: str): - file_srt = f"save/clips/{slug}/transcribe_{slug}.srt" - - # if srt transcribe file not exists - if not os.path.exists(file_srt): - print(f"Transcribe file not found {file_srt}") - return [] - - import youtube_hashtags - - return youtube_hashtags.get_top_hashtags(file_srt) - -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}/{title}.mp4") - output2 = run_linux_command(f"TwitchDownloaderCLI chatdownload --id {record_id} -o save/vods/{record_id}/{title}_chat.json -E") - 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}/{title}.mp4") - - 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 = " #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp" - file_path = "" - title = "" - description = "" - categoryId = CategoryId.GAMING - privatcyStatus = 'private' - base_tags = ['gaming', 'TeamPGP', 'twitch', 'Level1Techs', 'twitch'] - - 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}/{title}.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 in unuploaded: - print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}") - - file_path = f"save/clips/{slug}/{title}.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) + # 1. Load credentials from external JSON file + client_id, client_secret, manual_token = load_secrets("twitch_secrets.json") + + # 2. Assign or generate OAuth Access Token + if manual_token: + print("Using manual access token from JSON config file...") + access_token = manual_token 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 - } - } - } - } + print("No manual token found. Attempting to contact Twitch Auth Server...") + access_token = get_app_access_token(client_id, client_secret) + + # 3. Setup Headers required by Twitch Helix API + headers = { + "Client-ID": client_id, + "Authorization": f"Bearer {access_token}" } - } - """ + + # 4. Translate Username to User ID + user_id = get_user_id(TWITCH_USERNAME, headers) + print(f"Successfully retrieved ID for {TWITCH_USERNAME}: {user_id}\n") + + # 5. Fetch and Print VOD details + vods = get_channel_vods(user_id, headers, limit=5) + + if not vods: + print(f"No VODs found for {TWITCH_USERNAME}.") + return - 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 = "" - if DB.table == "vods": - query_string = vods_query_string - limit = 50 - elif DB.table == "clips": - query_string = clips_query_string - limit = 40 - # A simplified query string that hardcodes the ARCHIVE type filter into the request structure - payload = [{ - "operationName": "GetChannelVideos", - "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 [] + print(f"--- Latest VODs for {TWITCH_USERNAME} ---") + for vod in vods: + print(f"Title: {vod['title']}") + print(f"URL: {vod['url']}") + print(f"Published At: {vod['published_at']}") + print(f"Duration: {vod['duration']}") + print(f"Views: {vod['view_count']}") + print("-" * 40) - 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) - #insert_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 (FileNotFoundError, KeyError) as config_err: + print(f"Configuration Error: {config_err}") + except requests.exceptions.HTTPError as err: + print(f"HTTP Error detail: {err.response.text if err.response else err}") except Exception as e: - print(f"An unexpected error occurred: {e}") - if 'response' in locals(): - print(f"Server Response Text: {response.text}") - return [] + print(f"An error occurred: {e}") if __name__ == "__main__": - get_vod_ids_simplified(CHANNEL_NAME) - download_vods() - upload_vods() - DB.close_database() \ No newline at end of file + main() \ No newline at end of file diff --git a/twitch_vod.py b/twitch_vod.py new file mode 100644 index 0000000..89d9f27 --- /dev/null +++ b/twitch_vod.py @@ -0,0 +1,165 @@ +import asyncio +import json +import requests +from twitchAPI.twitch import Twitch +from twitchAPI.helper import first +# Import the explicit VideoType Enum to prevent the AttributeError +from twitchAPI.type import VideoType + +SECRETS = None +TWITCH = None +USER = None +GAME_CACHE = {} # Local cache dictionary to store game_id -> game_name mapping + +CHANNEL_NAME = "teampgp" + +async def get_twitch(): + global SECRETS + global TWITCH + global 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 get_game_name_by_id(game_id: str) -> str: + """Helper function to fetch game names and cache them locally.""" + if not game_id: + return "Unknown / No Category" + if game_id in GAME_CACHE: + return GAME_CACHE[game_id] + try: + game_generator = TWITCH.get_games(game_ids=[game_id]) + game = await first(game_generator) + if game: + GAME_CACHE[game_id] = game.name + return game.name + except Exception: + pass + return "Unknown Game" + +async def get_vod_game_name(vod_id: str) -> str: + game_id = 0 + game_name = "Unknown Game" + url = "https://gql.twitch.tv/gql" + + headers = { + "Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko", + "Content-Type": "application/json" + } + + payload = [{ + "operationName": "VideoMetadata", + "variables": { + "channelLogin": "", + "videoID": vod_id + }, + "extensions": { + "persistedQuery": { + "version": 1, + "sha256Hash": "45111672eea2e507f8ba44d101a61862f9c56b11dee09a15634cb75cb9b9084d" + } + } + }] + + response = requests.post(url, headers=headers, json=payload) + data = response.json() + + # Parsing the Game ID out of the response array + video_info = data[0]['data']['video'] + if video_info and video_info.get('game'): + game_id = video_info['game']['id'] + game_name = video_info['game']['displayName'] + print(f"Game: {game_name} (ID: {game_id})") + else: + print("No game information found for this VOD.") + + return game_id, game_name + +async def get_streamer_vods(): + await get_twitch() + + print(f"Starting VOD extraction for {USER.display_name}...") + vod_generator = TWITCH.get_videos(user_id=USER.id, first=100, video_type=VideoType.ALL) + + all_vods = [] + + async for v in vod_generator: + # FIXED: Resolving game category using the automatic stream markers + game_id, game_name = await get_vod_game_name(v.id) + + #print(f"{v}") + + vod_data = { + "id": v.id, + "title": v.title, + "created_at": str(v.published_at), + "view_count": v.view_count, + "duration": v.duration, + "url": v.url, + "thumbnail_url": v.thumbnail_url, + "game_id": game_id, + "game_name": game_name, + "stream_id": v.stream_id, + "creator_name": CHANNEL_NAME + } + all_vods.append(vod_data) + print(f"Collected VOD: {v.title} | Category: {game_name} ({v.duration})") + + print(f"\nFinished extracting VODs. Total gathered: {len(all_vods)}") + return all_vods + +async def get_streamer_clips(): + await get_twitch() + + print(f"Starting clip extraction for {USER.display_name}...") + + clip_generator = TWITCH.get_clips(broadcaster_id=USER.id, first=100) + + all_clips = [] + + async for c in clip_generator: + # Clips DO have game_id attributes natively supported + game_name = await get_game_name_by_id(c.game_id) + + clip_data = { + "id": c.id, + "title": c.title, + "created_at": str(c.created_at), + "view_count": c.view_count, + "duration": c.duration, + "url": c.url, + "thumbnail_url": c.thumbnail_url, + "game_id": c.game_id, + "game_name": game_name, + "stream_id": "0", + "creator_name": c.creator_name, + } + all_clips.append(clip_data) + print(f"Collected clip: {c.title} | Category: {game_name} ({c.view_count} views)") + + print(f"\nFinished extracting clips. Total gathered: {len(all_clips)}") + return all_clips + +async def main(): + print("--- Script Started ---") + + # 1. Pull clips (with categories) + #clips_list = await get_streamer_clips() + #print(f"\nSuccessfully received a list of {len(clips_list)} clips in main().") + #if clips_list: + # print(f"Top clip: '{clips_list[0]['title']}' (Game: {clips_list[0]['game_name']})") + + # 2. Pull VODs (without categories) + vods_list = await get_streamer_vods() + #print(f"\nSuccessfully received a list of {len(vods_list)} VODs in main().") + if vods_list: + print(f"Recent VOD: '{vods_list[0]['title']}'") + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/youtube_short.py b/youtube_short.py index c975ac5..3a7c3df 100644 --- a/youtube_short.py +++ b/youtube_short.py @@ -1,7 +1,7 @@ #!/usr/bin/env python3 import argparse import sys -import subprocess +import linux import tempfile from pathlib import Path from moviepy import VideoFileClip, ColorClip, CompositeVideoClip @@ -36,13 +36,8 @@ def fit_to_9_16_letterbox(input_path: str, top_text: str = "TOP TEXT", bottom_te background_layer = None try: - cleanup_cmd = [ - "ffmpeg", "-y", "-i", str(input_file), - "-map_chapters", "-1", "-sn", - "-c", "copy", temp_path - ] - subprocess.run(cleanup_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - + linux.run_command(f"ffmpeg -y -i {input_file} -map_chapters -1 -sn -c copy {temp_path}") + # Load video clip = VideoFileClip(temp_path)