forked from Ethanscharlie/Happy-Ripper
365 lines
10 KiB
Python
365 lines
10 KiB
Python
from dataclasses import dataclass
|
|
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
|
|
|
|
MUSIC_FOLDER = os.environ["HAPPY_RIPPER_MUSIC_DIR"]
|
|
ACCESS_URL = os.environ["HAPPY_RIPPER_ACCESS_URL"]
|
|
PORT = int(os.environ["HAPPY_RIPPER_MUSIC_PORT"])
|
|
|
|
|
|
@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({"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,
|
|
"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"]
|
|
|
|
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]
|
|
re.sub(r"[^\w_. -]", "_", artist)
|
|
title = info.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))
|
|
if len(os.listdir(tmp_path)) == 0:
|
|
raise RipError("No content was downloaded")
|
|
|
|
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)
|
|
|
|
|
|
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 = ""
|
|
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("/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,
|
|
"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.run(host="0.0.0.0", port=PORT)
|