youtube.download_strategy split_video (#76)

* youtube.download_strategy split_video

* almost working, need to deal with track_title and other overwrites

* working test

* ci ffmpeg

* sudo -f

* fix code block

.github/workflows/ci.yaml

@@ -21,6 +21,7 @@ jobs:
         python-version: ${{ matrix.python-version }}
     - name: Install dependencies
       run: |
+        sudo apt-get install -y ffmpeg
         python -m pip install --upgrade pip
         pip install -e .[lint,test]
     - name: Run linters

docs/config.rst

@@ -79,6 +79,17 @@ _____
 
 -------------------------------------------------------------------------------
 
+split_video
+___________
+.. autoclass:: ytdl_sub.downloaders.youtube.split_video.YoutubeSplitVideoDownloaderOptions()
+  :members:
+  :member-order: bysource
+  :inherited-members:
+
+.. autofunction:: ytdl_sub.downloaders.youtube.split_video.YoutubeSplitVideoDownloader.ytdl_option_defaults()
+
+-------------------------------------------------------------------------------
+
 soundcloud
 ''''''''''
 Download strategies for downloading music from Soundcloud. See

src/ytdl_sub/config/preset_class_mappings.py

@@ -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.split_video import YoutubeSplitVideoDownloader
 from ytdl_sub.downloaders.youtube_downloader import YoutubeChannelDownloader
 from ytdl_sub.downloaders.youtube_downloader import YoutubePlaylistDownloader
 from ytdl_sub.downloaders.youtube_downloader import YoutubeVideoDownloader
@@ -23,6 +24,7 @@ class DownloadStrategyMapping:
             "video": YoutubeVideoDownloader,
             "playlist": YoutubePlaylistDownloader,
             "channel": YoutubeChannelDownloader,
+            "split_video": YoutubeSplitVideoDownloader,
         },
         "soundcloud": {
             "albums_and_singles": SoundcloudAlbumsAndSinglesDownloader,

src/ytdl_sub/downloaders/youtube/split_video.py

@@ -0,0 +1,234 @@
+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.ffmpeg import FFMPEG
+from ytdl_sub.utils.thumbnail import convert_download_thumbnail
+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 _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
+) -> List[str]:
+    timestamp_begin = timestamps[idx]
+    timestamp_end = timestamps[idx + 1] if idx + 1 < len(timestamps) else ""
+
+    cmd = ["-i", input_file, "-ss", timestamp_begin]
+    if timestamp_end:
+        cmd += ["-to", timestamp_end]
+    cmd += ["-vcodec", "copy", "-acodec", "copy", output_file]
+    return cmd
+
+
+###############################################################################
+# 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:: markdown
+
+       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 = {"video_url", "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 _create_split_video_entry(
+        self, source_entry_dict: Dict, title: str, idx: int, split_video_count: int
+    ) -> YoutubePlaylistVideo:
+        """
+        Runs ffmpeg to create the split video
+        """
+        entry_dict = copy.deepcopy(source_entry_dict)
+        entry_dict["title"] = title
+        entry_dict["playlist_index"] = idx + 1
+        entry_dict["playlist_count"] = split_video_count
+        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"]
+
+        return YoutubePlaylistVideo(entry_dict=entry_dict, working_directory=self.working_directory)
+
+    def download(self) -> List[YoutubePlaylistVideo]:
+        """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
+        )
+        entry_dict = self.extract_info(url=self.download_options.video_url)
+
+        entry = YoutubeVideo(entry_dict=entry_dict, working_directory=self.working_directory)
+        # convert the entry thumbnail early so we do not have to guess the thumbnail extension
+        # when copying it
+        convert_download_thumbnail(entry=entry)
+
+        for idx, title in enumerate(titles):
+            new_uid = _split_video_uid(source_uid=entry.uid, idx=idx)
+
+            # Get the input/output file paths
+            input_file = entry.get_download_file_path()
+            output_file = str(Path(self.working_directory) / f"{new_uid}.{entry.ext}")
+            output_thumbnail_file = str(
+                Path(self.working_directory) / f"{new_uid}.{entry.thumbnail_ext}"
+            )
+
+            # 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
+                )
+            )
+            # Copy the thumbnail
+            copyfile(src=entry.get_download_thumbnail_path(), dst=output_thumbnail_file)
+
+            # Format the split video as a YoutubePlaylistVideo
+            split_videos.append(
+                self._create_split_video_entry(
+                    source_entry_dict=entry_dict,
+                    title=title,
+                    idx=idx,
+                    split_video_count=len(timestamps),
+                )
+            )
+
+        return split_videos

src/ytdl_sub/utils/ffmpeg.py

@@ -0,0 +1,35 @@
+import subprocess
+from typing import List
+
+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: List[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 = ["ffmpeg"]
+        cmd.extend(ffmpeg_args)
+        logger.debug("Running %s", " ".join(cmd))
+        subprocess.run(cmd, check=True)

tests/e2e/youtube/test_split_video.py

@@ -0,0 +1,126 @@
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+
+import mergedeep
+import pytest
+from conftest import assert_debug_log
+from e2e.expected_download import ExpectedDownload
+
+import ytdl_sub.downloaders.downloader
+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 split_timestamps_file_path():
+    timestamps = [
+        "0:00 Intro\n",
+        "00:10 Part 1\n",
+        "0:20 Part 2\n",
+        "00:30 Part 3\n",
+        "0:00:40 Part 4\n",
+        "00:01:01 Part 5\n",
+    ]
+
+    with NamedTemporaryFile(mode="w", encoding="utf-8", suffix=".txt") as tmp:
+        tmp.writelines(timestamps)
+        tmp.seek(0)
+        yield tmp.name
+
+
+@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, split_timestamps_file_path):
+    return {
+        "preset": "yt_music_video",
+        "youtube": {
+            "download_strategy": "split_video",
+            "video_url": "https://youtube.com/watch?v=HKTNxEqsN3Q",
+            "split_timestamps": split_timestamps_file_path,
+        },
+        # override the output directory with our fixture-generated dir
+        "output_options": {
+            "output_directory": output_directory,
+            "file_name": "{channel_sanitized} - {playlist_index}-{playlist_size}"
+            ".{title_sanitized}.{ext}",
+        },
+        # download the worst format so it is fast
+        "ytdl_options": {
+            "format": "worst[ext=mp4]",
+        },
+    }
+
+
+####################################################################################################
+# SINGLE VIDEO FIXTURES
+
+
+@pytest.fixture
+def single_video_subscription(config, subscription_name, subscription_dict):
+    single_video_preset = Preset.from_dict(
+        config=config,
+        preset_name=subscription_name,
+        preset_dict=subscription_dict,
+    )
+
+    return Subscription.from_preset(
+        preset=single_video_preset,
+        config=config,
+    )
+
+
+@pytest.fixture
+def expected_single_video_download():
+    # turn off black formatter here for readability
+    # fmt: off
+    return ExpectedDownload(
+        expected_md5_file_hashes={
+            Path('Project Zombie - 1-6.Intro.mp4'): "eaec6f50f364b13ef1a201e736ec9c05",
+            Path('Project Zombie - 2-6.Part 1.mp4'): "5850b19acb250cc13db36f80fa1bba5a",
+            Path('Project Zombie - 3-6.Part 2.mp4'): "445d95eba437db6df284df7e1ab633e8",
+            Path('Project Zombie - 4-6.Part 3.mp4'): "2b6e7532d515c9e64ed2a33d850cf199",
+            Path('Project Zombie - 5-6.Part 4.mp4'): "842bf3c4d1fcc4c5ab110635935dac66",
+            Path('Project Zombie - 6-6.Part 5.mp4'): "238de99f00f829ab72f042b79da9a33a",
+            Path('Project Zombie - Intro.jpg'): "e87282e4115baa8b5c727fb4de15316d",
+            Path('Project Zombie - Intro.nfo'): "ded59ac906f579312cc3cf98a57e7ea3",
+            Path('Project Zombie - Part 1.jpg'): "e87282e4115baa8b5c727fb4de15316d",
+            Path('Project Zombie - Part 1.nfo'): "70ff5cd0092b8bc22dc4db93a824789b",
+            Path('Project Zombie - Part 2.jpg'): "e87282e4115baa8b5c727fb4de15316d",
+            Path('Project Zombie - Part 2.nfo'): "54450c18a2cbb9d6d2ee5d0a1fb3f279",
+            Path('Project Zombie - Part 3.jpg'): "e87282e4115baa8b5c727fb4de15316d",
+            Path('Project Zombie - Part 3.nfo'): "0effb13fc4039363a95969d1048dde57",
+            Path('Project Zombie - Part 4.jpg'): "e87282e4115baa8b5c727fb4de15316d",
+            Path('Project Zombie - Part 4.nfo'): "74bd0d7c12105469838768a0cc323a8c",
+            Path('Project Zombie - Part 5.jpg'): "e87282e4115baa8b5c727fb4de15316d",
+            Path('Project Zombie - Part 5.nfo'): "a8cf2e77721335ea7c18e22734e7996c",
+        }
+    )
+    # fmt: on
+
+
+class TestPlaylistAsKodiMusicVideo:
+    """
+    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_single_video_download(
+        self, single_video_subscription, expected_single_video_download, output_directory
+    ):
+        single_video_subscription.download()
+        expected_single_video_download.assert_files_exist(relative_directory=output_directory)