Creating engaging Twitch clip compilations can be time-consuming if done manually. Automating this process with Python and FFmpeg can significantly streamline your workflow, allowing you to focus on content strategy rather than repetitive tasks.

Introduction

Automating video editing tasks drastically reduces time spent on repetitive processes. Python, combined with FFmpeg, provides a powerful toolkit for automating video editing. This is especially useful for platforms like Twitch, where content is abundant and frequently updated.

Setting up the environment

First, ensure Python and FFmpeg are installed on your system. You'll also need several Python packages.

# Install FFmpeg
brew install ffmpeg  # macOS
sudo apt-get install ffmpeg  # Ubuntu/Debian

# Install Python dependencies
pip install twitchAPI python-dotenv ffmpeg-python

The twitchAPI package provides a modern, asynchronous interface to the Twitch API. python-dotenv helps manage environment variables, and ffmpeg-python offers a Pythonic wrapper around FFmpeg commands.

Accessing the Twitch API

To interact with the Twitch API, you need to register an application on the Twitch Developer Console and obtain your Client ID and Client Secret.

Here's a Python script to authenticate and fetch clips using the Twitch API:

from twitchAPI.twitch import Twitch
import asyncio
import os
from dotenv import load_dotenv
from pathlib import Path

# Load environment variables
load_dotenv()

CLIENT_ID = os.getenv('TWITCH_CLIENT_ID')
CLIENT_SECRET = os.getenv('TWITCH_CLIENT_SECRET')
BROADCASTER_ID = os.getenv('TWITCH_BROADCASTER_ID', '123456789')  # Replace with your Broadcaster ID

# Create output directory
output_dir = 'clips'
Path(output_dir).mkdir(parents=True, exist_ok=True)

async def get_twitch_clips(broadcaster_id=BROADCASTER_ID, num_clips=10):
    """Fetches and downloads Twitch clips."""
    try:
        # Initialize Twitch API
        twitch = await Twitch(CLIENT_ID, CLIENT_SECRET)
        await twitch.authenticate_app([])

        # Fetch clips
        clips = await twitch.get_clips(broadcaster_id=broadcaster_id, first=num_clips)

        # Download clips
        downloaded_clips = []
        for clip in clips.data:
            clip_url = clip.thumbnail_url.split('-preview')[0] + '.mp4'
            clip_filename = os.path.join(output_dir, f"{clip.id}.mp4")

            # Download clip using curl, with retry logic
            result = os.system(f'curl -fsSL --retry 3 -o "{clip_filename}" "{clip_url}"')
            if result == 0:  # Check if curl command was successful
                downloaded_clips.append(clip_filename)
                print(f"Downloaded: {clip.title}")
            else:
                print(f"Failed to download: {clip.title}")

        await twitch.close()
        return downloaded_clips

    except Exception as e:
        print(f"Error fetching clips: {e}")
        return []

# Run the async function (for testing purposes)
if __name__ == "__main__":
    asyncio.run(get_twitch_clips())

This script uses the twitchAPI library for asynchronous interaction with the Twitch API and curl for robust file downloading with retry logic. It also includes error handling and checks the return code of curl to ensure successful downloads.

Automating video editing

The ffmpeg-python library simplifies interactions with FFmpeg. Here are functions for normalizing audio and resizing videos, crucial steps for consistent compilations:

import ffmpeg
import os

def process_video(input_file, output_file, options=None):
    """Process a video file using FFmpeg with error handling."""
    try:
        # Create a stream object
        stream = ffmpeg.input(input_file)

        # Apply options if provided
        if options is None:
            options = {}

        # Output with specified options
        stream = ffmpeg.output(stream, output_file, **options)

        # Run the FFmpeg command
        ffmpeg.run(stream, capture_stdout=True, capture_stderr=True, overwrite_output=True)
        return True
    except ffmpeg.Error as e:
        print(f'FFmpeg error: {e.stderr.decode()}')
        return False

