python_scripts/twitch_download_clips.py

#!/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()