# -*- 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']