105 lines
3.5 KiB
Python
105 lines
3.5 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
通用JSON/字典解析工具
|
||
"""
|
||
|
||
import json
|
||
from typing import Any, Dict, Optional
|
||
from utils.logger import get_logger
|
||
|
||
logger = get_logger("json_parser")
|
||
|
||
|
||
def safe_parse_json(param: str, script_id: Optional[str] = None, default_return: Any = None) -> Any:
|
||
"""安全解析JSON或Python字典格式的字符串
|
||
|
||
尝试多种解析方式:
|
||
1. 首先尝试标准JSON解析
|
||
2. 如果JSON解析失败，尝试使用eval解析Python字典格式
|
||
|
||
Args:
|
||
param: 要解析的字符串
|
||
script_id: 脚本ID，用于日志记录
|
||
default_return: 解析失败时的默认返回值
|
||
|
||
Returns:
|
||
解析后的数据，失败时返回default_return
|
||
"""
|
||
if not isinstance(param, str):
|
||
if script_id:
|
||
logger.error(f"脚本 {script_id} 输入参数不是字符串类型: {type(param)}")
|
||
else:
|
||
logger.error(f"输入参数不是字符串类型: {type(param)}")
|
||
return default_return
|
||
|
||
# 1. 首先尝试标准JSON解析
|
||
try:
|
||
data = json.loads(param)
|
||
return data
|
||
except json.JSONDecodeError:
|
||
# 2. 如果JSON解析失败，尝试使用eval解析Python字典或列表格式
|
||
try:
|
||
# 安全地使用eval解析字典或列表字符串
|
||
stripped_param = param.strip()
|
||
if (stripped_param.startswith('{') and stripped_param.endswith('}')) or \
|
||
(stripped_param.startswith('[') and stripped_param.endswith(']')):
|
||
data = eval(param, {"__builtins__": {}}, {})
|
||
return data
|
||
else:
|
||
if script_id:
|
||
logger.error(f"脚本 {script_id} 参数格式不正确: {param}")
|
||
else:
|
||
logger.error(f"参数格式不正确: {param}")
|
||
return default_return
|
||
except Exception as eval_error:
|
||
if script_id:
|
||
logger.error(f"脚本 {script_id} 参数解析失败: {str(eval_error)}, 参数: {param}")
|
||
else:
|
||
logger.error(f"参数解析失败: {str(eval_error)}, 参数: {param}")
|
||
return default_return
|
||
|
||
|
||
def safe_parse_dict(param: str, script_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
||
"""安全解析为字典类型
|
||
|
||
Args:
|
||
param: 要解析的字符串
|
||
script_id: 脚本ID，用于日志记录
|
||
|
||
Returns:
|
||
解析后的字典，失败时返回None
|
||
"""
|
||
data = safe_parse_json(param, script_id, None)
|
||
|
||
if not isinstance(data, dict):
|
||
if script_id:
|
||
logger.error(f"脚本 {script_id} 参数格式错误，期望字典类型: {type(data)}")
|
||
else:
|
||
logger.error(f"参数格式错误，期望字典类型: {type(data)}")
|
||
return None
|
||
|
||
return data
|
||
|
||
|
||
def safe_parse_list(param: str, script_id: Optional[str] = None) -> Optional[list]:
|
||
"""安全解析为列表类型
|
||
|
||
Args:
|
||
param: 要解析的字符串
|
||
script_id: 脚本ID，用于日志记录
|
||
|
||
Returns:
|
||
解析后的列表，失败时返回None
|
||
"""
|
||
data = safe_parse_json(param, script_id, None)
|
||
|
||
if not isinstance(data, list):
|
||
if script_id:
|
||
logger.error(f"脚本 {script_id} 参数格式错误，期望列表类型: {type(data)}")
|
||
else:
|
||
logger.error(f"参数格式错误，期望列表类型: {type(data)}")
|
||
return None
|
||
|
||
return data |