272 lines
12 KiB
Python
272 lines
12 KiB
Python
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"]
|