feat: VIDEO_EDIT input type for video trim/crop rich widgets

server.py

@@ -31,6 +31,7 @@ from io import BytesIO
 
 import aiohttp
 from aiohttp import web
+import av
 import logging
 
 import mimetypes
@@ -212,6 +213,52 @@ def create_block_external_middleware():
     return block_external_middleware
 
 
+def resolve_view_media_path(request, user_manager):
+    if "filename" not in request.rel_url.query:
+        return web.Response(status=400)
+    filename = request.rel_url.query["filename"]
+
+    # The frontend's LoadImage combo widget uses asset_hash values
+    # (e.g. "blake3:...") as widget values. When litegraph renders the
+    # node preview, it constructs /view?filename=<asset_hash>, so this
+    # endpoint must resolve blake3 hashes to their on-disk file paths.
+    if filename.startswith("blake3:"):
+        owner_id = user_manager.get_request_user_id(request)
+        result = resolve_hash_to_path(filename, owner_id=owner_id)
+        if result is None:
+            return web.Response(status=404)
+        return result.abs_path, result.download_name, result.content_type
+
+    filename, output_dir = folder_paths.annotated_filepath(filename)
+
+    if not filename:
+        return web.Response(status=400)
+
+    # validation for security: prevent accessing arbitrary path
+    if filename[0] == '/' or '..' in filename:
+        return web.Response(status=400)
+
+    if output_dir is None:
+        type = request.rel_url.query.get("type", "output")
+        output_dir = folder_paths.get_directory_by_type(type)
+
+    if output_dir is None:
+        return web.Response(status=400)
+
+    if "subfolder" in request.rel_url.query:
+        subfolder = request.rel_url.query["subfolder"]
+        if os.path.isabs(subfolder) or os.path.splitdrive(subfolder)[0]:
+            return web.Response(status=403)
+        base_dir = os.path.abspath(output_dir)
+        full_output_dir = os.path.join(base_dir, subfolder)
+        if os.path.commonpath((os.path.abspath(full_output_dir), base_dir)) != base_dir:
+            return web.Response(status=403)
+        output_dir = full_output_dir
+
+    filename = os.path.basename(filename)
+    return os.path.join(output_dir, filename), filename, None
+
+
 class PromptServer():
     def __init__(self, loop):
         PromptServer.instance = self
@@ -516,44 +563,10 @@ class PromptServer():
         @routes.get("/view")
         async def view_image(request):
             if "filename" in request.rel_url.query:
-                filename = request.rel_url.query["filename"]
-
-                # The frontend's LoadImage combo widget uses asset_hash values
-                # (e.g. "blake3:...") as widget values. When litegraph renders the
-                # node preview, it constructs /view?filename=<asset_hash>, so this
-                # endpoint must resolve blake3 hashes to their on-disk file paths.
-                if filename.startswith("blake3:"):
-                    owner_id = self.user_manager.get_request_user_id(request)
-                    result = resolve_hash_to_path(filename, owner_id=owner_id)
-                    if result is None:
-                        return web.Response(status=404)
-                    file, filename, resolved_content_type = result.abs_path, result.download_name, result.content_type
-                else:
-                    resolved_content_type = None
-                    filename, output_dir = folder_paths.annotated_filepath(filename)
-
-                    if not filename:
-                        return web.Response(status=400)
-
-                    # validation for security: prevent accessing arbitrary path
-                    if filename[0] == '/' or '..' in filename:
-                        return web.Response(status=400)
-
-                    if output_dir is None:
-                        type = request.rel_url.query.get("type", "output")
-                        output_dir = folder_paths.get_directory_by_type(type)
-
-                    if output_dir is None:
-                        return web.Response(status=400)
-
-                    if "subfolder" in request.rel_url.query:
-                        full_output_dir = os.path.join(output_dir, request.rel_url.query["subfolder"])
-                        if os.path.commonpath((os.path.abspath(full_output_dir), output_dir)) != output_dir:
-                            return web.Response(status=403)
-                        output_dir = full_output_dir
-
-                    filename = os.path.basename(filename)
-                    file = os.path.join(output_dir, filename)
+                resolved = resolve_view_media_path(request, self.user_manager)
+                if isinstance(resolved, web.Response):
+                    return resolved
+                file, filename, resolved_content_type = resolved
 
                 if os.path.isfile(file):
                     if 'preview' in request.rel_url.query:
@@ -660,6 +673,54 @@ class PromptServer():
 
             return web.Response(status=404)
 
+        @routes.get("/video_metadata")
+        async def get_video_metadata(request):
+            resolved = resolve_view_media_path(request, self.user_manager)
+            if isinstance(resolved, web.Response):
+                return resolved
+            file = resolved[0]
+
+            if not os.path.isfile(file):
+                return web.Response(status=404)
+
+            def probe_video_metadata():
+                with av.open(file) as container:
+                    stream = next((s for s in container.streams if s.type == "video"), None)
+                    if stream is None:
+                        return None
+
+                    fps = float(stream.average_rate) if stream.average_rate else None
+                    duration = None
+                    if stream.duration is not None and stream.time_base is not None:
+                        duration = float(stream.duration * stream.time_base)
+                    elif container.duration is not None:
+                        duration = float(container.duration * av.time_base)
+                    frame_count = stream.frames or None
+                    if frame_count is None and duration is not None and fps is not None:
+                        frame_count = round(duration * fps)
+
+                    return {
+                        "fps": fps,
+                        "duration": duration,
+                        "frame_count": frame_count,
+                        "width": stream.codec_context.width,
+                        "height": stream.codec_context.height,
+                        "size": os.path.getsize(file),
+                    }
+
+            try:
+                metadata = await asyncio.to_thread(probe_video_metadata)
+            except FileNotFoundError:
+                return web.Response(status=404)
+            except PermissionError:
+                return web.Response(status=403)
+            except av.error.FFmpegError:
+                return web.Response(status=415)
+            if metadata is None:
+                return web.Response(status=415)
+
+            return web.json_response(metadata)
+
         @routes.get("/view_metadata/{folder_name}")
         async def view_metadata(request):
             folder_name = request.match_info.get("folder_name", None)