From cc58529e9933e60067c123740da8c738184d9f81 Mon Sep 17 00:00:00 2001 From: "Ethan J. Hadley" Date: Mon, 7 Sep 2026 11:52:46 -0600 Subject: [PATCH] Downloading playlists that aren't albums --- happy_ripper.py | 92 ++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 72 insertions(+), 20 deletions(-) diff --git a/happy_ripper.py b/happy_ripper.py index b248136..2ee1b58 100644 --- a/happy_ripper.py +++ b/happy_ripper.py @@ -1,14 +1,20 @@ from dataclasses import dataclass - -from flask import Flask, render_template, request, jsonify -from threading import Thread -import time -from yt_dlp import YoutubeDL -from yt_dlp.utils import DownloadError, ExtractorError -from collections.abc import Callable import pathlib import os import shutil +import urllib.request +from threading import Thread +import time +from collections.abc import Callable + +from yt_dlp import YoutubeDL +from yt_dlp.postprocessor import PostProcessor +from yt_dlp.utils import DownloadError, ExtractorError +import mutagen +from mutagen.easyid3 import EasyID3 +from mutagen.mp3 import MP3 +from mutagen.id3 import ID3, APIC +from flask import Flask, render_template, request, jsonify @dataclass @@ -29,6 +35,49 @@ class ReleaseStatus: class RipError(Exception): def __init__(self, msg): super().__init__(msg) + self.msg = msg + + +class MetadataPostProcessor(PostProcessor): + def __init__(self, info: ReleaseInfo, art_path: str): + super().__init__() + self.info = info + self.art_path = art_path + + def run(self, information): + file = pathlib.Path(information["filepath"]) + + audio = EasyID3(file) + audio.delete() + audio["title"] = information["title"] + audio["artist"] = information["uploader"] + + audio["album"] = self.info.title + + index = information.get("playlist_index") + if index is None: + index = 1 + audio["tracknumber"] = str(index) + + audio.save() + + id3audio = MP3(file, ID3=ID3) + id3audio_tags = id3audio.tags + + if id3audio_tags is None: + raise RipError("ID3 tags missing") + + id3audio_tags.add( + APIC( + mime="image/jpeg", + type=3, + desc="Cover", + data=open(self.art_path, "rb").read(), + ) + ) + id3audio.save() + + return [], information def create_or_wipe_folder(path: str) -> None: @@ -46,22 +95,24 @@ def get_content_json(url: str): return ydl.extract_info(url, download=False) -def download_content(path: str, url: str) -> None: +def download_content(path: str, url: str, post_processor: PostProcessor) -> None: try: with YoutubeDL( { - "format": "m4a/bestaudio/best", + "format": "bestaudio/best", + "ignoreerrors": True, "postprocessors": [ { "key": "FFmpegExtractAudio", - "preferredcodec": "m4a", + "preferredcodec": "mp3", } ], "outtmpl": f"{path}/%(title)s.%(ext)s", } ) as ydl: + ydl.add_post_processor(post_processor) error_code = ydl.download([url]) - if error_code: + if error_code > 1: raise RipError(f"Content download failed ({error_code})") except DownloadError: @@ -113,9 +164,6 @@ def find_release_info(url: str) -> ReleaseInfo: art_url = thumbnails[1]["url"] - if not is_album: - raise RipError("Failed, not an album") - return ReleaseInfo(artist, title, art_url, is_album, is_topic) @@ -132,7 +180,7 @@ def rip_worker(rip_queue: list[str], status_callback: Callable[[ReleaseInfo | st time.sleep(1) continue - status_callback("Gathering Info", False) + status_callback("Gathering info", False) url = rip_queue.pop(0) create_or_wipe_folder(tmp_path) @@ -141,19 +189,23 @@ def rip_worker(rip_queue: list[str], status_callback: Callable[[ReleaseInfo | st info = find_release_info(url) status_callback(info, False) - status_callback("Downloading Content", False) + status_callback("Downloading art", False) + art_path = os.path.join(tmp_path, "cover.jpg") + urllib.request.urlretrieve(info.art_url, art_path) + if not os.path.exists(art_path): + raise RipError("Art download failed") - download_content(tmp_path, url) + status_callback("Downloading content", False) + download_content(tmp_path, url, MetadataPostProcessor(info, art_path)) except RipError as e: - status_callback(repr(e), True) + status_callback(e.msg, True) else: - status_callback("Finished Sucessfully", True) + status_callback("Finished sucessfully", True) if __name__ == "__main__": - app = Flask(__name__) rip_queue: list[str] = [] finished_queue: list[ReleaseStatus] = []