| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263 |
- # -*- coding: utf-8 -*-
- r"""响应归一化：把算法函数的返回（numpy / pandas / NaN / 时间 / Path …）压成**确定性**的 JSON 结构。
- 为什么要确定性：P10 的验收是"HTTP 取回的内容 == 进程内直调的内容"（逐值对拍），所以两边必须走**同一条**
- 归一化路径，且字典键排序、数值类型固定（NaN/Inf → null，numpy 标量 → python 标量）。
- """
- from __future__ import annotations
- import dataclasses
- import datetime as _dt
- import json
- import math
- import pathlib
- from typing import Any
- def to_jsonable(x: Any) -> Any:
- """递归归一化；不可识别的对象退回 `str()`（并保留类型名，便于排查）。"""
- if x is None or isinstance(x, (bool, int, str)):
- return x
- if isinstance(x, float):
- return None if (math.isnan(x) or math.isinf(x)) else x
- if isinstance(x, (_dt.datetime, _dt.date, _dt.time)):
- return x.isoformat()
- if isinstance(x, pathlib.Path):
- return str(x)
- if isinstance(x, dict):
- return {str(k): to_jsonable(v) for k, v in sorted(x.items(), key=lambda kv: str(kv[0]))}
- if isinstance(x, (list, tuple, set, frozenset)):
- seq = list(x)
- if isinstance(x, (set, frozenset)):
- seq = sorted(seq, key=lambda v: str(v))
- return [to_jsonable(v) for v in seq]
- if dataclasses.is_dataclass(x) and not isinstance(x, type):
- return to_jsonable(dataclasses.asdict(x))
- # numpy / pandas：不硬依赖，按鸭子类型处理
- for attr, kind in (('to_dict', 'dict'), ('tolist', 'list')):
- if hasattr(x, attr):
- try:
- got = getattr(x, attr)() if attr != 'to_dict' else x.to_dict(orient='records')
- return to_jsonable(got)
- except Exception: # noqa: BLE001 落到 str()
- pass
- if hasattr(x, 'item'): # numpy 标量
- try:
- return to_jsonable(x.item())
- except Exception: # noqa: BLE001
- pass
- if hasattr(x, 'shape') and hasattr(x, 'columns'): # DataFrame 兜底
- try:
- return {'columns': [str(c) for c in x.columns], 'rows': to_jsonable(x.values.tolist()),
- 'shape': list(x.shape)}
- except Exception: # noqa: BLE001
- pass
- return {'__type__': type(x).__name__, 'repr': str(x)[:500]}
- def dumps(payload: Any) -> str:
- """确定性 JSON 文本（键排序、非 ASCII 原样、紧凑分隔）—— 对拍与缓存键都用它。"""
- return json.dumps(to_jsonable(payload), ensure_ascii=False, sort_keys=True, separators=(',', ':'))
- __all__ = ['to_jsonable', 'dumps']
|
