How to Record Twitch Streams Automatically in Python

Using the power of streamlink and ffmpeg, twitch stream can be recorded every time user goes online.

good read
Using the power of streamlink and ffmpeg, twitch stream can be recorded every time user goes online.
Automatic twitch recording

Let's say you like to watch people playing game. Or maybe you like to study a game strategy from someone else. Or maybe you just find a player funny while playing a game. Of course you're going to search those videos on Internet. Some videos are streamable at any time, like YouTube videos, but some are just live stream with a certain schedule, like Twitch.

If it's streamable anytime, it's not a problem because we can watch it anytime we want. But what if it's a live stream and you're busy, or at work, or at other place that make you can't watch video right at the time. By the time you're home and plan to watch the video, it probably has ended.

Because you're reading this post right now, I assume you like to watch Twitch streams. Luckily, if you're busy and don't have time to follow your favorite Twitch streamers, there is a way to record those live streams anytime you want even 24/7.

This method requires you to understand command line though. Because we're going to use two awesome command line tools: streamlink and ffmpeg. Also, if possible, you need understand Python 3.x programming too, even though it's not important because you can just copy and paste the code here. But I'll also give some explanations for every code I write here in case you're trying understand how it works.

Update on livestreamer inactivity

Seems like livestreamer isn't active project anymore. But luckily there is an active project forked from it called streamlink. Just replace every livestreamer in the code with streamlink and it should work just fine. This tutorial code is updated with streamlink.

Installation

First thing first, let's install our command line tools. I'm using macOS Sierra to do the job, but it can also works for Windows and some Linux Distro. Assuming you already have Python 3.x installed on your system (if not then just go ahead to python website), we're going to install streamlink first using pip by typing this command:

$ pip3 install streamlink

streamlink will be used to record active live stream from Twitch. After the live streaming session ended, we're going to need a post processing tool for produced videos, because some errors could be happened when on recording session. For that we're going to use ffmpeg. We can install ffmpeg on macOS by using Homebrew by typing this command:

$ brew install ffmpeg

If you're on windows, you can download zeranoe builds here and extract it wherever you want, but I suggest to place it in C:\.

After all required tools installed, it's time for python scripting.

Please note that the code below was modified from slicktechies. The original code doesn't work anymore because livestreamer (or in this case streamlink) needs oauth token to work. That is why I make this tutorial and make the original code better. To see the original code, see references below.

Configuration

Write following code at the beginning of your script. This code will import every library we need to make the code works. Let's for now name our file twitch-recorder.py.

import requests
import os
import time
import json
import sys
import subprocess
import datetime
import getopt

Before we start the core programming, we're going to need some configuration variables. For that we're going to create a new class called TwitchRecorder. In this class constructor, some configuration variables are defined. Here I'll explain one by one what we're creating.

class TwitchRecorder:

    def __init__(self):
        # global configuration
        self.client_id = "jzkbprff40iqj646a697cyrvl0zt2m6" # don't change this
        # get oauth token value by typing `streamlink --twitch-oauth-authenticate` in terminal
        self.oauth_token = "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
        self.ffmpeg_path = 'ffmpeg'
        self.refresh = 30.0
        self.root_path = "/Users/junian/Documents/twitch"
        
        # user configuration
        self.username = "juniantr"
        self.quality = "best"

self.client_id is a value for Client-ID header needed to access Twitch API. So how can the value jzkbprff40iqj646a697cyrvl0zt2m6? Well, it turns out people on livestreamer issues found the default Client-ID for Twitch website itself. So we're going to use that to make it simple.

self.oauth_token is a value for streamlink needed to access Twitch video. You need to have a Twitch account to use this. This token is received by typing this command from command line:

$ streamlink --twitch-oauth-authenticate

After the command executed, you need to accept / allow Twitch for authentication. Then the browser will redirect you to something like this url:

https://streamlink.github.io/twitch_oauth.html#access_token=thisisyourtokenvalue&scope=user_read+user_subscriptions

Now see the url parameter with format access_token=thisisyourtokenvalue. That's your token value. Now you need to replace xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx with your token thisisyourtokenvalue to make it works.

You can leave self.ffmpeg_path as it is if you're on macOS or Linux. But in case you installed ffmpeg on different location, for example when you're on Windows, you can change self.ffmpeg_path to your ffmpeg binary path like this:

