Happy-Ripper/happy_ripper.py

from dataclasses import dataclass

from flask import Flask, render_template, request, jsonify
from threading import RLock, 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
    status: str


class YtdlpErrorCode(Exception):
    def __init__(self, code):
        super().__init__(f"Some videos failed with code {code}")
        self.code = code


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:
    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 YtdlpErrorCode(error_code)


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],], finish_callback: Callable[[ReleaseInfo],]
):
    tmp_path = find_tmp_path()

    while True:
        if len(rip_queue) == 0:
            time.sleep(1)
            continue

        create_or_wipe_folder(tmp_path)

        current_url = rip_queue.pop(0)
        release_info = ReleaseInfo(
            "",
            current_url,
            "https://imgs.search.brave.com/LeS4HHKZ1oz1T15VY5MwiUjWDjLiYKj0vgRABB3D2BY/rs:fit:860:0:0:0/g:ce/aHR0cHM6Ly90My5m/dGNkbi5uZXQvanBn/LzA2Lzg2LzE5LzM0/LzM2MF9GXzY4NjE5/MzQwN19ESFp3amV5/ZEJPUjF0RURrTEF6/d00zdzVrWXN0Unp6/Qi5qcGc",
            "Starting",
        )
        status_callback(release_info)

        is_album = False
        is_topic = False

        # Info json --------------------------------------------------

        release_info.status = "Gathering Info"
        status_callback(release_info)

        try:
            info = get_content_json(current_url)

        except DownloadError:
            release_info.status = "Info Download Failed"
            finish_callback(release_info)
            continue

        except ExtractorError:
            release_info.status = "Info Extraction Failed"
            finish_callback(release_info)
            continue

        entries = info.get("entries")
        if entries is None:
            release_info.status = "Entries is none"
            finish_callback(release_info)
            continue

        first_entry = next(iter(entries), None)
        if not first_entry:
            release_info.status = "Entries is empty"
            finish_callback(release_info)
            continue

        artist = first_entry.get("uploader")
        if artist is None:
            release_info.status = "Uploader tag is none"
            finish_callback(release_info)
            continue

        if " - Topic" in artist:
            is_topic = True
            artist = artist.replace(" - Topic", "")

        release_info.artist = artist

        title = info.get("title")
        if title is None:
            release_info.status = "Title tag is none"
            finish_callback(release_info)
            continue

        if "Album - " in title:
            is_album = True
            title = title.replace("Album - ", "")

        release_info.title = title

        thumbnails = info.get("thumbnails")
        if thumbnails is None:
            release_info.status = "Thumbnails is none"
            finish_callback(release_info)
            continue

        art_url = thumbnails[1]["url"]
        release_info.art_url = art_url

        if not is_album:
            release_info.status = "Failed, not an album"
            finish_callback(release_info)
            continue

        release_info.status = "Finished gathering info"
        status_callback(release_info)

        # Info json --------------------------------------------------

        release_info.status = "Downloading"
        status_callback(release_info)

        try:
            download_content(tmp_path, current_url)

        except DownloadError:
            release_info.status = "Content Download Failed"
            finish_callback(release_info)
            continue

        except YtdlpErrorCode:
            release_info.status = "Content Download Failed with code"
            finish_callback(release_info)
            continue

        except ExtractorError:
            release_info.status = "Content Extraction Failed"
            finish_callback(release_info)
            continue

        release_info.status = "Writting Metadata"
        status_callback(release_info)

        release_info.status = "Finished Successfully"
        finish_callback(release_info)


if __name__ == "__main__":

    app = Flask(__name__)
    rip_queue = []
    finished_queue = []

    current_release_info = ReleaseInfo("", "", "", "Waiting")

    def get_current_release_info(release_info: ReleaseInfo):
        global current_release_info
        current_release_info = release_info

    def get_current_release_info_on_finish(release_info: ReleaseInfo):
        global current_release_info
        global finished_queue
        current_release_info = ReleaseInfo("", "", "", "Waiting")
        finished_queue.append(release_info)
        if len(finished_queue) > 30:
            finished_queue.pop(0)

    ripping_thread = Thread(
        target=rip_worker, args=(rip_queue, get_current_release_info, get_current_release_info_on_finish), daemon=True
    ).start()

    @app.route("/")
    def index():
        return render_template("index.html")

    @app.route("/rip", methods=["POST"])
    def rip():
        url = request.form.get("url")
        rip_queue.append(url)
        return jsonify({"success": True, "message": f"Added, {url}!"})

    @app.route("/status", methods=["GET"])
    def status():
        return jsonify(
            {
                "success": True,
                "queue": rip_queue,
                "finishedQueue": [
                    {"artist": q.artist, "title": q.title, "status": q.status, "art_url": q.art_url}
                    for q in finished_queue
                ],
                "currentReleaseInfo": {
                    "artist": current_release_info.artist,
                    "title": current_release_info.title,
                    "status": current_release_info.status,
                    "art_url": current_release_info.art_url,
                },
            }
        )

    app.run()