Add support concurrent fragment video downloading

Author: code-yeongyu

twitter_video_tools/platform_video_downloader.py

@@ -1,4 +1,5 @@
-from typing import Optional
+from multiprocessing import cpu_count as get_cpu_count
+from typing import Optional, Union
 
 from playwright.sync_api import sync_playwright
 from yt_dlp.utils import YoutubeDLError
@@ -41,12 +42,16 @@ def download_twitter_video(
                 youtube_dl_option = self._make_youtube_dl_option(video_filename)
                 youtube_dl_wrapper.download([video_link], youtube_dl_option)
 
-    def _make_youtube_dl_option(self, video_filename: Optional[str] = None) -> dict[str, str]:
+    def _make_youtube_dl_option(self, video_filename: Optional[str] = None) -> dict[str, Union[str, int]]:
         youtube_dl_output_template = \
             f'{self.video_output_path}/{video_filename}' \
                 if video_filename else \
             f'{self.video_output_path}/%(title)s.%(ext)s'
-        return {'format': 'bestvideo/best', 'outtmpl': youtube_dl_output_template}
+        return {
+            'format': 'bestvideo/best',
+            'outtmpl': youtube_dl_output_template,
+            'concurrent_fragment_downloads': get_cpu_count(),
+        }
 
     def _get_twitter_private_videos_info(
         self,

twitter_video_tools/utils/execute_parallel.py

@@ -1,8 +1,8 @@
-import multiprocessing
 import multiprocessing.dummy
+from multiprocessing import cpu_count as get_cpu_count
 from typing import Any, Callable
 
 
 def execute_parallel(func: Callable[[str], None], *args: Any, **kwargs: Any) -> None:
-    pool = multiprocessing.dummy.Pool(multiprocessing.cpu_count())
+    pool = multiprocessing.dummy.Pool(get_cpu_count())
     pool.starmap(func, *args, **kwargs)

twitter_video_tools/utils/youtube_dl_wrapper.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, Union
 
 import yt_dlp
 
@@ -13,7 +13,7 @@ def _youtube_dl_download_video(self, link: str, youtube_dl_option: Optional[dict
             youtube_dl_downloader: yt_dlp.YoutubeDL
             youtube_dl_downloader.download([link])
 
-    def download(self, links: list[str], youtube_dl_option: Optional[dict[str, str]] = None) -> None:
+    def download(self, links: list[str], youtube_dl_option: Optional[dict[str, Union[str, int]]] = None) -> None:
         arguments = [(link, youtube_dl_option) for link in links]
         execute_parallel(self._youtube_dl_download_video, arguments)