Update app.py

app.py

@@ -1,104 +1,125 @@
 import os
+from fastapi import FastAPI, UploadFile, File, HTTPException
+from fastapi.responses import JSONResponse, PlainTextResponse
+from pydantic import BaseModel
+from faster_whisper import WhisperModel
+import tempfile
 import subprocess
-from typing import Optional
+import math
 
-from fastapi import FastAPI, File, UploadFile
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse
-from pydantic import BaseModel
+# ---------- Writable caches (prevents PermissionError on /.cache) ----------
+CACHE_ROOT = os.environ.get("HF_HOME", "/data/hf")
+os.environ["HF_HOME"] = CACHE_ROOT
+os.environ["HUGGINGFACE_HUB_CACHE"] = CACHE_ROOT
+os.environ["TRANSFORMERS_CACHE"] = CACHE_ROOT
+os.makedirs(CACHE_ROOT, exist_ok=True)
 
-# ---- CACHES: point to writable dirs, ensure perms ----
-os.environ.setdefault("HF_HOME", "/data/hf")
-os.environ.setdefault("HUGGINGFACE_HUB_CACHE", "/data/hf")
-os.environ.setdefault("XDG_CACHE_HOME", "/data/.cache")
-os.environ.setdefault("HF_HUB_DISABLE_TELEMETRY", "1")
-os.environ.setdefault("HF_HUB_DISABLE_PROGRESS_BARS", "1")
+# ---------- App ----------
+app = FastAPI(title="Nuvia Free Transcriber", version="1.4.0")
 
-for p in ("/data", "/data/hf", "/data/.cache", "/tmp"):
-    try:
-        os.makedirs(p, exist_ok=True)
-        os.chmod(p, 0o777)
-    except Exception:
-        pass
+# Root route (avoid 404 at "/")
+@app.get("/", response_class=PlainTextResponse)
+def root():
+    return "Nuvia Free Transcriber · try POST /transcribe or GET /health"
 
-# ---- Transcriber (CPU) ----
-from faster_whisper import WhisperModel
-MODEL_NAME = os.environ.get("WHISPER_MODEL", "tiny.en")
+# Health route used by your GPT Action
+class HealthOut(BaseModel):
+    ok: bool
 
-app = FastAPI(title="Nuvia Free Transcriber", version="1.3.0")
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"], allow_credentials=True,
-    allow_methods=["*"], allow_headers=["*"],
-)
+@app.get("/health", response_model=HealthOut)
+def health():
+    return {"ok": True}
+
+# ---------- Load model (tiny.en = fastest on CPU) ----------
+# You can switch to "base.en" if you want a bit more accuracy (slower).
+MODEL_NAME = os.environ.get("WHISPER_REPO", "Systran/faster-whisper-tiny.en")
+COMPUTE_TYPE = os.environ.get("WHISPER_COMPUTE", "int8")
+
+# Ensure the cache dir exists and is writable before model download
+os.makedirs(CACHE_ROOT, exist_ok=True)
 
-# IMPORTANT:
-# local_files_only=True prevents any runtime writes/downloads
 model = WhisperModel(
     MODEL_NAME,
     device="cpu",
-    compute_type="int8",
-    download_root="/data/hf",
-    local_files_only=True,
+    compute_type=COMPUTE_TYPE,
+    download_root=CACHE_ROOT,  # <— keeps models inside /data/hf
 )
 
-class HealthOut(BaseModel):
-    ok: bool
-
-class TranscribeOut(BaseModel):
-    text: str
-    duration_sec: Optional[float] = None
-    wpm: Optional[float] = None
-
-def ffprobe_duration(path: str) -> Optional[float]:
+# ---------- Helpers ----------
+def ffprobe_duration(path: str) -> float | None:
+    """Return duration in seconds using ffprobe, or None on failure."""
     try:
         out = subprocess.check_output(
-            ["ffprobe", "-v", "error", "-show_entries", "format=duration",
-             "-of", "default=noprint_wrappers=1:nokey=1", path]
+            [
+                "ffprobe",
+                "-v", "error",
+                "-show_entries", "format=duration",
+                "-of", "default=noprint_wrappers=1:nokey=1",
+                path,
+            ],
+            stderr=subprocess.STDOUT,
         )
-        return float(out.decode("utf-8").strip())
+        return float(out.decode().strip())
     except Exception:
         return None
 
-def word_count(text: str) -> int:
-    return len([w for w in text.split() if w.strip()])
-
-@app.get("/", response_model=HealthOut)
-def root():
-    return HealthOut(ok=True)
+def estimate_wpm(text: str, duration_sec: float | None) -> float | None:
+    if not text or not duration_sec or duration_sec <= 0:
+        return None
+    words = len(text.strip().split())
+    minutes = duration_sec / 60.0
+    if minutes <= 0:
+        return None
+    return words / minutes
 
-@app.get("/health", response_model=HealthOut)
-def health():
-    return HealthOut(ok=True)
+# ---------- Schemas ----------
+class TranscribeOut(BaseModel):
+    text: str
+    duration_sec: float | None = None
+    wpm: float | None = None
 
+# ---------- API ----------
 @app.post("/transcribe", response_model=TranscribeOut)
 async def transcribe(file: UploadFile = File(...)):
-    # Save upload
-    ext = os.path.splitext(file.filename or "")[1].lower() or ".bin"
-    tmp_in = f"/tmp/in{ext}"
-    raw = await file.read()
-    with open(tmp_in, "wb") as f:
-        f.write(raw)
-
-    # Convert to mono 16k WAV
-    tmp_wav = "/tmp/in.wav"
-    try:
-        subprocess.check_call(
-            ["ffmpeg", "-y", "-i", tmp_in, "-ar", "16000", "-ac", "1", tmp_wav],
-            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
-        )
-    except subprocess.CalledProcessError:
-        return JSONResponse(status_code=400, content={"error": "ffmpeg failed to decode audio"})
-
-    duration = ffprobe_duration(tmp_wav)
+    if not file.filename:
+        raise HTTPException(400, "Missing file name")
+    suffix = os.path.splitext(file.filename)[1].lower()
+    if suffix not in [".mp3", ".m4a", ".wav", ".aac", ".flac"]:
+        # allow anyway; faster-whisper handles most formats via ffmpeg
+        pass
 
-    # Transcribe (no downloads, model already cached)
-    segments, _ = model.transcribe(tmp_wav, language="en")
-    text = "".join(seg.text for seg in segments).strip()
+    # Save upload to a temp file
+    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
+        content = await file.read()
+        tmp.write(content)
+        tmp_path = tmp.name
+
+    # Duration via ffprobe (best effort)
+    duration = ffprobe_duration(tmp_path)
+
+    # Transcribe
+    # NOTE: beam_size=1 and vad_filter=True for speed/legibility on CPU Spaces
+    segments, info = model.transcribe(
+        tmp_path,
+        language="en",
+        beam_size=1,
+        vad_filter=True,
+        vad_parameters=dict(min_silence_duration_ms=600)
+    )
+
+    # Concatenate text
+    parts = []
+    for seg in segments:
+        # You can keep timestamps if you want: f"[{seg.start:.2f}-{seg.end:.2f}] {seg.text}"
+        parts.append(seg.text.strip())
+    full_text = " ".join([p for p in parts if p])
+
+    # Compute WPM if possible
+    wpm = estimate_wpm(full_text, duration)
 
-    wpm = None
-    if duration and duration > 0:
-        wc = word_count(text)
-        wpm = round((wc / (duration / 60.0)), 1)
+    try:
+        os.unlink(tmp_path)
+    except Exception:
+        pass
 
-    return TranscribeOut(text=text, duration_sec=duration, wpm=wpm)
+    return TranscribeOut(text=full_text, duration_sec=duration, wpm=wpm)