Fabric PySpark Notebook中解析含转义双引号的JSON字符串问题

我完全明白你现在遇到的棘手问题——从Pipeline传过来的JSON字符串里，Details字段的错误信息嵌套了未转义的双引号（比如"Internal Server Error"里的引号），导致json.loads直接报错，之前处理换行、制表符的逻辑完全没问题，但这个引号的处理确实需要精准区分JSON结构的引号和内容里的引号。

问题根源 #

本质上是这份输入的JSON格式不规范：JSON字符串内部的双引号必须用\转义才能被解析器识别，但Pipeline输出的字符串里，Details这类字段的内容里直接用了未转义的"，混淆了JSON的结构边界，所以我们需要在解析前先修正这个格式问题。

解决方案 #

核心思路是：精准识别并转义JSON值内容内部的未转义双引号，同时保留JSON键值对结构本身的合法引号。以下是调整后的可运行代码：

import re
import json

# Input String Received from Pipeline
var_str_log_data = "[{\"ApplicationName\": \"APP\",\"WorkspaceId\": \"0addb382-fa0d-4ce1-9c3c-95e32b957955\",\"Environment\": \"DEV\",\"Level\": \"ERROR\",\"Severity\": \"SEV 4\",\"Component\": \"PL_APP_Ingest\",\"Operation\": \"ETL\",\"Run_Id\": \"88276e4f-4e60-47ec-a6d6-dd9e475d2c3a\",\"SessionId\": \"\",\"Message\": \"[DEV] - APP - Pipeline Execution Failed - 20250329\",\"Status\": \"Success\",\"Details\": \"Pipeline Name: PL_APP_Ingest\\nRun Id: 88257e4f-4e60-34ec-a6d6-dd9e475d2c3a\\nError Message: Notebook execution failed at Notebook service with http status code - '200', please check the Run logs on Notebook, additional details - 'Error name - Py4JJavaError, Error value - An error occurred while calling o4628.load.\n: java.util.concurrent.ExecutionException: Operation failed: \"Internal Server Error\", 500, HEAD, http://msit-onelake.dfs.fabric.microsoft.com/MYWORKSPACE/dataverse_name_cds2_workspace_unq60feaf930cbf236eb2519b27f2d6b.Lakehouse/Tables/app_riskparcel/_delta_log?upn=false&action=getStatus&timeout=90\n  at com.google.common.util.concurrent.AbstractFuture$Sync.getValue(AbstractFuture.java:306)\n    at com.google.common.util.concurrent.AbstractFuture$Sync.get(AbstractFuture.java:293)\n at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3599)\n  at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2379)\n  ... 37 more\n' : \\nExecution URL: https://msit.powerbi.com/workloads/data-pipeline/artifacts/workspaces/0addb122-fa0d-4ce1-9c3c-95e25b957955/pipelines/PL_APP_Ingest/88223e4f-4e60-47ec-a6d6-dd9e475d2c3a?experience=power-bi\\nApp Name: APP\",\"CorrelationId\": \"\",\"User\": \"System\"}]"

# 1. 保留你原有的特殊字符处理逻辑，确保换行、制表符、回车被正确转义
var_str_log_data = re.sub(r'(?<!\\)\n', r'\\n', var_str_log_data)
var_str_log_data = re.sub(r'(?<!\\)\t', r'\\t', var_str_log_data)
var_str_log_data = re.sub(r'(?<!\\)\r', r'\\r', var_str_log_data)

# 2. 核心修复：转义JSON值内容内部的未转义双引号
# 正则说明：
# - (?<=\\":\\"): 匹配JSON键结束、值开始的位置（即": "的转义形式）
# - (?:[^"\\]|\\.)*?: 匹配值的内容，包含换行符，直到遇到未转义的"
# - \K: 重置匹配起点，只对后面的未转义"做替换
var_str_log_data = re.sub(
    r'(?<=\\":\\")(?:[^"\\]|\\.)*?\K(?<!\\)"',
    r'\\"',
    var_str_log_data,
    flags=re.DOTALL
)

print(f"Log Data Format and Content: {type(var_str_log_data)} - {var_str_log_data}.")

try:
    data = json.loads(var_str_log_data)
    print("JSON loaded successfully")
    # 验证Details字段是否正确解析
    print("\nParsed Details content snippet:", data[0]["Details"][:200] + "...")
except json.JSONDecodeError as e:
    error_pos = e.pos
    print("Invalid JSON:", e)
    print(f"Character at position {error_pos}: {repr(var_str_log_data[error_pos-1])}")
    start = max(0, error_pos - 20)
    end = min(len(var_str_log_data), error_pos + 20)
    print("Surrounding text:", repr(var_str_log_data[start:end]))

补充建议 #

1. 从根源解决：如果有条件的话，最好协调Pipeline的开发方，让他们在生成JSON字符串时就自动转义内容内部的双引号，这样能避免后续的格式修复工作，也更可靠。
2. 健壮性优化：如果你的日志里还有其他类似的特殊格式问题，可以考虑先把Python字符串字面量转成真实的字符串（用ast.literal_eval），再进行格式修正，比如：

   import ast
   # 先解析Python字符串字面量为真实的JSON字符串
   real_json_str = ast.literal_eval(var_str_log_data)
   # 再对real_json_str进行后续的转义处理
   

备注：内容来源于stack exchange，提问作者Prabhat