Refactor process_new_user_message to simplify file handling and improve readability

app.py

@@ -42,43 +42,44 @@ def downsample_video(video_path: str) -> list[tuple[Image.Image, float]]:
 
 
 def process_new_user_message(message: dict) -> list[dict]:
-    if message["files"]:
-        if "<image>" in message["text"]:
-            content = []
-            logger.debug(f"{message['files']=}")
-            parts = re.split(r"(<image>)", message["text"])
-            image_index = 0
-            logger.debug(f"{parts=}")
-            for part in parts:
-                logger.debug(f"{part=}")
-                if part == "<image>":
-                    content.append({"type": "image", "url": message["files"][image_index]})
-                    logger.debug(f"file: {message['files'][image_index]}")
-                    image_index += 1
-                elif part.strip():
-                    content.append({"type": "text", "text": part.strip()})
-                elif isinstance(part, str) and part != "<image>":
-                    content.append({"type": "text", "text": part})
-            logger.debug(f"{content=}")
-            return content
-        if message["files"][0].endswith(".mp4"):
-            content = []
-            video = message["files"].pop(0)
-            frames = downsample_video(video)
-            for frame in frames:
-                pil_image, timestamp = frame
-                with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp_file:
-                    pil_image.save(temp_file.name)
-                    content.append({"type": "text", "text": f"Frame {timestamp}:"})
-                    content.append({"type": "image", "url": temp_file.name})
-            logger.debug(f"{content=}")
-            return content
-        # non interleaved images
-        return [
-            {"type": "text", "text": message["text"]},
-            *[{"type": "image", "url": path} for path in message["files"]],
-        ]
-    return [{"type": "text", "text": message["text"]}]
+    if not message["files"]:
+        return [{"type": "text", "text": message["text"]}]
+
+    if "<image>" in message["text"]:
+        content = []
+        logger.debug(f"{message['files']=}")
+        parts = re.split(r"(<image>)", message["text"])
+        image_index = 0
+        logger.debug(f"{parts=}")
+        for part in parts:
+            logger.debug(f"{part=}")
+            if part == "<image>":
+                content.append({"type": "image", "url": message["files"][image_index]})
+                logger.debug(f"file: {message['files'][image_index]}")
+                image_index += 1
+            elif part.strip():
+                content.append({"type": "text", "text": part.strip()})
+            elif isinstance(part, str) and part != "<image>":
+                content.append({"type": "text", "text": part})
+        logger.debug(f"{content=}")
+        return content
+    if message["files"][0].endswith(".mp4"):
+        content = []
+        video = message["files"].pop(0)
+        frames = downsample_video(video)
+        for frame in frames:
+            pil_image, timestamp = frame
+            with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp_file:
+                pil_image.save(temp_file.name)
+                content.append({"type": "text", "text": f"Frame {timestamp}:"})
+                content.append({"type": "image", "url": temp_file.name})
+        logger.debug(f"{content=}")
+        return content
+    # non interleaved images
+    return [
+        {"type": "text", "text": message["text"]},
+        *[{"type": "image", "url": path} for path in message["files"]],
+    ]
 
 
 def process_history(history: list[dict]) -> list[dict]: