chzzk-recorder-test.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)
# Chzzk adaptation by CuriousTorvald (https://gist.github.com/curioustorvald/240e67a53f8131d7166ad5269044c719)

# Copyright (c) 2017, 2019, 2020, 2022, 2024 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 >= 6.5

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

import requests

FILE_NAME_FORMAT = "{record_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):
    liveTitle: str
    status: str
    concurrentUserCount: int
    accumulateCount: int
    paidPromotion: bool
    adult: bool
    chatChannelId: str
    categoryType: str
    liveCategory: str
    livePollingStatusJson: str
    faultStatus: int
    chatActive: bool
    chatAvailableGroup: str
    chatAvailableCondition: str
    minFollowerMinute: int



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

        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)

        time.sleep(self.refresh)


    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_streaming(self) -> Union[StreamData, None]:
        """Get stream info (https://dev.twitch.tv/docs/api/reference#get-streams)"""
        try:
            header = {
            }
            resp = requests.get(f"https://api.chzzk.naver.com/polling/v2/channels/{self.username}/live-status", headers=header, timeout=15)
            if resp.status_code != 200:
                logger.error("HTTP ERROR: %s", resp.status_code)
                return
            data = resp.json().get("content", [])
            if not data:
                logger.error("Search result is empty!")
                return

            return data
        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 or stream_data["status"] != "OPEN":
                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["liveTitle"])),
                    "record_started": datetime.datetime.now().strftime(TIME_FORMAT)
                }
                file_name = FILE_NAME_FORMAT.format(**_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", "https://chzzk.naver.com/live/" + 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 Chzzk 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()