grovergol-comfyui-grover-nodes/open_path_node.py

import os
import subprocess
import platform
import folder_paths
from server import PromptServer
from aiohttp import web
import json


class OpenPathNode:
    """
    A ComfyUI node that takes a path string as input and provides a button to open it in the default file explorer.
    """

    # Class variable to store last executed paths by node ID
    _last_executed_paths = {}

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {},
            "optional": {
                "path": ("STRING", {
                    "default": "",
                    "multiline": False,
                    "placeholder": "Enter file or folder path...",
                    "forceInput": False  # Allows both connection and manual input
                }),
            },
        }

    RETURN_TYPES = ("STRING",)
    RETURN_NAMES = ("path",)
    FUNCTION = "open_path"
    CATEGORY = "Grover/System"
    OUTPUT_NODE = True

    def open_path(self, path=""):
        """Process the path and return it (the button action is handled on the frontend)"""
        # Ensure we have a valid string path
        if path is None:
            path = ""

        path_str = str(path)

        # Store the last executed path for this node instance
        # We'll use a combination of the path and current time as a simple ID
        import time
        node_key = f"{id(self)}_{int(time.time())}"
        OpenPathNode._last_executed_paths[node_key] = path_str

        # Also store with a simpler key based on the path content for lookup
        if path_str:
            OpenPathNode._last_executed_paths[f"path_{hash(path_str)}"] = path_str
        return (path_str,)

    @classmethod
    def get_last_executed_path(cls, path_hint=""):
        """Get the last executed path, optionally filtered by a hint"""
        if not cls._last_executed_paths:
            return None

        # If we have a specific path hint, try to find it
        if path_hint:
            path_key = f"path_{hash(path_hint)}"
            if path_key in cls._last_executed_paths:
                return cls._last_executed_paths[path_key]

        # Otherwise return the most recent one
        # Get all non-hash keys (the timestamped ones)
        timestamped_keys = [
            k for k in cls._last_executed_paths.keys() if not k.startswith("path_")]
        if timestamped_keys:
            # Sort by timestamp (last part of the key) and get the most recent
            latest_key = max(timestamped_keys,
                             key=lambda x: int(x.split('_')[-1]))
            return cls._last_executed_paths[latest_key]

        return None    @ staticmethod

    def open_file_explorer(path):
        """Open the given path in the default file explorer"""
        try:
            # Normalize the path
            normalized_path = os.path.normpath(path)

            # Check if path exists
            if not os.path.exists(normalized_path):
                print(f"Path does not exist: {normalized_path}")
                return False

            # Get the system platform
            system = platform.system()

            if system == "Windows":
                # Windows - use explorer
                # Note: Don't use check=True as explorer can return non-zero exit codes even on success
                if os.path.isfile(normalized_path):
                    # If it's a file, select it in explorer
                    result = subprocess.run(
                        ['explorer', '/select,', normalized_path], capture_output=True)
                    # For files, we can check if explorer process started successfully
                    # Only treat severe errors (>1) as failures
                    if result.returncode > 1:
                        print(
                            f"Explorer returned code {result.returncode} for file: {normalized_path}")
                        print(
                            f"stderr: {result.stderr.decode() if result.stderr else 'None'}")
                        return False
                else:
                    # If it's a directory, open it
                    result = subprocess.run(
                        ['explorer', normalized_path], capture_output=True)
                    # For directories, explorer often returns 1 even on success, so only fail on severe errors
                    # Only treat severe errors (>1) as failures
                    if result.returncode > 1:
                        print(
                            f"Explorer returned code {result.returncode} for directory: {normalized_path}")
                        print(
                            f"stderr: {result.stderr.decode() if result.stderr else 'None'}")
                        return False
            elif system == "Darwin":  # macOS
                # macOS - use open command
                subprocess.run(['open', normalized_path], check=True)
            elif system == "Linux":
                # Linux - try various file managers
                try:
                    subprocess.run(['xdg-open', normalized_path], check=True)
                except FileNotFoundError:
                    # Fallback to other common file managers
                    for fm in ['nautilus', 'dolphin', 'thunar', 'pcmanfm']:
                        try:
                            subprocess.run([fm, normalized_path], check=True)
                            break
                        except FileNotFoundError:
                            continue
                    else:
                        print("No suitable file manager found for Linux")
                        return False
            else:
                print(f"Unsupported operating system: {system}")
                return False

            return True

        except Exception as e:
            print(f"Error opening path '{path}': {str(e)}")
            return False


# Web endpoint for handling button clicks
@PromptServer.instance.routes.post("/grover/open_path")
async def open_path_endpoint(request):
    """Handle the open path button click from the frontend"""
    try:
        data = await request.json()
        path = data.get("path", "")

        # If no direct path provided, try to get the last executed path
        if not path or path.strip() == "":
            last_path = OpenPathNode.get_last_executed_path()
            if last_path:
                path = last_path
            else:
                return web.json_response({
                    "success": False,
                    "error": "No path available. Please execute the workflow first (Queue Prompt) or enter a path manually."
                })

        success = OpenPathNode.open_file_explorer(path.strip())

        if success:
            return web.json_response({"success": True, "message": f"Opened path: {path}"})
        else:
            return web.json_response({"success": False, "error": f"Failed to open path: {path}"})

    except Exception as e:
        return web.json_response({"success": False, "error": str(e)})


# Node registration
NODE_CLASS_MAPPINGS = {
    "OpenPathNode": OpenPathNode
}

NODE_DISPLAY_NAME_MAPPINGS = {
    "OpenPathNode": "📁 Open Path"
}