Update app_server.py

app_server.py

@@ -1,76 +1,57 @@
-# app_server.py — BubbleGuard API + Dating-style Web Chat (Static UI)
+# app_server.py — BubbleGuard API + Web UI
 # Version: 1.7.1 (/api/* routes + repo-root UI support)
 
 import io, os, re, uuid, pathlib, tempfile, subprocess, unicodedata
 from typing import Dict, Optional
-
 from fastapi import FastAPI, UploadFile, File, Form, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.staticfiles import StaticFiles
 from fastapi.responses import PlainTextResponse
-
 import torch, joblib, torchvision
 from torchvision import transforms
 from transformers import RobertaTokenizerFast, AutoModelForSequenceClassification
 from PIL import Image
 from faster_whisper import WhisperModel
 
-# -------------------------- Paths & Config --------------------------
 BASE = pathlib.Path(__file__).resolve().parent
 TEXT_DIR = BASE / "Text"
 IMG_DIR  = BASE / "Image"
 AUD_DIR  = BASE / "Audio"
-STATIC_DIR = BASE / "static"
 
 DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
-
 IMG_UNSAFE_THR   = float(os.getenv("IMG_UNSAFE_THR", "0.5"))
 IMG_UNSAFE_INDEX = int(os.getenv("IMG_UNSAFE_INDEX", "1"))
-
 WHISPER_MODEL_NAME = os.getenv("WHISPER_MODEL", "base")
-
 TEXT_UNSAFE_THR        = float(os.getenv("TEXT_UNSAFE_THR", "0.60"))
 SHORT_MSG_MAX_TOKENS   = int(os.getenv("SHORT_MSG_MAX_TOKENS", "6"))
 SHORT_MSG_UNSAFE_THR   = float(os.getenv("SHORT_MSG_UNSAFE_THR", "0.90"))
-
 AUDIO_UNSAFE_INDEX = int(os.getenv("AUDIO_UNSAFE_INDEX", "1"))
 AUDIO_UNSAFE_THR   = float(os.getenv("AUDIO_UNSAFE_THR", "0.50"))
 
 app = FastAPI(title="BubbleGuard API", version="1.7.1")
+app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
 
-app.add_middleware(
-    CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]
-)
-
-# -------------------------- Text Classifier -------------------------
-if not TEXT_DIR.exists():
-    raise RuntimeError(f"Text model dir not found: {TEXT_DIR}. Run download_assets first.")
-
+# ---------- Text model ----------
+if not TEXT_DIR.exists(): raise RuntimeError(f"Missing Text dir: {TEXT_DIR}")
 tok = RobertaTokenizerFast.from_pretrained(TEXT_DIR, local_files_only=True)
 txtM = AutoModelForSequenceClassification.from_pretrained(TEXT_DIR, local_files_only=True).to(DEVICE).eval()
 
-# -------- Label mapping (robust) --------
-SAFE_LABEL_HINTS   = {"safe", "ok", "clean", "benign", "non-toxic", "non_toxic", "non toxic"}
-UNSAFE_LABEL_HINTS = {"unsafe", "toxic", "abuse", "harm", "offense", "nsfw", "not_safe", "not safe"}
+SAFE_LABEL_HINTS   = {"safe","ok","clean","benign","non-toxic","non_toxic","non toxic"}
+UNSAFE_LABEL_HINTS = {"unsafe","toxic","abuse","harm","offense","nsfw","not_safe","not safe"}
 
-def _infer_ids_by_name(model) -> (Optional[int], Optional[int]):
+def _infer_ids_by_name(model):
     try:
-        id2label = getattr(model.config, "id2label", None)
-        if not isinstance(id2label, dict):
-            return None, None
+        id2label = getattr(model.config, "id2label", {})
         norm = {}
         for k, v in id2label.items():
-            try:
-                ki = int(k)
-            except Exception:
-                try:
-                    ki = int(str(k).strip())
-                except Exception:
-                    continue
+            try: ki = int(k)
+            except Exception: 
+                try: ki = int(str(k).strip())
+                except Exception: continue
             norm[ki] = str(v).lower()
         s = u = None
         for i, name in norm.items():
-            if any(h in name for h in SAFE_LABEL_HINTS):   s = i
+            if any(h in name for h in SAFE_LABEL_HINTS): s = i
             if any(h in name for h in UNSAFE_LABEL_HINTS): u = i
         if s is not None and u is None: u = 1 - s
         if u is not None and s is None: s = 1 - u
