sponsors

rwx/sw/yt_dlp/__init__.py

@@ -21,6 +21,8 @@ from rwx import Object
 from rwx.log import stream as log
 from yt_dlp import YoutubeDL
 
+URL = "https://youtube.com"
+
 
 class Tab(Object, ABC):
     """YouTube Tab."""
@@ -154,13 +156,88 @@ class Videos(Tab):
         return None
 
 
-def download(video_id: str | None) -> None:
+def download_video(video_id: str | None) -> None:
     if video_id:
-        Tab.yt_dl(
+        ytdl(
             {
                 "format": "bestvideo[ext=mp4]+bestaudio[ext=mp4]",
                 "outtmpl": "%(id)s.%(ext)s",
+                "postprocessors": [
+                    {
+                        "key": "SponsorBlock",
+                        "categories": ["sponsor"],
+                    },
+                    {
+                        "key": "ModifyChapters",
+                        "remove_sponsor_segments": ["sponsor"],
+                    },
+                ],
                 "writesubtitles": True,
                 "writethumbnail": True,
             }
-        ).download([f"{Tab.URL_ROOT}/watch?v={video_id}"])
+        ).download([f"{URL}/watch?v={video_id}"])
+
+
+def extract(opt: dict, url: str) -> dict:
+    """Return extracted dict.
+
+    :rtype: dict
+    """
+    return ytdl({
+        **opt,
+        "extract_flat": True,
+        "skip_download": True,
+    }).extract_info(url, download=False)
+
+
+def extract_channel(channel_id: str) -> dict:
+    """Return extracted channel dict.
+
+    :rtype: dict
+    """
+    d = extract({
+    }, f"{URL}/channel/{channel_id}")
+    log.info(d)
+    return {
+    }
+
+
+def extract_channel_videos(channel_id: str) -> list[str]:
+    """Return extracted channel videos dict.
+
+    :rtype: dict
+    """
+    d = extract({
+    }, f"{URL}/channel/{channel_id}/videos")
+    log.info(d)
+    return [entry["id"] for entry in reversed(d["entries"])]
+
+
+def extract_video(video_id: str) -> dict:
+    """Return extracted video dict.
+
+    :rtype: dict
+    """
+    d = extract({
+    }, f"{URL}/watch?v={video_id}")
+    log.info(d)
+    return {
+        "title": d["title"],
+    }
+
+
+def next_download(videos: list[str]) -> str | None:
+    for video_id in videos:
+        if not Path(f"{video_id}.mp4").exists():
+            return video_id
+    return None
+
+
+def ytdl(opt: dict) -> YoutubeDL:
+    options = {
+        **opt,
+        "ignoreerrors": False,
+        "quiet": False,
+    }
+    log.info(options)
+    return YoutubeDL(options)