Update app.py

app.py

@@ -1,6 +1,4 @@
 #!/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
@@ -44,7 +42,6 @@ def ext_from_src(src: str) -> str:
 
 def fetch_bytes(src: str, stream_threshold=STREAM_THRESHOLD, timeout=60) -> bytes:
     if is_remote(src):
-        # HEAD first to make preview/size decision; fallback to GET if HEAD fails
         try:
             h = requests.head(src, timeout=6, allow_redirects=True)
             h.raise_for_status()
@@ -98,19 +95,10 @@ def convert_to_jpeg_bytes(img_bytes: bytes, base_h: int = 480) -> bytes:
 def b64_jpeg(img_bytes: bytes) -> str:
     return base64.b64encode(img_bytes).decode("utf-8")
 
-def extract_best_frames_bytes(media_path: str, sample_count: int = 5, timeout_probe: int = 10, timeout_extract: int = 15) -> list:
+def extract_best_frames_bytes(media_path: str, sample_count: int = 5, timeout_extract: int = 15) -> list:
     if not FFMPEG_BIN or not os.path.exists(media_path):
         return []
-    # Use ffprobe fallback if available; avoid unsupported flags
-    try:
-        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:
-        pass
-    # generate timestamps heuristically
-    timestamps = [0.5, 1.0, 2.0][:sample_count]
+    timestamps = [0.5, 1.0, 2.0, 3.0, 4.0][:sample_count]
     frames = []
     for i, t in enumerate(timestamps):
         fd, tmp = tempfile.mkstemp(suffix=f"_{i}.jpg")
@@ -164,7 +152,7 @@ def extract_text_from_response(res, parts: list):
         return
     try:
         first = choices[0]
-        msg = first.message if hasattr(first, "message") else first.get("message", first)
+        msg = first.message if hasattr(first, "message") else (first.get("message") if isinstance(first, dict) else first)
         if isinstance(msg, dict):
             content = msg.get("content")
         else:
@@ -183,7 +171,7 @@ def chat_complete(client: Mistral, model: str, messages: list) -> str:
     return "".join(parts).strip()
 
 def analyze_image(client: Mistral, img_bytes: bytes, prompt: str) -> str:
-    jpeg = convert_to_jpeg_bytes(img_bytes, base_h=480)
+    jpeg = convert_to_jpeg_bytes(img_bytes, base_h=640)
     b64 = b64_jpeg(jpeg)
     msgs = build_messages_for_image(prompt, b64)
     return chat_complete(client, PIXTRAL_MODEL, msgs)
@@ -240,10 +228,7 @@ def process_media(src: str, custom_prompt: str, api_key: str) -> str:
         try:
             try:
                 file_id = upload_file_to_mistral(client, tmp_path, filename=os.path.basename(src.split("?")[0]))
-                extra = (
-                    f"Uploaded video to Mistral Files with id: {file_id}\n\n"
-                    "Instruction: Analyze the video contents using the uploaded file id. Do not invent frames not present."
-                )
+                extra = f"Uploaded video to Mistral Files with id: {file_id}\n\nInstruction: Analyze the video contents using the uploaded file id. Do not invent frames not present."
                 msgs = build_messages_for_text(prompt, extra)
                 return chat_complete(client, VIDEO_MODEL, msgs)
             except Exception:
@@ -260,39 +245,36 @@ 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
+        return gr.update(value=None, visible=False), gr.update(value=None, visible=False)
     if not is_remote(url) and os.path.exists(url):
         ext = ext_from_src(url)
         if ext in VIDEO_EXTS:
-            return None, os.path.abspath(url), "Video"
+            return gr.update(value=None, visible=False), gr.update(value=os.path.abspath(url), visible=True)
         if ext in IMAGE_EXTS:
             try:
                 img = Image.open(url)
                 if getattr(img, "is_animated", False):
                     img.seek(0)
-                return img.convert("RGB"), None, "Image"
+                return gr.update(value=img.convert("RGB"), visible=True), gr.update(value=None, visible=False)
             except Exception:
-                return None, None, "Preview failed"
-    # Remote: HEAD then GET fallback; ensure returned types match Gradio Image/Video expectations
+                return gr.update(value=None, visible=False), gr.update(value=None, visible=False)
     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"
+        if h.ok:
+            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 gr.update(value=None, visible=False), gr.update(value=url, visible=True)
     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))
         if getattr(img, "is_animated", False):
             img.seek(0)
-        return img.convert("RGB"), None, "Image"
+        return gr.update(value=img.convert("RGB"), visible=True), gr.update(value=None, visible=False)
     except Exception:
-        return None, None, "Preview failed"
+        return gr.update(value=None, visible=False), gr.update(value=None, visible=False)
 
 with gr.Blocks(title="Flux Multimodal (fixed)", css=css) as demo:
     with gr.Row():
@@ -307,21 +289,9 @@ with gr.Blocks(title="Flux Multimodal (fixed)", css=css) as demo:
         with gr.Column(scale=2):
             final_md = gr.Markdown(value="")
 
-    def preview_update(url):
-        img, vid, label = load_preview(url)
-        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)
-
-    # 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])
-
+    url_input.change(fn=load_preview, inputs=[url_input], outputs=[preview_image, preview_video])
     def submit_wrapper(url, prompt, key):
         return process_media(url, prompt, key)
-
     submit_btn.click(fn=submit_wrapper, inputs=[url_input, custom_prompt, api_key], outputs=[final_md])
 
 if __name__ == "__main__":