scense/app/_archive_disabled_20261028_151505/body_restore.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""请求体「读后回填」的唯一实现（P0 崩点修复，QC 意见 #2）。

要解决的问题（目标机 traceback 原文已写进本迭代 P0 单 _HCoWd3He1axFKKa8ae91）
--------------------------------------------------------------------------
`app/http_semantics_middleware.py` 的 `restore_request_body()` 为了让我方读完 body 后
下游 `processorResource.getPostData()` 还能再解析一次，做了：

    from aiohttp.payload import BytesPayload
    request._payload = BytesPayload(body)

在 aiohttp 3.10.10 里 `Payload` / `BytesPayload` **没有** `at_eof()` 方法，而
`aiohttp.web_request.BaseRequest.can_read_body` 的实现是：

    return not self._payload.at_eof()

于是本中间件一旦回填过 `_payload`，下游任何一次 `can_read_body` 读取都会抛
`AttributeError: 'BytesPayload' object has no attribute 'at_eof'` → 该请求 **HTTP 500**。
影响面：全站 `*.dspy` 500 = 37/215，其中 POST/PUT 写类端点（CRUD `add_*` /
`create_*` / `entity_create` / `runtime_control` / `add_world` 等 42 条）必崩，
本迭代的 500 门禁（须为 0）不可能清零。

两种改法（QC 给了 A1/A2 二选一）
------------------------------
* A1：完全不替换 `_payload`，只保留 `request._read_bytes` 缓存，并把所有下游读取点
  改为优先走缓存。**问题**：下游读取点在基座 `ahserver.processorResource`（本应用
  `referenced_modules`，按 project-directory-spec §三「引用模块写保护」只读，
  改了会破坏所有依赖 ahserver 的项目），本角色无权修改。
* A2（本文件采用）：改用 `aiohttp.streams.StreamReader` + `feed_data/feed_eof`
  重建 `_payload`。`StreamReader` **有** `at_eof()`（`self._eof and not self._data`），
  因此 `can_read_body` 正常返回 bool；同时新流未 EOF 前可被 `post()/multipart()`
  重新消费，等价于原意图。

三条硬约束
----------
1. **绝不构造 `BytesPayload`**：构造失败/类型不可用时，宁可**不碰** `_payload`
   （原 `_payload` 本身就是 `StreamReader`，有 `at_eof`），也绝不塞一个没有
   `at_eof` 的对象进去——把 AttributeError 崩点换成"下游读不到 body"是倒退。
2. 回填同时保留（并写回）`request._read_bytes`，让 `await request.read()` 幂等。
3. 本模块**顶层不 import aiohttp**：aiohttp 只在函数内部按需导入，因此可以脱离
   web 运行时做离线自测（见 `scripts/selftest_body_restore_r50.py`）。
"""

from __future__ import annotations

import logging

log = logging.getLogger("yuanjing.body_restore")

#: 回填结果码
RESTORED = "restored"            # _payload 已用 StreamReader 重建，下游可重读
CACHE_ONLY = "cache_only"        # 只写了 _read_bytes 缓存，未碰 _payload（安全降级）
SKIPPED_EMPTY = "skipped_empty"  # body 为空，无需回填
FAILED = "failed"                # 回填抛异常（已吞异常，仅告警）


def _make_stream_reader(body, template=None, loop=None):
    """构造一个已喂入 body 并到达 EOF 的 aiohttp StreamReader。

    返回 (reader, "") 或 (None, reason)。构造成功后必须满足：
      * 有 `at_eof()`（`can_read_body` 不再 AttributeError）；
      * `at_eof()` 在 EOF 后返回 True（不谎报"还有 body 可读"）。
    """
    try:
        from aiohttp.streams import StreamReader
    except Exception as exc:                                    # noqa: BLE001
        return None, "no aiohttp.streams.StreamReader: %s" % exc
    kwargs = {}
    # 尽量沿用原 payload 的分片/上限/事件循环参数，避免行为漂移
    for attr, key in (("_limit", "limit"), ("_chunk", "chunk"), ("_timer", "timer"),
                      ("_loop", "loop")):
        val = getattr(template, attr, None) if template is not None else None
        if val is None and key == "loop":
            val = loop
        if val is not None:
            kwargs[key] = val
    try:
        reader = StreamReader(**kwargs)
    except Exception:                                           # noqa: BLE001
        # 参数签名随版本变化：退回到无参构造（ aiohttp 全版本支持）
        try:
            reader = StreamReader()
        except Exception as exc:                                # noqa: BLE001
            return None, "StreamReader() failed: %s" % exc
    try:
        reader.feed_data(body)
        reader.feed_eof()
    except Exception as exc:                                    # noqa: BLE001
        return None, "feed_data/feed_eof failed: %s" % exc
    if not callable(getattr(reader, "at_eof", None)):
        return None, "constructed reader has no at_eof()"
    # 自检：at_eof() 必须**可安全求值并返回 bool**（这正是线上 500 的位置）。
    # 注意：喂入 body + feed_eof 之后 at_eof() 仍应为 False（body 未被消费），
    # 这正是让下游 `can_read_body=True → read()` 能重读到 body 的语义。
    try:
        if not isinstance(reader.at_eof(), bool):
            return None, "at_eof() did not return bool"
    except Exception as exc:                                    # noqa: BLE001
        return None, "at_eof() raised: %s" % exc
    return reader, ""


def restore_request_payload(request, body):
    """把已读走的 body 安全回填给下游。返回 (ok: bool, code: str)。

    安全顺序（任一步失败都不影响前面已完成的步骤）：
      1. 写 `request._read_bytes`（aiohttp 自己的读缓存，重复读幂等）；
      2. 用 `StreamReader` 重建 `request._payload`（A2 方案，有 at_eof）；
      3. 第 2 步不可用时**保持 `_payload` 原样**（= CACHE_ONLY 降级），
         绝不写 `BytesPayload`（那正是 P0 500 的崩点）。
    """
    if not body:
        return True, SKIPPED_EMPTY
    if not isinstance(body, (bytes, bytearray)):
        return False, FAILED

    # 1) 读缓存（一定先做：即使后面 payload 重建失败，handler 走 read() 仍有内容）
    cache_ok = False
    try:
        request._read_bytes = bytes(body)
        cache_ok = True
    except Exception as exc:                                    # noqa: BLE001
        log.warning("[body_restore] 写 _read_bytes 失败 path=%s: %s",
                    getattr(request, "path", ""), exc)

    # 2) 用 StreamReader 重建 _payload
    template = None
    loop = None
    try:
        template = getattr(request, "_payload", None)
    except Exception:                                           # noqa: BLE001
        template = None
    try:
        loop = getattr(request, "_loop", None) or getattr(template, "_loop", None)
    except Exception:                                           # noqa: BLE001
        loop = None
    reader, reason = _make_stream_reader(bytes(body), template=template, loop=loop)
    if reader is None:
        # 3) 安全降级：**不碰** _payload
        log.warning("[body_restore] 跳过 _payload 重建（保持原流，避免 at_eof 缺失）"
                    "path=%s reason=%s cache_ok=%s",
                    getattr(request, "path", ""), reason, cache_ok)
        return (cache_ok, CACHE_ONLY)

    try:
        request._payload = reader
        # 让 aiohttp 的 post()/multipart() 重新解析（缓存的旧解析结果已不属于新流）
        for attr in ("_post_params", "_body_attrs_set"):
            try:
                if getattr(request, attr, None) is not None:
                    setattr(request, attr, None)
            except Exception:                                   # noqa: BLE001
                pass
        try:
            request._reading = False
        except Exception:                                       # noqa: BLE001
            pass
    except Exception as exc:                                    # noqa: BLE001
        log.warning("[body_restore] 赋 _payload 失败 path=%s: %s",
                    getattr(request, "path", ""), exc)
        return (cache_ok, CACHE_ONLY)

    # 自检：回填后 can_read_body 必须可安全求值（这正是线上 500 的位置）
    try:
        request.can_read_body
    except Exception as exc:                                    # noqa: BLE001
        log.error("[body_restore] 回填后 can_read_body 仍抛异常 path=%s: %s "
                  "→ 回退为仅缓存", getattr(request, "path", ""), exc)
        try:
            if template is not None:
                request._payload = template
        except Exception:                                       # noqa: BLE001
            pass
        return (cache_ok, CACHE_ONLY)
    return True, RESTORED


def describe(request):
    """诊断串（不含任何 body 内容/凭据，只有类型名），用于日志与自测。"""
    payload = getattr(request, "_payload", None)
    return "payload_type=%s has_at_eof=%s read_bytes_len=%s" % (
        type(payload).__name__,
        callable(getattr(payload, "at_eof", None)),
        len(getattr(request, "_read_bytes", b"") or b""),
    )