From e3461d1cd2e76ba3abc473b11701b3a2696f4958 Mon Sep 17 00:00:00 2001 From: Jesse Bannon Date: Thu, 9 Sep 2021 23:51:10 -0700 Subject: [PATCH] albums working for sc --- requirements.txt | 1 + ytdl_subscribe/parse.py | 8 +- ytdl_subscribe/subscriptions.py | 158 ++++++++++++++++++++++++-------- 3 files changed, 126 insertions(+), 41 deletions(-) diff --git a/requirements.txt b/requirements.txt index e5c73bfe..864e2c5b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ dicttoxml==1.7.4 +mergedeep==1.3.4 music-tag==0.4.3 pathvalidate==2.4.1 Pillow==8.3.2 diff --git a/ytdl_subscribe/parse.py b/ytdl_subscribe/parse.py index af244fd4..05a32f55 100644 --- a/ytdl_subscribe/parse.py +++ b/ytdl_subscribe/parse.py @@ -1,6 +1,9 @@ import yaml import time from getpass import getpass + +from mergedeep import mergedeep + from ytdl_subscribe.subscriptions import Subscription from ytdl_subscribe.enums import YAMLSection @@ -26,8 +29,11 @@ def parse_subscriptions(subscription_yaml_path): subscriptions = [] for name, subscription in yaml_dict[YAMLSection.SUBSCRIPTIONS_KEY].items(): + preset = {} if subscription.get('preset') in presets: preset = presets[subscription['preset']] - subscriptions.append(Subscription.from_dict(dict(preset, **subscription))) + subscription = mergedeep.merge({}, preset, subscription) + subscriptions.append(Subscription.from_dict(name, subscription)) + return subscriptions \ No newline at end of file diff --git a/ytdl_subscribe/subscriptions.py b/ytdl_subscribe/subscriptions.py index 1cf081e8..c56db3de 100644 --- a/ytdl_subscribe/subscriptions.py +++ b/ytdl_subscribe/subscriptions.py @@ -1,7 +1,6 @@ -from datetime import datetime import os from shutil import copyfile -import urllib + import music_tag from sanitize_filename import sanitize import dicttoxml @@ -29,16 +28,36 @@ class Subscription(object): source = None - def __init__(self, url, ytdl_opts, post_process, overrides, output_path): - self.url = url + def __init__(self, name, options, ytdl_opts, post_process, overrides, output_path): + """ + Parameters + ---------- + name: str + Name of the subscription + options: dict + Dictionary of ytdl options, specific to the source type + ytdl_opts: dict + Dictionary of options passed directly to ytdl. See `youtube_dl.YoutubeDL.YoutubeDL` for options. + post_process: dict + Dictionary of ytdl-subscribe post processing options + overrides: dict + Dictionary that overrides every ytdl entry. Be careful what you override! + output_path: str + Base path to save files to + """ + self.name = name + self.options = options self.ytdl_opts = ytdl_opts or dict() self.post_process = post_process self.overrides = overrides self.output_path = output_path + # Separate each subscription's working directory + self.WORKING_DIRECTORY += f"{'/' if self.WORKING_DIRECTORY else ''}{self.name}" + # Always set outtmpl to the id and extension. Will be renamed using the subscription's output_path value self.ytdl_opts['outtmpl'] = self.WORKING_DIRECTORY + '/%(id)s.%(ext)s' - + self.ytdl_opts['writethumbnail'] = True def parse_entry(self, entry): # Add overrides to the entry @@ -46,7 +65,6 @@ class Subscription(object): # Add the file path to the entry, assert it exists entry['file_path'] = f"{self.WORKING_DIRECTORY}/{entry['id']}.{entry['ext']}" - assert os.path.isfile(entry['file_path']) # Add sanitized values entry['sanitized_artist'] = sanitize(entry['artist']) @@ -75,6 +93,12 @@ class Subscription(object): with open(nfo_file_path, 'wb') as f: f.write(xml) + def extract_info(self): + """ + Extracts only the info of the source, does not download it + """ + raise NotImplemented('Each source needs to implement how it extracts info') + def post_process_entry(self, entry): if 'tagging' in self.post_process: self._post_process_tagging(entry) @@ -85,64 +109,107 @@ class Subscription(object): # Download the thumbnail if its present if 'thumbnail_name' in self.post_process: - thumbnail_file_path = _ffile(self.post_process['thumbnail_name'], self.output_path, entry, makedirs=True) - urllib.request.urlretrieve(entry['thumbnail_url'], thumbnail_file_path) - if 'convert_thumbnail' in self.post_process: - # TODO: Clean with yaml definitions - if self.post_process['convert_thumbnail'] == 'jpg': - self.post_process['convert_thumbnail'] = 'jpeg' - im = Image.open(thumbnail_file_path).convert("RGB") - im.save(thumbnail_file_path, self.post_process['convert_thumbnail']) + thumbnail_dest_path = _ffile(self.post_process['thumbnail_name'], self.output_path, entry, makedirs=True) + if not os.path.isfile(thumbnail_dest_path): + thumbnail_file_path = f"{self.WORKING_DIRECTORY}/{entry['id']}.{entry['thumbnail_ext']}" + if os.path.isfile(thumbnail_file_path): + copyfile(thumbnail_file_path, thumbnail_dest_path) + + if 'convert_thumbnail' in self.post_process: + # TODO: Clean with yaml definitions + if self.post_process['convert_thumbnail'] == 'jpg': + self.post_process['convert_thumbnail'] = 'jpeg' + im = Image.open(thumbnail_dest_path).convert("RGB") + im.save(thumbnail_dest_path, self.post_process['convert_thumbnail']) if 'nfo' in self.post_process: self._post_process_nfo(entry) - def extract_info(self): - """ - Extracts only the info of the source, does not download it - """ - with ytdl.YoutubeDL(self.ytdl_opts) as ytd: - info = ytd.extract_info(self.url) - - entries = [self.parse_entry(e) for e in info['entries']] - for e in entries: - self.post_process_entry(e) - - return - @classmethod - def from_dict(cls, d): - source = d['source'] - if source == SubscriptionSource.SOUNDCLOUD: + def from_dict(cls, name, d): + # TODO: Make sure there is only one source, verify url + if SubscriptionSource.SOUNDCLOUD in d: dclass = SoundcloudSubscription - elif source == SubscriptionSource.YOUTUBE: + elif SubscriptionSource.YOUTUBE in d: dclass = YoutubeSubscription else: raise ValueError('dne') return dclass( - url=d['url'], + name=name, + options=d[dclass.source], ytdl_opts=d['ytdl_opts'], post_process=d['post_process'], overrides=d['overrides'], - output_path=d['output_path'] + output_path=d['output_path'], ) class SoundcloudSubscription(Subscription): source = SubscriptionSource.SOUNDCLOUD + def is_entry_skippable(self, entry): + return self.options['skip_premiere_tracks'] and '/preview/' in entry['url'] + def parse_entry(self, entry): entry = super(SoundcloudSubscription, self).parse_entry(entry) - entry['upload_year'] = datetime.fromtimestamp(entry['timestamp']).year + entry['upload_year'] = entry['upload_date'][:4] - # Add thumbnail values - original_thumbnail = [t for t in entry['thumbnails'] if t['id'] == 'original'][0] - entry['thumbnail_url'] = original_thumbnail['url'] - entry['thumbnail_ext'] = original_thumbnail['url'].split('.')[-1] + # Add thumbnail ext value + entry['thumbnail_ext'] = entry['thumbnail'].split('.')[-1] + + # If the entry does not have album fields, set them to be the track fields + if 'album' not in entry: + entry['album'] = entry['title'] + entry['sanitized_album'] = entry['sanitized_title'] + entry['album_year'] = entry['upload_year'] + entry['tracknumber'] = 1 + entry['tracknumberpadded'] = f'{1:02d}' return entry + def parse_album_entry(self, album_entry): + album_year = min([int(e['upload_date'][:4]) for e in album_entry['entries']]) + for track_number, e in enumerate(album_entry['entries'], start=1): + e['album'] = album_entry['title'] + e['sanitized_album'] = sanitize(album_entry['title']) + e['album_year'] = album_year + e['tracknumber'] = track_number + e['tracknumberpadded'] = f'{track_number:02d}' + return album_entry + + def extract_info(self): + """ + Extracts only the info of the source, does not download it + """ + base_url = f"https://soundcloud.com/{self.options['username']}" + tracks = [] + + if self.options.get('download_strategy') == 'albums_then_tracks': + track_ytdl_opts = { + 'download_archive': self.WORKING_DIRECTORY + '/ytdl-download-archive.txt', + 'forcejson': True, + } + # Get the album tracks first, but do not download. Unfortunately we cannot use download_archive for + # this be + with ytdl.YoutubeDL(self.ytdl_opts) as ytd: + info = ytd.extract_info(base_url + '/albums', download=False) + + album_entries = [self.parse_album_entry(a) for a in info['entries']] + for album_entry in album_entries: + tracks += [self.parse_entry(e) for e in album_entry['entries'] if not self.is_entry_skippable(e)] + + with ytdl.YoutubeDL(dict(self.ytdl_opts, **track_ytdl_opts)) as ytd: + # Get the rest of the tracks that are part of the album tracks + info = ytd.extract_info(base_url + '/tracks') + album_track_ids = [t['id'] for t in tracks] + tracks += [self.parse_entry(e) for e in info['entries'] if e['id'] not in album_track_ids and not self.is_entry_skippable(e)] + + for e in tracks: + self.post_process_entry(e) + else: + raise ValueError('Invalid download_strategy field for Soundcloud') + class YoutubeSubscription(Subscription): source = SubscriptionSource.YOUTUBE @@ -151,9 +218,20 @@ class YoutubeSubscription(Subscription): entry = super(YoutubeSubscription, self).parse_entry(entry) entry['upload_year'] = entry['upload_date'][:4] - entry['thumbnail_url'] = entry['thumbnail'] entry['thumbnail_ext'] = 'webp' # Try to get the track, fall back on title entry['sanitized_track'] = sanitize(entry.get('track', entry['title'])) - return entry \ No newline at end of file + return entry + + def extract_info(self): + """ + Extracts only the info of the source, does not download it + """ + url = f"https://youtube.com/playlist?list={self.options['playlist_id']}" + with ytdl.YoutubeDL(self.ytdl_opts) as ytd: + info = ytd.extract_info(url) + + entries = [self.parse_entry(e) for e in info['entries']] + for e in entries: + self.post_process_entry(e) \ No newline at end of file