Jonathan
57 lines
2.1 KiB
Python
57 lines
2.1 KiB
Python
import os
|
|
from typing import Callable
|
|
|
|
import yt_dlp
|
|
|
|
|
|
class YtDlpClient:
|
|
"""Real YtClient backed by yt_dlp. Not unit-tested offline; see test_youtube_live.py."""
|
|
|
|
def search_album(self, artist: str, album: str) -> list[dict]:
|
|
query = f"ytsearch5:{artist} {album} full album"
|
|
opts = {
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
"extract_flat": "in_playlist",
|
|
"skip_download": True,
|
|
}
|
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
|
info = ydl.extract_info(query, download=False)
|
|
results: list[dict] = []
|
|
for e in (info.get("entries") or []):
|
|
if not e:
|
|
continue
|
|
results.append(
|
|
{
|
|
"source_ref": e.get("url") or e.get("webpage_url") or e.get("id"),
|
|
"title": e.get("title") or "",
|
|
"artist": e.get("uploader") or e.get("channel") or "",
|
|
"track_count": e.get("playlist_count") or 1,
|
|
}
|
|
)
|
|
return results
|
|
|
|
def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
|
|
os.makedirs(dest, exist_ok=True)
|
|
|
|
def hook(d: dict) -> None:
|
|
if d.get("status") == "downloading":
|
|
total = d.get("total_bytes") or d.get("total_bytes_estimate")
|
|
if total:
|
|
on_progress(min(1.0, d.get("downloaded_bytes", 0) / total))
|
|
elif d.get("status") == "finished":
|
|
on_progress(1.0)
|
|
|
|
opts = {
|
|
"quiet": True,
|
|
"no_warnings": True,
|
|
"format": "bestaudio/best",
|
|
"outtmpl": os.path.join(dest, "%(playlist_index)s-%(title)s.%(ext)s"),
|
|
"progress_hooks": [hook],
|
|
"postprocessors": [{"key": "FFmpegExtractAudio", "preferredcodec": "opus"}],
|
|
}
|
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
|
info = ydl.extract_info(source_ref, download=True)
|
|
entries = info.get("entries") if info.get("entries") is not None else [info]
|
|
return {"track_count": len([e for e in entries if e]), "path": dest}
|