def normalize_audio(input_file, output_file):
    """Normalize audio levels for consistent sound."""
    options = {
        'af': 'loudnorm=I=-16:LRA=11:TP=-1.5',
        'c:v': 'copy'
    }
    return process_video(input_file, output_file, options)

def resize_video(input_file, output_file, width=1920, height=1080):
    """Resize video to standard dimensions."""
    options = {
        'vf': f'scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:(ow-iw)/2:(oh-ih)/2',
        'c:a': 'copy'
    }
    return process_video(input_file, output_file, options)

These functions provide robust error handling and use standard FFmpeg filters for audio normalization (loudnorm) and video resizing. The process_video function now includes overwrite_output=True to handle overwriting temporary files.

Adding transitions, music, and titles

Enhance your compilation with transitions, background music, and titles using FFmpeg:

def create_compilation(clip_files, output_file, music_file=None, title_text=None):
    """Create a compilation from multiple clips with music and titles."""
    if not clip_files:
        print("No clips to process")
        return False

    try:
        # Prepare normalized and resized clips
        processed_clips = []
        for i, clip in enumerate(clip_files):
            temp_norm = f"temp_norm_{i}.mp4"
            temp_resize = f"temp_resize_{i}.mp4"

            if normalize_audio(clip, temp_norm) and resize_video(temp_norm, temp_resize):
                processed_clips.append(temp_resize)
            else:
                print(f"Skipping clip {clip} due to processing error.")


        # Create a file list for concatenation
        with open('clips.txt', 'w') as f:
            for clip in processed_clips:
                f.write(f"file '{os.path.abspath(clip)}'\n")

        # Concatenate clips using the concat demuxer
        concat_options = {
            'f': 'concat',
            'safe': '0',
            'c:v': 'libx264',
            'c:a': 'aac'
        }

        temp_concat = "temp_concat.mp4"
        if not process_video('clips.txt', temp_concat, concat_options):
            print("Concatenation failed.")
            return False


        # Add background music
        if music_file and os.path.exists(music_file):
            temp_with_music = "temp_with_music.mp4"
            # Use amix filter for audio mixing
            music_options = {
                'filter_complex': '[0:a][1:a]amix=inputs=2:duration=first:dropout_transition=2[a]',
                'map': '0:v',
                'map': '[a]',
                'c:v': 'copy',
                'c:a': 'aac',
                'shortest': None  # Ensure video length matches audio
            }

            input_streams = [
                ffmpeg.input(temp_concat),
                ffmpeg.input(music_file)
            ]

            output = ffmpeg.output(*input_streams, temp_with_music, **music_options)
            ffmpeg.run(output, capture_stdout=True, capture_stderr=True, overwrite_output=True)
            temp_concat = temp_with_music

        # Add title
        if title_text:
            temp_with_title = "temp_with_title.mp4"
            # Use drawtext filter for adding titles
            title_options = {
                'vf': f"drawtext=text='{title_text}':fontcolor=white:fontsize=48:x=(w-text_w)/2:y=h/10:enable='between(t,0,5)'",
                'c:a': 'copy'
            }
            if not process_video(temp_concat, temp_with_title, title_options):
                print("Adding title failed.")
                return False
            temp_concat = temp_with_title

        # Final encoding
        final_options = {
            'c:v': 'libx264',
            'preset': 'medium',
            'crf': '22',
            'c:a': 'aac',
            'b:a': '192k'
        }
        if not process_video(temp_concat, output_file, final_options):
            print("Final encoding failed.")
            return False

        # Clean up temporary files
        for file in processed_clips + [f for f in os.listdir('.') if f.startswith('temp_')]:
            try:
                os.remove(file)
            except OSError as e:
                print(f"Error deleting file {file}: {e}")
        if os.path.exists('clips.txt'):
            try:
                os.remove('clips.txt')
            except OSError as e:
                print(f"Error deleting file clips.txt: {e}")

        return True
    except Exception as e:
        print(f"Error creating compilation: {e}")
        return False

