Add analyze_video_async: background analysis with callback to AdVisor, single GPU lock

app.py

@@ -9,9 +9,11 @@ import os
 import subprocess
 import sys
 import tempfile
+import threading
 import time
 import traceback
 from pathlib import Path
+from urllib.parse import urlparse
 
 # Disable hf_transfer — uvx sandboxes don't have it
 os.environ["HF_HUB_ENABLE_HF_TRANSFER"] = "0"
@@ -464,8 +466,18 @@ def trim_video(input_path: str, max_seconds: int = MAX_DURATION) -> str:
     return input_path
 
 
+# One analysis at a time on the GPU, whether it came from the UI/sync API or
+# from the async queue below. Threads waiting here are the queue.
+ANALYSIS_LOCK = threading.Lock()
+
+
 def analyze_video(video_url: str, filename: str = "video.mp4", ad_context_json: str = "{}") -> str:
-    """Main analysis — uses full TRIBE v2 pipeline with WhisperX on GPU."""
+    """Main analysis — uses full TRIBE v2 pipeline with WhisperX on GPU. Serialized."""
+    with ANALYSIS_LOCK:
+        return _analyze_video_locked(video_url, filename, ad_context_json)
+
+
+def _analyze_video_locked(video_url: str, filename: str = "video.mp4", ad_context_json: str = "{}") -> str:
     try:
         # Parse ad context for tailored insights
         try:
@@ -567,6 +579,84 @@ def analyze_video(video_url: str, filename: str = "video.mp4", ad_context_json:
         return json.dumps({"error": error_msg, "traceback": traceback.format_exc()})
 
 
+# ---------------------------------------------------------------------------
+# Async API for AdVisor: accept a job, return at once, analyse in the
+# background, POST the report to the caller's callback URL.
+#
+# Only hosts listed in ADVISOR_CALLBACK_HOSTS (comma-separated, default: the
+# AdVisor Supabase project) may receive callbacks, so this public endpoint
+# cannot be pointed at arbitrary servers. The caller supplies job_ref and
+# job_token; both are echoed back and the caller verifies the token, so no
+# shared secret has to live in this Space.
+# ---------------------------------------------------------------------------
+ALLOWED_CALLBACK_HOSTS = {
+    h.strip().lower()
+    for h in os.environ.get("ADVISOR_CALLBACK_HOSTS", "emyiuzsgvnnzgtqwmttb.supabase.co").split(",")
+    if h.strip()
+}
+CALLBACK_RETRIES = 4
+
+
+def _post_callback(callback_url: str, payload: dict) -> bool:
+    delay = 5
+    for attempt in range(1, CALLBACK_RETRIES + 1):
+        try:
+            resp = requests.post(
+                callback_url,
+                json=payload,
+                headers={"Content-Type": "application/json", "X-Job-Token": payload.get("job_token", "")},
+                timeout=60,
+            )
+            if resp.ok:
+                print(f"[async] callback delivered for {payload.get('job_ref')} ({resp.status_code})")
+                return True
+            print(f"[async] callback HTTP {resp.status_code} for {payload.get('job_ref')}: {resp.text[:200]}")
+            if 400 <= resp.status_code < 500 and resp.status_code != 429:
+                return False  # our payload is wrong; retrying will not help
+        except Exception as e:  # network
+            print(f"[async] callback attempt {attempt} failed: {e}")
+        time.sleep(delay)
+        delay = min(delay * 2, 60)
+    return False
+
+
+def _run_async_job(video_url: str, filename: str, ad_context_json: str, callback_url: str, job_ref: str, job_token: str):
+    started = time.time()
+    try:
+        raw = analyze_video(video_url, filename, ad_context_json)
+        report = json.loads(raw)
+        if isinstance(report, dict) and report.get("error"):
+            payload = {"job_ref": job_ref, "job_token": job_token, "status": "failed", "error": str(report.get("error"))[:1000]}
+        else:
+            payload = {"job_ref": job_ref, "job_token": job_token, "status": "complete", "report": report}
+    except Exception as e:  # never let a job die silently
+        payload = {"job_ref": job_ref, "job_token": job_token, "status": "failed", "error": f"{type(e).__name__}: {e}"[:1000]}
+    payload["elapsed_seconds"] = round(time.time() - started, 1)
+    _post_callback(callback_url, payload)
+
+
+def analyze_video_async(video_url: str, filename: str, ad_context_json: str, callback_url: str, job_ref: str, job_token: str) -> str:
+    """Queue an analysis and return immediately. The report arrives at callback_url."""
+    try:
+        host = (urlparse(callback_url).hostname or "").lower()
+        if urlparse(callback_url).scheme != "https" or host not in ALLOWED_CALLBACK_HOSTS:
+            return json.dumps({"accepted": False, "error": f"callback host not allowed: {host or '(none)'}"})
+        if not video_url or not job_ref or not job_token:
+            return json.dumps({"accepted": False, "error": "video_url, job_ref and job_token are required"})
+        t = threading.Thread(
+            target=_run_async_job,
+            args=(video_url, filename or "video.mp4", ad_context_json or "{}", callback_url, job_ref, job_token),
+            daemon=True,
+            name=f"tribe-async-{job_ref[:12]}",
+        )
+        t.start()
+        waiting = sum(1 for th in threading.enumerate() if th.name.startswith("tribe-async-"))
+        print(f"[async] accepted job {job_ref} ({waiting} in flight)")
+        return json.dumps({"accepted": True, "job_ref": job_ref, "in_flight": waiting})
+    except Exception as e:
+        return json.dumps({"accepted": False, "error": f"{type(e).__name__}: {e}"})
+
+
 app = gr.Blocks(title="TRIBE v2 Neural Engagement Analyzer")
 with app:
     gr.Markdown("# TRIBE v2 Neural Engagement Analyzer")
@@ -578,6 +668,23 @@ with app:
     analyze_btn = gr.Button("Analyze", variant="primary")
     output = gr.JSON(label="Neural Engagement Report")
     analyze_btn.click(fn=analyze_video, inputs=[video_url_input, filename_input, ad_context_input], outputs=output)
+
+    # Hidden controls back the async API (gradio_api/call/analyze_video_async).
+    with gr.Row(visible=False):
+        async_video_url = gr.Textbox(label="video_url")
+        async_filename = gr.Textbox(label="filename", value="video.mp4")
+        async_ad_context = gr.Textbox(label="ad_context_json", value="{}")
+        async_callback_url = gr.Textbox(label="callback_url")
+        async_job_ref = gr.Textbox(label="job_ref")
+        async_job_token = gr.Textbox(label="job_token")
+        async_btn = gr.Button("Queue")
+        async_out = gr.Textbox(label="ack")
+    async_btn.click(
+        fn=analyze_video_async,
+        inputs=[async_video_url, async_filename, async_ad_context, async_callback_url, async_job_ref, async_job_token],
+        outputs=async_out,
+        api_name="analyze_video_async",
+    )
 app.queue()
 if __name__ == "__main__":
     app.launch()