MIF_E31231708/ml/api_app.py

from pathlib import Path
import tempfile

from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field

from audio_utils_api import SUPPORTED_AUDIO_EXTENSIONS, cleanup_temp_files
from database import (
    authenticate_user,
    create_user,
    get_database_label,
    init_db,
    list_prediction_results,
    save_prediction_result,
)
from predict_api import MODEL_NOT_FOUND_MESSAGE, get_model_info, predict_audio


app = FastAPI(title="ConfiVoice Prediction API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class PredictionSaveRequest(BaseModel):
    user_id: int | None = None
    student_name: str
    student_gender: str | None = None
    predicted_label: str | None = None
    label: str | None = None
    description: str | None = None
    confidence: float = 0
    probability_pd: float = 0
    probability_tpd: float = 0
    is_valid_audio: bool = True
    error_message: str | None = None
    audio_quality: dict = Field(default_factory=dict)
    voice_indicators: dict = Field(default_factory=dict)


class AuthRequest(BaseModel):
    username: str
    password: str


class RegisterRequest(AuthRequest):
    full_name: str


@app.on_event("startup")
def startup():
    init_db()


@app.get("/")
def read_root():
    return {
        "name": "ConfiVoice Prediction API",
        "status": "ok",
        "predict_endpoint": "/predict",
        "predictions_endpoint": "/predictions",
        "database": get_database_label(),
    }


@app.get("/model")
def read_model_info():
    try:
        return get_model_info()
    except FileNotFoundError as error:
        raise HTTPException(status_code=404, detail=MODEL_NOT_FOUND_MESSAGE) from error
    except Exception as error:
        raise HTTPException(status_code=500, detail=str(error)) from error


@app.get("/predictions")
def read_predictions(limit: int = 200, user_id: int | None = None):
    return {
        "database": get_database_label(),
        "data": list_prediction_results(limit=limit, user_id=user_id),
    }


@app.post("/register")
def register(payload: RegisterRequest):
    full_name = " ".join(payload.full_name.strip().split())
    username = payload.username.strip().lower()
    password = payload.password

    if not full_name:
        raise HTTPException(status_code=400, detail="Nama lengkap wajib diisi.")
    if len(username) < 3:
        raise HTTPException(status_code=400, detail="Username minimal 3 karakter.")
    if len(password) < 6:
        raise HTTPException(status_code=400, detail="Password minimal 6 karakter.")

    try:
        user = create_user(full_name, username, password)
    except Exception as error:
        message = str(error).lower()
        if "duplicate" in message or "unique" in message:
            raise HTTPException(status_code=409, detail="Username sudah terdaftar.") from error
        raise HTTPException(status_code=500, detail=str(error)) from error

    return {"status": "registered", "user": user}


@app.post("/login")
def login(payload: AuthRequest):
    username = payload.username.strip().lower()
    user = authenticate_user(username, payload.password)
    if not user:
        raise HTTPException(status_code=401, detail="Username atau password salah.")
    return {"status": "logged_in", "user": user}


@app.post("/predictions")
def create_prediction(payload: PredictionSaveRequest):
    student_name = payload.student_name.strip()
    if not student_name:
        raise HTTPException(status_code=400, detail="Nama siswa/i wajib diisi.")

    result = payload.model_dump()
    prediction_id = save_prediction_result(student_name, result)
    return {
        "status": "saved",
        "prediction_id": prediction_id,
        "student_name": student_name,
    }


@app.post("/save")
def save_prediction(payload: PredictionSaveRequest):
    return create_prediction(payload)


@app.post("/predict")
async def predict(
    file: UploadFile = File(...),
    student_name: str = Form("Tanpa Nama"),
    student_gender: str = Form(""),
):
    suffix = Path(file.filename or "").suffix.lower()
    if suffix not in SUPPORTED_AUDIO_EXTENSIONS:
        allowed = ", ".join(sorted(SUPPORTED_AUDIO_EXTENSIONS))
        raise HTTPException(
            status_code=400,
            detail=f"Format audio tidak didukung. Format yang didukung: {allowed}",
        )

    temp_input = tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
    temp_input_path = Path(temp_input.name)

    try:
        content = await file.read()
        temp_input.write(content)
        temp_input.close()

        if not content:
            raise HTTPException(status_code=400, detail="File audio kosong.")

        clean_student_name = student_name.strip() or "Tanpa Nama"
        result = predict_audio(temp_input_path)
        result["student_name"] = clean_student_name
        result["student_gender"] = student_gender.strip() or None
        return result
    except HTTPException:
        raise
    except FileNotFoundError as error:
        raise HTTPException(status_code=404, detail=MODEL_NOT_FOUND_MESSAGE) from error
    except Exception as error:
        raise HTTPException(status_code=500, detail=str(error)) from error
    finally:
        temp_input.close()
        cleanup_temp_files(temp_input_path)