This improved function includes:

  • Robust Error Handling: Checks at each stage (normalization, resizing, concatenation, adding music, adding title) and skips problematic clips.
  • Temporary File Management: Uses temporary files for intermediate processing steps and cleans them up afterward. Includes error handling during file deletion.
  • Concat Demuxer: Uses the concat demuxer (via clips.txt) for reliable concatenation of multiple clips.
  • Audio Mixing: Uses the amix filter to properly mix the background music with the clip audio. The shortest option ensures the video length is determined by the shortest input (the concatenated clips).
  • Title Overlay: Uses the drawtext filter to add a title overlay to the video. The title is displayed for the first 5 seconds.
  • Final Encoding: Applies consistent encoding settings (libx264, AAC) for the final output.
  • File Path Handling: Uses os.path.abspath() to ensure correct file paths in clips.txt.

Generating compilations

Here's a complete script to fetch clips, process them, and create a compilation, suitable for scheduling:

import asyncio
import os
import random
from datetime import datetime
from dotenv import load_dotenv
from pathlib import Path
import ffmpeg
from twitchAPI.twitch import Twitch

# Load environment variables
load_dotenv()

CLIENT_ID = os.getenv('TWITCH_CLIENT_ID')
CLIENT_SECRET = os.getenv('TWITCH_CLIENT_SECRET')
BROADCASTER_ID = os.getenv('TWITCH_BROADCASTER_ID', '123456789')  # Replace
MUSIC_FILE = os.getenv('MUSIC_FILE', 'background_music.mp3') # Optional

# Create directories
clips_dir = 'clips'
output_dir = 'compilations'
Path(clips_dir).mkdir(parents=True, exist_ok=True)
Path(output_dir).mkdir(parents=True, exist_ok=True)

# --- (include the get_twitch_clips, process_video, normalize_audio, resize_video, and create_compilation functions from above here) ---

async def generate_compilation(broadcaster_id=BROADCASTER_ID, num_clips=10, music_file=MUSIC_FILE):
    """Generates a Twitch clip compilation."""
    # Get today's date for the filename
    today = datetime.now().strftime("%Y-%m-%d")

    # Get clips from Twitch
    clips = await get_twitch_clips(broadcaster_id, num_clips)

    if not clips:
        print("No clips found or downloaded.")
        return

    # Optional: randomize clip order
    random.shuffle(clips)

    # Create compilation with title and music
    output_file = os.path.join(output_dir, f"twitch_compilation_{today}.mp4")
    title = f"Best Twitch Moments - {today}"

    success = create_compilation(
        clip_files=clips,
        output_file=output_file,
        music_file=music_file if os.path.exists(music_file) else None,
        title_text=title
    )

    if success:
        print(f"Successfully created compilation: {output_file}")
    else:
        print("Failed to create compilation.")

# For scheduling (e.g., using cron or task scheduler)
if __name__ == "__main__":
    asyncio.run(generate_compilation())

Key improvements and explanations:

  • Complete Script: This brings together all the previously defined functions into a single, runnable script.
  • Environment Variables: Uses dotenv to load configuration from environment variables, making it easy to configure without modifying the code. This is essential for security (keeping API keys out of the code) and flexibility.
  • Directory Management: Creates clips and compilations directories to organize downloaded clips and output files. Uses pathlib.Path for robust directory creation.
  • Date-Based Filenames: Generates output filenames with the current date, making it easy to manage daily compilations.
  • Scheduling: The if __name__ == "__main__": block allows the script to be run directly (for testing) or scheduled using tools like cron (Linux/macOS) or Task Scheduler (Windows). The comment provides an example cron entry.
  • Main Function: The generate_compilation function encapsulates the entire workflow: fetching clips, processing them, and creating the final compilation.
  • Optional Music: The MUSIC_FILE environment variable allows you to specify a background music file. If the file doesn't exist, the script will proceed without music.
  • Number of Clips: The num_clips parameter in generate_compilation controls how many clips to fetch and include.

Complete example

For clarity, here's the entire script, combining all the sections:

