''' I use this version for my personal use in a vps, edit it if you want to use it. it should work fine if rclone is not used, else edit function in line 235 ''' import requests, os, time, json, sys, subprocess, getopt, pathlib, locale, re from colorama import Fore, Style from datetime import datetime, timedelta locale.setlocale(locale.LC_TIME, "es_ES") from pytz import timezone from dotenv import load_dotenv, find_dotenv class TwitchArchive: def __init__(self): # user configuration self.username = "KalathrasLolweapon" # Twitch streamer username self.quality = "best" # Qualities options: best/source high/720p medium/540p low/360p # global configuration self.root_path = r"archive" # Path where this script saves everything self.refresh = 60 # Time between checking, avoid less than 1.0 self.downloadVOD = 1 # 0 - disable VOD downloading after stream finished, 1 - enable VOD downloading after stream finished (this option downloads the latest public vod) self.downloadCHAT = 1 # 0 - disable chat downloading and rendering, 1 - enable chat downloading and rendering self.downloadClips = 1 # Downloads the top 10 clips made during the stream. self.downloadMuted = 1 # Downalods the muted parts separately if twitch mutes the vod before being downloaded self.downloadChatHTML = 1 # uses the 'chatupdate' mode of TwitchDownloader, parses the json file to a readeble html file eg: https://vod.kalathrasarchives.com/file/kala-help/chat_html/20221213_15h57m24s.html self.uploadCloud = 1 # 0 - disable upload to remote cloud, 1 - enable upload to remote cloud self.deleteFiles = 0 # 0 - disable the deleting of files from current seccion after being uploaded to the cloud, 1 - enable the deleting files of files from current seccion after being uploaded to the cloud (BE CAREFUL WITH THIS OPTION) self.hls_segmentsVOD = 10 # 1-10 for downloading vod, it's possible to use multiple threads to potentially increase the throughput def run(self): self.os = self.get_OS() if load_dotenv(find_dotenv()): load_dotenv(find_dotenv()) else: print(f'{Fore.RED}\033[1mCREATE .env file with variables{Style.RESET_ALL}') quit() self.correct_user() print('Twitch-Archive --- ONLY VOD/CHAT') print('Configuration:') print(f'Root path: {Fore.GREEN}' + str(pathlib.Path(self.root_path).resolve()) + f'{Style.RESET_ALL}') print(f'Refresh rate: {Fore.GREEN} {str(self.refresh)}{Style.RESET_ALL}') if self.downloadVOD == 1: print(f'VOD downloading {Fore.GREEN}Enabled{Style.RESET_ALL}') else: print(f'VOD downloading: {Fore.RED}Disabled{Style.RESET_ALL}') if self.downloadCHAT == 1: print(f'Chat downloading {Fore.GREEN}Enabled{Style.RESET_ALL}') else: print(f'Chat downloading: {Fore.RED}Disabled{Style.RESET_ALL}') if self.uploadCloud == 1: print(f'Upload to cloud storage: {Fore.GREEN}Enabled{Style.RESET_ALL}') else: print(f'Upload to cloud storage: {Fore.RED}Disabled{Style.RESET_ALL}') if self.deleteFiles == 1: print(f'{Fore.RED}'+'\033[1m'+f'CAREFUL FILES ARE CONFIGURATED TO BE DELETED{Style.RESET_ALL}') else: print(f'{Fore.GREEN}'+'\033[1m'+f'Files will NOT be deleted{Style.RESET_ALL}') if self.uploadCloud == 0 and self.deleteFiles == 1: print(f'{Fore.RED}'+'\033[1m'+f'FILES WILL BE DELETED AND NO UPLOADED {Style.RESET_ALL}{Fore.GREEN}\n"CTRL + C"{Style.RESET_ALL}{Fore.RED}'+'\033[1m'+f' TO STOP AND CHANGED CONFIGURATION{Style.RESET_ALL}') if not os.path.exists(os.path.join(str(pathlib.Path(__file__).parent.resolve())+'/bin/temp/', ".log")): with open(os.path.join(str(pathlib.Path(__file__).parent.resolve())+'/bin/temp/', ".log"), 'w'): pass print(f"Checking for {Fore.GREEN}{self.username}{Style.RESET_ALL} every {Fore.GREEN}{self.refresh}{Style.RESET_ALL} seconds to download VOD/CHAT") self.loopcheck() def get_OS(self): if sys.platform.startswith('win32'): return 'windows' elif sys.platform.startswith('linux'): return 'linux' else: print('OS no supported') return def get_oauth_token(self): try: return requests.post(f"https://id.twitch.tv/oauth2/token?client_id={os.getenv('CLIENT-ID')}&client_secret={os.getenv('CLIENT-SECRET')}&grant_type=client_credentials").json()['access_token'] except: print(f'{Fore.RED}\033[1mCheck your client-id and secret{Style.RESET_ALL}') quit() def correct_user(self): try: url = f'https://api.twitch.tv/helix/users?login={self.username}' response = requests.get(url,headers = {"Authorization" : "Bearer " + self.get_oauth_token(), "Client-ID": os.getenv('CLIENT-ID')}, timeout = 15).json() if response['data'] == []: print(f'{Fore.RED}\033[1mUse a correct username{Style.RESET_ALL}') quit() except requests.exceptions.RequestException as e: print(e) def check_user(self): query = 'query{user(login: "' + self.username + '") {stream{archiveVideo{id}title createdAt}}}' try: response = requests.post('https://gql.twitch.tv/gql',json={'query': query},headers={"Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko"}) return json.loads(response.text) except requests.exceptions.RequestException as e: print(e) quit() def get_vod(self): query = 'query {user(login: "' + self.username + '"){videos(first: 1){edges {node {id title recordedAt lengthSeconds tags muteInfo{ mutedSegmentConnection{ nodes{ duration offset }}} topClips(first: 10) { edges{ node{ id slug viewCount title createdAt curator { displayName } durationSeconds url thumbnailURL(width: 480, height: 272)}}}}}}}}' try: response = requests.post('https://gql.twitch.tv/gql', json={'query': query}, headers={"Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko"}) return json.loads(response.text) except requests.exceptions.RequestException as e: print(e) def loopcheck(self): while True: is_live = self.check_user()['data']['user']['stream'] if is_live is not None: is_live_ready = self.check_user()['data']['user']['stream']['archiveVideo'] if is_live_ready is not None: bin_path = str(pathlib.Path(__file__).parent.resolve())+"/bin" live_temp_path = os.path.join(bin_path+'/temp/', "live_temp.ts") with open(os.path.join(bin_path+'/temp/', ".log")) as logs: logs = logs.read() log_id = is_live["createdAt"] + " - " + self.username + " - " + is_live["title"] if log_id in logs: time.sleep(self.refresh) with open(os.path.join(bin_path+'/temp/', ".log"), "r+") as logs: log_id = is_live["createdAt"] + " - " + self.username + " - " + is_live["title"] for line in logs: if log_id in line: break else: logs.write(is_live["createdAt"] + " - " + self.username + " - " + is_live["title"] +"\n") subprocess.call(['streamlink', 'twitch.tv/'+ self.username, self.quality, '--twitch-api-header', 'Authorization=OAuth ' + os.getenv('OAUTH-PRIVATE-TOKEN'), '--hls-live-restart', '--retry-streams', str(self.refresh), '--twitch-disable-reruns', '-o', live_temp_path]) os.remove(live_temp_path) current_vod = self.get_vod()['data']['user']['videos']['edges'][0]['node'] vod_date = datetime.strptime(current_vod["recordedAt"],'%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=timezone('UTC')).astimezone(tz=None).replace(tzinfo=None) vod_raw_filename = datetime.strftime(vod_date,'%Y%m%d_%Hh%Mm%Ss') if is_live_ready['id'] == current_vod['id']: print('VOD AND CHAT AVAILABLE') self.vod_path = str(pathlib.Path(os.path.join("VODS",self.root_path, self.username, vod_raw_filename ,"vod")).absolute()) self.clips_path = str(pathlib.Path(os.path.join("VODS",self.root_path, self.username, vod_raw_filename , "clips")).absolute()) self.muted_path = str(pathlib.Path(os.path.join("VODS",self.root_path, self.username, vod_raw_filename , "muted")).absolute()) self.json_path = str(pathlib.Path(os.path.join("VODS",self.root_path, self.username, vod_raw_filename ,"chat", "json")).absolute()) self.chat_path = str(pathlib.Path(os.path.join("VODS",self.root_path, self.username, vod_raw_filename , "chat")).absolute()) self.html_path = str(pathlib.Path(os.path.join("VODS",self.root_path, self.username, vod_raw_filename ,"chat", "html")).absolute()) if(os.path.isdir(self.vod_path) is False): os.makedirs(self.vod_path) if(os.path.isdir(self.clips_path) is False): os.makedirs(self.clips_path) if(os.path.isdir(self.muted_path) is False): os.makedirs(self.muted_path) if(os.path.isdir(self.json_path) is False): os.makedirs(self.json_path) if(os.path.isdir(self.chat_path) is False): os.makedirs(self.chat_path) if(os.path.isdir(self.html_path) is False): os.makedirs(self.html_path) vod_raw_path = os.path.join(bin_path+'/temp', "vod_temp.ts") vod_proc_path = os.path.join(self.vod_path, vod_raw_filename + ".mp4") chat_json_path = os.path.join(self.json_path, vod_raw_filename + ".json") chat_video_path = os.path.join(self.chat_path, vod_raw_filename + ".mp4") with open(os.path.join(str(pathlib.Path(os.path.join("VODS",self.root_path, self.username, vod_raw_filename)).absolute()), 'metadata.json'), 'w', encoding='utf-8') as f: json.dump(current_vod, f, ensure_ascii=False, indent=4) if self.username == 'KalathrasLolweapon': file_date = datetime.strptime(vod_raw_filename, '%Y%m%d_%Hh%Mm%Ss').date() week_first = file_date - timedelta(days=file_date.weekday()) week_last = week_first + timedelta(days=6) vod_year = 'VOD - ' + str(file_date.year) vod_month = f'{file_date.month:02d} - ' + file_date.strftime("%B").upper() vod_week = file_date.strftime("%B").capitalize() + ' ' + str(week_first.day) + '-' + str(week_last.day) chat_year = 'Chat - ' + str(file_date.year) chat_month = f'{file_date.month:02d} - ' + file_date.strftime("%B") vod_path = str(pathlib.Path(os.path.join("VODS",vod_year,vod_month,vod_week)).absolute()) chat_path = str(pathlib.Path(os.path.join("Chat",chat_year,chat_month)).absolute()) if(os.path.isdir(vod_path) is False): os.makedirs(vod_path) if(os.path.isdir(chat_path) is False): os.makedirs(chat_path) clean_vod_title = re.sub(r'[/\\:*?"<>|]', '_', current_vod['title']) if len(clean_vod_title) > 202: dif = len(clean_vod_title) - 202 clean_vod_title[:-dif] chat_video_path = os.path.join(chat_path, vod_raw_filename + ".mp4") vod_proc_path = os.path.join(vod_path, vod_raw_filename + '_'+ clean_vod_title + ".mp4") if self.downloadVOD == 1: print('Downloading VOD: ' + current_vod["title"]) try: subprocess.call([bin_path+"/TwitchDownloaderCLI.exe", 'videoDownload', '-u', str(current_vod["id"]), '-q', self.quality, "-t", str(self.hls_segmentsVOD), "--ffmpeg-path", bin_path +"/ffmpeg.exe", '--temp-path', bin_path+"/temp", "-o", vod_proc_path]) except Exception as e: print('Error', 'A ERROR has ocurred and the VOD will not be downloaded.\n') if self.downloadMuted == 1: if current_vod['muteInfo']['mutedSegmentConnection'] is not None: mutedSegments = current_vod['muteInfo']['mutedSegmentConnection']['nodes'] print('Downloading ' + len(mutedSegments) + ' muted segmes') for muted in mutedSegments: beg = time.strftime('%Hh%Mm%Ss', time.gmtime(muted['offset'])) end = time.strftime('%Hh%Mm%Ss', time.gmtime(muted['offset'] + muted['duration'])) muted_filename_path = os.path.join(self.muted_path, beg + '-' + end + ".mp4") subprocess.call([bin_path+"/TwitchDownloaderCLI.exe", 'videoDownload', '-u', current_vod['id'], '-q', self.quality, '-b', str(muted['offset']), '-e', str(muted['offset'] + muted['duration']), "-t", str(self.hls_segmentsVOD), "--ffmpeg-path", bin_path+"/ffmpeg.exe", '--temp-path', bin_path+"/temp" '-o', muted_filename_path]) else: print('The VOD has no muted segments') if self.downloadClips == 1: topClips = current_vod['topClips']['edges'] if topClips != []: print('Downloading Clips') for clips in topClips: clip = clips['node'] clean_title = re.sub(r'[/\\:*?"<>|]', '_', clip['title']) clip_filename_path = os.path.join(self.clips_path, clip['viewCount'] + '_' + clean_title + '_' + clip['id'] +".mp4") subprocess.call([bin_path+"/TwitchDownloaderCLI.exe", 'clipDownload', '-u', clip['slug'], '-q', self.quality, '-o', clip_filename_path]) else: print('No Clips has being made during the stream') if self.downloadCHAT == 1: print('Downloading CHAT: ' + current_vod["title"]) chat_settings = ["--background-color", "#FF111111", "-w", "500", "-h", "1080", "--outline", "true", "-f", "Arial", "--font-size", "22", "--update-rate", "1.0", "--offline", "--ffmpeg-path", f"{bin_path}/ffmpeg.exe", "--temp-path", f"{bin_path}/temp"] try: if self.os == 'windows': subprocess.call([bin_path + '/TwitchDownloaderCLI.exe', 'chatdownload', '--id', current_vod["id"], '-o', chat_json_path, '-E']) print('Rendering CHAT: ' + current_vod["title"]) subprocess.call([bin_path + '/TwitchDownloaderCLI.exe', 'chatrender', '-i', chat_json_path, '-o', chat_video_path] + chat_settings) elif self.os == 'linux': subprocess.call([bin_path + '/TwitchDownloaderCLI', 'chatdownload', '--id', current_vod["id"], '-o', chat_json_path, '-E']) subprocess.call([bin_path + '/TwitchDownloaderCLI', 'chatrender', '-i', chat_json_path, '-o', chat_video_path] + chat_settings) except Exception as e: print("A ERROR has ocurred and chat will need to be downloaded and rendered manually\n") if self.downloadChatHTML == 1: print('Downloading chat to html format') chat_html_path = os.path.join(self.html_path, vod_raw_filename + ".html") try: if self.os == 'windows': subprocess.call([bin_path+"/TwitchDownloaderCLI.exe", "chatupdate", "-i", chat_json_path, "-o", chat_html_path, "-E", "--temp-path", f"{bin_path}/temp"]) elif self.os == 'linux': subprocess.call([bin_path+"/TwitchDownloaderCLI", "chatupdate", "-i", chat_json_path, "-o", chat_html_path, "-E", "--temp-path", f"{bin_path}/temp"]) if self.username == 'KalathrasLolweapon': print('Uploading html chat to b2 bucket') subprocess.call(['rclone', 'copy', chat_html_path, 'b2:kala-help/chat_html', '--progress']) except Exception as e: print('A ERROR has ocurred and chat will need to be updated to html manually') if self.uploadCloud == 1: print('Uploading files:') if self.os == 'windows': if self.username == 'KalathrasLolweapon': subprocess.call(['rclone', 'copy', str(pathlib.Path(__file__).parent.resolve())+'/VODS', 'GD:VODS', '--progress']) subprocess.call(['rclone', 'copy', str(pathlib.Path(__file__).parent.resolve())+'/Chat', 'GD:Chat', '--progress']) else:subprocess.call(['rclone', 'copy', str(pathlib.Path(__file__).parent.resolve())+'/VODS', 'GD:VODS', '--progress']) elif self.os == 'linux':subprocess.call([bin_path+'/upload.sh', str(pathlib.Path(self.root_path).resolve()),self.username]) if self.deleteFiles == 1: print(f'{Fore.RED}DELETING FILES{Style.RESET_ALL}') if self.downloadVOD == 1: if(os.path.exists(vod_raw_path) is True): print(f'{Fore.RED}Deleting ' + vod_raw_path + f'{Style.RESET_ALL}') os.remove(vod_raw_path) if(os.path.exists(vod_proc_path) is True): print(f'{Fore.RED}Deleting ' + vod_proc_path + f'{Style.RESET_ALL}') os.remove(vod_proc_path) if self.downloadCHAT == 1: if(os.path.exists(chat_json_path) is True): print(f'{Fore.RED}Deleting ' + chat_json_path + f'{Style.RESET_ALL}') os.remove(chat_json_path) if(os.path.exists(chat_video_path) is True): print(f'{Fore.RED}Deleting ' + chat_video_path + f'{Style.RESET_ALL}') os.remove(chat_video_path) else: print('THE VOD/CHAT FOR CURRENT LIVESTREAM IS NOT AVAILABLE\nThe current livestream date: ' + is_live["createdAt"] + '\nThe VOD date: ' + current_vod["recordedAt"]) print('CURRENT SECCION HAVE FINISHED GOING BACK TO CHECKING') time.sleep(self.refresh) else: time.sleep(self.refresh) else: time.sleep(self.refresh) def main(argv): twitch_archive = TwitchArchive() help_msg = 'Twitch-Archive\nPython script to download the VOD and/or chat and render it, upload them to any cloud storage.\n -h, --help Display this information\n -u, --username Twitch channel username\n -q, --quality best/source high/720p medium/480p worst/360p\n -v, --vod <1/0> Download vod\n -c, --chat <1/0> Download chat and render it\n -r, --upload <1/0> Upload to cloud storage\n -d, --delete <1/0> Delete all files after upload (CAREFUL with this arg)\n' try: opts, args = getopt.getopt(argv,"h:u:q:v:c:r:d",["username=","quality=","vod=","chat=","metadata=","upload=","delete=","notifications="]) except getopt.GetoptError: print (help_msg) sys.exit(2) for opt, arg in opts: if opt in ('-h', '--help'): print(help_msg) sys.exit() elif opt in ("-u", "--username"): twitch_archive.username = arg elif opt in ("-q", "--quality"): twitch_archive.quality = arg elif opt in ("-v", "--vod"): twitch_archive.quality = int(arg) elif opt in ("-c", "--chat"): twitch_archive.quality = int(arg) elif opt in ("-r", "--upload"): twitch_archive.quality = int(arg) elif opt in ("-d", "--delete"): twitch_archive.quality = int(arg) twitch_archive.run() if __name__ == "__main__": main(sys.argv[1:])