yt2mp3/server/py/fastapi_server/main.py

from fastapi import FastAPI, HTTPException, Request, Response
#from fastapi import Body, Cookie, File, Form, Header, Path, Query
from contextlib import asynccontextmanager
from pydantic import BaseModel, ValidationError
from typing import Optional
import uvicorn
import aiomysql
import os
import bcrypt
import asyncio
import subprocess

from yt_dlp import YoutubeDL
from yt_dlp.utils import DownloadError

#from pytube import YouTube
#from moviepy.audio.io import AudioFileClip

#import asyncio

# In this FastAPI example, we skip BaseModel verification cause NuxtJS does that for us
# In other words, this backend only recieves requests from nuxt js backend
# So we can trust it

class db:
    pool = None

    host = "192.168.1.208"
    user = "suricata"
    db_name = "yt2mp3_nuxt"
    pwd = "19_P@NDCATA"  


@asynccontextmanager
async def lifespan(app: FastAPI):
    db.pool = await aiomysql.create_pool(host=db.host,user=db.user,password=db.pwd,db=db.db_name, minsize=1, maxsize=10)
    try:
        yield
    finally: 
        db.pool.close()
        await db.pool.wait_closed()

app = FastAPI(lifespan=lifespan)


async def fetch_user_info(email: str):
    async with db.pool.acquire() as conn:
        async with conn.cursor(aiomysql.DictCursor) as cur:
            await cur.execute("SELECT * FROM users WHERE email = %s", (email,))
            return await cur.fetchone()

def download_audio(video_url: str, output_folder: str) -> str:
    ydl_opts = {
        'format': 'bestaudio/best',
        'outtmpl': os.path.join(output_folder, '%(title)s.%(ext)s'),
        'postprocessors': [{
            'key': 'FFmpegExtractAudio',
            'preferredcodec': 'mp3',
            'preferredquality': '192',
        }, {
            'key': 'FFmpegMetadata',
            'add_metadata': True,
        }],
        'quiet': True,
        'noplaylist': True
    }

    with YoutubeDL(ydl_opts) as ydl:
        info = ydl.extract_info(video_url, download=True)
        filename = ydl.prepare_filename(info)
        return os.path.splitext(filename)[0] + ".mp3"  # path to final MP3

def set_metadata_and_rename(mp3_file: str, name: str, artist: str = None, album: str = None) -> str:
    new_file = os.path.abspath(os.path.join(os.path.dirname(mp3_file), f"{name}.mp3"))
    
    cmd = ["ffmpeg", "-y", "-i", mp3_file]
    
    # Only add metadata if provided
    cmd += ["-metadata", f"title={name}"]
    if artist:
        cmd += ["-metadata", f"artist={artist}"]
    if album:
        cmd += ["-metadata", f"album={album}"]
    
    cmd += [new_file]
    
    subprocess.run(cmd, check=True)
    os.remove(mp3_file)  # clean original
    return new_file


# Configuration (customize these!)
DOWNLOAD_FOLDER = "downloads"  # Where the converted files will be saved
if not os.path.exists(DOWNLOAD_FOLDER):
    os.makedirs(DOWNLOAD_FOLDER)

async def delay_delete_file(delay:float, file:str):
    await asyncio.sleep(delay)
    if os.path.exists(file): os.unlink(file)

@app.post("/convert")
async def create_conversion(req: Request):
    """Converts a YouTube video to MP3."""
    body = await req.json()
    print(body)
    video_url = body.get("video_url")
    user_id = body.get("user_id")
    
    if (video_url is None): raise HTTPException(400, "Please provide an URL on body")
    if (body.get("user_id") is None): raise HTTPException(401, "No User ID")

    USER_FOLDER = os.path.join(DOWNLOAD_FOLDER, str(user_id))
    if not os.path.exists(USER_FOLDER): os.mkdir(USER_FOLDER)
    try:
        file = await asyncio.to_thread(download_audio, video_url, USER_FOLDER)
    except DownloadError as e:
        raise HTTPException(500, "Error while downloading " + e.msg)
    final_file = await asyncio.to_thread(set_metadata_and_rename, file, body.get("song"), body.get("artist"), body.get("album"))
    file_name = final_file.split("/")[-1]

    print("File served: " + final_file)
    asyncio.create_task(delay_delete_file(300, final_file))

    return { "file" : file_name }



@app.post("/register")
async def create_user(req: Request):
    body = await req.json()

    # Auto exceptions give more verbose to Nuxt.

    body["password"] = bcrypt.hashpw(body["password"].encode(), bcrypt.gensalt()).decode()
    async with db.pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("INSERT INTO users (`first_name`,`last_name`,`email`,`password`) VALUES (%s,%s,%s,%s)", (body.first_name, body.last_name, body.email, body.password))
            await conn.commit()   



@app.post("/login")
async def login_endpoint(req: Request):
    
    body = await req.json()
    #print(body)
    try:
        user_data = await fetch_user_info(body["email"])
        if (user_data is None): 
            raise HTTPException(status_code=401, detail="User doesn't exist")
        else:
            pwd = user_data.get("password")
            if (bcrypt.checkpw(body["password"].encode(), user_data.get("password").encode())):
                return user_data
            else:
                raise HTTPException(status_code=401)
    except KeyError:
        raise HTTPException(status_code=400)
    #except ValidationError:
    #    raise HTTPException(status_code=400)


@app.get("/me")
async def me_get_endpoint(req: Request):
    """
    Retrieves user profile data.

    Trusts Nuxt's backend to allow it or not.
    """
    args = dict(req.query_params)
    #print(args)
    try:
        me = await fetch_user_info(args["email"])
        if (me is None):
            raise HTTPException(404, "User doesn't exist")
        else: return me
    except KeyError:
        raise HTTPException(400, "Please provide email")




if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8003)