from dataclasses import dataclass from http.client import NETWORK_AUTHENTICATION_REQUIRED import pathlib import os import re 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 from ytmusicapi import YTMusic MUSIC_FOLDER: str = os.environ.get("HAPPY_RIPPER_MUSIC_DIR", "Music") ACCESS_URL: str = os.environ.get("HAPPY_RIPPER_ACCESS_URL", "https://test.test.test") PORT = int(os.environ.get("HAPPY_RIPPER_MUSIC_PORT", 5000)) DENO_PATH: str = os.environ.get("HAPPY_RIPPER_DENO_PATH", str(pathlib.Path.home() / ".deno/bin/deno")) JS_RUNTIME_KEYS = {"deno": {"path": DENO_PATH}} @dataclass class ReleaseInfo: artist: str title: str art_url: str is_album: bool is_topic: bool @dataclass class ReleaseStatus: info: ReleaseInfo | None status: str destinations: set[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, on_finish: Callable): super().__init__() self.info = info self.art_path = art_path self.on_finish = on_finish def run(self, information): file = pathlib.Path(information["filepath"]) audio = EasyID3(file) audio.delete() audio["title"] = information["title"] audio["artist"] = information["uploader"].replace(" - Topic", "") 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() self.on_finish() 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({"js_runtimes": JS_RUNTIME_KEYS, "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( { "js_runtimes": JS_RUNTIME_KEYS, "format": "bestaudio/best", "ignoreerrors": True, "restrictfilenames": 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: first_entry = info else: 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"] if not is_album: artist = "Various Artists" 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 sort_into_music(path: str, info: ReleaseInfo) -> set[str]: destinations: set[str] = set() try: for file_name in os.listdir(path): if file_name == "cover.jpg": continue audio = EasyID3(os.path.join(path, file_name)) artist_list = audio.get("artist") if artist_list is None: raise RipError("Artist tag was None") artist = artist_list[0] artist = re.sub(r"[^\w_. -]", "_", artist) title = info.title title = re.sub(r"[^\w_. -]", "_", title) artist_path = os.path.join(MUSIC_FOLDER, artist) release_path = os.path.join(MUSIC_FOLDER, artist, title) if not os.path.exists(artist_path): os.makedirs(artist_path) if not os.path.exists(release_path): os.mkdir(release_path) dst_file_path = os.path.join(release_path, file_name) dst_file_path_cover = os.path.join(release_path, "cover.jpg") if os.path.exists(dst_file_path): os.remove(dst_file_path) if os.path.exists(dst_file_path_cover): os.remove(dst_file_path_cover) source_file_path = os.path.join(path, file_name) shutil.move(source_file_path, release_path) source_file_path_cover = os.path.join(path, "cover.jpg") shutil.copy(source_file_path_cover, release_path) destinations.add(release_path) except Exception: raise RipError("Filesystem error") return destinations def rip_worker(rip_queue: list[str], status_callback: Callable): 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 (0)", False) on_index = 0 def on_download_finish() -> None: nonlocal on_index on_index += 1 status_callback(f"Downloading content ({on_index})", False) download_content(tmp_path, url, MetadataPostProcessor(info, art_path, on_download_finish)) filelist = os.listdir(tmp_path) filelist.remove("cover.jpg") if len(filelist) == 0: raise RipError("Nothing ripped, Try again") status_callback("Sorting content", False) destinations = sort_into_music(tmp_path, info) status_callback(destinations, False) except RipError as e: status_callback(e.msg, True) except Exception: status_callback("Unknown Error", True) else: status_callback("Finished sucessfully", True) def get_all_playlists_from_channel(browseId: str) -> list[str]: urls: list[str] = [] yt = YTMusic() artist = yt.get_artist(browseId) albums_info = artist["albums"] if "params" in albums_info: albums = yt.get_artist_albums( albums_info["browseId"], albums_info["params"], limit=None, ) for album in albums: playlist_id = album.get("playlistId") if playlist_id: urls.append(f"https://music.youtube.com/playlist?list={playlist_id}") else: for album in artist["albums"]["results"]: playlist_id = album.get("audioPlaylistId") if playlist_id: urls.append(f"https://music.youtube.com/playlist?list={playlist_id}") return urls if __name__ == "__main__": app = Flask(__name__) rip_queue: list[str] = [] finished_queue: list[ReleaseStatus] = [] yt_music = YTMusic() current_release_info: ReleaseInfo | None = None current_release_status_message: str = "" current_release_destinations: set[str] = set() def get_current_release_status(x: ReleaseInfo | str | set[str], finished: bool): global current_release_info global current_release_status_message global current_release_destinations global finished_queue if type(x) is ReleaseInfo: current_release_info = x elif type(x) is str: current_release_status_message = x elif type(x) is set: current_release_destinations = x.copy() if finished: if current_release_info is None: current_release_info = ReleaseInfo("Unknown", "Unknown", "url", False, False) finished_queue.append( ReleaseStatus(current_release_info, current_release_status_message, current_release_destinations.copy()) ) current_release_info = None current_release_status_message = "Waiting" current_release_destinations = set() 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("/browse") def browse(): return render_template("browse.html") @app.route("/rip", methods=["POST"]) def rip(): print("RIP") url = request.form.get("url") if url is None: return jsonify({"success": False, "message": "Url is None"}) is_channel = request.form.get("is-channel-check") is_channel = is_channel is not None if is_channel: channel_playlists = get_all_playlists_from_channel(url) if len(channel_playlists) == 0: print("ERROR: Found no playlists in channel") return jsonify({"success": False, "message": "channel has no playlists"}) for playlist_url in channel_playlists: rip_queue.append(playlist_url) else: 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, "destinations": list(q.destinations), } ) 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.route("/search", methods=["POST"]) def search(): global yt_music query = request.form.get("search") if query is None: return jsonify({"success": False, "message": "Search query is None"}) responce = yt_music.search(query) data = [] for release in responce: release_data = {} # Category =============================================================== category = release.get("resultType") if category is None: return jsonify({"success": False, "message": "Category is None"}) release_data["category"] = category # Artist =============================================================== if "artist" in release: release_data["artist"] = release.get("artist", "Unknown") elif "artists" in release: artists = release.get("artists") if artists is None: return jsonify({"success": False, "message": "Artists exists but is None"}) release_data["artist"] = artists[0].get("name", "Unknown") print(artists) elif "author" in release: release_data["artist"] = release.get("author", "Unknown") else: release_data["artist"] = "Unknown" # Title =============================================================== if category == "artist": release_data["title"] = release_data["artist"] else: release_data["title"] = release.get("title", "Unknown") # Thumbnails =============================================================== if "thumbnails" in release: release_data["thumbnails"] = release.get("thumbnails") else: return jsonify({"success": False, "message": "Thumbnails is None"}) if "playlistId" in release: release_data["url"] = f"https://music.youtube.com/playlist?list={release['playlistId']}" elif "videoId" in release: release_data["url"] = f"https://music.youtube.com/watch?v={release['videoId']}" elif "browseId" in release and category == "artist": release_data["url"] = release.get("browseId") elif category == "artist" and "artists" in release: release_data["url"] = release.get("artists")[0].get("id") data.append(release_data) return jsonify(data) app.run(host="0.0.0.0", port=PORT)