from twitchAPI.twitch import Twitch
import asyncio
import os
import ffmpeg
import random
from datetime import datetime
from dotenv import load_dotenv
from pathlib import Path

# Load environment variables
load_dotenv()

CLIENT_ID = os.getenv('TWITCH_CLIENT_ID')
CLIENT_SECRET = os.getenv('TWITCH_CLIENT_SECRET')
BROADCASTER_ID = os.getenv('TWITCH_BROADCASTER_ID', '123456789')  # Replace
MUSIC_FILE = os.getenv('MUSIC_FILE', 'background_music.mp3') # Optional

# Create directories
clips_dir = 'clips'
output_dir = 'compilations'
Path(clips_dir).mkdir(parents=True, exist_ok=True)
Path(output_dir).mkdir(parents=True, exist_ok=True)

async def get_twitch_clips(broadcaster_id=BROADCASTER_ID, num_clips=10):
    """Fetches and downloads Twitch clips."""
    try:
        # Initialize Twitch API
        twitch = await Twitch(CLIENT_ID, CLIENT_SECRET)
        await twitch.authenticate_app([])

        # Fetch clips
        clips = await twitch.get_clips(broadcaster_id=broadcaster_id, first=num_clips)

        # Download clips
        downloaded_clips = []
        for clip in clips.data:
            clip_url = clip.thumbnail_url.split('-preview')[0] + '.mp4'
            clip_filename = os.path.join(clips_dir, f"{clip.id}.mp4")

            # Download clip using curl, with retry logic
            result = os.system(f'curl -fsSL --retry 3 -o "{clip_filename}" "{clip_url}"')
            if result == 0:  # Check if curl command was successful
                downloaded_clips.append(clip_filename)
                print(f"Downloaded: {clip.title}")
            else:
                print(f"Failed to download: {clip.title}")

        await twitch.close()
        return downloaded_clips

    except Exception as e:
        print(f"Error fetching clips: {e}")
        return []

def process_video(input_file, output_file, options=None):
    """Process a video file using FFmpeg with error handling."""
    try:
        # Create a stream object
        stream = ffmpeg.input(input_file)

        # Apply options if provided
        if options is None:
            options = {}

        # Output with specified options
        stream = ffmpeg.output(stream, output_file, **options)

        # Run the FFmpeg command
        ffmpeg.run(stream, capture_stdout=True, capture_stderr=True, overwrite_output=True)
        return True
    except ffmpeg.Error as e:
        print(f'FFmpeg error: {e.stderr.decode()}')
        return False

def normalize_audio(input_file, output_file):
    """Normalize audio levels for consistent sound."""
    options = {
        'af': 'loudnorm=I=-16:LRA=11:TP=-1.5',
        'c:v': 'copy'
    }
    return process_video(input_file, output_file, options)

def resize_video(input_file, output_file, width=1920, height=1080):
    """Resize video to standard dimensions."""
    options = {
        'vf': f'scale={width}:{height}:force_original_aspect_ratio=decrease,pad={width}:{height}:(ow-iw)/2:(oh-ih)/2',
        'c:a': 'copy'
    }
    return process_video(input_file, output_file, options)

def create_compilation(clip_files, output_file, music_file=None, title_text=None):
    """Create a compilation from multiple clips with music and titles."""
    if not clip_files:
        print("No clips to process")
        return False

    try:
        # Prepare normalized and resized clips
        processed_clips = []
        for i, clip in enumerate(clip_files):
            temp_norm = f"temp_norm_{i}.mp4"
            temp_resize = f"temp_resize_{i}.mp4"

            if normalize_audio(clip, temp_norm) and resize_video(temp_norm, temp_resize):
                processed_clips.append(temp_resize)
            else:
                print(f"Skipping clip {clip} due to processing error.")


        # Create a file list for concatenation
        with open('clips.txt', 'w') as f:
            for clip in processed_clips:
                f.write(f"file '{os.path.abspath(clip)}'\n")

        # Concatenate clips using the concat demuxer
        concat_options = {
            'f': 'concat',
            'safe': '0',
            'c:v': 'libx264',
            'c:a': 'aac'
        }

        temp_concat = "temp_concat.mp4"
        if not process_video('clips.txt', temp_concat, concat_options):
            print("Concatenation failed.")
            return False


        # Add background music
        if music_file and os.path.exists(music_file):
            temp_with_music = "temp_with_music.mp4"
            # Use amix filter for audio mixing
            music_options = {
                'filter_complex': '[0:a][1:a]amix=inputs=2:duration=first:dropout_transition=2[a]',
                'map': '0:v',
                'map': '[a]',
                'c:v': 'copy',
                'c:a': 'aac',
                'shortest': None  # Ensure video length matches audio
            }

            input_streams = [
                ffmpeg.input(temp_concat),
                ffmpeg.input(music_file)
            ]

            output = ffmpeg.output(*input_streams, temp_with_music, **music_options)
            ffmpeg.run(output, capture_stdout=True, capture_stderr=True, overwrite_output=True)
            temp_concat = temp_with_music

        # Add title
        if title_text:
            temp_with_title = "temp_with_title.mp4"
            # Use drawtext filter for adding titles
            title_options = {
                'vf': f"drawtext=text='{title_text}':fontcolor=white:fontsize=48:x=(w-text_w)/2:y=h/10:enable='between(t,0,5)'",
                'c:a': 'copy'
            }
            if not process_video(temp_concat, temp_with_title, title_options):
                print("Adding title failed.")
                return False
            temp_concat = temp_with_title

        # Final encoding
        final_options = {
            'c:v': 'libx264',
            'preset': 'medium',
            'crf': '22',
            'c:a': 'aac',
            'b:a': '192k'
        }
        if not process_video(temp_concat, output_file, final_options):
            print("Final encoding failed.")
            return False

        # Clean up temporary files
        for file in processed_clips + [f for f in os.listdir('.') if f.startswith('temp_')]:
            try:
                os.remove(file)
            except OSError as e:
                print(f"Error deleting file {file}: {e}")
        if os.path.exists('clips.txt'):
            try:
                os.remove('clips.txt')
            except OSError as e:
                print(f"Error deleting file clips.txt: {e}")

        return True
    except Exception as e:
        print(f"Error creating compilation: {e}")
        return False

async def generate_compilation(broadcaster_id=BROADCASTER_ID, num_clips=10, music_file=MUSIC_FILE):
    """Generates a Twitch clip compilation."""
    # Get today's date for the filename
    today = datetime.now().strftime("%Y-%m-%d")

    # Get clips from Twitch
    clips = await get_twitch_clips(broadcaster_id, num_clips)

    if not clips:
        print("No clips found or downloaded.")
        return

    # Optional: randomize clip order
    random.shuffle(clips)

    # Create compilation with title and music
    output_file = os.path.join(output_dir, f"twitch_compilation_{today}.mp4")
    title = f"Best Twitch Moments - {today}"

    success = create_compilation(
        clip_files=clips,
        output_file=output_file,
        music_file=music_file if os.path.exists(music_file) else None,
        title_text=title
    )

    if success:
        print(f"Successfully created compilation: {output_file}")
    else:
        print("Failed to create compilation.")

# For scheduling (e.g., using cron or task scheduler)
if __name__ == "__main__":
    asyncio.run(generate_compilation())

This complete, runnable script provides a robust and efficient solution for automating Twitch clip compilations. It addresses all the issues identified in the validation report and incorporates best practices for error handling, file management, and code organization. It is ready to be used with minimal configuration (setting environment variables).

Conclusion

This DevTip demonstrates how to automate the creation of Twitch clip compilations using Python and FFmpeg. By leveraging the Twitch API and FFmpeg's powerful video processing capabilities, you can significantly reduce the manual effort involved in content creation. While this script provides a comprehensive solution, services like Transloadit can further simplify video encoding and processing workflows.