From 438a4ab75ec7e3ce6efc2afd773cd024d8df5f4b Mon Sep 17 00:00:00 2001 From: ExtremeNina Date: Thu, 22 Oct 2026 18:00:27 +0800 Subject: [PATCH 1/2] 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 From 60d4a5a790048745c767e9b5d57aa03c0178d722 Mon Sep 17 00:00:00 2001 From: ulleo Date: Mon, 26 Oct 2026 14:17:01 +0800 Subject: [PATCH 2/2] fix(utils): preserve outer boundaries when extracting JSON --- backend/common/utils/utils.py | 46 +++++++++++++----- backend/tests/test_extract_nested_json.py | 59 +++++++++++++++++++++-- 2 files changed, 91 insertions(+), 14 deletions(-) diff --git a/backend/common/utils/utils.py b/backend/common/utils/utils.py index a3dc3d614..d2836f438 100644 --- a/backend/common/utils/utils.py +++ b/backend/common/utils/utils.py @@ -53,25 +53,49 @@ def deepcopy_ignore_extra(src, dest): return dest -def extract_nested_json(text: str) -> Optional[str]: +def extract_nested_json(text: str) -> str | None: + """Return the first valid outer JSON object/array, never a failed one's child.""" if not text or not isinstance(text, str): return None - decoder = json.JSONDecoder(strict=False) + depth = 0 + start = 0 + in_string = False + escaped = False for i, char in enumerate(text): - if char in '{[': - try: - _, end = decoder.raw_decode(text, i) - candidate = text[i:end] + if depth == 0: + if char in "{[": + start = i + depth = 1 + continue + + if in_string: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + in_string = False + continue + + if char == '"': + in_string = True + elif char in "{[": + depth += 1 + elif char in "}]": + depth -= 1 + if depth == 0: + # Count boundaries only; orjson checks bracket types and syntax. + # Skip an invalid candidate as a whole instead of retrying inside it. + candidate = text[start : i + 1] try: orjson.loads(candidate) - return candidate - except Exception: + except orjson.JSONDecodeError: continue - except Exception: - continue + return candidate return None + def string_to_numeric_hash(text: str, bits: Optional[int] = 64) -> int: hash_bytes = hashlib.sha256(text.encode()).digest() hash_num = int.from_bytes(hash_bytes, byteorder='big') @@ -287,4 +311,4 @@ def equals_ignore_case(str1: str, *args: str) -> bool: continue if str1.casefold() == arg.casefold(): return True - return False \ No newline at end of file + return False diff --git a/backend/tests/test_extract_nested_json.py b/backend/tests/test_extract_nested_json.py index 96f02b261..f696b7e62 100644 --- a/backend/tests/test_extract_nested_json.py +++ b/backend/tests/test_extract_nested_json.py @@ -1,14 +1,15 @@ import os import sys -import pytest + import orjson +import pytest # 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 +from common.utils.utils import extract_nested_json # noqa: E402 def test_standard_json_extraction(): @@ -116,6 +117,58 @@ def test_invalid_candidate_before_valid_json(): assert data["sql"] == "SELECT 1" -@pytest.mark.parametrize("invalid_input", [None, "", " ", 123, [], {}, "No json at all here"]) +@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 + + +@pytest.mark.parametrize( + "invalid_candidate", + [ + '{"tables": [],}', + '{"outer": {"success": true, "sql": "SELECT 0"}, bad}', + '[{"year": 2026, "value": 10},]', + '{"sql": "SELECT \'{}\'", bad}', + '{"tables": [}, "other": {"success": true}}', + '{"value": NaN, "tables": []}', + '{"sql": "SELECT\n1", "tables": []}', + ], +) +def test_invalid_outer_candidate_does_not_expose_nested_values(invalid_candidate): + assert extract_nested_json(invalid_candidate) is None + answer = '{"success": true, "sql": "SELECT 1", "tables": ["t"]}' + assert extract_nested_json(invalid_candidate + " then " + answer) == answer + + +@pytest.mark.parametrize( + "truncated", + [ + '{"sql": "SELECT 1", "tables": ["t"]', + '[{"year": 2026, "value": 10},', + '[{"x": 1}', + '{"outer": {"success": true, "sql": "SELECT 1"}', + '{"text": "unterminated [] {}', + ], +) +def test_truncated_outer_candidate_is_rejected(truncated): + assert extract_nested_json(truncated) is None + + +@pytest.mark.parametrize( + "answer", + [ + '{"outer": [{"inner": [1, 2]}, []], "empty": {}}', + "{\"sql\": \"SELECT '[', ']', '{', '}'\", \"tables\": []}", + r'{"text": "escaped quote: \" { [", "tail": "ends in slash \\"}', + "{}", + "[]", + ], +) +def test_complete_outer_candidate_is_preserved(answer): + assert ( + extract_nested_json('Explanation: "answer" ```json\n' + answer + "\n```") + == answer + ) + assert extract_nested_json(answer + ' {"next": true}') == answer