DumasNodes/dumas_json_nodes.py

import copy
import json


def _parse_json_path(path):
    if not path.strip():
        return []

    segments = []
    for raw_segment in path.split("."):
        segment = raw_segment.strip()
        if not segment:
            continue
        if segment.isdigit():
            segments.append(int(segment))
        else:
            segments.append(segment)
    return segments


def _get_json_path_value(data, path):
    current = data
    for segment in _parse_json_path(path):
        if isinstance(segment, int):
            if not isinstance(current, list) or segment >= len(current):
                return None
            current = current[segment]
            continue

        if not isinstance(current, dict) or segment not in current:
            return None
        current = current[segment]
    return current


def _set_json_path_value(data, path, value):
    if not isinstance(data, dict):
        return None

    segments = _parse_json_path(path)
    if not segments:
        return value

    result = copy.deepcopy(data)
    current = result

    for index, segment in enumerate(segments[:-1]):
        next_segment = segments[index + 1]
        if isinstance(segment, int):
            if not isinstance(current, list):
                return None
            while segment >= len(current):
                current.append({} if not isinstance(next_segment, int) else [])
            current = current[segment]
            continue

        if segment not in current or not isinstance(current[segment], (dict, list)):
            current[segment] = [] if isinstance(next_segment, int) else {}
        current = current[segment]

    last_segment = segments[-1]
    if isinstance(last_segment, int):
        if not isinstance(current, list):
            return None
        while last_segment >= len(current):
            current.append(None)
        current[last_segment] = value
        return result

    if not isinstance(current, dict):
        return None
    current[last_segment] = value
    return result


def _merge_json_objects(base_object, overlay_object):
    if not isinstance(base_object, dict) or not isinstance(overlay_object, dict):
        return None

    merged = copy.deepcopy(base_object)
    for key, value in overlay_object.items():
        if isinstance(merged.get(key), dict) and isinstance(value, dict):
            merged[key] = _merge_json_objects(merged[key], value)
        else:
            merged[key] = copy.deepcopy(value)
    return merged


class DumasJSONStringToObjectNode:
    CATEGORY = "Dumas/JSON"
    RETURN_TYPES = ("JSON",)
    FUNCTION = "convert_string_to_json"

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "json_string": ("STRING", {"multiline": True}),
            }
        }

    def convert_string_to_json(self, json_string):
        try:
            json_object = json.loads(json_string)
            return (json_object,)
        except json.JSONDecodeError as exc:
            print(f"[DumasNodes] Error decoding JSON: {exc}")
            return (None,)


class DumasJSONObjectToStringNode:
    CATEGORY = "Dumas/JSON"
    RETURN_TYPES = ("STRING",)
    FUNCTION = "convert_object_to_string"

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "json_object": ("JSON",),
                "pretty": ("BOOLEAN", {"default": True}),
                "sort_keys": ("BOOLEAN", {"default": False}),
            }
        }

    def convert_object_to_string(self, json_object, pretty, sort_keys):
        indent = 2 if pretty else None
        separators = None if pretty else (",", ":")
        return (json.dumps(json_object, indent=indent, sort_keys=sort_keys, separators=separators),)


class DumasJSONGetValueNode:
    CATEGORY = "Dumas/JSON"
    RETURN_TYPES = ("JSON",)
    FUNCTION = "get_value"

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "json_object": ("JSON",),
                "path": ("STRING", {"default": "", "multiline": False}),
            }
        }

    def get_value(self, json_object, path):
        return (_get_json_path_value(json_object, path),)


class DumasJSONSetValueNode:
    CATEGORY = "Dumas/JSON"
    RETURN_TYPES = ("JSON",)
    FUNCTION = "set_value"

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "json_object": ("JSON",),
                "path": ("STRING", {"default": "", "multiline": False}),
                "value_json": ("STRING", {"multiline": True, "default": "null"}),
            }
        }

    def set_value(self, json_object, path, value_json):
        try:
            value = json.loads(value_json)
        except json.JSONDecodeError as exc:
            print(f"[DumasNodes] Error decoding JSON value: {exc}")
            return (None,)

        return (_set_json_path_value(json_object, path, value),)


class DumasJSONMergeObjectsNode:
    CATEGORY = "Dumas/JSON"
    RETURN_TYPES = ("JSON",)
    FUNCTION = "merge_objects"

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "base_object": ("JSON",),
                "overlay_object": ("JSON",),
            }
        }

    def merge_objects(self, base_object, overlay_object):
        return (_merge_json_objects(base_object, overlay_object),)


class DumasJSONKeysNode:
    CATEGORY = "Dumas/JSON"
    RETURN_TYPES = ("JSON", "INT")
    RETURN_NAMES = ("keys", "count")
    FUNCTION = "list_keys"

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "json_object": ("JSON",),
            }
        }

    def list_keys(self, json_object):
        if not isinstance(json_object, dict):
            return (None, 0)
        keys = list(json_object.keys())
        return (keys, len(keys))


class DumasJSONArrayLengthNode:
    CATEGORY = "Dumas/JSON"
    RETURN_TYPES = ("INT",)
    FUNCTION = "get_length"

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "json_array": ("JSON",),
            }
        }

    def get_length(self, json_array):
        if not isinstance(json_array, list):
            return (0,)
        return (len(json_array),)


NODE_CLASS_MAPPINGS = {
    "DumasJSONStringToObject": DumasJSONStringToObjectNode,
    "DumasJSONObjectToString": DumasJSONObjectToStringNode,
    "DumasJSONGetValue": DumasJSONGetValueNode,
    "DumasJSONSetValue": DumasJSONSetValueNode,
    "DumasJSONMergeObjects": DumasJSONMergeObjectsNode,
    "DumasJSONKeys": DumasJSONKeysNode,
    "DumasJSONArrayLength": DumasJSONArrayLengthNode,
}

NODE_DISPLAY_NAME_MAPPINGS = {
    "DumasJSONStringToObject": "Dumas JSON String to Object",
    "DumasJSONObjectToString": "Dumas JSON Object to String",
    "DumasJSONGetValue": "Dumas JSON Get Value",
    "DumasJSONSetValue": "Dumas JSON Set Value",
    "DumasJSONMergeObjects": "Dumas JSON Merge Objects",
    "DumasJSONKeys": "Dumas JSON Keys",
    "DumasJSONArrayLength": "Dumas JSON Array Length",
}