mirror of
https://github.com/Cccc-owo/CheckInApp.git
synced 2026-06-17 14:06:28 +00:00
121 lines
3.0 KiB
Python
121 lines
3.0 KiB
Python
"""
|
|
JSON 处理辅助函数
|
|
|
|
提供安全的 JSON 解析和数据提取功能
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import Optional, Any, Dict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def safe_parse_json(json_str: Optional[str], default: Any = None, log_error: bool = True) -> Any:
|
|
"""
|
|
安全解析 JSON 字符串，失败时返回默认值
|
|
|
|
Args:
|
|
json_str: JSON 字符串
|
|
default: 解析失败时的默认值
|
|
log_error: 是否记录解析错误日志
|
|
|
|
Returns:
|
|
解析后的对象，失败时返回 default
|
|
"""
|
|
if not json_str:
|
|
return default
|
|
|
|
try:
|
|
return json.loads(str(json_str))
|
|
except (json.JSONDecodeError, AttributeError, TypeError) as e:
|
|
if log_error:
|
|
logger.debug(f"JSON 解析失败: {str(e)}, 原始数据: {json_str[:100]}...")
|
|
return default
|
|
|
|
|
|
def safe_parse_payload(payload_config: Optional[str], default: Optional[Dict] = None) -> Dict:
|
|
"""
|
|
安全解析 payload_config，失败时返回默认字典
|
|
|
|
Args:
|
|
payload_config: payload 配置字符串
|
|
default: 解析失败时的默认值
|
|
|
|
Returns:
|
|
解析后的字典
|
|
"""
|
|
result = safe_parse_json(payload_config, default or {})
|
|
# 确保返回值是字典类型
|
|
if not isinstance(result, dict):
|
|
logger.warning(f"payload_config 不是字典类型: {type(result)}")
|
|
return default or {}
|
|
return result
|
|
|
|
|
|
def extract_thread_id(payload_config: Optional[str]) -> Optional[str]:
|
|
"""
|
|
从 payload_config 中提取 ThreadId
|
|
|
|
Args:
|
|
payload_config: payload 配置字符串
|
|
|
|
Returns:
|
|
ThreadId 或 None
|
|
"""
|
|
payload = safe_parse_payload(payload_config)
|
|
return payload.get("ThreadId")
|
|
|
|
|
|
def extract_signature(payload_config: Optional[str]) -> Optional[str]:
|
|
"""
|
|
从 payload_config 中提取 Signature
|
|
|
|
Args:
|
|
payload_config: payload 配置字符串
|
|
|
|
Returns:
|
|
Signature 或 None
|
|
"""
|
|
payload = safe_parse_payload(payload_config)
|
|
return payload.get("Signature")
|
|
|
|
|
|
def build_task_info(task) -> Dict[str, str]:
|
|
"""
|
|
从 task 对象构建 task_info 字典（用于邮件通知等场景）
|
|
|
|
Args:
|
|
task: CheckInTask 对象
|
|
|
|
Returns:
|
|
包含 thread_id 和 name 的字典
|
|
"""
|
|
return {
|
|
"thread_id": resolve_task_thread_id(task) or "未知",
|
|
"name": getattr(task, "name", None) or f"Task-{getattr(task, 'id', 'Unknown')}",
|
|
}
|
|
|
|
|
|
def resolve_task_thread_id(task) -> Optional[str]:
|
|
"""
|
|
优先从显式字段读取任务 ThreadId，兼容旧数据时回退到 payload_config。
|
|
|
|
Args:
|
|
task: CheckInTask 或相似对象
|
|
|
|
Returns:
|
|
ThreadId 或 None
|
|
"""
|
|
thread_id = getattr(task, "thread_id", None)
|
|
if thread_id is not None:
|
|
value = str(thread_id).strip()
|
|
if value:
|
|
return value
|
|
|
|
legacy_thread_id = extract_thread_id(getattr(task, "payload_config", None))
|
|
if legacy_thread_id is None:
|
|
return None
|
|
value = str(legacy_thread_id).strip()
|
|
return value or None
|