|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
os.environ["XDG_CACHE_HOME"] = "/tmp" |
|
|
|
|
|
|
|
|
app = FastAPI() |
|
|
|
|
|
|
|
|
app.add_middleware( |
|
|
CORSMiddleware, |
|
|
allow_origins=["*"], |
|
|
allow_methods=["*"], |
|
|
allow_headers=["*"] |
|
|
) |
|
|
|
|
|
|
|
|
app.mount("/", StaticFiles(directory="static", html=True), name="static") |
|
|
|
|
|
|
|
|
model = whisper.load_model("base") |
|
|
|
|
|
@app.post("/api/analyze") |
|
|
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) |
|
|
|
|
|
|
|
|
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 |
|
|
|
|
|
|
|
|
wav_path = path + ".wav" |
|
|
subprocess.run(["ffmpeg", "-y", "-i", path, "-ar", "16000", wav_path], |
|
|
stderr=subprocess.DEVNULL, stdout=subprocess.DEVNULL) |
|
|
|
|
|
|
|
|
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 |
|
|
} |
|
|
|