diff --git a/ytdl_subscribe/parse.py b/ytdl_subscribe/parse.py index 1c4af1ad..147907ca 100644 --- a/ytdl_subscribe/parse.py +++ b/ytdl_subscribe/parse.py @@ -1,21 +1,14 @@ import yaml -import time -from getpass import getpass from mergedeep import mergedeep -from ytdl_subscribe.subscriptions import Subscription +from ytdl_subscribe.subscriptions.subscription import Subscription from ytdl_subscribe.enums import YAMLSection def _set_config_variables(config): Subscription.WORKING_DIRECTORY = config.get("working_directory", "") - Subscription.USERNAME = config.get("ytdl_username") - if Subscription.USERNAME: - print(f"Enter password for '{Subscription.USERNAME}': ") - time.sleep(0.2) - Subscription.PASSWORD = getpass() def parse_subscriptions(subscription_yaml_path): diff --git a/ytdl_subscribe/subscriptions.py b/ytdl_subscribe/subscriptions.py deleted file mode 100644 index 84d778eb..00000000 --- a/ytdl_subscribe/subscriptions.py +++ /dev/null @@ -1,276 +0,0 @@ -import json -import os -from shutil import copyfile - -import music_tag -from sanitize_filename import sanitize -import dicttoxml - -from PIL import Image -from ytdl_subscribe import SubscriptionSource - -import yt_dlp as ytdl - - -def _f(value, entry): - return value.format(**entry) - - -def _ffile(file_value, path, entry, makedirs=False): - file_name = _f(file_value, entry) - output_file_path = f"{path}/{file_name}" - if makedirs: - os.makedirs(os.path.dirname(output_file_path), exist_ok=True) - - return output_file_path - - -class Subscription(object): - WORKING_DIRECTORY = "" - - source = None - - 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 - entry = dict(entry, **self.overrides) - - # Add the file path to the entry, assert it exists - entry["file_path"] = f"{self.WORKING_DIRECTORY}/{entry['id']}.{entry['ext']}" - - # Add sanitized values - entry["sanitized_artist"] = sanitize(entry["artist"]) - entry["sanitized_title"] = sanitize(entry["title"]) - - return entry - - def _post_process_tagging(self, entry): - t = music_tag.load_file(entry["file_path"]) - for tag, tag_formatter in self.post_process["tagging"].items(): - t[tag] = _f(tag_formatter, entry) - t.save() - - def _post_process_nfo(self, entry): - nfo = {} - for tag, tag_formatter in self.post_process["nfo"].items(): - nfo[tag] = _f(tag_formatter, entry) - - xml = dicttoxml.dicttoxml( - obj=nfo, - root="nfo_root" in self.post_process, - custom_root=self.post_process.get("nfo_root"), - attr_type=False, - ) - nfo_file_path = _ffile( - self.post_process["nfo_name"], self.output_path, entry, makedirs=True - ) - 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) - - # Move the file after all direct file modifications are complete - output_file_path = _ffile( - self.post_process["file_name"], self.output_path, entry, makedirs=True - ) - copyfile(entry["file_path"], output_file_path) - - # Download the thumbnail if its present - if "thumbnail_name" in self.post_process: - 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) - - @classmethod - def from_dict(cls, name, d): - # TODO: Make sure there is only one source, verify url - if SubscriptionSource.SOUNDCLOUD in d: - dclass = SoundcloudSubscription - elif SubscriptionSource.YOUTUBE in d: - dclass = YoutubeSubscription - else: - raise ValueError("dne") - - return dclass( - name=name, - options=d[dclass.source], - ytdl_opts=d["ytdl_opts"], - post_process=d["post_process"], - overrides=d["overrides"], - 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"] = entry["upload_date"][:4] - - # 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": - # Get the album info first, but do not download. This tells us which track ids belong - # to an album. Unfortunately we cannot use download_archive or info.json for this - with ytdl.YoutubeDL(self.ytdl_opts) as ytd: - info = ytd.extract_info(base_url + "/albums", download=False) - - # For each album, parse each entry in the album - 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) - ] - - # Download the tracks now, and use download_archive to cache - track_ytdl_opts = { - "download_archive": self.WORKING_DIRECTORY - + "/ytdl-download-archive.txt", - } - with ytdl.YoutubeDL(dict(self.ytdl_opts, **track_ytdl_opts)) as ytd: - info = ytd.extract_info(base_url + "/tracks") - - # Skip parsing entries that have already been parsed when parsing albums - 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 - - def parse_entry(self, entry): - entry = super(YoutubeSubscription, self).parse_entry(entry) - entry["upload_year"] = entry["upload_date"][:4] - - entry["thumbnail_ext"] = entry["thumbnail"].split(".")[-1] - - # Try to get the track, fall back on title - entry["sanitized_track"] = sanitize(entry.get("track", entry["title"])) - return entry - - def extract_info(self): - playlist_id = self.options["playlist_id"] - url = f"https://youtube.com/playlist?list={playlist_id}" - track_ytdl_opts = { - "download_archive": self.WORKING_DIRECTORY + "/ytdl-download-archive.txt", - "writeinfojson": True, - } - - # Do not get entries from the extract info, let it write to the info.json file and - # load that instead. This is because if the video is already downloaded, it will - # not fetch the metadata (maybe there is a way??) - with ytdl.YoutubeDL(dict(self.ytdl_opts, **track_ytdl_opts)) as ytd: - _ = ytd.extract_info(url) - - # Load the entries from info.json, ignore the playlist entry - entries = [] - for file_name in os.listdir(self.WORKING_DIRECTORY): - if file_name.endswith(".info.json") and not file_name.startswith( - playlist_id - ): - with open(self.WORKING_DIRECTORY + "/" + file_name, "r") as f: - entries.append(json.load(f)) - - entries = [self.parse_entry(e) for e in entries] - for e in entries: - self.post_process_entry(e) diff --git a/ytdl_subscribe/subscriptions/__init__.py b/ytdl_subscribe/subscriptions/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/ytdl_subscribe/subscriptions/soundcloud.py b/ytdl_subscribe/subscriptions/soundcloud.py new file mode 100644 index 00000000..ca16ef8e --- /dev/null +++ b/ytdl_subscribe/subscriptions/soundcloud.py @@ -0,0 +1,79 @@ +import yt_dlp as ytdl +from sanitize_filename import sanitize + +from ytdl_subscribe import SubscriptionSource +from ytdl_subscribe.subscriptions.subscription import Subscription + + +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"] = entry["upload_date"][:4] + + # 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): + base_url = f"https://soundcloud.com/{self.options['username']}" + tracks = [] + + if self.options.get("download_strategy") == "albums_then_tracks": + # Get the album info first, but do not download. This tells us which track ids belong + # to an album. Unfortunately we cannot use download_archive or info.json for this + with ytdl.YoutubeDL(self.ytdl_opts) as ytd: + info = ytd.extract_info(base_url + "/albums", download=False) + + # For each album, parse each entry in the album + 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) + ] + + # Download the tracks now, and use download_archive to cache + track_ytdl_opts = { + "download_archive": self.WORKING_DIRECTORY + + "/ytdl-download-archive.txt", + } + with ytdl.YoutubeDL(dict(self.ytdl_opts, **track_ytdl_opts)) as ytd: + info = ytd.extract_info(base_url + "/tracks") + + # Skip parsing entries that have already been parsed when parsing albums + 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") diff --git a/ytdl_subscribe/subscriptions/subscription.py b/ytdl_subscribe/subscriptions/subscription.py new file mode 100644 index 00000000..a62affc2 --- /dev/null +++ b/ytdl_subscribe/subscriptions/subscription.py @@ -0,0 +1,185 @@ +import os +from shutil import copyfile + +import dicttoxml +import music_tag +from PIL import Image +from sanitize_filename import sanitize + +from ytdl_subscribe import SubscriptionSource + + +class Subscription(object): + WORKING_DIRECTORY = "" + + source = None + + 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 format_entry(self, value, entry): + """ + Parameters + ---------- + value: str + String with format variables that should be populated + entry: dict + Entry used to populate any format variables in the value + + Returns + ------- + str + """ + return value.format(**entry) + + def format_filepath(self, filepath, entry, makedirs=False): + """ + Convert a filepath value in the config to an actual filepath. + + Parameters + ---------- + filepath: str + File path relative to the specified output path + entry: dict + Entry used to populate any format variables in the filepath + makedirs: bool + Whether to create all directories in the final filepath. + + Returns + ------- + str + Full filepath. + """ + file_name = self.format_entry(filepath, entry) + output_file_path = f"{self.output_path}/{file_name}" + if makedirs: + os.makedirs(os.path.dirname(output_file_path), exist_ok=True) + + return output_file_path + + def parse_entry(self, entry): + # Add overrides to the entry + entry = dict(entry, **self.overrides) + + # Add the file path to the entry, assert it exists + entry["file_path"] = f"{self.WORKING_DIRECTORY}/{entry['id']}.{entry['ext']}" + + # Add sanitized values + entry["sanitized_artist"] = sanitize(entry["artist"]) + entry["sanitized_title"] = sanitize(entry["title"]) + + return entry + + def _post_process_tagging(self, entry): + t = music_tag.load_file(entry["file_path"]) + for tag, tag_formatter in self.post_process["tagging"].items(): + t[tag] = self.format_entry(tag_formatter, entry) + t.save() + + def _post_process_nfo(self, entry): + nfo = {} + for tag, tag_formatter in self.post_process["nfo"].items(): + nfo[tag] = self.format_entry(tag_formatter, entry) + + xml = dicttoxml.dicttoxml( + obj=nfo, + root="nfo_root" in self.post_process, + custom_root=self.post_process.get("nfo_root"), + attr_type=False, + ) + nfo_file_path = self.format_filepath( + self.post_process["nfo_name"], entry, makedirs=True + ) + 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) + + # Move the file after all direct file modifications are complete + output_file_path = self.format_filepath( + self.post_process["file_name"], entry, makedirs=True + ) + copyfile(entry["file_path"], output_file_path) + + # Download the thumbnail if its present + if "thumbnail_name" in self.post_process: + thumbnail_dest_path = self.format_filepath( + self.post_process["thumbnail_name"], + 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) + + @classmethod + def from_dict(cls, name, d): + # TODO: Make sure there is only one source, verify url + if SubscriptionSource.SOUNDCLOUD in d: + from ytdl_subscribe.subscriptions.soundcloud import SoundcloudSubscription + + dclass = SoundcloudSubscription + elif SubscriptionSource.YOUTUBE in d: + from ytdl_subscribe.subscriptions.youtube import YoutubeSubscription + + dclass = YoutubeSubscription + else: + raise ValueError("dne") + + return dclass( + name=name, + options=d[dclass.source], + ytdl_opts=d["ytdl_opts"], + post_process=d["post_process"], + overrides=d["overrides"], + output_path=d["output_path"], + ) diff --git a/ytdl_subscribe/subscriptions/youtube.py b/ytdl_subscribe/subscriptions/youtube.py new file mode 100644 index 00000000..c3437351 --- /dev/null +++ b/ytdl_subscribe/subscriptions/youtube.py @@ -0,0 +1,49 @@ +import json +import os + +import yt_dlp as ytdl +from sanitize_filename import sanitize + +from ytdl_subscribe import SubscriptionSource +from ytdl_subscribe.subscriptions.subscription import Subscription + + +class YoutubeSubscription(Subscription): + source = SubscriptionSource.YOUTUBE + + def parse_entry(self, entry): + entry = super(YoutubeSubscription, self).parse_entry(entry) + entry["upload_year"] = entry["upload_date"][:4] + + entry["thumbnail_ext"] = entry["thumbnail"].split(".")[-1] + + # Try to get the track, fall back on title + entry["sanitized_track"] = sanitize(entry.get("track", entry["title"])) + return entry + + def extract_info(self): + playlist_id = self.options["playlist_id"] + url = f"https://youtube.com/playlist?list={playlist_id}" + track_ytdl_opts = { + "download_archive": self.WORKING_DIRECTORY + "/ytdl-download-archive.txt", + "writeinfojson": True, + } + + # Do not get entries from the extract info, let it write to the info.json file and + # load that instead. This is because if the video is already downloaded, it will + # not fetch the metadata (maybe there is a way??) + with ytdl.YoutubeDL(dict(self.ytdl_opts, **track_ytdl_opts)) as ytd: + _ = ytd.extract_info(url) + + # Load the entries from info.json, ignore the playlist entry + entries = [] + for file_name in os.listdir(self.WORKING_DIRECTORY): + if file_name.endswith(".info.json") and not file_name.startswith( + playlist_id + ): + with open(self.WORKING_DIRECTORY + "/" + file_name, "r") as f: + entries.append(json.load(f)) + + entries = [self.parse_entry(e) for e in entries] + for e in entries: + self.post_process_entry(e)