self.ffmpeg_path = 'C:/ffmpeg-3.2.2-win32-static/bin/ffmpeg.exe'

Next thing to understand is self.refresh. We use this variable for interval of checking user status, whether it's currently live streaming or not. It is recommended to set this value not less than 15 seconds because it is not good to access Twitch API if the interval is too low.

Use self.root_path to set saved videos. This can be anything as long as you have permission to access that directory. No need to create directory manually because the code will create it automatically.

The last two variables are used to set Twitch username you wish to record and the quality of recording. Both self.username and self.quality can be set via command line argument. So you don't have to change it in the code.

Check for user availability status

Next step is to make a function that can check user status. This step requires us to consume Twitch kraken API. The API endpoint we're going to use is this:

https://api.twitch.tv/kraken/streams/<username>

To access this API, we also going to use self.client_id as request header. This API will send us back with a json-formatted value. Here is an example of json result after consuming that API when the user is not live streaming right now:

{
  "stream": null,
  "_links": {
    "self": "https://api.twitch.tv/kraken/streams/juniantr",
    "channel": "https://api.twitch.tv/kraken/channels/juniantr"
  }
}

After receiving the json value, we're going to set the status into 4 categories. If the user is online and streaming right at the time, we set status = 0. If the user is offline, we set status = 1. If the user doesn't exist (in case of typo or something), we set status = 2. For unknown error, we set status = 3.

We're going to return user status and json data if exists. This function is still under class TwitchRecorder Here is the complete function to check Twitch user status:

class TwitchRecorder:

    def check_user(self):
        # 0: online, 
        # 1: offline, 
        # 2: not found, 
        # 3: error
        url = 'https://api.twitch.tv/kraken/streams/' + self.username
        info = None
        status = 3
        try:
            r = requests.get(url, headers = {"Client-ID" : self.client_id}, timeout = 15)
            r.raise_for_status()
            info = r.json()
            if info['stream'] == None:
                status = 1
            else:
                status = 0
        except requests.exceptions.RequestException as e:
            if e.response:
                if e.response.reason == 'Not Found' or e.response.reason == 'Unprocessable Entity':
                    status = 2

        return status, info

Routine to record live stream video when user is online

Still under class TwitchRecorder, we're going to write a loop to check user status.

If user is online, we call streamlink subprocess using subprocess.call() function. After recording is finished, we cleanup the video from any possible errors (usually some error frames caused by bad connection) using subprocess.call() of ffmpeg. After that, repeat to check user status again.

If the user is offline or not found, we just write a message on screen, then sleep and check user status again.

This routine would be executed in infinite of time, unless something wrong with your machine or you press ctrl + c from your keyboard.

Here is the complete function for this loop:

class TwitchRecorder:
    
    def loopcheck(self):
        while True:
            status, info = self.check_user()
            if status == 2:
                print("Username not found. Invalid username or typo.")
                time.sleep(self.refresh)
            elif status == 3:
                print(datetime.datetime.now().strftime("%Hh%Mm%Ss")," ","unexpected error. will try again in 5 minutes.")
                time.sleep(300)
            elif status == 1:
                print(self.username, "currently offline, checking again in", self.refresh, "seconds.")
                time.sleep(self.refresh)
            elif status == 0:
                print(self.username, "online. Stream recording in session.")
                filename = self.username + " - " + datetime.datetime.now().strftime("%Y-%m-%d %Hh%Mm%Ss") + " - " + (info['stream']).get("channel").get("status") + ".mp4"
                
                # clean filename from unecessary characters
                filename = "".join(x for x in filename if x.isalnum() or x in [" ", "-", "_", "."])
                
                recorded_filename = os.path.join(self.recorded_path, filename)
                
                # start streamlink process
                subprocess.call(["streamlink", "--twitch-oauth-token", self.oauth_token, "twitch.tv/" + self.username, self.quality, "-o", recorded_filename])

                print("Recording stream is done. Fixing video file.")
                if(os.path.exists(recorded_filename) is True):
                    try:
                        subprocess.call([self.ffmpeg_path, '-err_detect', 'ignore_err', '-i', recorded_filename, '-c', 'copy', os.path.join(self.processed_path, filename)])
                        os.remove(recorded_filename)
                    except Exception as e:
                        print(e)
                else:
                    print("Skip fixing. File not found.")
                    
                print("Fixing is done. Going back to checking..")
                time.sleep(self.refresh)

Combine them all together

Still in class TwitchRecorder, we now need to define a main function to run them all. But before we run loopcheck(), we're going to make sure some variables and directories are created.

We need to define and create 2 separates folder: self.recorded_path and self.processed_path. Recorded path is path where streamlink will save its videos. After recording is complete, ffmpeg will fix the video, put it to processed directory, and delete the old file.

Also, before starting loopcheck, we're going to need fix all previous recorded videos in case user terminate the process so those last videos won't left behind.

Here is the complete function for this part:

class TwitchRecorder:

    def run(self):
        # path to recorded stream
        self.recorded_path = os.path.join(self.root_path, "recorded", self.username)

        # path to finished video, errors removed
        self.processed_path = os.path.join(self.root_path, "processed", self.username)

        # create directory for recordedPath and processedPath if not exist
        if(os.path.isdir(self.recorded_path) is False):
            os.makedirs(self.recorded_path)
        if(os.path.isdir(self.processed_path) is False):
            os.makedirs(self.processed_path)

        # make sure the interval to check user availability is not less than 15 seconds
        if(self.refresh < 15):
            print("Check interval should not be lower than 15 seconds.")
            self.refresh = 15
            print("System set check interval to 15 seconds.")
        
        # fix videos from previous recording session
        try:
            video_list = [f for f in os.listdir(self.recorded_path) if os.path.isfile(os.path.join(self.recorded_path, f))]
            if(len(video_list) > 0):
                print('Fixing previously recorded files.')
            for f in video_list:
                recorded_filename = os.path.join(self.recorded_path, f)
                print('Fixing ' + recorded_filename + '.')
                try:
                    subprocess.call([self.ffmpeg_path, '-err_detect', 'ignore_err', '-i', recorded_filename, '-c', 'copy', os.path.join(self.processed_path,f)])
                    os.remove(recorded_filename)
                except Exception as e:
                    print(e)
        except Exception as e:
            print(e)

        print("Checking for", self.username, "every", self.refresh, "seconds. Record with", self.quality, "quality.")
        self.loopcheck()

Accepting command line arguments

Now that we already have all the functions needed, we just need to create an object from TwitchRecorder class and call run(). But before that, I believe it's better to add one more feature.

So imagine if you'd like to record 2 or more users at the same time. It's ineffective if you had to duplicate the script and change its username for every user. That is why we're going to add to set username via command line parameter. So we're going to make it so we can run it to something like this:

$ python3 twitch-recorder.py --username=<username> --quality=<quality>

To do something like that, we're going to user getopt. This is used to easily read command line arguments. Once we create a TwitchRecorder object (let's name it twitch_recorder), we can set its username and quality from command line argument. This way, you can record multiple users at the same time without duplicating this code.

Here is the complete code looks like:

def main(argv):
    twitch_recorder = TwitchRecorder()
    usage_message = 'twitch-recorder.py -u <username> -q <quality>'

    try:
        opts, args = getopt.getopt(argv,"hu:q:",["username=","quality="])
    except getopt.GetoptError:
        print (usage_message)
        sys.exit(2)
    for opt, arg in opts:
        if opt == '-h':
            print(usage_message)
            sys.exit()
        elif opt in ("-u", "--username"):
            twitch_recorder.username = arg
        elif opt in ("-q", "--quality"):
            twitch_recorder.quality = arg

    twitch_recorder.run()

if __name__ == "__main__":
    main(sys.argv[1:])

Running the script

After step above finished, now you can run this script by typing something like this in terminal:

$ python3 twitch-recorder --user=juniantr

To record another user, you can type the similar command inside new terminal session or window (or you can use something like tmux).

Summary

That's the simple way to record live stream from Twitch. Please note that this script probably won't work if your computer in sleep mode or your hard drive is full. Use this code at your own risk, I don't understand the legality to record a live streaming in your country.

To download the complete code, you can get it from this gist.

If you find anything wrong, feel free to write in comments below.

References

This article is also published on CodeProject. If you find this article useful, feel free to leave a 5 star rating and review.


Except as otherwise noted, the article of this page is licensed under a Creative Commons Attribution 4.0 International License, and code samples are licensed under the MIT License. For details, see our Site Policies.

Subscribe to Junian.Net

Get the latest posts delivered right to your inbox.

Delivered by FeedBurner or subscribe via RSS with Feedly!