from dataclasses import dataclass 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 class ReleaseInfo: artist: str title: str art_url: str is_album: bool is_topic: bool @dataclass class ReleaseStatus: info: ReleaseInfo | None status: str 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: """Creates tmp directory, if it already exists, it clears it, and makes a new one.""" if not os.path.isdir(path): os.mkdir(path) else: shutil.rmtree(path) os.mkdir(path) def get_content_json(url: str): with YoutubeDL({"playlist_items": "1"}) as ydl: return ydl.extract_info(url, download=False) def download_content(path: str, url: str, post_processor: PostProcessor) -> None: try: with YoutubeDL( { "format": "bestaudio/best", "ignoreerrors": True, "postprocessors": [ { "key": "FFmpegExtractAudio", "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 > 1: raise RipError(f"Content download failed ({error_code})") except DownloadError: raise RipError("Content Download Failed") except ExtractorError: raise RipError("Content Extraction Failed") def find_release_info(url: str) -> ReleaseInfo: try: info = get_content_json(url) except DownloadError: raise RipError("Info Download Failed") except ExtractorError: raise RipError("Info Extraction Failed") entries = info.get("entries") if entries is None: raise RipError("Entries is none") first_entry = next(iter(entries), None) if not first_entry: raise RipError("Entries is empty") artist = first_entry.get("uploader") if artist is None: raise RipError("Uploader tag is none") is_topic = False if " - Topic" in artist: is_topic = True artist = artist.replace(" - Topic", "") title = info.get("title") if title is None: raise RipError("Title tag is none") is_album = False if "Album - " in title: is_album = True title = title.replace("Album - ", "") thumbnails = info.get("thumbnails") if thumbnails is None: raise RipError("Thumbnails is none") art_url = thumbnails[1]["url"] return ReleaseInfo(artist, title, art_url, is_album, is_topic) def find_tmp_path() -> str: project_root_path = pathlib.Path(__file__).parent.resolve() return f"{project_root_path}/tmp" def rip_worker(rip_queue: list[str], status_callback: Callable[[ReleaseInfo | str, bool],]): tmp_path = find_tmp_path() while True: if len(rip_queue) == 0: time.sleep(1) continue status_callback("Gathering info", False) url = rip_queue.pop(0) create_or_wipe_folder(tmp_path) try: info = find_release_info(url) status_callback(info, 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") status_callback("Downloading content", False) download_content(tmp_path, url, MetadataPostProcessor(info, art_path)) except RipError as e: status_callback(e.msg, True) else: status_callback("Finished sucessfully", True) if __name__ == "__main__": app = Flask(__name__) rip_queue: list[str] = [] finished_queue: list[ReleaseStatus] = [] current_release_info: ReleaseInfo | None = None current_release_status_message: str = "" def get_current_release_status(x: ReleaseInfo | str, finished: bool): global current_release_info global current_release_status_message global finished_queue if type(x) is ReleaseInfo: current_release_info = x elif type(x) is str: current_release_status_message = x if finished: finished_queue.append(ReleaseStatus(current_release_info, current_release_status_message)) current_release_info = None current_release_status_message = "Waiting" if len(finished_queue) > 30: finished_queue.pop(0) ripping_thread = Thread(target=rip_worker, args=(rip_queue, get_current_release_status), daemon=True).start() @app.route("/") def index(): return render_template("index.html") @app.route("/rip", methods=["POST"]) def rip(): url = request.form.get("url") if url is None: return jsonify({"success": False, "message": "Url is None"}) rip_queue.append(url) return jsonify({"success": True, "message": f"Added, {url}!"}) @app.route("/status", methods=["GET"]) def status(): data = {"success": True, "queue": rip_queue, "finishedQueue": [], "status": current_release_status_message} for q in finished_queue: if q.info is None: continue data["finishedQueue"].append( {"artist": q.info.artist, "title": q.info.title, "status": q.status, "art_url": q.info.art_url} ) if current_release_info is not None: data["currentReleaseInfo"] = { "artist": current_release_info.artist, "title": current_release_info.title, "art_url": current_release_info.art_url, } return jsonify(data) app.run()