diff --git a/coub-dl.py b/coub-dl.py index a563450..b8a894a 100644 --- a/coub-dl.py +++ b/coub-dl.py @@ -6,8 +6,8 @@ from enum import Enum from datetime import timedelta from math import floor from argparse import ArgumentParser -from functools import wraps, partial -from atexit import register +from functools import wraps +from tempfile import TemporaryDirectory @@ -70,11 +70,12 @@ def download_video_stream(url, file_dict): stdout=DEVNULL, stderr=DEVNULL) -def read_extensions(file_dict): - for file in listdir(): +def read_extensions(file_dict, directory): + for file in listdir(directory): for filename in file_dict: - if match('^{}.*'.format(file_dict[filename]), file): - file_dict[filename] = file + fullname = '{}/{}'.format(directory, file) + if match('^{}.*'.format(file_dict[filename]), fullname): + file_dict[filename] = fullname @call_verbose(before_message='Looping shorter stream... ') @@ -84,7 +85,7 @@ def loop_shorter_stream(file_dict, shorter, shorter_file, loop_fraction): stdout=DEVNULL, stderr=DEVNULL) # concat them - call(('ffmpeg', '-f', 'concat', '-i', file_dict[File.LIST], + call(('ffmpeg', '-f', 'concat', '-safe', '0', '-i', file_dict[File.LIST], '-c', 'copy', file_dict[File.LOOP]), stdout=DEVNULL, stderr=DEVNULL) @@ -98,15 +99,6 @@ def mux_streams(file_dict): stdout=DEVNULL, stderr=DEVNULL) -def cleanup(file_dict, outputs): - for key in file_dict: - if key not in outputs: - try: - remove(file_dict[key]) - except FileNotFoundError: - pass - - def get_length(file): data = get_duration(get_command_stderr(('ffprobe', file))).split(':') return timedelta(hours=float(data[0]), minutes=float(data[1]), seconds=float(data[2])) @@ -135,71 +127,76 @@ def yes_no_question(question, default): print("Please respond with 'yes'(y) or 'no'(n)!") +def run(url, files_dict, directory): + # download streams and update FILE dict with extensions + download_audio_stream(url, files_dict) + download_video_stream(url, files_dict) + read_extensions(files_dict, directory) -# parse arguments -parser = ArgumentParser(description='Download player-looped videos with youtube-dl & ffmpeg.') -parser.add_argument('-nv', '--nonverbose', action='store_true', help='Turn off non-critical messages to user.') -parser.add_argument('-o', '--output', default=None, help='Specify name of the output file (use -e for extension).') -parser.add_argument('-e', '--extension', default='mp4', help='Set the container to use for the output.') -parser.add_argument('url', type=str, help='The URL of the site containing the video to download.') + # get stream lengths via ffprobe + audioLen = get_length(files_dict[Stream.AUDIO]) + videoLen = get_length(files_dict[Stream.VIDEO]) -args = parser.parse_args() -args.extension = '.' + args.extension -VERBOSE = False if args.nonverbose else True + # decide which stream needs some looping + longer = audioLen if audioLen > videoLen else videoLen + shorter = audioLen if audioLen < videoLen else videoLen + shorterFile = files_dict[Stream.AUDIO] if audioLen < videoLen else files_dict[Stream.VIDEO] + files_dict[File.LOOP] += splitext(shorterFile)[1] + files_dict[File.FRACTION] += splitext(shorterFile)[1] + + times = longer.total_seconds() / shorter.total_seconds() + timesLoop_base = floor(times) + timesLoop_fraction = times % 1 + + # write concat helper file for ffmpeg + with open(files_dict[File.LIST], 'w') as f: + for i in range(timesLoop_base): + print("file '{}'".format(shorterFile), file=f) + print("file '{}'".format(files_dict[File.FRACTION]), file=f) + + loop_shorter_stream(files_dict, shorter, shorterFile, timesLoop_fraction) + mux_streams(files_dict) -FILES = {Stream.AUDIO: 'audio', Stream.VIDEO: 'video', - File.LIST: 'list.txt', File.LOOP: 'loop', File.FRACTION: 'fraction', - File.OUTPUT: 'output'+args.extension} -OUTPUT_KEYS = [File.OUTPUT] -URL = args.url +if __name__ == '__main__': + # parse arguments + parser = ArgumentParser(description='Download player-looped videos with youtube-dl & ffmpeg.') + parser.add_argument('-nv', '--nonverbose', action='store_true', help='Turn off non-critical messages to user.') + parser.add_argument('-o', '--output', default=None, help='Specify name of the output file (use -e for extension).') + parser.add_argument('-e', '--extension', default='mp4', help='Set the container to use for the output.') + parser.add_argument('url', type=str, help='The URL of the site containing the video to download.') -# clean up on exit -register(partial(cleanup, FILES, OUTPUT_KEYS)) + args = parser.parse_args() + args.extension = '.' + args.extension + VERBOSE = False if args.nonverbose else True + FILES = {Stream.AUDIO: 'audio', Stream.VIDEO: 'video', + File.LIST: 'list.txt', File.LOOP: 'loop', File.FRACTION: 'fraction', + File.OUTPUT: 'output'+args.extension} + OUTPUT_KEYS = [File.OUTPUT] + URL = args.url -# fetch video title if no filename was specified -if args.output is None: - FILES[File.OUTPUT] = check_output(('youtube-dl', '--get-title', args.url)).decode('utf-8').strip() -else: - FILES[File.OUTPUT] = args.output -FILES[File.OUTPUT] += args.extension - -# ask what to do if output exists -if exists(FILES[File.OUTPUT]): - answer = yes_no_question('A file named "{}" already exists! Overwrite?'.format(FILES[File.OUTPUT]), default='no') - if not answer: - print_opt('Exiting!') - exit() + # fetch video title if no filename was specified + if args.output is None: + FILES[File.OUTPUT] = check_output(('youtube-dl', '--get-title', args.url)).decode('utf-8').strip() else: - remove(FILES[File.OUTPUT]) + FILES[File.OUTPUT] = args.output + FILES[File.OUTPUT] += args.extension -# download streams and update FILE dict with extensions -download_audio_stream(URL, FILES) -download_video_stream(URL, FILES) -read_extensions(FILES) + # ask what to do if output exists + if exists(FILES[File.OUTPUT]): + answer = yes_no_question('A file named "{}" already exists! Overwrite?'.format(FILES[File.OUTPUT]), + default='no') + if not answer: + print_opt('Exiting!') + exit() + else: + remove(FILES[File.OUTPUT]) -# get stream lengths via ffprobe -audioLen = get_length(FILES[Stream.AUDIO]) -videoLen = get_length(FILES[Stream.VIDEO]) + with TemporaryDirectory() as dir: + for key in FILES: + if key not in OUTPUT_KEYS: + FILES[key] = '{}/{}'.format(dir, FILES[key]) -# decide which stream needs some looping -longer = audioLen if audioLen > videoLen else videoLen -shorter = audioLen if audioLen < videoLen else videoLen -shorterFile = FILES[Stream.AUDIO] if audioLen < videoLen else FILES[Stream.VIDEO] -FILES[File.LOOP] += splitext(shorterFile)[1] -FILES[File.FRACTION] += splitext(shorterFile)[1] - -times = longer.total_seconds() / shorter.total_seconds() -timesLoop_base = floor(times) -timesLoop_fraction = times % 1 - -# write concat helper file for ffmpeg -with open(FILES[File.LIST], 'w') as f: - for i in range(timesLoop_base): - print("file '{}'".format(shorterFile), file=f) - print("file '{}'".format(FILES[File.FRACTION]), file=f) - -loop_shorter_stream(FILES, shorter, shorterFile, timesLoop_fraction) -mux_streams(FILES) + run(URL, FILES, dir)