python_scripts/twitch_download_clips.py

#!/usr/bin/env python3
import requests
import subprocess
from datetime import datetime

from database import Database

import uploader
from uploader import CategoryId

CHANNEL_NAME = 'teampgp'  # Replace with the streamer's username
DB = Database("clips")

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 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_clips():
    """Loops over undownloaded metadata entries to write files down locally."""
    undownloaded_clips = DB.get_undownloaded()

    print("Downloading Clips...")

    for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt, uploaded_shorts_yt in undownloaded_clips:
        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")
        
        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("Finished Downloading Clips...")

def upload_clips():
    """Loops over the downloaded videos entries and uploaded them to youtube."""
    print(f"Uploading Clips...")

    unuploaded_clips = DB.get_unuploaded()

    categoryId = CategoryId.GAMING

    for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in unuploaded_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}, #Shorts #Clips #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp"
        categoryId = CategoryId.GAMING
        privatcyStatus = 'private'
        tags = ['shorts', 'gaming', 'TeamPGP', f'{game_name}', f'{clip_by}', 'twitch_clips', 'clips', 'Level1Techs', 'twitch']

        tags.extend(top_hashtags({slug}))

        output = uploader.upload_video(file_path, title, categoryId, description, privatcyStatus, tags)

        if output is True:
            print(f"Slug: {slug} | Was successfully uploaded.")
            DB.mark_as_uploaded(slug)
        else:
            print(f"Slug: {slug} | Upload Process failed.")
        
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']}")
            
            DB.insert_clips_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__":
    get_channel_clips(CHANNEL_NAME)
    download_clips()
    upload_clips()