src/backend/base/langflow/components/data/json_to_data.py

import json
from pathlib import Path

from json_repair import repair_json

from langflow.custom import Component
from langflow.io import FileInput, MessageTextInput, MultilineInput, Output
from langflow.schema import Data


class JSONToDataComponent(Component):
    display_name = "Load JSON"
    description = (
        "Convert a JSON file, JSON from a file path, or a JSON string to a Data object or a list of Data objects"
    )
    icon = "braces"
    name = "JSONtoData"
    legacy = True

    inputs = [
        FileInput(
            name="json_file",
            display_name="JSON File",
            file_types=["json"],
            info="Upload a JSON file to convert to a Data object or list of Data objects",
        ),
        MessageTextInput(
            name="json_path",
            display_name="JSON File Path",
            info="Provide the path to the JSON file as pure text",
        ),
        MultilineInput(
            name="json_string",
            display_name="JSON String",
            info="Enter a valid JSON string (object or array) to convert to a Data object or list of Data objects",
        ),
    ]

    outputs = [
        Output(name="data", display_name="Data", method="convert_json_to_data"),
    ]

    def convert_json_to_data(self) -> Data | list[Data]:
        if sum(bool(field) for field in [self.json_file, self.json_path, self.json_string]) != 1:
            msg = "Please provide exactly one of: JSON file, file path, or JSON string."
            self.status = msg
            raise ValueError(msg)

        json_data = None

        try:
            if self.json_file:
                resolved_path = self.resolve_path(self.json_file)
                file_path = Path(resolved_path)
                if file_path.suffix.lower() != ".json":
                    self.status = "The provided file must be a JSON file."
                else:
                    json_data = file_path.read_text(encoding="utf-8")

            elif self.json_path:
                file_path = Path(self.json_path)
                if file_path.suffix.lower() != ".json":
                    self.status = "The provided file must be a JSON file."
                else:
                    json_data = file_path.read_text(encoding="utf-8")

            else:
                json_data = self.json_string

            if json_data:
                # Try to parse the JSON string
                try:
                    parsed_data = json.loads(json_data)
                except json.JSONDecodeError:
                    # If JSON parsing fails, try to repair the JSON string
                    repaired_json_string = repair_json(json_data)
                    parsed_data = json.loads(repaired_json_string)

                # Check if the parsed data is a list
                if isinstance(parsed_data, list):
                    result = [Data(data=item) for item in parsed_data]
                else:
                    result = Data(data=parsed_data)
                self.status = result
                return result

        except (json.JSONDecodeError, SyntaxError, ValueError) as e:
            error_message = f"Invalid JSON or Python literal: {e}"
            self.status = error_message
            raise ValueError(error_message) from e

        except Exception as e:
            error_message = f"An error occurred: {e}"
            self.status = error_message
            raise ValueError(error_message) from e

        # An error occurred
        raise ValueError(self.status)