Record Twitch Streams Automatically in Python

.gitignore

defaults.json

README.md

Please run init.sh first from this directory and follow its directions to install dependencies. Then run the script by typing:

python3 twitch-recorder.py -u username

init.sh

#/bin/bash
set -eo pipefail
IFS=$'\n\t'

echo -n "Checking whether the script is running in twitch-recorder dir ... "
if [ -f twitch-recorder.py ]; then
    echo "it is!"
else
    echo -e "it is not.\n\nPlease change to the directory with twitch-recorder.py and run this again\n"
    exit -2
fi


echo -n "Checking for ffmpeg ... "
if dpkg -s ffmpeg | grep Status | grep installed; then
    echo "Found ffmpeg"
else
    echo -e "Did not find ffmpeg. Please install it using the command:\n\nsudo apt install ffmpeg"
    exit -1
fi

env_dir=~/venv/StreamTwitch

if [ -n "$(type -t deactivate)" ] && [ "$(type -t deactivate)" = function ]; then
    deactivate
fi

echo "Deleting and re-initalizing virtualenv"
rm -rf "$env_dir"
mkdir -p "$env_dir"

echo "Initializng virtualenv"
python3 -m venv "$env_dir"
. "$env_dir/bin/activate"
echo "Upgrading pip and installing deps"
pip3 install --upgrade pip

# In stall things in the virtualenv
pip3 install streamlink

# Authorize if necessary
echo -n "Checking for ~/.streamlinkrc ... "
if grep "twitch-oauth-token=" < ~/.streamlinkrc; then
    echo "contains the oauth token"
else
    touch ~/.streamlinkrc
    echo "No authorization to stream twitch as your user"
    echo -e "Please run the commands:\n\n"
    echo -e "source $env_dir/bin/activate\nstreamlink --twitch-oauth-authenticate\n\n"
    echo -e "Then paste the line it gives you into the file ~/.streamlinkrc"
    exit -1
fi

# Get the path where streams will be saved
echo "Please type the path where recorded streams should be saved"
read stream_path

cat << EOF > defaults.json
{
    "stream_path":"$stream_path",
    "quality":"480p"
}
EOF


# Print final instructions
echo -e "\nNow run\n\nsource $env_dir/bin/activate\npython3 twitch-recorder.py"

show_download_process.sh

ps -eo pid,args | grep streamlink

twitch-recorder.py

#!/usr/bin/env python3

# This code is based on a tutorial by slicktechies
# You can read more details at:
# https://www.junian.net/2017/01/how-to-record-twitch-streams.html
# original code is from
# https://slicktechies.com/how-to-watchrecord-twitch-streams-using-livestreamer/
#
# You must put the oauth token for your user in the streamlink config file
# You can get it by typing:
#
#     streamlink --twitch-oauth-authenticate
#
# in the terminal

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


class FixFilesInBackground:
    def __init__(self, recorded_filename, processed_filename):
        self.ffmpeg_path = 'ffmpeg'
        self.recorded_filename = recorded_filename
        self.processed_filename = processed_filename

        self.thread = threading.Thread(target=self.run, args=())
        self.thread.daemon = False
        self.thread.start()

    def run(self):
        if(os.path.exists(self.recorded_filename) is True):
            print('Fixing "{}".'.format(self.recorded_filename))
            try:
                subprocess.call([
                    self.ffmpeg_path, '-err_detect', 'ignore_err',
                    '-i', self.recorded_filename,
                    '-c', 'copy', self.processed_filename])
                os.remove(self.recorded_filename)
            except Exception as e:
                print(e)
        else:
            print('No need to fix "{}".'.format(self.recorded_filename))


class TwitchRecorder:
    def __init__(self, root_path, quality):
        # global configuration
        self.client_id = "jzkbprff40iqj646a697cyrvl0zt2m6"  # don't change this
        self.refresh = 30.0
        self.root_path = root_path
        self.stop_file = 'stop_recording'

        # user configuration
        self.channel_name = "TwitchPresents"
        self.quality = quality

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

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

        # 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:
            def recorded_fn(f): return os.path.join(self.recorded_path, f)
            video_list = [f for f in os.listdir(self.recorded_path)
                          if os.path.isfile(recorded_fn(f))]
            if len(video_list) > 0:
                print('Starting process to fix previously recorded files.')
                for f in video_list:
                    processed_filename = os.path.join(self.processed_path, f)
                    FixFilesInBackground(recorded_fn(f), processed_filename)
        except Exception as e:
            print(e)

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

    def check_user(self):
        # 0: online,
        # 1: offline,
        # 2: not found,
        # 3: error
        url = 'https://api.twitch.tv/kraken/streams/' + self.channel_name
        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'] is None:
                status = 1
            else:
                status = 0
        except requests.exceptions.RequestException as e:
            if e.response:
                if e.response.reason in {'Not Found', 'Unprocessable Entity'}:
                    status = 2

        return status, info

    def loopcheck(self):
        while True:
            print(datetime.datetime.now(), ':', end=" ")
            if os.path.isfile(self.stop_file):
                print('Stopping recording ... stop_file: "{}" exists'.format(
                    self.stop_file))
                return

            status, info = self.check_user()
            last_check = datetime.datetime.now()
            to_sleep = self.refresh
            if status == 2:
                print("Channel_name not found. Invalid channel_name or typo.")
            elif status == 3:
                print("{} unexpected error. will try again in "
                      "5 minutes.".format(
                          datetime.datetime.now().strftime("%Hh%Mm%Ss")))
                to_sleep = 300
            elif status == 1:
                print(self.channel_name, "currently offline, "
                      "checking again in", self.refresh, "seconds.")
            elif status == 0:
                print(self.channel_name, "online. "
                      "Stream recording in session.")
                filename = "{} - {} - {}.mp4".format(
                    self.channel_name,
                    datetime.datetime.now().strftime("%Y-%m-%d %Hh%Mm%Ss"),
                    (info['stream']).get("channel").get("status"))

                # 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.tv/" + self.channel_name,
                                 self.quality, "-o", recorded_filename])

                print("Recording stream is done. "
                      "Starting process to fix/copy video file if needed.")
                processed_filename = os.path.join(self.processed_path,
                                                  filename)
                FixFilesInBackground(recorded_filename, processed_filename)
                print("Process has started. Going back to checking..")

            # Sleep remaining time
            remaining_time = to_sleep - math.floor(
                (datetime.datetime.now() - last_check).total_seconds())
            if remaining_time > 0:
                time.sleep(remaining_time)


def main(argv):
    try:
        with open('defaults.json', encoding="utf8") as f:
            defaults = json.loads(f.read())
    except Exception as e:
        raise Exception('Could not read defaults.json - '
                        'you may not have run init.sh') from e

    twitch_recorder = TwitchRecorder(defaults['stream_path'],
                                     defaults['quality'])
    usage_message = 'twitch-recorder.py -c <channel_name> -q <quality>'

    try:
        opts, args = getopt.getopt(argv, "hc:q:",
                                   ["channel_name=", "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 ("-c", "--channel_name"):
            twitch_recorder.channel_name = arg
        elif opt in ("-q", "--quality"):
            twitch_recorder.quality = arg

    twitch_recorder.run()


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

Remember: go to https://www.junian.net/2017/01/how-to-record-twitch-streams.html for the original and explanations.