app.py

"""DHVANI inference API for Hugging Face Spaces."""

from __future__ import annotations

import logging
import tempfile
import time
import uuid
from collections import defaultdict
from pathlib import Path

from fastapi import FastAPI, File, HTTPException, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware

from analyzer import VoiceAnalyzer

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger("dhvani")

ALLOWED = {".wav", ".mp3", ".m4a", ".ogg", ".flac", ".webm"}
MAX_BYTES = 20 * 1024 * 1024
VERSION = "2.1.2"  # multi-segment ML + premium-TTS safeguard for unseen voices
RATE_LIMIT_WINDOW_SEC = 60
RATE_LIMIT_MAX = 20

app = FastAPI(title="DHVANI Inference", version=VERSION)
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

analyzer = VoiceAnalyzer()
_request_log: dict[str, list[float]] = defaultdict(list)


@app.on_event("startup")
def startup_warmup():
    try:
        analyzer.warmup()
        logger.info("Models warmed up: %s", analyzer.models_loaded)
    except Exception as exc:
        logger.error("Model warmup failed: %s", exc)


def rate_limited(client: str) -> bool:
    now = time.time()
    window = _request_log[client]
    _request_log[client] = [t for t in window if now - t < RATE_LIMIT_WINDOW_SEC]
    if len(_request_log[client]) >= RATE_LIMIT_MAX:
        return True
    _request_log[client].append(now)
    return False


@app.get("/")
@app.get("/health")
def health():
    custom = analyzer._custom  # noqa: SLF001
    custom_info = None
    if custom.enabled:
        custom_info = {
            "loaded": custom.loaded,
            "version": custom.version if custom.loaded else "pending",
            "metadata": custom.metadata if custom.loaded else {},
        }
    e2e = analyzer._e2e  # noqa: SLF001
    return {
        "status": "ok",
        "product": "DHVANI",
        "mode": "hf-space",
        "version": VERSION,
        "sovereign": bool(e2e.enabled and e2e.loaded),
        "mixed_media_aware": VERSION >= "2.0.1",
        "ensemble": analyzer.models_loaded or "loading",
        "e2e_model": e2e.model_id if e2e.loaded else None,
        "custom_model": custom_info,
    }


@app.post("/analyze")
async def analyze(request: Request, audio: UploadFile = File(...)):
    client = request.client.host if request.client else "unknown"
    if rate_limited(client):
        raise HTTPException(status_code=429, detail="Rate limit exceeded. Try again in a minute.")

    if not audio.filename:
        raise HTTPException(status_code=400, detail="Empty filename.")

    suffix = Path(audio.filename).suffix.lower()
    if suffix and suffix not in ALLOWED:
        raise HTTPException(status_code=400, detail=f"Unsupported file type: {suffix}")

    data = await audio.read()
    if not data:
        raise HTTPException(status_code=400, detail="No audio file provided.")
    if len(data) > MAX_BYTES:
        raise HTTPException(status_code=400, detail="File exceeds 20 MB limit.")

    saved_path = Path(tempfile.gettempdir()) / f"{uuid.uuid4().hex}{suffix or '.wav'}"
    saved_path.write_bytes(data)

    try:
        result = analyzer.analyze_file(saved_path)
        payload = result.to_api_dict()
        payload["backend_version"] = VERSION
        return payload
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    except Exception as exc:
        logger.exception("Analysis failed")
        raise HTTPException(status_code=500, detail=f"Analysis failed: {exc}") from exc
    finally:
        if saved_path.exists():
            saved_path.unlink()