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 not 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): normalized = normalize_path(path) return Path(normalized).expanduser().resolve() 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 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 find_files_by_ext(root, extensions): root_path = ensure_exists(root, "search root", is_dir=True) wanted = {ext.lower().lstrip(".") for ext in extensions} matches = [] for path in root_path.rglob("*"): if path.is_file() and path.suffix.lower().lstrip(".") in wanted: matches.append(path) return matches def create_task_dir(prefix="task"): output_root = get_comfy_output_dir() / "keysync" 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, "data_dir": task_dir / "data", "videos_dir": task_dir / "data" / "videos", "audios_dir": task_dir / "data" / "audios", "outputs_dir": task_dir / "outputs", "logs_dir": task_dir / "logs", } for key, value in paths.items(): if key.endswith("_dir"): value.mkdir(parents=True, exist_ok=True) return paths