fastapi_server.py

import base64
import logging
from pathlib import Path
from typing import Any

from fastapi import FastAPI, File, HTTPException, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

from api_server import (
    MODEL_STATUS,
    analyze_emotion_base64,
    build_error_response,
    get_cnn_model_repo,
    get_model_endpoint,
    get_wav2vec_model_id,
)


logger = logging.getLogger("voice-emotion-fastapi")

BASE_DIR = Path(__file__).resolve().parent
FRONTEND_DIR = BASE_DIR / "frontend"

app = FastAPI(title="Voice Emotion Analysis")

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

app.mount("/static", StaticFiles(directory=FRONTEND_DIR), name="static")


@app.get("/")
async def index() -> FileResponse:
    return FileResponse(FRONTEND_DIR / "index.html")


@app.api_route("/ping", methods=["GET", "HEAD"])
async def ping() -> dict[str, str]:
    """Lightweight endpoint for uptime bots (e.g. UptimeRobot)."""
    return {"status": "alive"}


@app.get("/health")
async def health() -> dict[str, Any]:
    return {
        "status": "healthy",
        "cnnEndpoint": get_model_endpoint("cnn") or "",
        "wav2vecEndpoint": get_model_endpoint("wav2vec") or "",
        "cnnModel": get_cnn_model_repo(),
        "wav2vecModel": get_wav2vec_model_id(),
        "modelStatus": MODEL_STATUS,
    }


def get_audio_from_payload(payload: dict[str, Any]) -> str:
    if "audio" in payload and isinstance(payload["audio"], str):
        return payload["audio"]

    data = payload.get("data")
    if isinstance(data, list) and data and isinstance(data[0], str):
        return data[0]

    raise ValueError("Request must include audio as base64 in `audio` or `data[0]`.")


@app.post("/api/predict")
async def predict(request: Request) -> JSONResponse:
    try:
        payload = await request.json()
        audio_base64 = get_audio_from_payload(payload)
        return JSONResponse(analyze_emotion_base64(audio_base64))
    except ValueError as error:
        return JSONResponse(
            build_error_response(str(error), warnings=[str(error)]),
            status_code=400,
        )
    except Exception as error:
        logger.exception("FastAPI prediction failed")
        return JSONResponse(
            build_error_response(
                f"Server error: {error}",
                warnings=[f"Server error: {error}"],
            ),
            status_code=500,
        )


@app.post("/api/analyze")
async def analyze_upload(audio: UploadFile = File(...)) -> JSONResponse:
    content = await audio.read()
    if not content:
        raise HTTPException(status_code=400, detail="Uploaded audio file is empty.")

    audio_base64 = base64.b64encode(content).decode("utf-8")
    return JSONResponse(analyze_emotion_base64(audio_base64))


if __name__ == "__main__":
    import uvicorn

    uvicorn.run("fastapi_server:app", host="0.0.0.0", port=7860, reload=False)