Happy-Ripper-Docker/happy_ripper.py

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


@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)


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) -> None:
    try:
        with YoutubeDL(
            {
                "format": "m4a/bestaudio/best",
                "postprocessors": [
                    {
                        "key": "FFmpegExtractAudio",
                        "preferredcodec": "m4a",
                    }
                ],
                "outtmpl": f"{path}/%(title)s.%(ext)s",
            }
        ) as ydl:
            error_code = ydl.download([url])
            if error_code:
                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"]

    if not is_album:
        raise RipError("Failed, not an album")

    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 Content", False)

            download_content(tmp_path, url)

        except RipError as e:
            status_callback(repr(e), 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()