from .dumas_image_nodes import ( NODE_CLASS_MAPPINGS as IMAGE_NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS as IMAGE_NODE_DISPLAY_NAME_MAPPINGS, _next_counter_for_relative_path, _folder_is_image, _list_folder_image_files, resolve_serve_token, ) from .dumas_json_nodes import ( NODE_CLASS_MAPPINGS as JSON_NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS as JSON_NODE_DISPLAY_NAME_MAPPINGS, ) from .dumas_h3_longvideos import ( NODE_CLASS_MAPPINGS as H3_LONGVIDEO_NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS as H3_LONGVIDEO_NODE_DISPLAY_NAME_MAPPINGS, ) from .dumas_h3_shot_length import ( NODE_CLASS_MAPPINGS as H3_SHOT_LENGTH_NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS as H3_SHOT_LENGTH_NODE_DISPLAY_NAME_MAPPINGS, ) from .dumas_h3_inspector import ( NODE_CLASS_MAPPINGS as H3_INSPECTOR_NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS as H3_INSPECTOR_NODE_DISPLAY_NAME_MAPPINGS, ) NODE_CLASS_MAPPINGS = {} NODE_CLASS_MAPPINGS.update(JSON_NODE_CLASS_MAPPINGS) NODE_CLASS_MAPPINGS.update(IMAGE_NODE_CLASS_MAPPINGS) NODE_CLASS_MAPPINGS.update(H3_LONGVIDEO_NODE_CLASS_MAPPINGS) NODE_CLASS_MAPPINGS.update(H3_SHOT_LENGTH_NODE_CLASS_MAPPINGS) NODE_CLASS_MAPPINGS.update(H3_INSPECTOR_NODE_CLASS_MAPPINGS) NODE_DISPLAY_NAME_MAPPINGS = {} NODE_DISPLAY_NAME_MAPPINGS.update(JSON_NODE_DISPLAY_NAME_MAPPINGS) NODE_DISPLAY_NAME_MAPPINGS.update(IMAGE_NODE_DISPLAY_NAME_MAPPINGS) NODE_DISPLAY_NAME_MAPPINGS.update(H3_LONGVIDEO_NODE_DISPLAY_NAME_MAPPINGS) NODE_DISPLAY_NAME_MAPPINGS.update(H3_SHOT_LENGTH_NODE_DISPLAY_NAME_MAPPINGS) NODE_DISPLAY_NAME_MAPPINGS.update(H3_INSPECTOR_NODE_DISPLAY_NAME_MAPPINGS) WEB_DIRECTORY = "./js" try: import asyncio import io import os import string import shutil import subprocess import sys import threading from PIL import Image, ImageOps from aiohttp import web from server import PromptServer _DUMAS_DIALOG_LOCK = threading.Lock() def _dumas_dialog_available(): return sys.platform == "win32" and shutil.which("powershell") is not None def _dumas_dialog_windows(start_path): ps = ( "Add-Type -AssemblyName System.Windows.Forms;" "$r='';" "$o=New-Object System.Windows.Forms.Form;" "$o.TopMost=$true;$o.ShowInTaskbar=$false;$o.FormBorderStyle='None';" "$o.Width=1;$o.Height=1;$o.Opacity=0;$o.StartPosition='CenterScreen';" "$o.Add_Shown({" "$o.Activate();" "$d=New-Object System.Windows.Forms.FolderBrowserDialog;" '$d.Description="Select a folder";$d.ShowNewFolderButton=$true;' "if($env:DUMAS_START){try{$d.SelectedPath=$env:DUMAS_START}catch{}};" "if($d.ShowDialog($o) -eq [System.Windows.Forms.DialogResult]::OK){$script:r=$d.SelectedPath};" "$o.Close()" "});" "[void]$o.ShowDialog();" "[Console]::Out.Write($r)" ) env = dict(os.environ) env["DUMAS_START"] = start_path or "" result = subprocess.run( ["powershell", "-NoProfile", "-STA", "-Command", ps], capture_output=True, text=True, timeout=300, env=env, creationflags=0x08000000, check=False, ) return (result.stdout or "").strip() def _dumas_native_folder_dialog(start_path=""): if not _DUMAS_DIALOG_LOCK.acquire(blocking=False): return None try: if sys.platform == "win32": return _dumas_dialog_windows(start_path) return "" finally: try: _DUMAS_DIALOG_LOCK.release() except Exception: pass def _dumas_is_path_under(child_path, parent_path): try: return os.path.commonpath([os.path.realpath(child_path), os.path.realpath(parent_path)]) == os.path.realpath(parent_path) except ValueError: return False def _dumas_make_thumb(full_path): image = Image.open(full_path) image = ImageOps.exif_transpose(image).convert("RGB") image.thumbnail((192, 192)) buffer = io.BytesIO() image.save(buffer, format="JPEG", quality=80) return buffer.getvalue() @PromptServer.instance.routes.get("/dumas/api/save_image/file") async def dumas_save_image_file(request): path = resolve_serve_token(request.query.get("t", "")) if not path or not os.path.isfile(path): return web.Response(status=404, text="unknown or expired preview token") return web.FileResponse(path) @PromptServer.instance.routes.get("/dumas/api/save_image/next_counter") async def dumas_save_image_next_counter(request): folder_raw = str(request.query.get("folder", "") or "").strip() name = str(request.query.get("name", "") or "") try: digits = max(1, min(8, int(request.query.get("digits", "3")))) except Exception: digits = 3 def _scan(): import folder_paths output_dir = folder_paths.get_output_directory() base = os.path.abspath(folder_raw) if folder_raw else output_dir parts = [part for part in name.replace("\\", "/").split("/") if part] if not parts: return 1, "" relative_template = "/".join(parts) counter = _next_counter_for_relative_path(base, relative_template) resolved_relative = relative_template.replace("%counter%", str(counter).zfill(digits)) return counter, resolved_relative try: loop = asyncio.get_running_loop() counter, resolved = await loop.run_in_executor(None, _scan) return web.json_response({"ok": True, "counter": counter, "resolved": resolved}) except Exception as exc: return web.json_response({"ok": False, "message": str(exc), "counter": 1, "resolved": ""}) @PromptServer.instance.routes.get("/dumas/api/pick_directory") async def dumas_pick_directory(request): try: if not _dumas_dialog_available(): return web.json_response({"ok": False, "message": "Native folder picker is unavailable here.", "unavailable": True}) start_path = str(request.query.get("path", "") or "").strip() if start_path and not os.path.isdir(start_path): start_path = "" loop = asyncio.get_running_loop() selected = await loop.run_in_executor(None, _dumas_native_folder_dialog, start_path) if selected is None: return web.json_response({"ok": False, "busy": True}) if not selected: return web.json_response({"ok": False, "cancelled": True}) return web.json_response({"ok": True, "path": selected}) except Exception as exc: return web.json_response({"ok": False, "message": str(exc)}) @PromptServer.instance.routes.get("/dumas/api/load_images_folder/list") async def dumas_load_images_folder_list(request): headers = {"Cache-Control": "no-store"} folder = request.query.get("path", "") recursive = request.query.get("recursive", "0") == "1" if not folder or not os.path.isdir(folder): return web.json_response({"ok": False, "message": "Folder not found.", "files": []}, headers=headers) real_folder = os.path.realpath(folder) try: loop = asyncio.get_running_loop() files = await loop.run_in_executor(None, _list_folder_image_files, real_folder, recursive) except Exception as exc: return web.json_response({"ok": False, "message": f"Could not read folder: {exc}", "files": []}, headers=headers) return web.json_response({"ok": True, "folder": real_folder, "files": files}, headers=headers) @PromptServer.instance.routes.get("/dumas/api/load_images_folder/thumb") async def dumas_load_images_folder_thumb(request): folder = request.query.get("path", "") rel_path = request.query.get("file", "") if not folder or not rel_path or not os.path.isdir(folder): return web.Response(status=404) full_path = os.path.realpath(os.path.join(folder, rel_path)) if not _dumas_is_path_under(full_path, folder) or not os.path.isfile(full_path) or not _folder_is_image(os.path.basename(full_path)): return web.Response(status=403) try: loop = asyncio.get_running_loop() body = await loop.run_in_executor(None, _dumas_make_thumb, full_path) return web.Response(body=body, content_type="image/jpeg", headers={"Cache-Control": "no-cache"}) except Exception: return web.Response(status=404) @PromptServer.instance.routes.get("/dumas/api/load_images_folder/browse") async def dumas_load_images_folder_browse(request): path = request.query.get("path", "") try: if not path: dirs = [] if os.name == "nt": for letter in string.ascii_uppercase: drive = f"{letter}:\\" if os.path.isdir(drive): dirs.append({"name": drive, "path": drive, "images": -1}) else: dirs.append({"name": "/", "path": "/", "images": -1}) return web.json_response({"ok": True, "path": "", "parent": None, "dirs": dirs}) if not os.path.isdir(path): return web.json_response({"ok": False, "message": "Folder not found.", "dirs": []}) real_path = os.path.realpath(path) parent = os.path.dirname(real_path) if parent == real_path: parent = "" subdirs = [] try: for name in sorted(os.listdir(real_path), key=str.lower): full_path = os.path.join(real_path, name) if os.path.isdir(full_path): subdirs.append((name, full_path)) except OSError as exc: return web.json_response({"ok": False, "message": f"Could not read folder: {exc}", "dirs": []}) should_count = len(subdirs) <= 60 dirs = [] for name, full_path in subdirs: images = -1 if should_count: try: images = sum(1 for entry in os.listdir(full_path) if _folder_is_image(entry)) except OSError: images = -1 dirs.append({"name": name, "path": full_path, "images": images}) return web.json_response({"ok": True, "path": real_path, "parent": parent, "dirs": dirs}) except Exception as exc: return web.json_response({"ok": False, "message": str(exc), "dirs": []}) @PromptServer.instance.routes.get("/dumas/api/load_images_folder/pick_native") async def dumas_load_images_folder_pick_native(request): try: if not _dumas_dialog_available(): return web.json_response({"ok": False, "unavailable": True}) start_path = str(request.query.get("path", "") or "").strip() if start_path and not os.path.isdir(start_path): start_path = "" loop = asyncio.get_running_loop() selected = await loop.run_in_executor(None, _dumas_native_folder_dialog, start_path) if selected is None: return web.json_response({"ok": False, "busy": True}) if selected and os.path.isdir(selected): return web.json_response({"ok": True, "path": selected}) return web.json_response({"ok": False, "cancelled": True}) except Exception as exc: return web.json_response({"ok": False, "message": str(exc)}) except Exception: pass __all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS", "WEB_DIRECTORY"]