From 438a4ab75ec7e3ce6efc2afd773cd024d8df5f4b Mon Sep 17 00:00:00 2001 From: ExtremeNina Date: Thu, 22 Oct 2026 18:00:27 +0800 Subject: [PATCH] fix(utils): ignore braces inside strings when extracting nested json (#1401) --- backend/common/utils/utils.py | 35 +++---- backend/tests/test_extract_nested_json.py | 121 ++++++++++++++++++++++ 2 files changed, 135 insertions(+), 21 deletions(-) create mode 100644 backend/tests/test_extract_nested_json.py diff --git a/backend/common/utils/utils.py b/backend/common/utils/utils.py index 3c3bdd0d8..a3dc3d614 100644 --- a/backend/common/utils/utils.py +++ b/backend/common/utils/utils.py @@ -53,30 +53,23 @@ def deepcopy_ignore_extra(src, dest): return dest -def extract_nested_json(text): - stack = [] - start_index = -1 - results = [] +def extract_nested_json(text: str) -> Optional[str]: + if not text or not isinstance(text, str): + return None + decoder = json.JSONDecoder(strict=False) for i, char in enumerate(text): if char in '{[': - if not stack: # 记录起始位置 - start_index = i - stack.append(char) - elif char in '}]': - if stack and ((char == '}' and stack[-1] == '{') or (char == ']' and stack[-1] == '[')): - stack.pop() - if not stack: # 栈空时截取完整JSON - json_str = text[start_index:i + 1] - try: - orjson.loads(json_str) # 验证有效性 - results.append(json_str) - except: - pass - else: - stack = [] # 括号不匹配则重置 - if len(results) > 0 and results[0]: - return results[0] + try: + _, end = decoder.raw_decode(text, i) + candidate = text[i:end] + try: + orjson.loads(candidate) + return candidate + except Exception: + continue + except Exception: + continue return None def string_to_numeric_hash(text: str, bits: Optional[int] = 64) -> int: diff --git a/backend/tests/test_extract_nested_json.py b/backend/tests/test_extract_nested_json.py new file mode 100644 index 000000000..96f02b261 --- /dev/null +++ b/backend/tests/test_extract_nested_json.py @@ -0,0 +1,121 @@ +import os +import sys +import pytest +import orjson + +# Ensure backend root is in sys.path +backend_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if backend_dir not in sys.path: + sys.path.insert(0, backend_dir) + +from common.utils.utils import extract_nested_json + + +def test_standard_json_extraction(): + raw = '{"sql": "SELECT * FROM users WHERE status = 1", "success": true}' + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == "SELECT * FROM users WHERE status = 1" + assert data["success"] is True + + +def test_standard_json_array_extraction(): + raw = '[{"id": 1, "name": "foo"}, {"id": 2, "name": "bar"}]' + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert len(data) == 2 + assert data[0]["id"] == 1 + + +def test_sql_with_single_left_brace(): + """Test Issue #1401: SQL containing an unmatched left curly brace '{'.""" + raw = """{"sql": "SELECT * FROM config WHERE path LIKE '{'", "success": true}""" + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == "SELECT * FROM config WHERE path LIKE '{'" + assert data["success"] is True + + +def test_sql_with_single_right_brace(): + """Test Issue #1401: SQL containing an unmatched right curly brace '}'.""" + raw = """{"sql": "SELECT * FROM logs WHERE note LIKE '%}'", "success": true}""" + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == "SELECT * FROM logs WHERE note LIKE '%}'" + assert data["success"] is True + + +def test_sql_with_balanced_braces_in_like_pattern(): + """Test Issue #1401: SQL with LIKE pattern containing '{pattern}'.""" + raw = """{"sql": "SELECT * FROM users WHERE tag LIKE '%{vip}%'", "success": true}""" + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == "SELECT * FROM users WHERE tag LIKE '%{vip}%'" + assert data["success"] is True + + +def test_sql_with_escaped_quotes_and_braces(): + """Test SQL with escaped double quotes and braces inside string literal.""" + raw = r"""{"sql": "SELECT \"col\" FROM t WHERE json_val = '{\"key\": \"val\"}'", "success": true}""" + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == 'SELECT "col" FROM t WHERE json_val = \'{"key": "val"}\'' + assert data["success"] is True + + +def test_sql_with_escaped_backslashes(): + """Test SQL containing escaped backslashes and path delimiters.""" + raw = r"""{"sql": "SELECT * FROM files WHERE path = 'C:\\Users\\default\\{tmp}'", "success": true}""" + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == r"SELECT * FROM files WHERE path = 'C:\Users\default\{tmp}'" + assert data["success"] is True + + +def test_markdown_code_block_wrapping(): + """Test JSON wrapped inside markdown code block.""" + raw = """Here is the query result: +```json +{ + "sql": "SELECT * FROM users WHERE template LIKE '%{test}%'", + "success": true +} +``` +Hope this helps!""" + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == "SELECT * FROM users WHERE template LIKE '%{test}%'" + assert data["success"] is True + + +def test_leading_explanation_with_braces_and_quotes(): + """Test text preceding the JSON that contains braces or quotes.""" + raw = """注意：参数模板例如 {workspace_id}，大模型返回的结果如下： +{"sql": "SELECT * FROM table WHERE col = '{foo}'", "success": true}""" + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == "SELECT * FROM table WHERE col = '{foo}'" + assert data["success"] is True + + +def test_invalid_candidate_before_valid_json(): + """Test that invalid braced block before valid JSON does not break extraction.""" + raw = '{invalid json here} Followed by valid: {"sql": "SELECT 1", "success": true}' + extracted = extract_nested_json(raw) + assert extracted is not None + data = orjson.loads(extracted) + assert data["sql"] == "SELECT 1" + + +@pytest.mark.parametrize("invalid_input", [None, "", " ", 123, [], {}, "No json at all here"]) +def test_invalid_or_empty_inputs(invalid_input): + assert extract_nested_json(invalid_input) is None