- "value": "import json\nfrom json import JSONDecodeError\n\nimport jq\nfrom json_repair import repair_json\n\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import HandleInput, MessageTextInput\nfrom lfx.io import Output\nfrom lfx.log.logger import logger\nfrom lfx.schema.data import Data\nfrom lfx.schema.message import Message\n\n\nclass ParseJSONDataComponent(Component):\n display_name = \"Parse JSON\"\n description = \"Convert and extract JSON fields.\"\n icon = \"braces\"\n name = \"ParseJSONData\"\n legacy: bool = True\n replacement = [\"processing.ParserComponent\"]\n\n inputs = [\n HandleInput(\n name=\"input_value\",\n display_name=\"Input\",\n info=\"Data object to filter.\",\n required=True,\n input_types=[\"Message\", \"Data\", \"JSON\"],\n ),\n MessageTextInput(\n name=\"query\",\n display_name=\"JQ Query\",\n info=\"JQ Query to filter the data. The input is always a JSON list.\",\n required=True,\n ),\n ]\n\n outputs = [\n Output(display_name=\"Filtered Data\", name=\"filtered_data\", method=\"filter_data\"),\n ]\n\n def _parse_data(self, input_value) -> str:\n if isinstance(input_value, Message) and isinstance(input_value.text, str):\n return input_value.text\n if isinstance(input_value, Data):\n return json.dumps(input_value.data)\n return str(input_value)\n\n def filter_data(self) -> list[Data]:\n to_filter = self.input_value\n if not to_filter:\n return []\n # Check if input is a list\n if isinstance(to_filter, list):\n to_filter = [self._parse_data(f) for f in to_filter]\n else:\n to_filter = self._parse_data(to_filter)\n\n # If input is not a list, don't wrap it in a list\n if not isinstance(to_filter, list):\n to_filter = repair_json(to_filter)\n try:\n to_filter_as_dict = json.loads(to_filter)\n except JSONDecodeError:\n try:\n to_filter_as_dict = json.loads(repair_json(to_filter))\n except JSONDecodeError as e:\n msg = f\"Invalid JSON: {e}\"\n raise ValueError(msg) from e\n else:\n to_filter = [repair_json(f) for f in to_filter]\n to_filter_as_dict = []\n for f in to_filter:\n try:\n to_filter_as_dict.append(json.loads(f))\n except JSONDecodeError:\n try:\n to_filter_as_dict.append(json.loads(repair_json(f)))\n except JSONDecodeError as e:\n msg = f\"Invalid JSON: {e}\"\n raise ValueError(msg) from e\n to_filter = to_filter_as_dict\n\n full_filter_str = json.dumps(to_filter_as_dict)\n\n logger.info(\"to_filter: %s\", to_filter)\n\n results = jq.compile(self.query).input_text(full_filter_str).all()\n logger.info(\"results: %s\", results)\n return [Data(data=value) if isinstance(value, dict) else Data(text=str(value)) for value in results]\n"
0 commit comments