Add ComfyUI HighSync nodes

highsync_wrapper/file_utils.py

@@ -0,0 +1,92 @@
+import os
+import re
+import shutil
+import uuid
+from datetime import datetime
+from pathlib import Path
+
+
+WINDOWS_DRIVE_RE = re.compile(r"^([A-Za-z]):[\\/](.*)$")
+
+
+def is_windows_drive_path(path):
+    return isinstance(path, str) and WINDOWS_DRIVE_RE.match(path.strip()) is not None
+
+
+def normalize_path(path):
+    if path is None:
+        return path
+
+    raw_path = str(path).strip().strip('"')
+    if raw_path == "":
+        return raw_path
+
+    match = WINDOWS_DRIVE_RE.match(raw_path)
+    if match and os.name != "nt":
+        drive = match.group(1).lower()
+        rest = match.group(2).replace("\\", "/")
+        return f"/mnt/{drive}/{rest}"
+
+    return os.path.abspath(os.path.expanduser(raw_path))
+
+
+def resolve_path(path):
+    return Path(normalize_path(path)).expanduser().resolve()
+
+
+def ensure_exists(path, label="path", is_dir=None):
+    if path is None or str(path).strip() == "":
+        raise FileNotFoundError(f"{label} is empty.")
+
+    resolved = resolve_path(path)
+    if is_dir is True and not resolved.is_dir():
+        raise FileNotFoundError(f"{label} does not exist: {resolved}")
+    if is_dir is False and not resolved.is_file():
+        raise FileNotFoundError(f"{label} does not exist: {resolved}")
+    if is_dir is None and not resolved.exists():
+        raise FileNotFoundError(f"{label} does not exist: {resolved}")
+    return resolved
+
+
+def safe_copy(src, dst):
+    src_path = ensure_exists(src, "source file", is_dir=False)
+    dst_path = resolve_path(dst)
+    dst_path.parent.mkdir(parents=True, exist_ok=True)
+    if src_path.resolve() != dst_path.resolve():
+        shutil.copy2(src_path, dst_path)
+    return dst_path
+
+
+def get_comfy_output_dir():
+    try:
+        import folder_paths
+
+        return Path(folder_paths.get_output_directory()).resolve()
+    except Exception:
+        current = Path(__file__).resolve()
+        for parent in current.parents:
+            if (parent / "custom_nodes").exists() and (parent / "output").exists():
+                return (parent / "output").resolve()
+        return Path("/mnt/d/ComfyUI/output").resolve()
+
+
+def create_task_dir(prefix="highsync"):
+    output_root = get_comfy_output_dir() / "highsync"
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    suffix = uuid.uuid4().hex[:8]
+    task_id = f"{prefix}_{timestamp}_{suffix}"
+    task_dir = output_root / task_id
+
+    paths = {
+        "task_id": task_id,
+        "task_dir": task_dir,
+        "media_dir": task_dir / "media",
+        "logs_dir": task_dir / "logs",
+        "frames_dir": task_dir / "frames",
+    }
+
+    for key, value in paths.items():
+        if key.endswith("_dir"):
+            value.mkdir(parents=True, exist_ok=True)
+
+    return paths