separate each type of subscription
This commit is contained in:
Jesse Bannon
parent
ed182d79f6
commit
155dbb6063
6 changed files with 314 additions and 284 deletions
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
0
ytdl_subscribe/subscriptions/__init__.py
Normal file
0
ytdl_subscribe/subscriptions/__init__.py
Normal file
79
ytdl_subscribe/subscriptions/soundcloud.py
Normal file
79
ytdl_subscribe/subscriptions/soundcloud.py
Normal file
|
|
@ -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")
|
||||
185
ytdl_subscribe/subscriptions/subscription.py
Normal file
185
ytdl_subscribe/subscriptions/subscription.py
Normal file
|
|
@ -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"],
|
||||
)
|
||||
49
ytdl_subscribe/subscriptions/youtube.py
Normal file
49
ytdl_subscribe/subscriptions/youtube.py
Normal file
|
|
@ -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)
|
||||
Loading…
Reference in a new issue