@@ -79,148 +60,92 @@ def _infer_ids_by_name(model) -> (Optional[int], Optional[int]):
         return None, None
 
 @torch.no_grad()
-def _infer_ids_by_probe(model, tok, device) -> (int, int):
-    samples = ["hi", "hello", "how are you", "nice to meet you", "thanks"]
-    enc = tok(samples, return_tensors="pt", truncation=True, padding=True, max_length=64)
-    enc = {k: v.to(device) for k, v in enc.items()}
+def _infer_ids_by_probe(model, tok, device):
+    enc = tok(["hi","hello","how are you","nice to meet you","thanks"], return_tensors="pt", truncation=True, padding=True, max_length=64)
+    enc = {k:v.to(device) for k,v in enc.items()}
     probs = torch.softmax(model(**enc).logits, dim=-1).mean(0)
-    safe_idx = int(torch.argmax(probs).item()); unsafe_idx = 1 - safe_idx
-    return safe_idx, unsafe_idx
+    s = int(torch.argmax(probs)); return s, 1 - s
 
-def _resolve_safe_unsafe_ids(model, tok, device) -> (int, int):
-    s_env = os.getenv("SAFE_ID"); u_env = os.getenv("UNSAFE_ID")
-    if s_env is not None and u_env is not None:
-        return int(s_env), int(u_env)
+def _resolve_ids(model, tok, device):
+    s_env, u_env = os.getenv("SAFE_ID"), os.getenv("UNSAFE_ID")
+    if s_env is not None and u_env is not None: return int(s_env), int(u_env)
     s, u = _infer_ids_by_name(model)
-    if s is not None and u is not None:
-        return s, u
-    return _infer_ids_by_probe(model, tok, device)
+    return (s, u) if (s is not None and u is not None) else _infer_ids_by_probe(model, tok, device)
 
-SAFE_ID, UNSAFE_ID = _resolve_safe_unsafe_ids(txtM, tok, DEVICE)
-print(f"[BubbleGuard] SAFE_ID={SAFE_ID}  UNSAFE_ID={UNSAFE_ID}  id2label={getattr(txtM.config, 'id2label', None)}")
+SAFE_ID, UNSAFE_ID = _resolve_ids(txtM, tok, DEVICE)
+print(f"[BubbleGuard] SAFE_ID={SAFE_ID}  UNSAFE_ID={UNSAFE_ID}  id2label={getattr(txtM.config,'id2label',None)}")
 
-# ------------------------ Text utils ------------------------
-def normalize(text: str) -> str:
-    if not isinstance(text, str): return ""
-    t = unicodedata.normalize("NFKC", text)
-    t = t.replace("’","'").replace("‘","'").replace("“",'"').replace("”",'"')
-    t = re.sub(r"[^a-z0-9\s']", " ", t.lower())
-    return re.sub(r"\s+", " ", t).strip()
+def normalize(t: str) -> str:
+    if not isinstance(t, str): return ""
+    t = unicodedata.normalize("NFKC", t).replace("’","'").replace("‘","'").replace("“",'"').replace("”",'"')
+    t = re.sub(r"[^a-z0-9\s']", " ", t.lower()); return re.sub(r"\s+", " ", t).strip()
 
-SAFE_PHRASES = [r"^i don'?t$", r"^i do not$", r"^don'?t$", r"^no$", r"^not really$", r"^i woulde?n'?t$", r"^i don'?t like$"]
-SAFE_RE = re.compile("|".join(SAFE_PHRASES))
+SAFE_RE = re.compile("|".join([r"^i don'?t$", r"^i do not$", r"^don'?t$", r"^no$", r"^not really$", r"^i woulde?n'?t$", r"^i don'?t like$"]))
 NEGATION_ONLY = re.compile(r"^(?:i\s+)?(?:do\s+not|don'?t|no|not)$")
 NEUTRAL_DISLIKE = re.compile(r"^i don'?t like(?:\s+to)?\b")
-
-SENSITIVE_TERMS = {"people","you","him","her","them","men","women","girls","boys",
-                   "muslim","christian","jew","jews","black","white","asian",
-                   "gay","lesbian","trans","transgender","disabled",
-                   "immigrants","refugees","poor","old","elderly","fat","skinny"}
+SENSITIVE_TERMS = {"people","you","him","her","them","men","women","girls","boys","muslim","christian","jew","jews","black","white","asian","gay","lesbian","trans","transgender","disabled","immigrants","refugees","poor","old","elderly","fat","skinny"}
 PROFANITY_TERMS = {"fuck","shit","bitch","pussy","dick","cunt","slut","whore"}
-
-GREETINGS = [r"^hi$", r"^hello$", r"^hey(?: there)?$", r"^how are (?:you|u)\b.*$",
-             r"^good (?:morning|afternoon|evening)\b.*$", r"^what'?s up\b.*$", r"^how'?s it going\b.*$"]
-GREETING_RE = re.compile("|".join(GREETINGS))
+GREETING_RE = re.compile("|".join([r"^hi$", r"^hello$", r"^hey(?: there)?$", r"^how are (?:you|u)\b.*$", r"^good (?:morning|afternoon|evening)\b.*$", r"^what'?s up\b.*$", r"^how'?s it going\b.*$"]))
 
 @torch.no_grad()
 def text_safe_payload(text: str) -> Dict:
     clean = normalize(text); toks = clean.split()
-
-    if len(toks) == 1 and toks[0] in PROFANITY_TERMS:
-        p = [0.0,0.0]; p[UNSAFE_ID]=1.0
-        return {"safe":False,"unsafe_prob":1.0,"label":"UNSAFE","probs":p,"tokens":1,"reason":"profanity_single_word"}
-
-    if len(toks) <= SHORT_MSG_MAX_TOKENS and any(t in PROFANITY_TERMS for t in toks):
-        p = [0.0,0.0]; p[UNSAFE_ID]=1.0
-        return {"safe":False,"unsafe_prob":1.0,"label":"UNSAFE","probs":p,"tokens":len(toks),"reason":"profanity_short_text"}
-
+    if len(toks)==1 and toks[0] in PROFANITY_TERMS:
+        p=[0,0]; p[UNSAFE_ID]=1.; return {"safe":False,"unsafe_prob":1.0,"label":"UNSAFE","probs":p,"tokens":1,"reason":"profanity_single_word"}
+    if len(toks)<=SHORT_MSG_MAX_TOKENS and any(t in PROFANITY_TERMS for t in toks):
+        p=[0,0]; p[UNSAFE_ID]=1.; return {"safe":False,"unsafe_prob":1.0,"label":"UNSAFE","probs":p,"tokens":len(toks),"reason":"profanity_short_text"}
     if SAFE_RE.match(clean) or NEGATION_ONLY.match(clean) or GREETING_RE.match(clean):
-        p=[0.0,0.0]; p[SAFE_ID]=1.0
-        return {"safe":True,"unsafe_prob":0.0,"label":"SAFE","probs":p,"tokens":len(toks),"reason":"allow_or_greeting"}
-
+        p=[0,0]; p[SAFE_ID]=1.; return {"safe":True,"unsafe_prob":0.0,"label":"SAFE","probs":p,"tokens":len(toks),"reason":"allow_or_greeting"}
     if NEUTRAL_DISLIKE.match(clean):
         if not any(t in clean for t in SENSITIVE_TERMS) and not any(t in clean for t in PROFANITY_TERMS):
-            enc = tok(text, return_tensors="pt", truncation=True, padding=True, max_length=512)
-            enc = {k: v.to(DEVICE) for k, v in enc.items()}
-            probs = torch.softmax(txtM(**enc).logits[0], dim=-1).cpu().tolist()
-            up = float(probs[UNSAFE_ID]); safe = up < 0.98
-            return {"safe":bool(safe),"unsafe_prob":up,"label":"SAFE" if safe else "UNSAFE",
-                    "probs":probs,"tokens":int(enc["input_ids"].shape[1]),"reason":"neutral_dislike_relaxed"}
-
-    enc = tok(text, return_tensors="pt", truncation=True, padding=True, max_length=512)
-    enc = {k: v.to(DEVICE) for k, v in enc.items()}
-    logits = txtM(**enc).logits[0]
-    probs = torch.softmax(logits, dim=-1).cpu().tolist()
-    up = float(probs[UNSAFE_ID]); toks = int(enc["input_ids"].shape[1])
-
-    safe = up < (SHORT_MSG_UNSAFE_THR if toks <= SHORT_MSG_MAX_TOKENS else TEXT_UNSAFE_THR)
-    return {"safe":bool(safe),"unsafe_prob":up,"label":str(int(torch.argmax(logits))),
-            "probs":probs,"tokens":toks,"reason":"short_msg_threshold" if toks<=SHORT_MSG_MAX_TOKENS else "global_threshold"}
-
-# -------------------------- Image Classifier ------------------------
+            enc = tok(text, return_tensors="pt", truncation=True, padding=True, max_length=512); enc = {k:v.to(DEVICE) for k,v in enc.items()}
+            probs = torch.softmax(txtM(**enc).logits[0], dim=-1).cpu().tolist(); up=float(probs[UNSAFE_ID])
+            return {"safe": up<0.98, "unsafe_prob": up, "label":"SAFE" if up<0.98 else "UNSAFE", "probs": probs, "tokens": int(enc["input_ids"].shape[1]), "reason":"neutral_dislike_relaxed"}
+    enc = tok(text, return_tensors="pt", truncation=True, padding=True, max_length=512); enc = {k:v.to(DEVICE) for k,v in enc.items()}
+    logits = txtM(**enc).logits[0]; probs = torch.softmax(logits, dim=-1).cpu().tolist(); up=float(probs[UNSAFE_ID]); n=int(enc["input_ids"].shape[1])
+    thr = SHORT_MSG_UNSAFE_THR if n<=SHORT_MSG_MAX_TOKENS else TEXT_UNSAFE_THR
+    return {"safe": up<thr, "unsafe_prob": up, "label": str(int(torch.argmax(logits))), "probs": probs, "tokens": n, "reason": "short_msg_threshold" if n<=SHORT_MSG_MAX_TOKENS else "global_threshold"}
+
+# ---------- Image ----------
 class SafetyResNet(torch.nn.Module):
     def __init__(self):
         super().__init__()
         base = torchvision.models.resnet50(weights=None)
         self.feature_extractor = torch.nn.Sequential(*list(base.children())[:8])
         self.pool = torch.nn.AdaptiveAvgPool2d(1)
-        self.classifier = torch.nn.Sequential(
-            torch.nn.Linear(2048, 512), torch.nn.ReLU(True), torch.nn.Dropout(0.30), torch.nn.Linear(512, 2)
-        )
-    def forward(self, x):
-        x = self.pool(self.feature_extractor(x))
-        return self.classifier(torch.flatten(x, 1))
+        self.cls  = torch.nn.Sequential(torch.nn.Linear(2048,512), torch.nn.ReLU(True), torch.nn.Dropout(0.30), torch.nn.Linear(512,2))
+    def forward(self,x): return self.cls(torch.flatten(self.pool(self.feature_extractor(x)),1))
 
-if not IMG_DIR.exists():
-    raise RuntimeError(f"Image model dir not found: {IMG_DIR}. Run download_assets first.")
-
-imgM = SafetyResNet().to(DEVICE)
-imgM.load_state_dict(torch.load(IMG_DIR / "resnet_safety_classifier.pth", map_location=DEVICE), strict=True)
-imgM.eval()
-
-img_tf = transforms.Compose([
-    transforms.Resize(256, interpolation=transforms.InterpolationMode.BILINEAR),
-    transforms.CenterCrop(224),
-    transforms.ToTensor(),
-    transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
-])
+if not IMG_DIR.exists(): raise RuntimeError(f"Missing Image dir: {IMG_DIR}")
+imgM = SafetyResNet().to(DEVICE); imgM.load_state_dict(torch.load(IMG_DIR/"resnet_safety_classifier.pth", map_location=DEVICE), strict=True); imgM.eval()
+img_tf = transforms.Compose([ transforms.Resize(256, interpolation=transforms.InterpolationMode.BILINEAR), transforms.CenterCrop(224), transforms.ToTensor(), transforms.Normalize([0.485,0.456,0.406],[0.229,0.224,0.225]) ])
 
 @torch.no_grad()
 def image_safe_payload(pil: Image.Image) -> Dict:
     x = img_tf(pil.convert("RGB")).unsqueeze(0).to(DEVICE)
     probs = torch.softmax(imgM(x)[0], dim=0).cpu().tolist()
-    up = float(probs[IMG_UNSAFE_INDEX])
-    return {"safe": up < IMG_UNSAFE_THR, "unsafe_prob": up, "probs": probs}
+    up = float(probs[IMG_UNSAFE_INDEX]); return {"safe": up<IMG_UNSAFE_THR, "unsafe_prob": up, "probs": probs}
 
-# -------------------------- Audio (ASR -> NLP) ----------------------
-compute_type = "float16" if DEVICE == "cuda" else "int8"
+# ---------- Audio ----------
+compute_type = "float16" if DEVICE=="cuda" else "int8"
 asr = WhisperModel(WHISPER_MODEL_NAME, device=DEVICE, compute_type=compute_type)
+if not AUD_DIR.exists(): raise RuntimeError(f"Missing Audio dir: {AUD_DIR}")
+text_clf = joblib.load(AUD_DIR/"text_pipeline_balanced.joblib")
 
-if not AUD_DIR.exists():
-    raise RuntimeError(f"Audio pipeline dir not found: {AUD_DIR}. Run download_assets first.")
-text_clf = joblib.load(AUD_DIR / "text_pipeline_balanced.joblib")
-
-def _ffmpeg_to_wav(src_bytes: bytes) -> bytes:
+def _ffmpeg_to_wav(src: bytes) -> bytes:
     with tempfile.TemporaryDirectory() as td:
-        in_path = pathlib.Path(td) / f"in-{uuid.uuid4().hex}.bin"
-        out_path = pathlib.Path(td) / "out.wav"
-        in_path.write_bytes(src_bytes)
-        cmd = ["ffmpeg","-y","-i",str(in_path),"-ac","1","-ar","16000",str(out_path)]
+        ip = pathlib.Path(td)/"in"; op = pathlib.Path(td)/"out.wav"; ip.write_bytes(src)
         try:
-            subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-            return out_path.read_bytes()
-        except FileNotFoundError as e:
-            raise RuntimeError("FFmpeg not found on PATH.") from e
-        except subprocess.CalledProcessError:
-            return src_bytes
+            subprocess.run(["ffmpeg","-y","-i",str(ip),"-ac","1","-ar","16000",str(op)], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            return op.read_bytes()
+        except FileNotFoundError as e: raise RuntimeError("FFmpeg not found on PATH.") from e
+        except subprocess.CalledProcessError: return src
 
-def _transcribe_wav_bytes(wav_bytes: bytes) -> str:
-    td = tempfile.mkdtemp()
-    p = pathlib.Path(td) / "in.wav"
+def _transcribe_wav_bytes(w: bytes) -> str:
+    td = tempfile.mkdtemp(); p = pathlib.Path(td)/"in.wav"
     try:
-        p.write_bytes(wav_bytes)
-        segments, _ = asr.transcribe(str(p), beam_size=5, language="en")
-        return " ".join(s.text for s in segments).strip()
+        p.write_bytes(w); segs,_ = asr.transcribe(str(p), beam_size=5, language="en")
+        return " ".join(s.text for s in segs).strip()
     finally:
         try: p.unlink(missing_ok=True)
         except Exception: pass
@@ -228,74 +153,48 @@ def _transcribe_wav_bytes(wav_bytes: bytes) -> str:
         except Exception: pass
 
 def audio_safe_from_bytes(raw: bytes) -> Dict:
-    wav = _ffmpeg_to_wav(raw)
-    text = _transcribe_wav_bytes(wav)
-    proba = text_clf.predict_proba([text])[0].tolist()
-    up = float(proba[AUDIO_UNSAFE_INDEX])
-    return {"safe": up < AUDIO_UNSAFE_THR, "unsafe_prob": up, "text": text, "probs": proba}
+    wav = _ffmpeg_to_wav(raw); text = _transcribe_wav_bytes(wav)
+    proba = text_clf.predict_proba([text])[0].tolist(); up=float(proba[AUDIO_UNSAFE_INDEX])
+    return {"safe": up<AUDIO_UNSAFE_THR, "unsafe_prob": up, "text": text, "probs": proba}
 
-# ------------------------------ Routes (under /api) ------------------------------
+# ---------- Routes (/api/*) ----------
 @app.get("/api/health")
 def health():
-    return {
-        "ok": True,
-        "device": DEVICE,
-        "whisper": WHISPER_MODEL_NAME,
-        "img": {"unsafe_threshold": IMG_UNSAFE_THR, "unsafe_index": IMG_UNSAFE_INDEX},
-        "text_thresholds": {
-            "TEXT_UNSAFE_THR": TEXT_UNSAFE_THR,
-            "SHORT_MSG_MAX_TOKENS": SHORT_MSG_MAX_TOKENS,
-            "SHORT_MSG_UNSAFE_THR": SHORT_MSG_UNSAFE_THR,
-        },
-        "audio": {"unsafe_index": AUDIO_UNSAFE_INDEX, "unsafe_threshold": AUDIO_UNSAFE_THR},
-        "safe_unsafe_indices(text_model)": {"SAFE_ID": SAFE_ID, "UNSAFE_ID": UNSAFE_ID},
-    }
+    return {"ok":True,"device":DEVICE,"whisper":WHISPER_MODEL_NAME,
+            "img":{"unsafe_threshold":IMG_UNSAFE_THR,"unsafe_index":IMG_UNSAFE_INDEX},
+            "text_thresholds":{"TEXT_UNSAFE_THR":TEXT_UNSAFE_THR,"SHORT_MSG_MAX_TOKENS":SHORT_MSG_MAX_TOKENS,"SHORT_MSG_UNSAFE_THR":SHORT_MSG_UNSAFE_THR},
+            "audio":{"unsafe_index":AUDIO_UNSAFE_INDEX,"unsafe_threshold":AUDIO_UNSAFE_THR},
+            "safe_unsafe_indices(text_model)":{"SAFE_ID":SAFE_ID,"UNSAFE_ID":UNSAFE_ID}}
 
 @app.post("/api/check_text")
 def check_text(text: str = Form(...)):
-    if not text or not text.strip():
-        raise HTTPException(400, "Empty text")
-    try:
-        return text_safe_payload(text)
-    except Exception as e:
-        raise HTTPException(500, f"Text screening error: {e}")
+    if not text.strip(): raise HTTPException(400, "Empty text")
+    try: return text_safe_payload(text)
+    except Exception as e: raise HTTPException(500, f"Text screening error: {e}")
 
 @app.post("/api/check_image")
 async def check_image(file: UploadFile = File(...)):
     data = await file.read()
-    if not data:
-        raise HTTPException(400, "Empty image")
-    try:
-        pil = Image.open(io.BytesIO(data))
-    except Exception:
-        raise HTTPException(400, "Invalid image")
-    try:
-        return image_safe_payload(pil)
-    except Exception as e:
-        raise HTTPException(500, f"Image screening error: {e}")
+    if not data: raise HTTPException(400, "Empty image")
+    try: pil = Image.open(io.BytesIO(data))
+    except Exception: raise HTTPException(400, "Invalid image")
+    try: return image_safe_payload(pil)
+    except Exception as e: raise HTTPException(500, f"Image screening error: {e}")
 
 @app.post("/api/check_audio")
 async def check_audio(file: UploadFile = File(...)):
     raw = await file.read()
-    if not raw:
-        raise HTTPException(400, "Empty audio")
-    try:
-        return audio_safe_from_bytes(raw)
-    except RuntimeError as e:
-        raise HTTPException(500, f"{e}")
-    except Exception as e:
-        raise HTTPException(500, f"Audio processing error: {e}")
+    if not raw: raise HTTPException(400, "Empty audio")
+    try: return audio_safe_from_bytes(raw)
+    except RuntimeError as e: raise HTTPException(500, f"{e}")
+    except Exception as e: raise HTTPException(500, f"Audio processing error: {e}")
 
-# --------------------------- Static Mount ---------------------------
-# Serve UI from /static if present; otherwise from repo root (index.html at root).
+# ---------- Static ----------
 static_dir = BASE / "static"
-root_index = BASE / "index.html"
-
 if static_dir.exists():
     app.mount("/", StaticFiles(directory=str(static_dir), html=True), name="static")
-elif root_index.exists():
+elif (BASE/"index.html").exists():
     app.mount("/", StaticFiles(directory=str(BASE), html=True), name="static-root")
 else:
     @app.get("/", response_class=PlainTextResponse)
-    def _root_fallback():
-        return "BubbleGuard API is running. Put index.html at repo root or add a 'static/' folder."
+    def _root_fallback(): return "BubbleGuard API is running. Add index.html to repo root."