danvics/ytdl-sub
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add youtube.download_strategy: "merge_playlist" (#77)

Browse source 
* begin

* chapters class, trying to add chapter splitting and general ffmpeg metadata creation

* working split video

* so very close

* test passing, is reproducible

* lint fixed

* fix docs

* track no longer used in example

* if, then del

* multiple values

* github hash
This commit is contained in:
Jesse Bannon 2022-06-26 15:18:07 -07:00 committed by GitHub
parent 1620c3a42d
commit cb6dc8e034
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 613 additions and 96 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
docs/config.rst
View file

@ -90,6 +90,17 @@ ___________
-------------------------------------------------------------------------------
merge_playlist
______________
.. autoclass:: ytdl_sub.downloaders.youtube.merge_playlist.YoutubeMergePlaylistDownloaderOptions()
:members:
:member-order: bysource
:inherited-members:
.. autofunction:: ytdl_sub.downloaders.youtube.merge_playlist.YoutubeMergePlaylistDownloader.ytdl_option_defaults()
-------------------------------------------------------------------------------
soundcloud
''''''''''
Download strategies for downloading music from Soundcloud. See

4
examples/kodi_music_videos_config.yaml
View file

@ -47,7 +47,7 @@ presets:
nfo_root: "musicvideo"
tags:
artist: "{artist}"
title: "{track_title}"
title: "{title}"
album: "Music Videos"
year: "{upload_year}"
@ -56,7 +56,7 @@ presets:
# here, which gets reused above for the video, thumbnail, and NFO file.
overrides:
music_video_directory: "path/to/Music Videos"
music_video_name: "{artist_sanitized} - {track_title_sanitized}"
music_video_name: "{artist_sanitized} - {title_sanitized}"
# It is not always ideal to download all of an artist's music videos.
# Maybe you only like one song of theirs. We can reuse our preset above

2
setup.cfg
View file

@ -24,13 +24,13 @@ package_dir =
packages=find:
install_requires =
yt-dlp
argparse==1.4.0
dicttoxml==1.7.4
mergedeep==1.3.4
mediafile==0.9.0
Pillow==9.1.0
PyYAML==6.0
yt-dlp==2022.4.8
[options.packages.find]
where=src

4
src/ytdl_sub/config/config_file.py
View file

@ -57,13 +57,13 @@ class ConfigOptions(StrictDictValidator):
.. code-block:: bash
ytdl-sub dl --preset yt_music_video --youtube.video_url youtube.com/watch?v=a1b2c3
ytdl-sub dl --preset "yt_music_video" --youtube.video_url "youtube.com/watch?v=a1b2c3"
to
.. code-block:: bash
ytdl-sub dl --mv --v youtube.com/watch?v=a1b2c3
ytdl-sub dl --mv --v "youtube.com/watch?v=a1b2c3"
"""
if self._dl_aliases:
return self._dl_aliases.dict

2
src/ytdl_sub/config/preset_class_mappings.py
View file

@ -4,6 +4,7 @@ from typing import Type
from ytdl_sub.downloaders.downloader import Downloader
from ytdl_sub.downloaders.soundcloud_downloader import SoundcloudAlbumsAndSinglesDownloader
from ytdl_sub.downloaders.youtube.merge_playlist import YoutubeMergePlaylistDownloader
from ytdl_sub.downloaders.youtube.split_video import YoutubeSplitVideoDownloader
from ytdl_sub.downloaders.youtube_downloader import YoutubeChannelDownloader
from ytdl_sub.downloaders.youtube_downloader import YoutubePlaylistDownloader
@ -25,6 +26,7 @@ class DownloadStrategyMapping:
"playlist": YoutubePlaylistDownloader,
"channel": YoutubeChannelDownloader,
"split_video": YoutubeSplitVideoDownloader,
"merge_playlist": YoutubeMergePlaylistDownloader,
},
"soundcloud": {
"albums_and_singles": SoundcloudAlbumsAndSinglesDownloader,

2
src/ytdl_sub/downloaders/downloader.py
View file

@ -44,6 +44,8 @@ class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC):
downloader_options_type: Type[DownloaderValidator] = DownloaderValidator
downloader_entry_type: Type[Entry] = Entry
supports_download_archive: bool = True
@classmethod
def ytdl_option_overrides(cls) -> Dict:
"""Global overrides that even overwrite user input"""

145
src/ytdl_sub/downloaders/youtube/merge_playlist.py Normal file
View file

@ -0,0 +1,145 @@
from typing import Dict
from typing import List
from typing import Optional
from ytdl_sub.downloaders.youtube_downloader import YoutubeDownloader
from ytdl_sub.downloaders.youtube_downloader import YoutubePlaylistDownloaderOptions
from ytdl_sub.entries.youtube import YoutubeVideo
from ytdl_sub.utils.chapters import Chapters
from ytdl_sub.utils.chapters import Timestamp
from ytdl_sub.utils.ffmpeg import add_ffmpeg_metadata
from ytdl_sub.validators.validators import BoolValidator
class YoutubeMergePlaylistDownloaderOptions(YoutubePlaylistDownloaderOptions):
r"""
Downloads all videos in a playlist and merges them into a single video.
Usage:
.. code-block:: yaml
presets:
example_preset:
youtube:
# required
download_strategy: "merge_playlist"
playlist_url: "https://www.youtube.com/playlist?list=UCsvn_Po0SmunchJYtttWpOxMg"
# optional
add_chapters: False
CLI usage:
.. code-block:: bash
ytdl-sub dl \
--preset "example_preset" \
--youtube.playlist_url "https://www.youtube.com/playlist?list=UCsvn_Po0SmunchJYtttWpOxMg" \
--youtube.add_chapters True
"""
_required_keys = {"playlist_url"}
_optional_keys = {"add_chapters"}
def __init__(self, name, value):
super().__init__(name, value)
self._add_chapters = self._validate_key_if_present(
"add_chapters", validator=BoolValidator, default=False
).value
@property
def add_chapters(self) -> Optional[bool]:
"""
Optional. Whether to add chapters using each video's title in the merged playlist.
Defaults to False.
"""
return self._add_chapters
class YoutubeMergePlaylistDownloader(
YoutubeDownloader[YoutubeMergePlaylistDownloaderOptions, YoutubeVideo]
):
downloader_options_type = YoutubeMergePlaylistDownloaderOptions
downloader_entry_type = YoutubeVideo
supports_download_archive = False
@classmethod
def ytdl_option_defaults(cls) -> Dict:
"""
Default `ytdl_options`_ for ``merge_playlist``
.. code-block:: yaml
ytdl_options:
ignoreerrors: True # ignore errors like hidden videos, age restriction, etc
playlistreverse: True # Sort the playlist so it begins with the first entry
postprocessors:
# Convert the videos to mkv format
- key: "FFmpegVideoConvertor"
when: "post_process"
preferedformat: "mkv"
# Concatenate all the playlist videos into a single file
- key: "FFmpegConcat"
when: "playlist"
"""
return dict(
super().ytdl_option_defaults(),
**{
"playlistreverse": True,
"postprocessors": [
{
"key": "FFmpegVideoConvertor",
"when": "post_process",
"preferedformat": "mkv",
},
{
"key": "FFmpegConcat",
"when": "playlist",
},
],
},
)
@classmethod
def _add_chapters(cls, merged_video: YoutubeVideo) -> None:
titles: List[str] = []
timestamps: List[Timestamp] = []
current_timestamp_sec = 0
for video_entry in merged_video.kwargs("entries"):
timestamps.append(Timestamp(current_timestamp_sec))
titles.append(video_entry["title"])
current_timestamp_sec += video_entry["duration"]
add_ffmpeg_metadata(
file_path=merged_video.get_download_file_path(),
chapters=Chapters(timestamps=timestamps, titles=titles),
file_duration_sec=merged_video.kwargs("duration"),
)
def _to_merged_video(self, entry_dict: Dict) -> YoutubeVideo:
"""
Adds a few entries not included in a playlist entry to make it look like a merged video
entry_dict
"""
# Set the upload date to be the latest playlist video date
entry_dict["upload_date"] = max(
playlist_entry["upload_date"] for playlist_entry in entry_dict["entries"]
)
entry_dict["duration"] = sum(
playlist_entry["duration"] for playlist_entry in entry_dict["entries"]
)
entry_dict["ext"] = entry_dict["requested_downloads"][0]["ext"]
return YoutubeVideo(entry_dict=entry_dict, working_directory=self.working_directory)
def download(self) -> List[YoutubeVideo]:
"""Download a single Youtube video, then split it into multiple videos"""
merged_video = self._to_merged_video(
entry_dict=self.extract_info(url=self.download_options.playlist_url)
)
if self.download_options.add_chapters:
self._add_chapters(merged_video=merged_video)
return [merged_video]

86
src/ytdl_sub/downloaders/youtube/split_video.py
View file

@ -1,17 +1,16 @@
import copy
import os.path
import re
from pathlib import Path
from shutil import copyfile
from typing import Dict
from typing import List
from typing import Tuple
from ytdl_sub.downloaders.youtube_downloader import YoutubeDownloader
from ytdl_sub.downloaders.youtube_downloader import YoutubeVideoDownloaderOptions
from ytdl_sub.entries.youtube import YoutubePlaylistVideo
from ytdl_sub.entries.youtube import YoutubeVideo
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.utils.chapters import Chapters
from ytdl_sub.utils.chapters import Timestamp
from ytdl_sub.utils.ffmpeg import FFMPEG
from ytdl_sub.utils.thumbnail import convert_download_thumbnail
from ytdl_sub.validators.validators import StringValidator
@ -29,56 +28,11 @@ def _split_video_uid(source_uid: str, idx: int) -> str:
return f"{source_uid}___{idx}"
def _parse_split_timestamp_file(split_timestamp_path: str) -> Tuple[List[str], List[str]]:
"""
Returns two lists, one containing timestamps in HH:MM:SS format, and the other titles
"""
if not os.path.isfile(split_timestamp_path):
raise ValidationException(
f"split_timestamp file path '{split_timestamp_path}' does not exist."
)
with open(split_timestamp_path, "r", encoding="utf-8") as file:
lines = file.readlines()
timestamps: List[str] = []
titles: List[str] = []
idx = 0
for idx, line in enumerate(lines):
match = _SPLIT_TIMESTAMP_REGEX.match(line)
if not match:
break
timestamp = match.group(1)
title = match.group(2)
match len(timestamp):
case 4: # 0:00
timestamp = f"00:0{timestamp}"
case 5: # 00:00
timestamp = f"00:{timestamp}"
case 7: # 0:00:00
timestamp = f"0{timestamp}"
case _:
pass
assert len(timestamp) == 8
timestamps.append(timestamp)
titles.append(title)
if idx not in (len(lines) - 1, len(lines) - 2):
raise ValidationException(
f"split_timestamp file '{split_timestamp_path} is not formatted correctly. "
f"Each line must be formatted as '0:00 title' - a timestamp, space, then title."
)
return timestamps, titles
def _split_video_ffmpeg_cmd(
input_file: str, output_file: str, timestamps: List[str], idx: int
input_file: str, output_file: str, timestamps: List[Timestamp], idx: int
) -> List[str]:
timestamp_begin = timestamps[idx]
timestamp_end = timestamps[idx + 1] if idx + 1 < len(timestamps) else ""
timestamp_begin = timestamps[idx].timestamp_str
timestamp_end = timestamps[idx + 1].timestamp_str if idx + 1 < len(timestamps) else ""
cmd = ["-i", input_file, "-ss", timestamp_begin]
if timestamp_end:
@ -87,12 +41,8 @@ def _split_video_ffmpeg_cmd(
return cmd
###############################################################################
# Youtube split video downloader + options
class YoutubeSplitVideoDownloaderOptions(YoutubeVideoDownloaderOptions):
"""
r"""
Downloads a single youtube video, then splits in to separate videos using a file containing
timestamps. Each separate video will be formatted as if it was downloaded from a playlist.
This download strategy is intended for CLI usage performing a one-time download of a video,
@ -130,9 +80,8 @@ class YoutubeSplitVideoDownloaderOptions(YoutubeVideoDownloaderOptions):
16:39 Jokes
1:02:23 Ending
The above will create 5 videos in total. The first timestamp must start with ``0:00``
and the last timestamp, in this example, would create a video starting at ``1:02:23`` and
end at Youtube video's ending.
The above will create 5 videos in total. The last timestamp, in this example, would create a
video starting at ``1:02:23`` and end at Youtube video's ending.
"""
_required_keys = {"video_url", "split_timestamps"}
@ -183,8 +132,10 @@ class YoutubeSplitVideoDownloader(
entry_dict["id"] = _split_video_uid(source_uid=entry_dict["id"], idx=idx)
# Remove track and artist since its now split
del entry_dict["track"]
del entry_dict["artist"]
if "track" in entry_dict:
del entry_dict["track"]
if "artist" in entry_dict:
del entry_dict["artist"]
return YoutubePlaylistVideo(entry_dict=entry_dict, working_directory=self.working_directory)
@ -192,9 +143,7 @@ class YoutubeSplitVideoDownloader(
"""Download a single Youtube video, then split it into multiple videos"""
split_videos: List[YoutubePlaylistVideo] = []
timestamps, titles = _parse_split_timestamp_file(
split_timestamp_path=self.download_options.split_timestamps
)
chapters = Chapters.from_file(chapters_file_path=self.download_options.split_timestamps)
entry_dict = self.extract_info(url=self.download_options.video_url)
entry = YoutubeVideo(entry_dict=entry_dict, working_directory=self.working_directory)
@ -202,7 +151,7 @@ class YoutubeSplitVideoDownloader(
# when copying it
convert_download_thumbnail(entry=entry)
for idx, title in enumerate(titles):
for idx, title in enumerate(chapters.titles):
new_uid = _split_video_uid(source_uid=entry.uid, idx=idx)
# Get the input/output file paths
@ -215,7 +164,10 @@ class YoutubeSplitVideoDownloader(
# Run ffmpeg to create the split the video
FFMPEG.run(
_split_video_ffmpeg_cmd(
input_file=input_file, output_file=output_file, timestamps=timestamps, idx=idx
input_file=input_file,
output_file=output_file,
timestamps=chapters.timestamps,
idx=idx,
)
)
# Copy the thumbnail
@ -227,7 +179,7 @@ class YoutubeSplitVideoDownloader(
source_entry_dict=entry_dict,
title=title,
idx=idx,
split_video_count=len(timestamps),
split_video_count=len(chapters.timestamps),
)
)

4
src/ytdl_sub/entries/variables/youtube_variables.py
View file

@ -32,6 +32,8 @@ class YoutubeVideoVariables(EntryVariables):
Returns
-------
The track title of a music video if it is available, otherwise it falls back to the title.
NOTE: Even if a video has music metadata, this variable does not always get pulled via
yt-dlp. Use with caution.
"""
# Try to get the track, fall back on title
if self.kwargs_contains("track"):
@ -54,6 +56,8 @@ class YoutubeVideoVariables(EntryVariables):
Returns
-------
The artist of a music video if it is available, otherwise it falls back to the channel.
NOTE: Even if a video has music metadata, this variable does not always get pulled via
yt-dlp. Use with caution.
"""
if self.kwargs_contains("artist"):
return self.kwargs("artist")

22
src/ytdl_sub/subscriptions/subscription.py
View file

@ -132,6 +132,18 @@ class Subscription:
"""
return self.overrides.apply_formatter(formatter=self.output_options.output_directory)
@property
def maintain_download_archive(self) -> bool:
"""
Returns
-------
Whether to maintain a download archive
"""
return (
self.output_options.maintain_download_archive
and self.downloader_class.supports_download_archive
)
def _copy_file_to_output_directory(
self, entry: Entry, source_file_path: str, output_file_name: str
):
@ -152,7 +164,7 @@ class Subscription:
os.makedirs(os.path.dirname(destination_file_path), exist_ok=True)
copyfile(source_file_path, destination_file_path)
if self.output_options.maintain_download_archive:
if self.maintain_download_archive:
self._enhanced_download_archive.mapping.add_entry(entry, output_file_name)
def _copy_entry_files_to_output_directory(self, entry: Entry):
@ -207,14 +219,14 @@ class Subscription:
"""
Context manager to initialize the enhanced download archive
"""
if self.output_options.maintain_download_archive:
if self.maintain_download_archive:
self._enhanced_download_archive.prepare_download_archive()
yield
# If output options maintains stale file deletion, perform the delete here prior to saving
# the download archive
if self.output_options.maintain_download_archive:
if self.maintain_download_archive:
date_range_to_keep = self.output_options.get_upload_date_range_to_keep()
if date_range_to_keep:
self._enhanced_download_archive.remove_stale_files(date_range=date_range_to_keep)
@ -234,7 +246,7 @@ class Subscription:
output_directory=self.output_directory,
overrides=self.overrides,
enhanced_download_archive=self._enhanced_download_archive
if self.output_options.maintain_download_archive
if self.maintain_download_archive
else None,
)
@ -253,7 +265,7 @@ class Subscription:
download_options=self.downloader_options,
ytdl_options=self.ytdl_options.dict,
download_archive_file_name=self._enhanced_download_archive.archive_file_name
if self.output_options.maintain_download_archive
if self.maintain_download_archive
else None,
)

170
src/ytdl_sub/utils/chapters.py Normal file
View file

@ -0,0 +1,170 @@
import os
import re
from typing import List
from ytdl_sub.utils.exceptions import ValidationException
class Timestamp:
# Captures the following formats:
# 0:00 title
# 00:00 title
# 1:00:00 title
# 01:00:00 title
# where capture group 1 and 2 are the timestamp and title, respectively
_SPLIT_TIMESTAMP_REGEX = re.compile(r"^((?:\d\d:)?(?:\d:)?(?:\d)?\d:\d\d)$")
@classmethod
def _normalize_timestamp_str(cls, timestamp_str: str) -> str:
match = cls._SPLIT_TIMESTAMP_REGEX.match(timestamp_str)
if not match:
raise ValueError(f"Cannot parse youtube timestamp '{timestamp_str}'")
timestamp = match.group(1)
match len(timestamp):
case 4: # 0:00
timestamp = f"00:0{timestamp}"
case 5: # 00:00
timestamp = f"00:{timestamp}"
case 7: # 0:00:00
timestamp = f"0{timestamp}"
case _:
pass
assert len(timestamp) == 8
return timestamp
def __init__(self, timestamp_sec: int):
self._timestamp_sec = timestamp_sec
@property
def timestamp_sec(self) -> int:
"""
Returns
-------
Timestamp in seconds
"""
return self._timestamp_sec
@property
def timestamp_str(self) -> str:
"""
Returns
-------
The timestamp in 'HH:MM:SS' format
"""
seconds = self.timestamp_sec
hours = int(seconds / 3600)
seconds -= hours * 3600
minutes = int(seconds / 60)
seconds -= minutes * 60
return f"{str(hours).zfill(2)}:{str(minutes).zfill(2)}:{str(seconds).zfill(2)}"
@classmethod
def from_seconds(cls, timestamp_sec: int) -> "Timestamp":
"""
Parameters
----------
timestamp_sec
Timestamp in number of seconds
"""
return cls(timestamp_sec=timestamp_sec)
@classmethod
def from_str(cls, timestamp_str: str) -> "Timestamp":
"""
Parameters
----------
timestamp_str
Timestamp in the form of "HH:MM:SS"
Raises
------
ValueError
Invalid timestamp string format
"""
hour_minute_second = cls._normalize_timestamp_str(timestamp_str).split(":")
if len(hour_minute_second) != 3:
raise ValueError("Youtube timestamp must be in the form of 'HH:MM:SS'")
hour, minute, second = tuple(x for x in hour_minute_second)
try:
return cls(timestamp_sec=(int(hour) * 3600) + (int(minute) * 60) + int(second))
except ValueError as cast_exception:
raise ValueError(
"Youtube timestamp must be in the form of 'HH:MM:SS'"
) from cast_exception
class Chapters:
"""
Represents a list of (timestamps, titles)
"""
def __init__(
self,
timestamps: List[Timestamp],
titles: List[str],
):
self.timestamps = timestamps
self.titles = titles
for idx in range(len(timestamps) - 1):
if timestamps[idx].timestamp_sec >= timestamps[idx + 1].timestamp_sec:
raise ValueError("Timestamps must be in ascending order")
def contains_zero_timestamp(self) -> bool:
"""
Returns
-------
True if the first timestamp starts at 0. False otherwise.
"""
return self.timestamps[0].timestamp_sec == 0
@classmethod
def from_file(cls, chapters_file_path: str) -> "Chapters":
"""
Parameters
----------
chapters_file_path
Path to file containing chapters
Raises
------
ValidationException
File path does not exist or contains invalid formatting
"""
if not os.path.isfile(chapters_file_path):
raise ValidationException(
f"chapter/timestamp file path '{chapters_file_path}' does not exist."
)
with open(chapters_file_path, "r", encoding="utf-8") as file:
lines = file.readlines()
timestamps: List[Timestamp] = []
titles: List[str] = []
for idx, line in enumerate(lines):
line_split = line.strip().split(maxsplit=1)
# Allow the last line to be blank
if idx == len(lines) - 1 and not line.strip():
break
if len(line_split) != 2:
raise ValidationException(
f"Chapter/Timestamp file '{chapters_file_path}' could not parse '{line}': "
f"must be in the format of 'HH:MM:SS title"
)
timestamp_str, title = tuple(x for x in line_split)
timestamps.append(Timestamp.from_str(timestamp_str))
titles.append(title)
return cls(timestamps=timestamps, titles=titles)

96
src/ytdl_sub/utils/ffmpeg.py
View file

@ -1,11 +1,25 @@
import shutil
import subprocess
import tempfile
from typing import List
from typing import Optional
from ytdl_sub.utils.chapters import Chapters
from ytdl_sub.utils.exceptions import ValidationException
from ytdl_sub.utils.logger import Logger
logger = Logger.get(name="ffmpeg")
_FFMPEG_METADATA_SPECIAL_CHARS = ["=", ";", "#", "\n", "\\"]
def _ffmpeg_metadata_escape(str_to_escape: str) -> str:
# backslash at the end of the list is intentional
for special_char in _FFMPEG_METADATA_SPECIAL_CHARS:
str_to_escape.replace(special_char, f"\\{special_char}")
return str_to_escape
class FFMPEG:
@classmethod
@ -33,3 +47,85 @@ class FFMPEG:
cmd.extend(ffmpeg_args)
logger.debug("Running %s", " ".join(cmd))
subprocess.run(cmd, check=True)
def _create_metadata_chapter_entry(start_sec: int, end_sec: int, title: str) -> List[str]:
return [
"",
"[CHAPTER]",
"TIMEBASE=1/1000",
f"START={start_sec * 1000}",
f"END={end_sec * 1000}",
f"title={_ffmpeg_metadata_escape(title)}",
]
def _create_metadata_chapters(chapters: Chapters, file_duration_sec: int) -> List[str]:
lines: List[str] = []
if not chapters.contains_zero_timestamp():
lines += _create_metadata_chapter_entry(
start_sec=0,
end_sec=chapters.timestamps[0].timestamp_sec,
title="Intro", # TODO: make this configurable
)
for idx in range(len(chapters.timestamps) - 1):
lines += _create_metadata_chapter_entry(
start_sec=chapters.timestamps[idx].timestamp_sec,
end_sec=chapters.timestamps[idx + 1].timestamp_sec,
title=chapters.titles[idx],
)
# Add the last chapter using the file duration
lines += _create_metadata_chapter_entry(
start_sec=chapters.timestamps[-1].timestamp_sec,
end_sec=file_duration_sec,
title=chapters.titles[-1],
)
return lines
def add_ffmpeg_metadata(
file_path: str, chapters: Optional[Chapters], file_duration_sec: int
) -> None:
"""
Adds ffmetadata to a file. TODO: support more than just chapters
Parameters
----------
file_path
Full path to the file to add metadata to
chapters
Chapters to embed in the file. If a chapter for 0:00 does not exist, one is created
file_duration_sec
Length of the file in seconds
"""
lines = [";FFMETADATA1"]
if chapters:
lines += _create_metadata_chapters(chapters=chapters, file_duration_sec=file_duration_sec)
file_path_ext = file_path.split(".")[-1]
output_file_path = f"{file_path}.out.{file_path_ext}"
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", encoding="utf-8") as metadata_file:
metadata_file.write("\n".join(lines))
metadata_file.flush()
FFMPEG.run(
[
"-i",
file_path,
"-i",
metadata_file.name,
"-map_metadata",
"1",
"-bitexact", # for reproducibility
"-codec",
"copy",
output_file_path,
]
)
shutil.move(src=output_file_path, dst=file_path)

4
src/ytdl_sub/utils/thumbnail.py
View file

@ -10,8 +10,8 @@ from ytdl_sub.entries.entry import Entry
def _get_downloaded_thumbnail_path(entry: Entry) -> Optional[str]:
thumbnails = entry.kwargs("thumbnails")
possible_thumbnail_exts = set()
thumbnails = entry.kwargs("thumbnails") or []
possible_thumbnail_exts = {"jpg", "webp"} # Always check for jpg and webp thumbs
# The source `thumbnail` value and the actual downloaded thumbnail extension sometimes do
# not match. Find all possible extensions by checking all available thumbnails.

23
tests/e2e/expected_download.py
View file

@ -2,15 +2,21 @@ import hashlib
import os.path
from pathlib import Path
from typing import Dict
from typing import List
from typing import Optional
from typing import Union
class ExpectedDownload:
"""
To test ytdl-sub downloads work, we compare each downloaded file's md5 hash to an
expected md5 hash defined in this class
expected md5 hash defined in this class.
If the hash value is None, only assert the file exists. If the hash value is a list,
try all the hashes (used in case the GitHub env produces different deterministic value).
"""
def __init__(self, expected_md5_file_hashes: Dict[Path, str]):
def __init__(self, expected_md5_file_hashes: Dict[Path, Optional[Union[str, List[str]]]]):
self.expected_md5_file_hashes = expected_md5_file_hashes
@property
@ -39,9 +45,16 @@ class ExpectedDownload:
full_path
), f"Expected {str(relative_path)} to be a file but it is not"
if expected_md5_hash is None:
continue
with open(full_path, "rb") as file:
md5_hash = hashlib.md5(file.read()).hexdigest()
assert (
md5_hash == expected_md5_hash
), f"MD5 hash for {str(relative_path)} does not match"
if isinstance(expected_md5_hash, str):
expected_md5_hash = [expected_md5_hash]
assert md5_hash in expected_md5_hash, (
f"MD5 hash for {str(relative_path)} does not match: "
f"{md5_hash} != {expected_md5_hash}"
)

2
tests/e2e/youtube/test_channel_as_kodi_tv_show.py
View file

@ -88,7 +88,7 @@ def expected_full_channel_download():
Path("pz/Season 2011/s2011.e0201 - Jesse's Minecraft Server [Trailer - Feb.1].mp4"): "e66287b9832277b6a4d1554e29d9fdcc",
Path("pz/Season 2011/s2011.e0201 - Jesse's Minecraft Server [Trailer - Feb.1].nfo"): "f7c0de89038f8c491bded8a3968720a2",
Path("pz/Season 2011/s2011.e0227 - Jesse's Minecraft Server [Trailer - Feb.27].jpg"): "2e58e4d5f06ce5d1c3336fa493470135",
Path("pz/Season 2011/s2011.e0227 - Jesse's Minecraft Server [Trailer - Feb.27].jpg"): None,
Path("pz/Season 2011/s2011.e0227 - Jesse's Minecraft Server [Trailer - Feb.27].mp4"): "04ab5cb3cc12325d0c96a7cd04a8b91d",
Path("pz/Season 2011/s2011.e0227 - Jesse's Minecraft Server [Trailer - Feb.27].nfo"): "ee1eda78fa0980bc703e602b5012dd1f",

88
tests/e2e/youtube/test_merge_playlist.py Normal file
View file

@ -0,0 +1,88 @@
from pathlib import Path
import pytest
from e2e.expected_download import ExpectedDownload
from ytdl_sub.config.config_file import ConfigFile
from ytdl_sub.config.preset import Preset
from ytdl_sub.subscriptions.subscription import Subscription
@pytest.fixture
def config_path():
return "examples/kodi_music_videos_config.yaml"
@pytest.fixture
def subscription_name():
return "jmc"
@pytest.fixture
def config(config_path):
return ConfigFile.from_file_path(config_path=config_path)
@pytest.fixture
def subscription_dict(output_directory, subscription_name):
return {
"preset": "yt_music_video_playlist",
"youtube": {
"download_strategy": "merge_playlist",
"playlist_url": "https://youtube.com/playlist?list=PL5BC0FC26BECA5A35",
"add_chapters": True,
},
# override the output directory with our fixture-generated dir
"output_options": {"output_directory": output_directory},
# download the worst format so it is fast
"ytdl_options": {
"format": "best[height<=480]",
"postprocessor_args": {"ffmpeg": ["-bitexact"]}, # Must add this for reproducibility
},
"overrides": {"artist": "JMC"},
}
####################################################################################################
# PLAYLIST FIXTURES
@pytest.fixture
def playlist_subscription(config, subscription_name, subscription_dict):
playlist_preset = Preset.from_dict(
config=config,
preset_name=subscription_name,
preset_dict=subscription_dict,
)
return Subscription.from_preset(
preset=playlist_preset,
config=config,
)
@pytest.fixture
def expected_playlist_download():
return ExpectedDownload(
expected_md5_file_hashes={
Path("JMC - Jesse's Minecraft Server.jpg"): "348e3007fc590d0b1e2f6682501b0b5f",
Path("JMC - Jesse's Minecraft Server.mkv"): [
"6053c47a8690519b0a33c13fa4b01ac0",
"3ab42b3e6be0a44deb3a9a28e6ebaf16",
],
Path("JMC - Jesse's Minecraft Server.nfo"): "10df5dcdb65ab18ecf21b3503c77e48b",
}
)
class TestYoutubeMergePlaylist:
"""
Downloads my old minecraft youtube channel, pretends they are music videos. Ensure the above
files exist and have the expected md5 file hashes.
"""
def test_playlist_download(
self, playlist_subscription, expected_playlist_download, output_directory
):
playlist_subscription.download()
expected_playlist_download.assert_files_exist(relative_directory=output_directory)

22
tests/e2e/youtube/test_playlist_as_kodi_music_videos.py
View file

@ -66,20 +66,20 @@ def expected_playlist_download():
return ExpectedDownload(
expected_md5_file_hashes={
# Download mapping
Path(".ytdl-sub-jmc-download-archive.json"): "7541aa75606b86bff5ff276895520cf0",
Path(".ytdl-sub-jmc-download-archive.json"): "d8e784353c7c3006cb755a034c965160",
# Entry files
Path("JMC - Jesse's Minecraft Server [Trailer - Feb.1].jpg"): "048a19cf0f674437351872c3f312ebf1",
Path("JMC - Jesse's Minecraft Server [Trailer - Feb.1].jpg"): None,
Path("JMC - Jesse's Minecraft Server [Trailer - Feb.1].mp4"): "e66287b9832277b6a4d1554e29d9fdcc",
Path("JMC - Jesse's Minecraft Server [Trailer - Feb.1].nfo"): "3d272fe58487b6011ad049b6000b046f",
Path("JMC - Given to Fly.jpg"): "2e58e4d5f06ce5d1c3336fa493470135",
Path("JMC - Given to Fly.mp4"): "04ab5cb3cc12325d0c96a7cd04a8b91d",
Path("JMC - Given to Fly.nfo"): "0dc578bf5f1ceb6e069a57d329894f35",
Path("JMC - Jesse's Minecraft Server [Trailer - Feb.27].jpg"): None,
Path("JMC - Jesse's Minecraft Server [Trailer - Feb.27].mp4"): "04ab5cb3cc12325d0c96a7cd04a8b91d",
Path("JMC - Jesse's Minecraft Server [Trailer - Feb.27].nfo"): "6f99af10bef67276a507d1d9770c5e92",
Path("JMC - Indifference (Remastered).jpg"): "9baaddc6b62f5b9ae3781eb4eef0e3b3",
Path("JMC - Indifference (Remastered).mp4"): "025de6099a5c98e6397153c7a62d517d",
Path("JMC - Indifference (Remastered).nfo"): "061b86d9dc8fb39d39feab3292dafeb0",
Path("JMC - Jesse's Minecraft Server [Trailer - Mar.21].jpg"): None,
Path("JMC - Jesse's Minecraft Server [Trailer - Mar.21].mp4"): "025de6099a5c98e6397153c7a62d517d",
Path("JMC - Jesse's Minecraft Server [Trailer - Mar.21].nfo"): "beec3c1326654bd8c858cecf4e40977a",
}
)
# fmt: on
@ -120,9 +120,9 @@ def expected_single_video_download():
# fmt: off
return ExpectedDownload(
expected_md5_file_hashes={
Path("JMC - Whale & Wasp.jpg"): "b58377dfe7c39527e1990a24b36bbd77",
Path("JMC - Whale & Wasp.mp4"): "931a705864c57d21d6fedebed4af6bbc",
Path("JMC - Whale & Wasp.nfo"): "6c2f085adb847c1dcc47c19514c454d8",
Path("JMC - Oblivion Mod 'Falcor' p.1.jpg"): None,
Path("JMC - Oblivion Mod 'Falcor' p.1.mp4"): "931a705864c57d21d6fedebed4af6bbc",
Path("JMC - Oblivion Mod 'Falcor' p.1.nfo"): "89f509a8a3d9003e22a9091abeeae5dc",
}
)
# fmt: on

22
tests/unit/utils/test_chapters.py Normal file
View file

@ -0,0 +1,22 @@
import pytest
from ytdl_sub.utils.chapters import Timestamp
class TestTimestamp:
@pytest.mark.parametrize(
"timestamp_str, timestamp_int",
[
("0:00", 0),
("0:24", 24),
("1:11", 71),
("01:11", 71),
("00:22", 22),
("1:01:01", 3600 + 60 + 1),
("01:01:01", 3600 + 60 + 1),
("00:00:00", 0),
],
)
def test_timestamp_from_str(self, timestamp_str, timestamp_int):
ts = Timestamp.from_str(timestamp_str=timestamp_str)
assert ts.timestamp_sec == timestamp_int