diff --git a/src/ytdl_sub/downloaders/youtube/__init__.py b/src/ytdl_sub/downloaders/youtube/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/ytdl_sub/downloaders/youtube/split_video.py b/src/ytdl_sub/downloaders/youtube/split_video.py new file mode 100644 index 00000000..7198168c --- /dev/null +++ b/src/ytdl_sub/downloaders/youtube/split_video.py @@ -0,0 +1,187 @@ +import copy +import os.path +import re +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.utils.exceptions import ValidationException +from ytdl_sub.utils.ffmpeg import FFMPEG +from ytdl_sub.validators.validators import StringValidator + +# 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) (.+)$") + + +def _parse_split_timestamp_file(split_timestamp_path: str) -> List[Tuple[str, str]]: + """ + Returns list of (00:00:00 title), each timestamp as HH:MM:SS + """ + 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() + + timestamp_titles: List[Tuple[str, 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 + timestamp_titles.append((timestamp, 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 timestamp_titles + + +############################################################################### +# Youtube split video downloader + options + + +class YoutubeSplitVideoDownloaderOptions(YoutubeVideoDownloaderOptions): + """ + 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, + not a subscription. + + Usage: + + .. code-block:: yaml + + presets: + example_preset: + youtube: + # required + download_strategy: "split_video" + video_url: "youtube.com/watch?v=VMAPTo7RVDo" + split_timestamps: path/to/timestamps.txt + + + CLI usage: + + .. code-block:: bash + + ytdl-sub dl \ + --preset "example_preset" \ + --youtube.video_url "youtube.com/watch?v=VMAPTo7RVDo" \ + --youtube.split_timestamps "path/to/timestamps.txt" + + ``split_timestamps`` file format: + + .. code-block:: txt + + 0:00 Intro + 0:24 Blackwater Park + 10:23 Bleak + 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. + """ + + _required_keys = super()._required_keys.union("split_timestamps") + + def __init__(self, name, value): + super().__init__(name, value) + self._split_timestamps = self._validate_key("split_timestamps", StringValidator).value + + @property + def split_timestamps(self) -> str: + """ + Required. The path to the file containing the split timestamps. + """ + return self._split_timestamps + + +class YoutubeSplitVideoDownloader( + YoutubeDownloader[YoutubeSplitVideoDownloaderOptions, YoutubePlaylistVideo] +): + downloader_options_type = YoutubeSplitVideoDownloaderOptions + downloader_entry_type = YoutubePlaylistVideo + + @classmethod + def ytdl_option_defaults(cls) -> Dict: + """ + Default `ytdl_options`_ for ``split_video`` + + .. code-block:: yaml + + ytdl_options: + ignoreerrors: True # ignore errors like hidden videos, age restriction, etc + """ + return dict( + super().ytdl_option_defaults(), + **{"break_on_existing": True}, + ) + + def download(self) -> List[YoutubePlaylistVideo]: + """Download a single Youtube video, then split it into multiple videos""" + split_videos: List[YoutubePlaylistVideo] = [] + + timestamp_titles = _parse_split_timestamp_file( + split_timestamp_path=self.download_options.split_timestamps + ) + entry_dict = self.extract_info(url=self.download_options.video_url) + uid = entry_dict["id"] + ext = entry_dict["ext"] + + idx = 0 + for timestamp, title in timestamp_titles: + entry_dict_ = copy.deepcopy(entry_dict) + new_uid = f"{uid}___{idx}" + entry_dict_["title"] = title + entry_dict_["playlist_index"] = idx + 1 + entry_dict_["playlist_count"] = len(timestamp_titles) + entry_dict_["id"] = new_uid + + timestamp_begin_arg = f"-ss {timestamp}" + timestamp_end_arg = ( + f"-to {timestamp_titles[idx + 1][0]}" if idx + 1 < len(timestamp_titles) else "" + ) + + FFMPEG.run( + f"-i {uid}.{ext} " + f"{timestamp_begin_arg} {timestamp_end_arg} " + f"-vcodec copy -acodec copy {new_uid}.{ext}" + ) + + split_videos.append( + YoutubePlaylistVideo( + entry_dict=entry_dict, working_directory=self.working_directory + ) + ) + + return split_videos diff --git a/src/ytdl_sub/utils/ffmpeg.py b/src/ytdl_sub/utils/ffmpeg.py new file mode 100644 index 00000000..c8f742ff --- /dev/null +++ b/src/ytdl_sub/utils/ffmpeg.py @@ -0,0 +1,33 @@ +import subprocess + +from ytdl_sub.utils.exceptions import ValidationException +from ytdl_sub.utils.logger import Logger + +logger = Logger.get(name="ffmpeg") + + +class FFMPEG: + @classmethod + def _ensure_installed(cls): + try: + subprocess.check_output(["which", "ffmpeg"]) + except subprocess.CalledProcessError as subprocess_error: + raise ValidationException( + "Trying to use a feature which requires ffmpeg, but it cannot be found" + ) from subprocess_error + + @classmethod + def run(cls, ffmpeg_args: str) -> None: + """ + Runs an ffmpeg command. Should not include 'ffmpeg' as the beginning argument. + + Parameters + ---------- + ffmpeg_args: + Arguments to pass to ffmpeg. Each one will be separated by a space. + """ + cls._ensure_installed() + + cmd = f"ffmpeg {ffmpeg_args}" + logger.debug("Running %s", cmd) + subprocess.run(cmd, check=True)