Update app.py

app.py

@@ -417,7 +417,7 @@ def create_demo():
             with gr.Column(scale=1):
                 preview_image = gr.Image(label="Preview Image", type="pil", elem_classes="preview_media", visible=False)
                 preview_video = gr.Video(label="Preview Video", elem_classes="preview_media", visible=False, format="mp4")
-                preview_status = gr.Textbox(label="Preview status", interactive=False, lines=1, value="", visible=True)
+                preview_status = gr.Textbox(label="Preview status", interactive=False, lines=2, value="", visible=True)
             with gr.Column(scale=2):
                 url_input = gr.Textbox(label="Image / Video URL", placeholder="https://...", lines=1)
                 with gr.Accordion("Prompt (optional)", open=False):
@@ -431,43 +431,30 @@ def create_demo():
                 output_md = gr.Markdown("")
                 status_state = gr.State("idle")
 
-        def _set_preview_text(text: str):
-            return gr.update(value=text)
-
         def load_preview(url: str):
             if not url:
                 return gr.update(value=None, visible=False), gr.update(value=None, visible=False), gr.update(value="")
-            # If remote and looks like a video by extension, show video immediately and try async conversion for browser compatibility
-            if is_remote(url) and any(url.lower().endswith(ext) for ext in VIDEO_EXTS):
-                return gr.update(value=None, visible=False), gr.update(value=url, visible=True), gr.update(value="Remote video detected — showing preview if browser-playable; converting if needed...")
-            # Try fetching as image first (quick check). Provide verbose status messages.
+            # Quick content-type/extension heuristic first
             try:
                 if is_remote(url):
                     head = safe_head(url)
                     if head:
                         ctype = (head.headers.get("content-type") or "").lower()
-                        if ctype.startswith("video/"):
-                            return gr.update(value=None, visible=False), gr.update(value=url, visible=True), gr.update(value=f"Remote video (content-type={ctype}) — showing preview if browser-playable.")
-                        if ctype.startswith("image/"):
-                            r = safe_get(url, timeout=10)
-                            img_bytes = r.content
-                        else:
-                            # try GET anyway
-                            r = safe_get(url, timeout=10)
-                            img_bytes = r.content
-                    else:
-                        r = safe_get(url, timeout=10)
-                        img_bytes = r.content
+                        if ctype.startswith("video/") or any(url.lower().endswith(ext) for ext in VIDEO_EXTS):
+                            return gr.update(value=None, visible=False), gr.update(value=url, visible=True), gr.update(value=f"Remote video detected (content-type={ctype}). Showing preview if browser-playable; converting if necessary.")
+                # Try image fetch (fast)
+                if is_remote(url):
+                    r = safe_get(url, timeout=10)
+                    img_bytes = r.content
                 else:
                     with open(url, "rb") as f:
                         img_bytes = f.read()
                 img = Image.open(BytesIO(img_bytes))
                 if getattr(img, "is_animated", False):
                     img.seek(0)
-                return gr.update(value=img.convert("RGB"), visible=True), gr.update(value=None, visible=False), gr.update(value="Image preview loaded")
+                return gr.update(value=img.convert("RGB"), visible=True), gr.update(value=None, visible=False), gr.update(value="Image preview loaded.")
             except Exception as e:
-                # Fallback: if file might be video but extension missing, show helpful message
-                msg = f"Preview load failed: {e}. If this is a video, ensure URL ends with a video extension or upload a playable mp4/webm."
+                msg = f"Preview load failed: {e}. If this is a video, ensure URL ends with .mp4/.webm or upload a playable file."
                 return gr.update(value=None, visible=False), gr.update(value=None, visible=False), gr.update(value=msg)
 
         url_input.change(fn=load_preview, inputs=[url_input], outputs=[preview_image, preview_video, preview_status])
@@ -477,10 +464,9 @@ def create_demo():
 
         clear_btn.click(fn=clear_all, inputs=[], outputs=[url_input, preview_image, preview_video, status_state, progress_md, output_md])
 
-        def _convert_video_for_preview(path: str, progress_cb=None) -> str:
+        def _convert_video_for_preview(path: str) -> str:
             if not FFMPEG_BIN or not os.path.exists(path):
                 return path
