143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
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
|
|
|
|
from pytube import YouTube
|
|
from moviepy.editor 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()
|
|
|
|
# Configuration (customize these!)
|
|
DOWNLOAD_FOLDER = "downloads" # Where the converted files will be saved
|
|
if not os.path.exists(DOWNLOAD_FOLDER):
|
|
os.makedirs(DOWNLOAD_FOLDER)
|
|
|
|
|
|
@app.post("/convert")
|
|
async def create_conversion(req: Request):
|
|
"""Converts a YouTube video to MP3."""
|
|
video_url = req.json().get("url")
|
|
if (video_url is None): raise HTTPException(400, "Please provide an URL on body")
|
|
try:
|
|
yt = YouTube(video_url)
|
|
stream = yt.streams.filter(only_audio=True).first()
|
|
if not stream:
|
|
raise HTTPException(status_code=400, detail="No audio stream found.")
|
|
|
|
# Download the audio stream
|
|
temp_file = os.path.join(DOWNLOAD_FOLDER, f"{yt.title}_temp.mp4") # temporary file
|
|
stream.download(output_path=DOWNLOAD_FOLDER, filename=f"{yt.title}_temp")
|
|
|
|
# Convert to MP3 using MoviePy
|
|
audio_clip = AudioFileClip(os.path.join(DOWNLOAD_FOLDER, f"{yt.title}_temp.mp4"))
|
|
mp3_file = os.path.join(DOWNLOAD_FOLDER, f"{yt.title}.mp3")
|
|
audio_clip.write_audiofile(mp3_file)
|
|
|
|
# Clean up the temporary file
|
|
os.remove(os.path.join(DOWNLOAD_FOLDER, f"{yt.title}_temp.mp4"))
|
|
|
|
return mp3_file # Return the filename of the converted MP3
|
|
except Exception as e:
|
|
print(f"Error converting video: {e}") # Log the error
|
|
raise HTTPException(status_code=500, detail=f"Conversion failed: {str(e)}")
|
|
|
|
class register_model(BaseModel):
|
|
first_name:str
|
|
last_name:str
|
|
email:str
|
|
password:str
|
|
|
|
@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) |