Update app.py

app.py

@@ -1,4 +1,6 @@
 #!/usr/bin/env python3
+# Single-file Gradio app: reliable preview, image->image model, video->video model or frame fallback,
+# streaming disabled for safety; preserves default prompt text unchanged.
 import os
 import subprocess
 import tempfile
@@ -42,9 +44,9 @@ def ext_from_src(src: str) -> str:
 
 def fetch_bytes(src: str, stream_threshold=STREAM_THRESHOLD, timeout=60) -> bytes:
     if is_remote(src):
-        # Try HEAD first to detect size; fallback to GET
+        # HEAD first to make preview/size decision; fallback to GET if HEAD fails
         try:
-            h = requests.head(src, timeout=8, allow_redirects=True)
+            h = requests.head(src, timeout=6, allow_redirects=True)
             h.raise_for_status()
             cl = h.headers.get("content-length")
             if cl and int(cl) > stream_threshold:
@@ -62,7 +64,6 @@ def fetch_bytes(src: str, stream_threshold=STREAM_THRESHOLD, timeout=60) -> byte
                     finally:
                         try: os.remove(path)
                         except Exception: pass
-            # small content or no content-length: simple GET
         except Exception:
             pass
         with requests.get(src, timeout=timeout) as r:
@@ -100,22 +101,21 @@ def b64_jpeg(img_bytes: bytes) -> str:
 def extract_best_frames_bytes(media_path: str, sample_count: int = 5, timeout_probe: int = 10, timeout_extract: int = 15) -> list:
     if not FFMPEG_BIN or not os.path.exists(media_path):
         return []
-    probe_cmd = [FFMPEG_BIN, "-v", "error", "-show_entries", "format=duration",
-                 "-of", "default=noprint_wrappers=1:nokey=1", media_path]
+    # Use ffprobe fallback if available; avoid unsupported flags
     try:
-        out = subprocess.check_output(probe_cmd, timeout=timeout_probe).strip()
-        duration = float(out) if out else None
+        probe_cmd = [FFMPEG_BIN, "-v", "error", "-i", media_path, "-hide_banner"]
+        subprocess.run(probe_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=timeout_probe)
+        # best-effort timestamps if probing failed to get duration
+        # We'll still try a few timestamps
     except Exception:
-        duration = None
-    if duration and duration > 0:
-        timestamps = [(duration * i) / (sample_count + 1) for i in range(1, sample_count + 1)]
-    else:
-        timestamps = [0.5, 1.0, 2.0][:sample_count]
+        pass
+    # generate timestamps heuristically
+    timestamps = [0.5, 1.0, 2.0][:sample_count]
     frames = []
     for i, t in enumerate(timestamps):
         fd, tmp = tempfile.mkstemp(suffix=f"_{i}.jpg")
         os.close(fd)
-        cmd = [FFMPEG_BIN, "-nostdin", "-y", "-i", media_path, "-ss", str(t), "-frames:v", "1", "-q:v", "2", tmp]
+        cmd = [FFMPEG_BIN, "-nostdin", "-y", "-ss", str(t), "-i", media_path, "-frames:v", "1", "-q:v", "2", tmp]
         try:
             subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=timeout_extract)
             if os.path.exists(tmp) and os.path.getsize(tmp) > 0:
@@ -156,14 +156,15 @@ def build_messages_for_text(prompt: str, extra: str) -> list:
 
 def extract_text_from_response(res, parts: list):
     try:
-        choices = getattr(res, "choices", None) or res.get("choices", [])
+        choices = getattr(res, "choices", None) or (res.get("choices") if isinstance(res, dict) else [])
     except Exception:
         choices = []
     if not choices:
         parts.append(str(res))
         return
     try:
-        msg = choices[0].message
+        first = choices[0]
+        msg = first.message if hasattr(first, "message") else first.get("message", first)
         if isinstance(msg, dict):
             content = msg.get("content")
         else:
@@ -210,7 +211,8 @@ def process_media(src: str, custom_prompt: str, api_key: str) -> str:
     is_video = ext in VIDEO_EXTS
     if is_remote(src):
         try:
-            h = requests.head(src, timeout=8, allow_redirects=True)
+            h = requests.head(src, timeout=6, allow_redirects=True)
+            h.raise_for_status()
             ctype = (h.headers.get("content-type") or "").lower()
             if ctype.startswith("video/"):
                 is_video = True; is_image = False
@@ -259,6 +261,7 @@ css = ".preview_media img, .preview_media video { max-width: 100%; height: auto;
 def load_preview(url: str):
     if not url:
         return None, None, ""
+    # Local file path handling
     if not is_remote(url) and os.path.exists(url):
         ext = ext_from_src(url)
         if ext in VIDEO_EXTS:
@@ -271,15 +274,17 @@ def load_preview(url: str):
                 return img.convert("RGB"), None, "Image"
             except Exception:
                 return None, None, "Preview failed"
-    # Remote: prefer HEAD, then GET fallback for image bytes
+    # Remote: HEAD then GET fallback; ensure returned types match Gradio Image/Video expectations
     try:
         h = requests.head(url, timeout=6, allow_redirects=True)
+        h.raise_for_status()
         ctype = (h.headers.get("content-type") or "").lower()
         if ctype.startswith("video/") or any(url.lower().split("?")[0].endswith(ext) for ext in VIDEO_EXTS):
             return None, url, "Video"
     except Exception:
         pass
     try:
+        # GET and try to open as image
         r = requests.get(url, timeout=15)
         r.raise_for_status()
         img = Image.open(BytesIO(r.content))
@@ -304,9 +309,15 @@ with gr.Blocks(title="Flux Multimodal (fixed)", css=css) as demo:
 
     def preview_update(url):
         img, vid, label = load_preview(url)
-        return (img if label == "Image" else None, vid if label == "Video" else None, label == "Image", label == "Video")
+        show_img = label == "Image"
+        show_vid = label == "Video"
+        # Return values in order matching outputs assigned below
+        return (img if show_img else None, vid if show_vid else None, show_img, show_vid)
 
-    url_input.change(fn=preview_update, inputs=[url_input], outputs=[preview_image, preview_video, preview_image, preview_video])
+    # wire change: outputs must match component types exactly
+    url_input.change(fn=preview_update,
+                     inputs=[url_input],
+                     outputs=[preview_image, preview_video, preview_image, preview_video])
 
     def submit_wrapper(url, prompt, key):
         return process_media(url, prompt, key)