-            # ensure mp4 with h264 for browser playability
             out_fd, out_path = tempfile.mkstemp(suffix=".mp4")
             os.close(out_fd)
             cmd = [
@@ -501,23 +487,19 @@ def create_demo():
                 if not url:
                     return ("error", "**Error:** No URL provided.")
                 progress(0.01, desc="Starting processing...")
-                # Fetch header to decide media type early for verbose feedback
                 progress(0.03, desc="Checking URL / content-type...")
                 is_img, is_vid = determine_media_type(url, progress=progress)
                 progress(0.06, desc=f"Determined media type: image={is_img}, video={is_vid}")
+                client = get_client(key)
                 if is_vid:
-                    progress(0.08, desc="Fetching video bytes (this may take a while)...")
+                    progress(0.08, desc="Fetching video bytes (may take a while)...")
                     raw = fetch_bytes(url, timeout=120, progress=progress)
                     tmp = save_bytes_to_temp(raw, suffix=ext_from_src(url) or ".mp4")
-                    progress(0.15, desc="Saved video to temp file; preparing for preview/analysis...")
-                    # Convert for preview to ensure browser-playable format (non-destructive)
-                    try:
-                        preview_tmp = _convert_video_for_preview(tmp, progress_cb=progress)
-                    except Exception:
-                        preview_tmp = tmp
-                    # Update progress text for preview step (front-end preview will show converted file path)
-                    progress(0.25, desc="Video ready for preview and analysis. Starting analysis...")
-                    res = analyze_video_cohesive(get_client(key), tmp, prompt or "", progress=progress)
+                    progress(0.18, desc="Saved video to temp; converting for preview if needed...")
+                    preview_tmp = _convert_video_for_preview(tmp)
+                    # Return preview path to frontend by placing it into preview_video (separate event)
+                    progress(0.25, desc="Starting video analysis...")
+                    res = analyze_video_cohesive(client, tmp, prompt or "", progress=progress)
                     progress(0.98, desc="Finalizing result...")
                     try:
                         if preview_tmp != tmp:
@@ -532,31 +514,31 @@ def create_demo():
                     progress(0.08, desc="Fetching image bytes...")
                     raw = fetch_bytes(url, progress=progress)
                     progress(0.18, desc="Analyzing image...")
-                    res = analyze_image_structured(get_client(key), raw, prompt or "", progress=progress)
+                    try:
+                        res = analyze_image_structured(client, raw, prompt or "", progress=progress)
+                    except UnidentifiedImageError:
+                        return ("error", "Error: provided file is not a valid image.")
                     progress(0.98, desc="Finalizing result...")
                     status = "done" if not (isinstance(res, str) and res.lower().startswith("error")) else "error"
                     return (status, res if isinstance(res, str) else str(res))
                 else:
-                    # Unknown: try fetch and try image first, then video fallback
-                    progress(0.07, desc="Unknown media type — fetching bytes for heuristic check...")
+                    progress(0.07, desc="Unknown media type — fetching bytes for heuristics...")
                     raw = fetch_bytes(url, timeout=120, progress=progress)
-                    # try image
                     try:
                         progress(0.15, desc="Attempting to interpret as image...")
-                        _ = Image.open(BytesIO(raw))
+                        Image.open(BytesIO(raw))
                         progress(0.2, desc="Image detected — analyzing...")
-                        res = analyze_image_structured(get_client(key), raw, prompt or "", progress=progress)
+                        res = analyze_image_structured(client, raw, prompt or "", progress=progress)
                         status = "done" if not (isinstance(res, str) and res.lower().startswith("error")) else "error"
                         return (status, res if isinstance(res, str) else str(res))
                     except Exception:
-                        # Treat as video
                         fd, tmp = tempfile.mkstemp(suffix=ext_from_src(url) or ".mp4")
                         os.close(fd)
                         with open(tmp, "wb") as fh:
                             fh.write(raw)
                         try:
                             progress(0.3, desc="Saved fallback video file; analyzing...")
-                            res = analyze_video_cohesive(get_client(key), tmp, prompt or "", progress=progress)
+                            res = analyze_video_cohesive(client, tmp, prompt or "", progress=progress)
                             status = "done" if not (isinstance(res, str) and res.lower().startswith("error")) else "error"
                             return (status, res if isinstance(res, str) else str(res))
                         finally:
@@ -572,7 +554,7 @@ def create_demo():
             return _btn_label_for_status(s)
 
         status_state.change(fn=btn_label_from_state, inputs=[status_state], outputs=[submit_btn])
-        # Show verbose progress messages in progress_md by listening to a lightweight helper that maps state to text.
+
         def status_to_progress_text(s):
             return {"idle":"Idle","busy":"Processing…","done":"Completed","error":"Error — see output"}.get(s, s)
         status_state.change(fn=status_to_progress_text, inputs=[status_state], outputs=[progress_md])