Move download logic into own file

src/ytdl_sub/downloaders/downloader.py

@@ -1,13 +1,8 @@
 import abc
 import contextlib
-import copy
-import json
 import os
-import time
 from abc import ABC
-from contextlib import contextmanager
 from pathlib import Path
-from typing import Callable
 from typing import Dict
 from typing import Generic
 from typing import Iterable
@@ -19,17 +14,14 @@ from typing import Tuple
 from typing import Type
 from typing import TypeVar
 
-import yt_dlp as ytdl
-from yt_dlp.utils import ExistingVideoReached
-from yt_dlp.utils import MaxDownloadsReached
-from yt_dlp.utils import RejectedVideoReached
-
 from ytdl_sub.config.preset_options import AddsVariablesMixin
 from ytdl_sub.config.preset_options import Overrides
 from ytdl_sub.downloaders.generic.validators import MultiUrlValidator
 from ytdl_sub.downloaders.generic.validators import UrlThumbnailListValidator
 from ytdl_sub.downloaders.generic.validators import UrlValidator
 from ytdl_sub.downloaders.ytdl_options_builder import YTDLOptionsBuilder
+from ytdl_sub.downloaders.ytdlp import extract_info_via_info_json
+from ytdl_sub.downloaders.ytdlp import extract_info_with_retry
 from ytdl_sub.entries.entry import Entry
 from ytdl_sub.entries.entry_parent import EntryParent
 from ytdl_sub.entries.variables.kwargs import COMMENTS
@@ -39,14 +31,11 @@ from ytdl_sub.entries.variables.kwargs import REQUESTED_SUBTITLES
 from ytdl_sub.entries.variables.kwargs import SOURCE_ENTRY
 from ytdl_sub.entries.variables.kwargs import SPONSORBLOCK_CHAPTERS
 from ytdl_sub.entries.variables.kwargs import UPLOAD_DATE_INDEX
-from ytdl_sub.thread.log_entries_downloaded_listener import LogEntriesDownloadedListener
-from ytdl_sub.utils.exceptions import FileNotDownloadedException
 from ytdl_sub.utils.file_handler import FileHandler
-from ytdl_sub.utils.file_handler import FileMetadata
 from ytdl_sub.utils.logger import Logger
 from ytdl_sub.utils.thumbnail import ThumbnailTypes
 from ytdl_sub.utils.thumbnail import convert_download_thumbnail
-from ytdl_sub.utils.thumbnail import convert_url_thumbnail
+from ytdl_sub.utils.thumbnail import download_and_convert_url_thumbnail
 from ytdl_sub.validators.strict_dict_validator import StrictDictValidator
 from ytdl_sub.ytdl_additions.enhanced_download_archive import DownloadArchiver
 from ytdl_sub.ytdl_additions.enhanced_download_archive import EnhancedDownloadArchive
@@ -132,9 +121,6 @@ class YtDlpDownloader(BaseDownloader[DownloaderOptionsT], ABC):
     and should translate that to list of Entry objects.
     """
 
-    _extract_entry_num_retries: int = 5
-    _extract_entry_retry_wait_sec: int = 5
-
     @classmethod
     def ytdl_option_defaults(cls) -> Dict:
         """
@@ -203,18 +189,6 @@ class YtDlpDownloader(BaseDownloader[DownloaderOptionsT], ABC):
             .to_dict()
         )
 
-    @classmethod
-    @contextmanager
-    def ytdl_downloader(cls, ytdl_options_overrides: Dict) -> ytdl.YoutubeDL:
-        """
-        Context manager to interact with yt_dlp.
-        """
-        download_logger.debug("ytdl_options: %s", str(ytdl_options_overrides))
-        with Logger.handle_external_logs(name="yt-dlp"):
-            # Deep copy ytdl_options in case yt-dlp modifies the dict
-            with ytdl.YoutubeDL(copy.deepcopy(ytdl_options_overrides)) as ytdl_downloader:
-                yield ytdl_downloader
-
     @property
     def is_dry_run(self) -> bool:
         """
@@ -233,177 +207,6 @@ class YtDlpDownloader(BaseDownloader[DownloaderOptionsT], ABC):
         """
         return self.download_ytdl_options.get("writethumbnail", False)
 
