twitch-recorder-direct.py

#!/usr/bin/env python3

# This code is based on tutorial by slicktechies modified as needed to use oauth token from Twitch.
# 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/
# 24/7/365 NAS adaptation by CuriousTorvald (https://gist.github.com/curioustorvald/f7d1eefe1310efb8d41bee2f48a8e681)
# Twitch Helix API integration by Krepe.Z (https://gist.github.com/krepe90/22a0a6159b024ccf8f67ee034f94c1cc)

# Copyright (c) 2017, 2019, 2020, 2022 Junian, CuriousTorvald and Krepe.Z
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

# Only works for Streamlink version >= 5.0

import datetime
import logging
import os
import re
import subprocess
import sys
import argparse
import time
from typing import List, TypedDict, Union

import requests


TWITCH_API_CLIENT_ID = ""
TWITCH_API_CLIENT_SECRET = ""
FILE_NAME_FORMAT = "{user_login} - {stream_started} - {escaped_title}.ts"
TIME_FORMAT = "%Y-%m-%d %Hh%Mm%Ss"
RECORDING_SAVE_ROOT_DIR = "./"
INTERNET_TIMEOUT = 5

logger = logging.getLogger()
logger.setLevel(logging.INFO)
fmt = logging.Formatter("{asctime} {levelname} {name} {message}", style="{")
stream_hdlr = logging.StreamHandler()
stream_hdlr.setFormatter(fmt)
logger.addHandler(hdlr=stream_hdlr)


def escape_filename(s: str) -> str:
    """Remove invalid characters for the filename"""
    return re.sub(r"[/\\?%*:|\"<>.\n{}]", "", s)

def truncate_long_name(s: str) -> str:
    return (s[:75] + '..') if len(s) > 77 else s


class StreamData(TypedDict):
    id: str
    user_id: str
    user_login: str
    game_id: str
    game_name: str
    type: str
    title: str
    viewer_count: int
    started_at: str
    language: str
    thumbnail_url: str
    tag_ids: List[str]
    is_mature: bool


class TwitchRecorder:
    def __init__(self, username: str, quality: str) -> None:
        logger.info("Twitch Recorder initializing...")

        self.client_id = TWITCH_API_CLIENT_ID
        self.oauth_token = ""
        self._oauth_token_expires = 0

        self.ffmpeg_path = "ffmpeg"
        self.refresh = 2.0
        self.root_path = RECORDING_SAVE_ROOT_DIR

        self.username = username
        self.quality = quality

        self.file_dir = os.path.join(self.root_path, self.username)

        if not self.check_streamlink():
            sys.exit(1)
        self.token_acquired = self.get_oauth_token()
        while not self.token_acquired:
            time.sleep(INTERNET_TIMEOUT)
            self.token_acquired = self.get_oauth_token()
        if not self.check_user_exist():
            sys.exit(1)

        time.sleep(self.refresh)

    def get_oauth_token(self) -> bool:
        """Get oauth token from twitch api server using client id"""
        logger.info("Requesting OAuth token from the Twitch API server...")
        try:
            data = {
                "client_id": TWITCH_API_CLIENT_ID,
                "client_secret": TWITCH_API_CLIENT_SECRET,
                "grant_type": "client_credentials",
                "scope": ""
            }
            resp = requests.post("https://id.twitch.tv/oauth2/token", data=data)
            if resp.status_code != 200:
                return False
            resp_json = resp.json()
            access_token: str = resp_json["access_token"]
            token_type: str = resp_json["token_type"]
            self.oauth_token = f"{token_type.title()} {access_token}"
            self._oauth_token_expires = time.time() + resp_json["expires_in"]
            logger.debug("oauth_token is %s, expires at %d", self.oauth_token, self._oauth_token_expires)
        except requests.RequestException as e:
            logger.error("Failed to get OAuth token: %s", e)
            return False
        else:
            return True

    def check_streamlink(self) -> bool:
        """check if streamlink >= 5.0 is installed"""
        try:
            ret = subprocess.check_output(["streamlink", "--version"], universal_newlines=True)
            re_ver = re.search(r"streamlink (\d+)\.(\d+)\.(\d+)", ret, flags=re.IGNORECASE)
            if not re_ver:
                return False
            s_ver = tuple(map(int, re_ver.groups()))
            return s_ver[0] >= 5
        except FileNotFoundError:
            logger.error("Streamlink not found! Please install the Streamlink then re-start the script.")
            return False

    def check_oauth_token(self) -> None:
        """Auto re-request oauth token before it expires"""
        if time.time() + 3600 > self._oauth_token_expires:
            self.get_oauth_token()

    def check_user_exist(self) -> bool:
        """Check if username is vaild (https://dev.twitch.tv/docs/api/reference#get-users)"""
        logger.info("Checking username...")
        try:
            header = {
                "Client-ID": self.client_id,
                "Authorization": self.oauth_token
            }
            resp = requests.get(f"https://api.twitch.tv/helix/users?login={self.username}", headers=header)
            if resp.status_code != 200:
                logger.error("HTTP ERROR: %s", resp.status_code)
                logger.debug(resp.text)
                return False
            if not resp.json().get("data"):
                logger.error("Response is empty!")
                return False
        except requests.RequestException as e:
            logger.error("Failed to get user information: %s", e)
            return False
        else:
            return True

    def check_streaming(self) -> Union[StreamData, None]:
        """Get stream info (https://dev.twitch.tv/docs/api/reference#get-streams)"""
        try:
            header = {
                "Client-ID": self.client_id,
                "Authorization": self.oauth_token
            }
            resp = requests.get(f"https://api.twitch.tv/helix/streams?user_login={self.username}", headers=header, timeout=15)
            if resp.status_code != 200:
                logger.error("HTTP ERROR: %s", resp.status_code)
                return
            data = resp.json().get("data", [])
            if not data:
                logger.error("Search result is empty!")
                return
            return data[0]
        except requests.RequestException as e:
            logger.error("Fail to get stream info: %s", e)
            return

    def loop(self):
        """main loop function"""
        logger.info("Loop start!")
        while True:
            stream_data = self.check_streaming()
            if stream_data is None:
                logger.info("%s is currently offline, checking again in %.1f seconds.", self.username, self.refresh)
                time.sleep(self.refresh)
            else:
                logger.info("%s online. Stream recording in session.", self.username)
                _data = {
                    "escaped_title": truncate_long_name(escape_filename(stream_data["title"])),
                    "stream_started": datetime.datetime.fromisoformat(stream_data["started_at"].replace("Z", "+00:00")).astimezone().strftime(TIME_FORMAT),
                    "record_started": datetime.datetime.now().strftime(TIME_FORMAT)
                }
                file_name = FILE_NAME_FORMAT.format(**stream_data, **_data)
                file_path = os.path.join(self.file_dir, file_name)

                uq_num = 0
                while os.path.exists(file_path):
                    logger.warning("File already exists, will add numbers: %s", file_path)
                    uq_num += 1
                    file_path_no_ext, file_ext = os.path.splitext(file_path)
                    if uq_num > 1 and file_path_no_ext.endswith(f" ({uq_num - 1})"):
                        file_path_no_ext = file_path_no_ext.removesuffix(f" ({uq_num - 1})")
                    file_path = f"{file_path_no_ext} ({uq_num}){file_ext}"

                # start streamlink process
                logger.info("The video will be saved on %s", file_path)
                # ret = subprocess.call(["streamlink", "--twitch-disable-hosting", "--twitch-disable-ads", "twitch.tv/" + self.username, self.quality, "-o", file_path])
                # AD skipping interferes with Twitch's "NotLikeThis" filler? Trying to disabling one --20230215 CuriousTorvald
                ret = subprocess.call(["streamlink", "--twitch-disable-hosting", "twitch.tv/" + self.username, self.quality, "-o", file_path])

                if ret != 0:
                    logger.warning("Unexpected error. Will try again.")
                else:
                    # end streamlink process
                    logger.info("Recording finished. Going back to checking...")
                    
                time.sleep(self.refresh)

    def run(self):
        """run"""
        if self.refresh < 2:
            print("Check interval should not be lower than 2 seconds; setting the interval to 2.")
            self.refresh = 2
        # create directory for recordedPath and processedPath if not exist
        if not os.path.isdir(self.file_dir):
            os.makedirs(self.file_dir)
        self.loop()


def main():
    parser = argparse.ArgumentParser(description="Simple Twitch recording script")
    parser.add_argument("-u", "--username", required=True)
    parser.add_argument("-q", "--quality", default="best")
    # parser.add_argument("--logging-telegram", action="store_true")
    args = parser.parse_args()
    print(args)

    #logger.setLevel(logging.DEBUG)

    recorder = TwitchRecorder(args.username, args.quality)
    recorder.run()


if __name__ == "__main__":
    main()

For someone who wants to convert multiple TS to multiple MP4, use the following code in the terminal.
It can be easily done with the bash script and FFmpeg.

When you run this, all the .ts files in the current directory will be converted into mp4 in the subdirectory mp4.
This command will also skip the conversion by the -n argument if there's already the same output file exists.

for i in *.ts; do ffmpeg -n -i "$i" -acodec copy -vcodec copy "mp4/${i%.*}.mp4"; done