This commit is contained in:
Marc Beninca 2025-03-18 11:41:06 +01:00
parent e75692ee26
commit 899b1383de
Signed by: marc.beninca
GPG key ID: 9C7613450C80C24F

View file

@ -17,13 +17,28 @@
from abc import ABC, abstractmethod
from pathlib import Path
from typing import Any
from rwx import Object
from rwx.log import stream as log
from yt_dlp import YoutubeDL
from rwx import Object
from rwx.log import stream as log
URL = "https://youtube.com"
class Channel(Object):
def __init__(self, d: dict) -> None:
self.identifier = 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"]
self.url = d["channel_url"]
# TODO entries
class Tab(Object, ABC):
"""YouTube Tab."""
@ -156,6 +171,11 @@ class Videos(Tab):
return None
# ╭──────────╮
# │ download │
# ╰──────────╯
def download_video(video_id: str | None) -> None:
if video_id:
ytdl(
@ -178,51 +198,63 @@ def download_video(video_id: str | None) -> None:
).download([f"{URL}/watch?v={video_id}"])
def extract(opt: dict, url: str) -> dict:
# ╭─────────╮
# │ extract │
# ╰─────────╯
def extract(opt: dict[str, Any], url: str) -> dict[str, Any]:
"""Return extracted dict.
:rtype: dict
"""
return ytdl({
return ytdl(
{
**opt,
"extract_flat": True,
"skip_download": True,
}).extract_info(url, download=False)
}
).extract_info(url, download=False)
def extract_channel(channel_id: str) -> dict:
"""Return extracted channel dict.
:param channel_id: channel identifier
:type channel_id: str
:rtype: dict
"""
d = extract({
}, f"{URL}/channel/{channel_id}")
log.info(d)
return {
}
def extract_videos(channel_id: str) -> list[str]:
"""Return extracted channel videos dict.
:rtype: dict
"""
d = extract({
}, f"{URL}/channel/{channel_id}/videos")
d = extract({}, f"{URL}/channel/{channel_id}")
return d
def extract_videos(channel_id: str) -> dict:
"""Return extracted videos dict.
:param channel_id: channel identifier
:type channel_id: str
:rtype: dict
"""
d = extract({}, f"{URL}/channel/{channel_id}/videos")
return d
return [entry["id"] for entry in reversed(d["entries"])]
def extract_video(video_id: str) -> dict:
"""Return extracted video dict.
:param video_id: video identifier
:type video_id: str
:rtype: dict
"""
d = extract({
}, f"{URL}/watch?v={video_id}")
d = extract({}, f"{URL}/watch?v={video_id}")
return d
# ╭────────╮
# │ filter │
# ╰────────╯
def filter_video(d: dict) -> dict:
"""Return filtered video dict.
@ -245,16 +277,59 @@ def filter_videos(d: dict) -> list[str]:
return [entry["id"] for entry in reversed(d["entries"])]
# ╭──────╮
# │ next │
# ╰──────╯
def next_download(videos: list[str]) -> str | None:
index = 0
for video_id in videos:
index += 1
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_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_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,