102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
import os
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status, UploadFile, File, Form
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.deps import get_db, get_current_active_superuser
|
|
from app.models.user import User
|
|
from app.core.config import settings
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/song", status_code=status.HTTP_201_CREATED)
|
|
async def upload_song(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
file: UploadFile = File(...),
|
|
current_user: User = Depends(get_current_active_superuser),
|
|
) -> Any:
|
|
"""
|
|
Upload a song file. Only superusers can upload songs.
|
|
"""
|
|
# Check file type
|
|
if not file.content_type.startswith("audio/"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="File must be an audio file",
|
|
)
|
|
|
|
# Generate unique filename to avoid collisions
|
|
filename = f"{uuid.uuid4()}{os.path.splitext(file.filename)[1]}"
|
|
|
|
# Create directory if it doesn't exist
|
|
audio_dir = settings.AUDIO_DIR
|
|
audio_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Save file
|
|
file_path = audio_dir / filename
|
|
|
|
try:
|
|
contents = await file.read()
|
|
with open(file_path, "wb") as f:
|
|
f.write(contents)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error uploading file: {str(e)}",
|
|
)
|
|
|
|
# Return the file path (to be saved in Song model)
|
|
return {"file_path": str(file_path)}
|
|
|
|
|
|
@router.post("/image", status_code=status.HTTP_201_CREATED)
|
|
async def upload_image(
|
|
*,
|
|
db: Session = Depends(get_db),
|
|
file: UploadFile = File(...),
|
|
image_type: str = Form(...), # "artist" or "album"
|
|
current_user: User = Depends(get_current_active_superuser),
|
|
) -> Any:
|
|
"""
|
|
Upload an image file for artist or album. Only superusers can upload images.
|
|
"""
|
|
# Check file type
|
|
if not file.content_type.startswith("image/"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="File must be an image",
|
|
)
|
|
|
|
# Check image type
|
|
if image_type not in ["artist", "album"]:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Image type must be 'artist' or 'album'",
|
|
)
|
|
|
|
# Generate unique filename to avoid collisions
|
|
filename = f"{uuid.uuid4()}{os.path.splitext(file.filename)[1]}"
|
|
|
|
# Create directory if it doesn't exist
|
|
images_dir = settings.IMAGES_DIR
|
|
images_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Save file
|
|
file_path = images_dir / filename
|
|
|
|
try:
|
|
contents = await file.read()
|
|
with open(file_path, "wb") as f:
|
|
f.write(contents)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error uploading file: {str(e)}",
|
|
)
|
|
|
|
# Return the file path (to be saved in Artist or Album model)
|
|
return {"file_path": str(file_path)} |