added playlist items concurrency support.

README.md

@@ -275,7 +275,7 @@ If you feel like donating and appreciate my work, you can do so by donating to c
 Certain configuration values can be set via environment variables, using the `-e` parameter on the docker command line, or the `environment:` section in `compose.yaml` file.
 
 | Environment Variable           | Description                                                        | Default                            |
-| ------------------------- | ------------------------------------------------------------------ | ---------------------------------- |
+| ------------------------------ | ------------------------------------------------------------------ | ---------------------------------- |
 | YTP_OUTPUT_TEMPLATE            | The template for the filenames of the downloaded videos            | `%(title)s.%(ext)s`                |
 | YTP_DEFAULT_PRESET             | The default preset to use for the download                         | `default`                          |
 | YTP_INSTANCE_TITLE             | The title of the instance                                          | `empty string`                     |
@@ -313,4 +313,5 @@ Certain configuration values can be set via environment variables, using the `-e
 | YTP_BASE_PATH                  | Set this if you are serving YTPTube from sub-folder                | `/`                                |
 | YTP_PREVENT_LIVE_PREMIERE      | Prevents the initial youtube premiere stream from being downloaded | `false`                            |
 | YTP_TASKS_HANDLER_TIMER        | The cron expression for the tasks handler timer.                   | `*15 */1 * * *`                    |
+| YTP_PLAYLIST_ITEMS_CONCURRENCY | The number of playlist items be to processed at same time.         | `1`                                |
 

app/library/DownloadQueue.py

@@ -202,6 +202,54 @@ class DownloadQueue(metaclass=Singleton):
                 LOG.error(f"Failed to cancel downloads. {e!s}")
 
     async def _process_playlist(self, entry: dict, item: Item, already=None):
+        if 1 == self.config.playlist_items_concurrency:
+            return await self._process_playlist_old(entry=entry, item=item, already=already)
+
+        LOG.info(f"Playlist '{entry.get('id')}: {entry.get('title')}' processing.")
+        entries = entry.get("entries", [])
+        playlistCount = int(entry.get("playlist_count", len(entries)))
+        results = []
+
+        semaphore = asyncio.Semaphore(self.config.playlist_items_concurrency)
+
+        async def process_entry(i, etr):
+            extras = {
+                "playlist": entry.get("id"),
+                "playlist_index": f"{{0:0{len(str(playlistCount))}d}}".format(i),
+                "playlist_autonumber": i,
+            }
+
+            for property in ("id", "title", "uploader", "uploader_id"):
+                if property in entry:
+                    extras[f"playlist_{property}"] = entry.get(property)
+
+            LOG.debug(f"Processing entry {i}/{playlistCount} - ID: {etr.get('id')} - Title: {etr.get('title')}")
+
+            if "thumbnail" not in etr and "youtube:" in entry.get("extractor", ""):
+                extras["thumbnail"] = f"https://img.youtube.com/vi/{etr['id']}/maxresdefault.jpg"
+
+            async with semaphore:
+                return await self.add(
+                    item=item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras),
+                    already=already,
+                )
+
+        tasks = [process_entry(i, etr) for i, etr in enumerate(entries, start=1)]
+        results = await asyncio.gather(*tasks)
+
+        LOG.info(
+            f"Playlist '{entry.get('id')}: {entry.get('title')}' processing completed with '{len(results)}' entries."
+        )
+
+        if any("error" == res["status"] for res in results):
+            return {
+                "status": "error",
+                "msg": ", ".join(res["msg"] for res in results if res["status"] == "error" and "msg" in res),
+            }
+
+        return {"status": "ok"}
+
+    async def _process_playlist_old(self, entry: dict, item: Item, already=None):
         LOG.info(f"Playlist '{entry.get('id')}: {entry.get('title')}' processing.")
         entries = entry.get("entries", [])
         playlistCount = int(entry.get("playlist_count", len(entries)))
@@ -221,6 +269,8 @@ class DownloadQueue(metaclass=Singleton):
             if "thumbnail" not in etr and "youtube:" in entry.get("extractor", ""):
                 extras["thumbnail"] = f"https://img.youtube.com/vi/{etr['id']}/maxresdefault.jpg"
 
+            LOG.debug(f"Processing entry {i}/{playlistCount} - ID: {etr.get('id')} - Title: {etr.get('title')}")
+
             results.append(
                 await self.add(
                     item=item.new_with(url=etr.get("url") or etr.get("webpage_url"), extras=extras),

app/library/config.py

@@ -167,6 +167,9 @@ class Config:
     prevent_live_premiere: bool = False
     """Prevent downloading of the initial premiere live broadcast."""
 
+    playlist_items_concurrency: int = 1
+    """The number of concurrent playlist items to be processed at same time."""
+
     pictures_backends: list[str] = [
         "https://unsplash.it/1920/1080?random",
         "https://picsum.photos/1920/1080",
@@ -199,6 +202,7 @@ class Config:
         "socket_timeout",
         "extract_info_timeout",
         "debugpy_port",
+        "playlist_items_concurrency",
     )
     "The variables that are integers."