Spaces:
Running
Running
from fastapi import FastAPI, File, UploadFile, Form | |
from fastapi.responses import JSONResponse | |
from fastapi.middleware.cors import CORSMiddleware | |
from fastapi.staticfiles import StaticFiles | |
import tempfile, subprocess, whisper, os | |
# Set writable cache dir | |
os.environ["XDG_CACHE_HOME"] = "/tmp" | |
# Instantiate app first | |
app = FastAPI() | |
# CORS middleware | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=["*"], | |
allow_methods=["*"], | |
allow_headers=["*"] | |
) | |
# Serve static frontend files | |
app.mount("/", StaticFiles(directory="static", html=True), name="static") | |
# Load whisper model | |
model = whisper.load_model("base") | |
async def analyze(file: UploadFile = File(None), url: str = Form(None)): | |
if not file and not url: | |
return JSONResponse({"error": "No input provided"}, status_code=400) | |
# Download or save uploaded file | |
if url: | |
tmp = tempfile.NamedTemporaryFile(delete=False, suffix='.mp4') | |
subprocess.run(["yt-dlp", "-o", tmp.name, url], check=True) | |
path = tmp.name | |
else: | |
tmp = tempfile.NamedTemporaryFile(delete=False, suffix=file.filename) | |
tmp.write(await file.read()) | |
tmp.close() | |
path = tmp.name | |
# Extract audio using ffmpeg | |
wav_path = path + ".wav" | |
subprocess.run(["ffmpeg", "-y", "-i", path, "-ar", "16000", wav_path], | |
stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL) | |
# Transcribe and detect cues | |
result = model.transcribe(wav_path) | |
transcript = [seg["text"].strip() for seg in result["segments"]] | |
flags = [] | |
for seg in result["segments"]: | |
if "buy" in seg["text"].lower() and seg["avg_logprob"] < -1: | |
flags.append({ | |
"type": "keyword_lowprob", | |
"timestamp": f"{seg['start']:.02f}s", | |
"content": seg['text'] | |
}) | |
summary = "No obvious subliminals" if not flags else "⚠ Potential subliminal cues found" | |
return { | |
"summary": summary, | |
"transcript": transcript, | |
"subliminal_flags": flags | |
} | |