mirror/yt-dlp
1
0
Fork 0
mirror of https://github.com/yt-dlp/yt-dlp.git synced 2025-03-09 12:50:23 -05:00
Code Activity

Fixed up tests and linting

Browse source
This commit is contained in:
Kieran Eglin 2024-04-24 10:57:39 -07:00
parent 44bb6c2056
commit ea2a085397
No known key found for this signature in database
GPG key ID: 193984967FCF432D
3 changed files with 29 additions and 134 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
test/test_YoutubeDL.py
View file

@ -874,32 +874,33 @@ def test_format_note(self):
}), r'^30fps$') }), r'^30fps$')
def test_postprocessors(self): def test_postprocessors(self):
filename = 'post-processor-testfile.mp4' filename = 'post-processor-testfile'
audiofile = filename + '.mp3' video_file = filename + '.mp4'
audio_file = filename + '.mp3'
class SimplePP(PostProcessor): class SimplePP(PostProcessor):
def run(self, info): def run(self, info):
with open(audiofile, 'w') as f: with open(audio_file, 'w') as f:
f.write('EXAMPLE') f.write('EXAMPLE')
return [info['filepath']], info return [info['filepath']], info
def run_pp(params, PP): def run_pp(params, PP):
with open(filename, 'w') as f: with open(video_file, 'w') as f:
f.write('EXAMPLE') f.write('EXAMPLE')
ydl = YoutubeDL(params) ydl = YoutubeDL(params)
ydl.add_post_processor(PP()) ydl.add_post_processor(PP())
ydl.post_process(filename, {'filepath': filename}) ydl.post_process(video_file, {'filepath': video_file})
run_pp({'keepvideo': True}, SimplePP) run_pp({'keepvideo': True, 'outtmpl': filename}, SimplePP)
self.assertTrue(os.path.exists(filename), '%s doesn\'t exist' % filename) self.assertTrue(os.path.exists(video_file), '%s doesn\'t exist' % video_file)
self.assertTrue(os.path.exists(audiofile), '%s doesn\'t exist' % audiofile) self.assertTrue(os.path.exists(audio_file), '%s doesn\'t exist' % audio_file)
os.unlink(filename) os.unlink(video_file)
os.unlink(audiofile) os.unlink(audio_file)
run_pp({'keepvideo': False}, SimplePP) run_pp({'keepvideo': False, 'outtmpl': filename}, SimplePP)
self.assertFalse(os.path.exists(filename), '%s exists' % filename) self.assertFalse(os.path.exists(video_file), '%s exists' % video_file)
self.assertTrue(os.path.exists(audiofile), '%s doesn\'t exist' % audiofile) self.assertTrue(os.path.exists(audio_file), '%s doesn\'t exist' % audio_file)
os.unlink(audiofile) os.unlink(audio_file)
class ModifierPP(PostProcessor): class ModifierPP(PostProcessor):
def run(self, info): def run(self, info):
@ -907,9 +908,9 @@ def run(self, info):
f.write('MODIFIED') f.write('MODIFIED')
return [], info return [], info
run_pp({'keepvideo': False}, ModifierPP) run_pp({'keepvideo': False, 'outtmpl': filename}, ModifierPP)
self.assertTrue(os.path.exists(filename), '%s doesn\'t exist' % filename) self.assertTrue(os.path.exists(video_file), '%s doesn\'t exist' % video_file)
os.unlink(filename) os.unlink(video_file)
def test_match_filter(self): def test_match_filter(self):
first = { first = {

3
yt_dlp/YoutubeDL.py
View file

@ -1982,7 +1982,7 @@ def __process_playlist(self, ie_result, download):
'playlist', ie_result, self.prepare_filename(ie_copy, 'pl_infojson')) 'playlist', ie_result, self.prepare_filename(ie_copy, 'pl_infojson'))
if _infojson_written is None: if _infojson_written is None:
return return
description_file = self._write_description('playlist', ie_result, self.prepare_filename(ie_copy, 'pl_description')) description_file = self._write_description('playlist', ie_result, self.prepare_filename(ie_copy, 'pl_description'))
if description_file is None: if description_file is None:
return return
@ -4265,7 +4265,6 @@ def _write_description(self, label, info_dict, filename):
self.to_screen(f'[info] Writing {label} description to: {filename}') self.to_screen(f'[info] Writing {label} description to: {filename}')
with open(filename, 'w', encoding='utf-8') as descfile: with open(filename, 'w', encoding='utf-8') as descfile:
descfile.write(info_dict['description']) descfile.write(info_dict['description'])
info_dict['description_filepath'] = filename
except OSError: except OSError:
self.report_error(f'Cannot write {label} description file {filename}') self.report_error(f'Cannot write {label} description file {filename}')
return None return None

