youtube.download_strategy split_video

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

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)