rwx/rwx/sw/ytdlp/__init__.py

"""YouTube DownLoad."""

from datetime import datetime
from pathlib import Path
from typing import Any

from yt_dlp import YoutubeDL

from rwx import Object
from rwx.log import stream as log

URL = "https://youtube.com"


# ╭─────────╮
# │ classes │
# ╰─────────╯


class Cache(Object):
    """YouTube local cache."""

    def __init__(self, root: Path) -> None:
        self.root = root


class Channel(Object):
    """YouTube channel."""

    def __init__(self, channel_id: str) -> None:
        """Set objects tree.

        :param channel_id: channel identifier
        :type channel_id: str
        """
        d = extract_videos(channel_id)
        # channel
        self.uid = d["channel_id"]
        self.title = d["channel"]
        self.followers = int(d["channel_follower_count"])
        self.description = d["description"]
        self.tags = d["tags"]
        # TODO thumbnails
        self.uploader_id = d["uploader_id"]
        self.uploader = d["uploader"]
        # videos
        self.videos_ids = [
            entry["id"]
            for entry in reversed(d["entries"])
            if entry["availability"] != "subscriber_only"
        ]
        # playlists
        d = extract_playlists(channel_id)
        self.playlists_ids = [playlist["id"] for playlist in reversed(d["entries"])]

    def load_videos(self) -> None:
        """Load videos extra info."""
        self.videos = []
        for index, video_id in enumerate(self.videos_ids):
            log.info(f"{index} ∕ {len(self.videos_ids)}")
            self.videos.append(Video(video_id))


# TODO Format
# TODO Playlist/basic,extra
# TODO Thumbnail


class Video(Object):
    """YouTube video."""

    def __init__(self, d: dict) -> None:
        """Set video info.

        :param d: video info
        :type d: dict
        """
        self.at = datetime.now().strftime("%Y%m%d%H%M%S")
        # info
        self.uid = d["id"]
        self.title = d["title"]
        self.description_cut = d["description"]
        self.duration = int(d["duration"])
        # TODO thumbnail from thumbnails

    def load_extra(self):
        d = extract_video(self.uid)
        # TODO formats
        # TODO thumbnail from thumbnails
        # TODO compare existing thumbnail
        self.description = d["description"]
        # TODO channel_id
        self.duration = int(d["duration"])
        self.views = int(d["view_count"])
        self.categories = d["categories"]
        self.tags = d["tags"]
        # TODO automatic_captions
        # TODO subtitles
        self.chapters = d["chapters"]
        self.likes = d["like_count"]
        self.timestamp = datetime.fromtimestamp(d["timestamp"]).strftime("%Y%m%d%H%M%S")
        self.fulltitle = d["fulltitle"]


# ╭──────────╮
# │ download │
# ╰──────────╯


def download_video(video_id: str | None) -> None:
    if video_id:
        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([url_video(video_id)])


# ╭─────────╮
# │ extract │
# ╰─────────╯


def extract(url: str) -> dict[str, Any]:
    """Return extracted dict.

    :rtype: dict
    """
    d = ytdl(
        {
            "extract_flat": True,
            "skip_download": True,
        },
    ).extract_info(url, download=False)
    log.info(d)
    return d


def extract_playlist(playlist_id: str) -> dict:
    """Return extracted playlist dict.

    :param playlist_id: playlist identifier
    :type playlist_id: str
    :rtype: dict
    """
    return extract(url_playlist(playlist_id))


def extract_playlists(channel_id: str) -> dict:
    """Return extracted playlists dict.

    :param channel_id: channel identifier
    :type channel_id: str
    :rtype: dict
    """
    return extract(url_playlists(channel_id))


def extract_video(video_id: str) -> dict:
    """Return extracted video dict.

    :param video_id: video identifier
    :type video_id: str
    :rtype: dict
    """
    return extract(url_video(video_id))


def extract_videos(channel_id: str) -> dict:
    """Return extracted videos dict.

    :param channel_id: channel identifier
    :type channel_id: str
    :rtype: dict
    """
    return extract(url_videos(channel_id))


# ╭──────╮
# │ next │
# ╰──────╯


def next_download(videos: list[str]) -> str | None:
    for index, video_id in enumerate(videos):
        if not Path(f"{video_id}.mp4").exists():
            log.info(f"{index} ∕ {len(videos)}")
            return video_id
    return None


# ╭─────╮
# │ url │
# ╰─────╯


def url_channel(channel_id: str) -> str:
    """Return channel URL.

    :param channel_id: channel identifier
    :type channel_id: str
    :rtype: str
    """
    return f"{URL}/channel/{channel_id}"


def url_playlist(playlist_id: str) -> str:
    """Return playlist URL.

    :param playlist_id: playlist identifier
    :type playlist_id: str
    :rtype: str
    """
    return f"{URL}/playlist?list={playlist_id}"


def url_playlists(channel_id: str) -> str:
    """Return playlists URL.

    :param channel_id: channel identifier
    :type channel_id: str
    :rtype: str
    """
    return f"{url_channel(channel_id)}/playlists"


def url_video(video_id: str) -> str:
    """Return video URL.

    :param video_id: video identifier
    :type video_id: str
    :rtype: str
    """
    return f"{URL}/watch?v={video_id}"


def url_videos(channel_id: str) -> str:
    """Return videos URL.

    :param channel_id: channel identifier
    :type channel_id: str
    :rtype: str
    """
    return f"{url_channel(channel_id)}/videos"


# ╭──────╮
# │ ytdl │
# ╰──────╯


def ytdl(opt: dict) -> YoutubeDL:
    options = {
        **opt,
        "ignoreerrors": False,
        "quiet": False,
    }
    log.info(options)
    return YoutubeDL(options)