125
yt_dlp/postprocessor/movefilesafterdownload.py
View file

@ -8,7 +8,7 @@
make_dir, make_dir,
replace_extension replace_extension
) )
import pdb
class MoveFilesAfterDownloadPP(PostProcessor): class MoveFilesAfterDownloadPP(PostProcessor):
TOP_LEVEL_KEYS = ['filepath'] TOP_LEVEL_KEYS = ['filepath']
@ -26,8 +26,9 @@ def __init__(self, downloader=None, downloaded=True):
@classmethod @classmethod
def pp_key(cls): def pp_key(cls):
return 'MoveFiles' return 'MoveFiles'
def move_file_and_write_to_info(self, info_dict, relevant_dict, output_file_type): def move_file_and_write_to_info(self, info_dict, relevant_dict=None, output_file_type=None):
relevant_dict = relevant_dict or info_dict
if 'filepath' not in relevant_dict: if 'filepath' not in relevant_dict:
return return
@ -39,14 +40,6 @@ def move_file_and_write_to_info(self, info_dict, relevant_dict, output_file_type
name_format = self._downloader.prepare_filename(info_dict, output_file_type) name_format = self._downloader.prepare_filename(info_dict, output_file_type)
final_filepath = replace_extension(name_format, extension) final_filepath = replace_extension(name_format, extension)
move_result = self.move_file(info_dict, current_filepath, final_filepath) move_result = self.move_file(info_dict, current_filepath, final_filepath)
print('*******************')
print("output_file_type", output_file_type)
print("name_format", name_format)
print("current_filepath", current_filepath)
print("final_filepath", final_filepath)
print("move_result", move_result)
print('*******************')
if move_result: if move_result:
relevant_dict['filepath'] = move_result relevant_dict['filepath'] = move_result
@ -56,9 +49,9 @@ def move_file_and_write_to_info(self, info_dict, relevant_dict, output_file_type
def move_file(self, info_dict, current_filepath, final_filepath): def move_file(self, info_dict, current_filepath, final_filepath):
if not current_filepath or not final_filepath: if not current_filepath or not final_filepath:
return return
dl_path, _dl_name = os.path.split(info_dict['filepath']) dl_parent_folder = os.path.split(info_dict['filepath'])[0]
finaldir = info_dict.get('__finaldir', os.path.abspath(dl_path)) finaldir = info_dict.get('__finaldir', os.path.abspath(dl_parent_folder))
if not os.path.isabs(current_filepath): if not os.path.isabs(current_filepath):
current_filepath = os.path.join(finaldir, current_filepath) current_filepath = os.path.join(finaldir, current_filepath)
@ -86,24 +79,12 @@ def move_file(self, info_dict, current_filepath, final_filepath):
make_dir(final_filepath, PostProcessingError) make_dir(final_filepath, PostProcessingError)
self.to_screen(f'Moving file "{current_filepath}" to "{final_filepath}"') self.to_screen(f'Moving file "{current_filepath}" to "{final_filepath}"')
shutil.move(current_filepath, final_filepath) # os.rename cannot move between volumes shutil.move(current_filepath, final_filepath) # os.rename cannot move between volumes
return final_filepath return final_filepath
def run(self, info): def run(self, info):
dl_path, dl_name = os.path.split(info['filepath']) # This represents the main media file (using the 'filepath' key)
finaldir = info.get('__finaldir', os.path.abspath(dl_path)) self.move_file_and_write_to_info(info)
# TODO: add one single key to infodict with ALL downloaded files
# TODO: test with --path temp and stuff
# TODO: make the below work with not-currently-written filepaths like description, annotations, etc
# - Descriptions work, have to do all the other ones too
# - I lied, this should become another post-processor
# TODO: [DONE] probably something with relative paths into absolute again?
# TODO: remove all __files_to_move stuff when done
# TODO: add net-new filepaths to `sanitize_info`
# TODO: consider adding a `infojson_filepath` key in addition to the `infojson_filename` key where the former is the fullpath
for filepath_key in self.TOP_LEVEL_KEYS:
self.move_file_and_write_to_info(info, info, None)
for key, output_file_type in self.CHILD_KEYS.items(): for key, output_file_type in self.CHILD_KEYS.items():
if key not in info: if key not in info:
@ -116,89 +97,3 @@ def run(self, info):
self.move_file_and_write_to_info(info, file_dict, output_file_type) self.move_file_and_write_to_info(info, file_dict, output_file_type)
return [], info return [], info
# class MoveFilesAfterDownloadPP(PostProcessor):
# FILETYPE_KEYS = ['media', 'thumbnails', 'requested_subtitles']
# def __init__(self, downloader=None, downloaded=True):
# PostProcessor.__init__(self, downloader)
# self._downloaded = downloaded
# @classmethod
# def pp_key(cls):
# return 'MoveFiles'
# def expand_relative_paths(self, files_to_move, finaldir):
# for filetype in self.FILETYPE_KEYS:
# if filetype not in files_to_move:
# files_to_move[filetype] = []
# for file_attrs in files_to_move[filetype]:
# if not os.path.isabs(file_attrs['final_filepath']):
# file_attrs['final_filepath'] = os.path.join(finaldir, file_attrs['final_filepath'])
# if not os.path.isabs(file_attrs['current_filepath']):
# file_attrs['current_filepath'] = os.path.abspath(file_attrs['current_filepath'])
# return files_to_move
# def write_filepath_into_info(self, info, filetype, file_attrs):
# if filetype == 'media':
# info['filepath'] = file_attrs['final_filepath']
# elif filetype == 'thumbnails':
# for filetype_dict in info[filetype]:
# if filetype_dict['id'] == file_attrs['id']:
# filetype_dict['filepath'] = file_attrs['final_filepath']
# elif filetype == 'requested_subtitles':
# lang = file_attrs['lang']
# if lang in info[filetype]:
# info[filetype][lang]['filepath'] = file_attrs['final_filepath']
# def run(self, info):
# dl_path, dl_name = os.path.split(info['filepath'])
# finaldir = info.get('__finaldir', os.path.abspath(dl_path))
# th = self._downloader.prepare_filename(info, 'thumbnail')
# pdb.set_trace()
# print("th", th)
# if self._downloaded:
# info['__files_to_move']['media'] = [{'current_filepath': info['filepath'], 'final_filepath': dl_name}]
# files_to_move = self.expand_relative_paths(info['__files_to_move'], finaldir)
# for filetype in self.FILETYPE_KEYS:
# for file_attrs in files_to_move[filetype]:
# current_filepath = file_attrs['current_filepath']
# final_filepath = file_attrs['final_filepath']
# if not current_filepath or not final_filepath:
# continue
# if current_filepath == final_filepath:
# # This ensures the infojson contains the full filepath even
# # when --no-overwrites is used
# self.write_filepath_into_info(info, filetype, file_attrs)
# continue
# if not os.path.exists(current_filepath):
# self.report_warning('File "%s" cannot be found' % current_filepath)
# continue
# if os.path.exists(final_filepath):
# if self.get_param('overwrites', True):
# self.report_warning('Replacing existing file "%s"' % final_filepath)
# os.remove(final_filepath)
# else:
# self.report_warning(
# 'Cannot move file "%s" out of temporary directory since "%s" already exists. '
# % (current_filepath, final_filepath))
# continue
# make_dir(final_filepath, PostProcessingError)
# self.to_screen(f'Moving file "{current_filepath}" to "{final_filepath}"')
# shutil.move(current_filepath, final_filepath) # os.rename cannot move between volumes
# self.write_filepath_into_info(info, filetype, file_attrs)
# return [], info