sqxp/backend/app/core/request_payload.py

"""统一解析接口请求载荷（明文 / 密文信封）。"""

from __future__ import annotations

import logging
from typing import Any, Union

from app.core import api_crypto
from app.core.config import settings


def resolve_request_payload(
    request_body: dict,
    current_user: dict,
    *,
    source: str = "unknown",
) -> Union[dict, list]:
    """兼容明文与密文信封，返回业务层 payload（多为 dict；批量数组接口解密后为 list）。"""
    logger = logging.getLogger("sqxp")
    payload = request_body if isinstance(request_body, dict) else {}
    if int(payload.get("enc", 0) or 0) == 1 and isinstance(payload.get("edata"), str):
        logger.warning(
            "%s received encrypted envelope in route handler; middleware may be bypassed",
            source,
        )
        try:
            user_id_for_crypto = str(current_user.get("user_id") or "")
            ts = int(payload.get("ts"))
            nonce = str(payload.get("nonce") or "")
            if user_id_for_crypto and nonce and settings.API_ENCRYPT_FIXED_SECRET:
                key = api_crypto.derive_key(
                    fixed_secret=settings.API_ENCRYPT_FIXED_SECRET,
                    user_id=user_id_for_crypto,
                    ts=ts,
                    nonce=nonce,
                )
                plain = api_crypto.decrypt_json(str(payload.get("edata")), key)
                payload = plain if isinstance(plain, dict) else {}
                logger.warning(
                    "%s decrypted payload via route fallback; please verify ApiCryptoMiddleware",
                    source,
                )
        except Exception as exc:
            logger.warning("%s resolve_request_payload decrypt failed: err=%s", source, exc)
            return {}
    dw = source.startswith("delivery_warning")
    if isinstance(payload, dict):
        inner = payload.get("data")
        if isinstance(inner, (dict, list)):
            if dw:
                if isinstance(inner, dict):
                    logger.info(
                        "%s unwrap: using inner dict keys=%s note_len=%s remark_len=%s",
                        source,
                        sorted(inner.keys()),
                        len(str(inner.get("note") or "")),
                        len(str(inner.get("remark") or "")),
                    )
                else:
                    logger.info("%s unwrap: inner is list len=%s", source, len(inner))
            return inner
        if dw:
            logger.info(
                "%s no_inner_data: payload_keys=%s (no dict/list data wrapper)",
                source,
                sorted(payload.keys()),
            )
    elif dw:
        logger.info("%s payload_not_dict type=%s", source, type(payload).__name__)
    return payload if isinstance(payload, dict) else {}