shanxia2.0/backend/app/core/admin_request_json.py

"""运营 Web（/web/admin）JSON 请求体解析。

与 ApiCryptoMiddleware 配合：中间件未改写 body 时，在路由内解密 `{ enc, ts, nonce, edata }`
并剥 `{ data: ... }`，避免 FastAPI 声明式 Body 在校验阶段只看到密文信封而 422。
"""

from __future__ import annotations

import logging
from typing import Any, Optional, Tuple

from fastapi import Request

from app.core.request_payload import resolve_request_payload

_log = logging.getLogger("sqxp")


async def parse_admin_request_json(
    request: Request,
    current_user: dict,
    *,
    source: str,
) -> Tuple[Any, Optional[str]]:
    """
    读取 JSON body；dict 时走 resolve_request_payload（明文 / 密文 / data 包装）。

    Returns:
        (payload, None) 成功；
        (_, err_msg) 密文解密失败（此时 payload 为原始 raw，调用方可忽略）。
    """
    dw = source.startswith("delivery_warning")
    try:
        raw: Any = await request.json()
    except Exception as ex:
        if dw:
            _log.warning("parse_admin_request_json source=%s request.json() failed: %s", source, ex)
        raw = {}
    if isinstance(raw, dict):
        if dw:
            ed = raw.get("edata")
            _log.info(
                "parse_admin_request_json source=%s raw_keys=%s enc=%s edata_len=%s user_id=%s",
                source,
                sorted(raw.keys()),
                int(raw.get("enc", 0) or 0),
                len(ed) if isinstance(ed, str) else (len(str(ed)) if ed is not None else 0),
                current_user.get("user_id"),
            )
        resolved = resolve_request_payload(raw, current_user, source=source)
        if dw:
            if isinstance(resolved, dict):
                _log.info(
                    "parse_admin_request_json source=%s resolved_keys=%s",
                    source,
                    sorted(resolved.keys()),
                )
            else:
                _log.info(
                    "parse_admin_request_json source=%s resolved_type=%s",
                    source,
                    type(resolved).__name__,
                )
        if resolved == {} and int(raw.get("enc", 0) or 0) == 1:
            if dw:
                _log.warning(
                    "parse_admin_request_json source=%s treating as decrypt failure (resolved empty, enc=1)",
                    source,
                )
            return raw, "密文解密失败"
        return resolved, None
    if dw:
        _log.warning(
            "parse_admin_request_json source=%s raw_not_dict type=%s",
            source,
            type(raw).__name__,
        )
    return raw, None