-    def extract_info(self, ytdl_options_overrides: Dict, **kwargs) -> Dict:
-        """
-        Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info
-        All kwargs will passed to the extract_info function.
-
-        Parameters
-        ----------
-        ytdl_options_overrides
-            Optional. Dict containing ytdl args to override other predefined ytdl args
-        **kwargs
-            arguments passed directory to YoutubeDL extract_info
-        """
-        with self.ytdl_downloader(ytdl_options_overrides) as ytdl_downloader:
-            return ytdl_downloader.extract_info(**kwargs)
-
-    def extract_info_with_retry(
-        self,
-        ytdl_options_overrides: Dict,
-        is_downloaded_fn: Optional[Callable[[], bool]] = None,
-        is_thumbnail_downloaded_fn: Optional[Callable[[], bool]] = None,
-        **kwargs,
-    ) -> Dict:
-        """
-        Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info
-        All kwargs will passed to the extract_info function.
-
-        This should be used when downloading a single entry. Checks if the entry's video
-        and thumbnail files exist - retry if they do not.
-
-        Parameters
-        ----------
-        ytdl_options_overrides
-            Dict containing ytdl args to override other predefined ytdl args
-        is_downloaded_fn
-            Optional. Function to check if the entry is downloaded
-        is_thumbnail_downloaded_fn
-            Optional. Function to check if the entry thumbnail is downloaded
-        **kwargs
-            arguments passed directory to YoutubeDL extract_info
-
-        Raises
-        ------
-        FileNotDownloadedException
-            If the entry fails to download
-        """
-        num_tries = 0
-        entry_files_exist = False
-        copied_ytdl_options_overrides = copy.deepcopy(ytdl_options_overrides)
-
-        while not entry_files_exist and num_tries < self._extract_entry_num_retries:
-            entry_dict = self.extract_info(
-                ytdl_options_overrides=copied_ytdl_options_overrides, **kwargs
-            )
-
-            is_downloaded = is_downloaded_fn is None or is_downloaded_fn()
-            is_thumbnail_downloaded = (
-                is_thumbnail_downloaded_fn is None or is_thumbnail_downloaded_fn()
-            )
-
-            if is_downloaded and is_thumbnail_downloaded:
-                return entry_dict
-
-            # If the video file is downloaded but the thumbnail is not, then do not download
-            # the video again
-            if is_downloaded and not is_thumbnail_downloaded:
-                copied_ytdl_options_overrides["skip_download"] = True
-                copied_ytdl_options_overrides["writethumbnail"] = True
-
-            time.sleep(self._extract_entry_retry_wait_sec)
-            num_tries += 1
-
-            # Remove the download archive so it can retry without thinking its already downloaded,
-            # even though it is not
-            if "download_archive" in copied_ytdl_options_overrides:
-                del copied_ytdl_options_overrides["download_archive"]
-
-            if num_tries < self._extract_entry_retry_wait_sec:
-                download_logger.debug(
-                    "Failed to download entry. Retrying %d / %d",
-                    num_tries,
-                    self._extract_entry_num_retries,
-                )
-
-        error_dict = {"ytdl_options": ytdl_options_overrides, "kwargs": kwargs}
-        raise FileNotDownloadedException(
-            f"yt-dlp failed to download an entry with these arguments: {error_dict}"
-        )
-
-    def _get_entry_dicts_from_info_json_files(self) -> List[Dict]:
-        """
-        Returns
-        -------
-        List of all info.json files read as JSON dicts
-        """
-        entry_dicts: List[Dict] = []
-        info_json_paths = [
-            Path(self.working_directory) / file_name
-            for file_name in os.listdir(self.working_directory)
-            if file_name.endswith(".info.json")
-        ]
-
-        for info_json_path in info_json_paths:
-            with open(info_json_path, "r", encoding="utf-8") as file:
-                entry_dicts.append(json.load(file))
-
-        return entry_dicts
-
-    @contextlib.contextmanager
-    def _listen_and_log_downloaded_info_json(self, log_prefix: Optional[str]):
-        """
-        Context manager that starts a separate thread that listens for new .info.json files,
-        prints their titles as they appear
-        """
-        if not log_prefix:
-            yield
-            return
-
-        info_json_listener = LogEntriesDownloadedListener(
-            working_directory=self.working_directory,
-            log_prefix=log_prefix,
-        )
-
-        info_json_listener.start()
-
-        try:
-            yield
-        finally:
-            info_json_listener.complete = True
-
-    def extract_info_via_info_json(
-        self,
-        ytdl_options_overrides: Dict,
-        log_prefix_on_info_json_dl: Optional[str] = None,
-        **kwargs,
-    ) -> List[Dict]:
-        """
-        Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info with infojson enabled. Entry dicts
-        are extracted via reading all info.json files in the working directory rather than
-        from the output of extract_info.
-
-        This allows us to catch RejectedVideoReached and ExistingVideoReached exceptions, and
-        simply ignore while still being able to read downloaded entry metadata.
-
-        Parameters
-        ----------
-        ytdl_options_overrides
-            Dict containing ytdl args to override other predefined ytdl args
-        log_prefix_on_info_json_dl
-            Optional. Spin a new thread to listen for new info.json files. Log
-            f'{log_prefix_on_info_json_dl} {title}' when a new one appears
-        **kwargs
-            arguments passed directory to YoutubeDL extract_info
-        """
-        try:
-            with self._listen_and_log_downloaded_info_json(log_prefix=log_prefix_on_info_json_dl):
-                _ = self.extract_info(ytdl_options_overrides=ytdl_options_overrides, **kwargs)
-        except RejectedVideoReached:
-            download_logger.debug(
-                "RejectedVideoReached, stopping additional downloads "
-                "(Can be disable by setting `ytdl_options.break_on_reject` to False)."
-            )
-        except ExistingVideoReached:
-            download_logger.debug(
-                "ExistingVideoReached, stopping additional downloads. "
-                "(Can be disable by setting `ytdl_options.break_on_existing` to False)."
-            )
-        except MaxDownloadsReached:
-            download_logger.info("MaxDownloadsReached, stopping additional downloads.")
-
-        return self._get_entry_dicts_from_info_json_files()
-
     ###############################################################################################
     # DOWNLOAD FUNCTIONS
 
@@ -463,13 +266,13 @@ class YtDlpDownloader(BaseDownloader[DownloaderOptionsT], ABC):
                 FileHandler.delete(info_json_file)
 
     def _extract_entry_info_with_retry(self, entry: Entry) -> Entry:
-        download_entry_dict = self.extract_info_with_retry(
+        download_entry_dict = extract_info_with_retry(
+            ytdl_options_overrides=self.download_ytdl_options,
             is_downloaded_fn=None if self.is_dry_run else entry.is_downloaded,
             is_thumbnail_downloaded_fn=None
             if (self.is_dry_run or not self.is_entry_thumbnails_enabled)
             else entry.is_thumbnail_downloaded,
             url=entry.webpage_url,
-            ytdl_options_overrides=self.download_ytdl_options,
         )
         return Entry(download_entry_dict, working_directory=self.working_directory)
 
@@ -531,10 +334,11 @@ class YtDlpDownloader(BaseDownloader[DownloaderOptionsT], ABC):
         url = self.overrides.apply_formatter(collection_url.url)
 
         with self._separate_download_archives():
-            entry_dicts = self.extract_info_via_info_json(
+            entry_dicts = extract_info_via_info_json(
+                working_directory=self.working_directory,
                 ytdl_options_overrides=self.metadata_ytdl_options,
-                url=url,
                 log_prefix_on_info_json_dl="Downloading metadata for",
+                url=url,
             )
 
         parents = EntryParent.from_entry_dicts(
@@ -595,6 +399,16 @@ class YtDlpDownloader(BaseDownloader[DownloaderOptionsT], ABC):
                 yield entry
 
     def download(self, entry: Entry) -> Entry:
+        """
+        Parameters
+        ----------
+        entry
+            Entry to download
+
+        Returns
+        -------
+        The entry that was downloaded successfully
+        """
         download_logger.info(
             "Downloading entry %d/%d: %s",
             self._url_state.entries_downloaded,
@@ -624,33 +438,6 @@ class YtDlpDownloader(BaseDownloader[DownloaderOptionsT], ABC):
 
         return entry
 
-    @classmethod
-    def _download_thumbnail(
-        cls,
-        thumbnail_url: str,
-        output_thumbnail_path: str,
-    ) -> Optional[bool]:
-        """
-        Downloads a thumbnail and stores it in the output directory
-
-        Parameters
-        ----------
-        thumbnail_url:
-            Url of the thumbnail
-        output_thumbnail_path:
-            Path to store the thumbnail after downloading
-
-        Returns
-        -------
-        True if the thumbnail converted. None if it is missing or failed.
-        """
-        if not thumbnail_url:
-            return None
-
-        return convert_url_thumbnail(
-            thumbnail_url=thumbnail_url, output_thumbnail_path=output_thumbnail_path
-        )
-
     def _download_parent_thumbnails(
         self,
         thumbnail_list_info: UrlThumbnailListValidator,
@@ -687,7 +474,7 @@ class YtDlpDownloader(BaseDownloader[DownloaderOptionsT], ABC):
                 download_logger.debug("Failed to find thumbnail id '%s'", thumbnail_id)
                 continue
 
-            if self._download_thumbnail(
+            if download_and_convert_url_thumbnail(
                 thumbnail_url=thumbnail_url,
                 output_thumbnail_path=str(Path(self.working_directory) / thumbnail_name),
             ):

src/ytdl_sub/downloaders/ytdlp.py

@@ -0,0 +1,217 @@
+import contextlib
+import copy
+import json
+import os
+import time
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Callable
+from typing import Dict
+from typing import List
+from typing import Optional
+
+import yt_dlp as ytdl
+from yt_dlp.utils import ExistingVideoReached
+from yt_dlp.utils import MaxDownloadsReached
+from yt_dlp.utils import RejectedVideoReached
+
+from ytdl_sub.thread.log_entries_downloaded_listener import LogEntriesDownloadedListener
+from ytdl_sub.utils.exceptions import FileNotDownloadedException
+from ytdl_sub.utils.logger import Logger
+
+download_logger = Logger.get(name="yt-dlp-downloader")
+
+_extract_entry_num_retries: int = 5
+_extract_entry_retry_wait_sec: int = 5
+
+
+@contextmanager
+def ytdlp_downloader(ytdl_options_overrides: Dict) -> ytdl.YoutubeDL:
+    """
+    Context manager to interact with yt_dlp.
+    """
+    download_logger.debug("ytdl_options: %s", str(ytdl_options_overrides))
+    with Logger.handle_external_logs(name="yt-dlp"):
+        # Deep copy ytdl_options in case yt-dlp modifies the dict
+        with ytdl.YoutubeDL(copy.deepcopy(ytdl_options_overrides)) as ytdl_downloader:
+            yield ytdl_downloader
+
+
+def extract_info(ytdl_options_overrides: Dict, **kwargs) -> Dict:
+    """
+    Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info
+    All kwargs will passed to the extract_info function.
+
+    Parameters
+    ----------
+    ytdl_options_overrides
+        Optional. Dict containing ytdl args to override other predefined ytdl args
+    **kwargs
+        arguments passed directory to YoutubeDL extract_info
+    """
+    with ytdlp_downloader(ytdl_options_overrides) as ytdlp:
+        return ytdlp.extract_info(**kwargs)
+
+
+def extract_info_with_retry(
+    ytdl_options_overrides: Dict,
+    is_downloaded_fn: Optional[Callable[[], bool]] = None,
+    is_thumbnail_downloaded_fn: Optional[Callable[[], bool]] = None,
+    **kwargs,
+) -> Dict:
+    """
+    Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info
+    All kwargs will passed to the extract_info function.
+
+    This should be used when downloading a single entry. Checks if the entry's video
+    and thumbnail files exist - retry if they do not.
+
+    Parameters
+    ----------
+    ytdl_options_overrides
+        Dict containing ytdl args to override other predefined ytdl args
+    is_downloaded_fn
+        Optional. Function to check if the entry is downloaded
+    is_thumbnail_downloaded_fn
+        Optional. Function to check if the entry thumbnail is downloaded
+    **kwargs
+        arguments passed directory to YoutubeDL extract_info
+
+    Raises
+    ------
+    FileNotDownloadedException
+        If the entry fails to download
+    """
+    num_tries = 0
+    entry_files_exist = False
+    copied_ytdl_options_overrides = copy.deepcopy(ytdl_options_overrides)
+
+    while not entry_files_exist and num_tries < _extract_entry_num_retries:
+        entry_dict = extract_info(ytdl_options_overrides=copied_ytdl_options_overrides, **kwargs)
+
+        is_downloaded = is_downloaded_fn is None or is_downloaded_fn()
+        is_thumbnail_downloaded = is_thumbnail_downloaded_fn is None or is_thumbnail_downloaded_fn()
+
+        if is_downloaded and is_thumbnail_downloaded:
+            return entry_dict
+
+        # If the video file is downloaded but the thumbnail is not, then do not download
+        # the video again
+        if is_downloaded and not is_thumbnail_downloaded:
+            copied_ytdl_options_overrides["skip_download"] = True
+            copied_ytdl_options_overrides["writethumbnail"] = True
+
+        time.sleep(_extract_entry_retry_wait_sec)
+        num_tries += 1
+
+        # Remove the download archive so it can retry without thinking its already downloaded,
+        # even though it is not
+        if "download_archive" in copied_ytdl_options_overrides:
+            del copied_ytdl_options_overrides["download_archive"]
+
+        if num_tries < _extract_entry_num_retries:
+            download_logger.debug(
+                "Failed to download entry. Retrying %d / %d",
+                num_tries,
+                _extract_entry_num_retries,
+            )
+
+    error_dict = {"ytdl_options": ytdl_options_overrides, "kwargs": kwargs}
+    raise FileNotDownloadedException(
+        f"yt-dlp failed to download an entry with these arguments: {error_dict}"
+    )
+
+
+def _get_entry_dicts_from_info_json_files(working_directory: str) -> List[Dict]:
+    """
+    Parameters
+    ----------
+    working_directory
+        Directory that info json files are located
+
+    Returns
+    -------
+    List of all info.json files read as JSON dicts
+    """
+    entry_dicts: List[Dict] = []
+    info_json_paths = [
+        Path(working_directory) / file_name
+        for file_name in os.listdir(working_directory)
+        if file_name.endswith(".info.json")
+    ]
+
+    for info_json_path in info_json_paths:
+        with open(info_json_path, "r", encoding="utf-8") as file:
+            entry_dicts.append(json.load(file))
+
+    return entry_dicts
+
+
+@contextlib.contextmanager
+def _listen_and_log_downloaded_info_json(working_directory: str, log_prefix: Optional[str]):
+    """
+    Context manager that starts a separate thread that listens for new .info.json files,
+    prints their titles as they appear
+    """
+    if not log_prefix:
+        yield
+        return
+
+    info_json_listener = LogEntriesDownloadedListener(
+        working_directory=working_directory,
+        log_prefix=log_prefix,
+    )
+
+    info_json_listener.start()
+
+    try:
+        yield
+    finally:
+        info_json_listener.complete = True
+
+
+def extract_info_via_info_json(
+    working_directory: str,
+    ytdl_options_overrides: Dict,
+    log_prefix_on_info_json_dl: Optional[str] = None,
+    **kwargs,
+) -> List[Dict]:
+    """
+    Wrapper around yt_dlp.YoutubeDL.YoutubeDL.extract_info with infojson enabled. Entry dicts
+    are extracted via reading all info.json files in the working directory rather than
+    from the output of extract_info.
+
+    This allows us to catch RejectedVideoReached and ExistingVideoReached exceptions, and
+    simply ignore while still being able to read downloaded entry metadata.
+
+    Parameters
+    ----------
+    working_directory
+        Directory that info json files reside in
+    ytdl_options_overrides
+        Dict containing ytdl args to override other predefined ytdl args
+    log_prefix_on_info_json_dl
+        Optional. Spin a new thread to listen for new info.json files. Log
+        f'{log_prefix_on_info_json_dl} {title}' when a new one appears
+    **kwargs
+        arguments passed directory to YoutubeDL extract_info
+    """
+    try:
+        with _listen_and_log_downloaded_info_json(
+            working_directory=working_directory, log_prefix=log_prefix_on_info_json_dl
+        ):
+            _ = extract_info(ytdl_options_overrides=ytdl_options_overrides, **kwargs)
+    except RejectedVideoReached:
+        download_logger.debug(
+            "RejectedVideoReached, stopping additional downloads "
+            "(Can be disable by setting `ytdl_options.break_on_reject` to False)."
+        )
+    except ExistingVideoReached:
+        download_logger.debug(
+            "ExistingVideoReached, stopping additional downloads. "
+            "(Can be disable by setting `ytdl_options.break_on_existing` to False)."
+        )
+    except MaxDownloadsReached:
+        download_logger.info("MaxDownloadsReached, stopping additional downloads.")
+
+    return _get_entry_dicts_from_info_json_files(working_directory=working_directory)

src/ytdl_sub/utils/thumbnail.py

@@ -47,7 +47,9 @@ def convert_download_thumbnail(entry: Entry, error_if_not_found: bool = True) ->
 
 
 @retry(times=3, exceptions=(Exception,))
-def convert_url_thumbnail(thumbnail_url: str, output_thumbnail_path: str) -> Optional[bool]:
+def download_and_convert_url_thumbnail(
+    thumbnail_url: Optional[str], output_thumbnail_path: str
+) -> Optional[bool]:
     """
     Downloads and converts a thumbnail from a url into a jpg
 
@@ -62,6 +64,9 @@ def convert_url_thumbnail(thumbnail_url: str, output_thumbnail_path: str) -> Opt
     -------
     True to indicate it converted the thumbnail from url. None if the retry failed.
     """
+    if not thumbnail_url:
+        return None
+
     # timeout after 8 seconds
     with urlopen(thumbnail_url, timeout=1.0) as file:
         with tempfile.NamedTemporaryFile(delete=False) as thumbnail: