Update app.py

app.py

@@ -6,7 +6,7 @@ import os, shutil, subprocess, tempfile, base64, json
 from io import BytesIO
 from typing import List, Tuple
 import requests
-from PIL import Image, ImageFile, UnidentifiedImageError
+from PIL import Image, ImageFile, UnidentifiedImageError, ImageSequence
 import gradio as gr
 
 # --- Config
@@ -164,7 +164,7 @@ def ffmpeg_make_browser_mp4(input_path: str, output_path: str, max_width: int =
         "-c:a", "aac", "-b:a", "128k",
         "-movflags", "+faststart",
         output_path
-    ]
+        ]
     try:
         subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=timeout, check=True)
         return os.path.exists(output_path) and os.path.getsize(output_path) > 0
@@ -184,13 +184,11 @@ def analyze_video_cohesive(client, video_path: str, prompt: str) -> str:
     except Exception:
         pass
 
-    # If upload failed or not available, try to make a browser-friendly MP4 for Gradio to play and for ffmpeg frame extraction.
     tmp_fixed = None
     try:
         tmp_fd, tmp_fixed = tempfile.mkstemp(suffix=".mp4"); os.close(tmp_fd)
         ok = ffmpeg_make_browser_mp4(video_path, tmp_fixed, max_width=1280, crf=28, preset="fast", timeout=120)
         if ok:
-            # Use frame extraction on the fixed file for analysis if upload isn't possible
             frames = extract_best_frames_bytes(tmp_fixed, sample_count=6)
         else:
             frames = extract_best_frames_bytes(video_path, sample_count=6)
@@ -225,6 +223,30 @@ def determine_media_type(src: str) -> Tuple[bool, bool]:
             elif ctype.startswith("video/"): is_video, is_image = True, False
     return is_image, is_video
 
+def save_bytes_to_temp(data: bytes, suffix: str = ".dat") -> str:
+    fd, path = tempfile.mkstemp(suffix=suffix)
+    os.close(fd)
+    with open(path, "wb") as fh:
+        fh.write(data)
+    return path
+
+def convert_to_jpeg_bytes(data: bytes, base_h: int = 1024) -> bytes:
+    buf = BytesIO(data)
+    img = Image.open(buf)
+    if getattr(img, "is_animated", False):
+        img = next(ImageSequence.Iterator(img))
+    img = img.convert("RGB")
+    w, h = img.size
+    if h > base_h:
+        new_w = int(w * (base_h / h))
+        img = img.resize((new_w, base_h), Image.LANCZOS)
+    out = BytesIO()
+    img.save(out, format="JPEG", quality=90, optimize=True)
+    return out.getvalue()
+
+def b64_bytes(data: bytes, mime: str = "image/jpeg") -> str:
+    return "data:" + mime + ";base64," + base64.b64encode(data).decode("ascii")
+
 def process_media(src: str, custom_prompt: str, api_key: str, progress=gr.Progress()) -> str:
     client = get_client(api_key)
     prompt = (custom_prompt or "").strip() or "Please provide a detailed visual review."
@@ -286,58 +308,3 @@ def create_demo():
                     clear_btn = gr.Button("Clear")
                 output_md = gr.Markdown("")
                 status_state = gr.State("idle")
-
-        def load_preview(url: str):
-            empty_img = gr.update(value=None, visible=False)
-            empty_vid = gr.update(value=None, visible=False)
-            if not url: return empty_img, empty_vid
-            if not is_remote(url) and os.path.exists(url):
-                ext = ext_from_src(url)
-                if ext in VIDEO_EXTS: return empty_img, 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 gr.update(value=img.convert("RGB"), visible=True), empty_vid
-                    except Exception: return empty_img, empty_vid
-            head = safe_head(url)
-            if head:
-                ctype = (head.headers.get("content-type") or "").lower()
-                if ctype.startswith("video/") or any(url.lower().endswith(ext) for ext in VIDEO_EXTS):
-                    return empty_img, gr.update(value=url, visible=True)
-            try:
-                r = safe_get(url, timeout=15)
-                img = Image.open(BytesIO(r.content))
-                if getattr(img, "is_animated", False): img.seek(0)
-                return gr.update(value=img.convert("RGB"), visible=True), empty_vid
-            except Exception:
-                return empty_img, empty_vid
-
-        url_input.change(fn=load_preview, inputs=[url_input], outputs=[preview_image, preview_video])
-
-        def clear_all():
-            return "", gr.update(value=None, visible=False), gr.update(value=None, visible=False), "idle", gr.update(value=_btn_label_for_status("idle"))
-        clear_btn.click(fn=clear_all, inputs=[], outputs=[url_input, preview_image, preview_video, status_state, submit_btn])
-
-        def start_busy():
-            s = "busy"
-            return s, gr.update(value=_btn_label_for_status(s))
-        submit_btn.click(fn=start_busy, inputs=[], outputs=[status_state, submit_btn])
-
-        def worker(url: str, prompt: str, key: str, progress=gr.Progress()):
-            return process_media(url or "", prompt or "", key or "", progress=progress)
-        submit_btn.click(fn=worker, inputs=[url_input, custom_prompt, api_key], outputs=[output_md], queue=True).then(
-            fn=lambda res: ("error", "**Error:** no result returned.") if not res else
-                           ("error", f"**Error:** {res}") if isinstance(res, str) and res.lower().startswith("error") else ("done", res),
-            inputs=[output_md],
-            outputs=[status_state, output_md],
-        )
-
-        def btn_label_for_state(s: str):
-            return _btn_label_for_status(s)
-        status_state.change(fn=btn_label_for_state, inputs=[status_state], outputs=[submit_btn])
-
-    return demo
-
-if __name__ == "__main__":
-    create_demo().launch()