First files
This commit is contained in:
+367
@@ -0,0 +1,367 @@
|
||||
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.get("HAPPY_RIPPER_MUSIC_DIR", "Music")
|
||||
ACCESS_URL = os.environ.get("HAPPY_RIPPER_ACCESS_URL", "https://test.test.test")
|
||||
PORT = int(os.environ.get("HAPPY_RIPPER_MUSIC_PORT", 5000))
|
||||
|
||||
|
||||
@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"]
|
||||
|
||||
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]
|
||||
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)
|
||||
Reference in New Issue
Block a user