feat: 请求体规则扩展(append/insert/regex_replace)、OAuth 代理节点支持与列表分页

- 请求体规则新增 append、insert、regex_replace 三种操作，路径语法支持数组索引
- OAuth 授权/导入/批量导入支持指定代理节点(proxy_node_id)，Key 级代理避免 IP 污染
- 密钥列表、模型映射、模型列表添加智能分页(useSmartPagination)
- AdvancedGuide 新增请求体规则使用指南与示例
- 简化 Codex enrich_codex 实现，README 添加 QQ 群二维码

src/api/admin/provider_oauth.py

@@ -140,6 +140,10 @@ class CompleteOAuthResponse(BaseModel):
 class ProviderCompleteOAuthRequest(BaseModel):
     callback_url: str = Field(..., min_length=5, description="浏览器地址栏中的完整回调 URL")
     name: str | None = Field(None, max_length=100, description="账号名称（可选）")
+    proxy_node_id: str | None = Field(
+        None,
+        description="代理节点 ID（可选）。设置后 token 交换及后续所有操作（刷新、额度查询）均走该代理，避免 IP 污染",
+    )
 
 
 class ProviderCompleteOAuthResponse(BaseModel):
@@ -162,6 +166,32 @@ def _require_fixed_provider(provider: Provider) -> str:
     return provider_type
 
 
+def _resolve_proxy_for_oauth(
+    provider_proxy: dict[str, Any] | None,
+    proxy_node_id: str | None,
+) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
+    """解析 OAuth 操作使用的代理配置。
+
+    当前端指定了 proxy_node_id 时，优先使用该代理进行 token 交换等操作，
+    并返回需要保存到 Key 上的代理配置。
+
+    Args:
+        provider_proxy: Provider 级别的代理配置
+        proxy_node_id: 前端指定的代理节点 ID（可选）
+
+    Returns:
+        (effective_proxy, key_proxy):
+            - effective_proxy: 本次操作实际使用的代理配置
+            - key_proxy: 需要保存到 Key 上的代理配置（None 表示不设置 Key 级代理）
+    """
+    if proxy_node_id and proxy_node_id.strip():
+        key_proxy: dict[str, Any] = {"node_id": proxy_node_id.strip(), "enabled": True}
+        # 本次操作使用 Key 级代理
+        return key_proxy, key_proxy
+    # 无 Key 级代理，使用 Provider 级代理
+    return provider_proxy, None
+
+
 def _pkce_s256(verifier: str) -> str:
     digest = hashlib.sha256(verifier.encode("utf-8")).digest()
     return base64.urlsafe_b64encode(digest).decode("utf-8").rstrip("=")
@@ -207,11 +237,14 @@ def _create_oauth_key(
     auth_config: dict[str, Any],
     api_formats: list[str],
     flush_only: bool = False,
+    proxy: dict[str, Any] | None = None,
 ) -> "ProviderAPIKey":
     """创建 OAuth Key 记录并持久化。
 
     Args:
         flush_only: True 时仅 flush（批量导入场景），False 时 commit + refresh。
+        proxy: Key 级别代理配置（如 {"node_id": "xxx", "enabled": True}），
+               创建时设置后，后续 token 刷新、额度刷新等操作立即走代理，避免 IP 污染。
     """
     from src.models.database import ProviderAPIKey as ProviderAPIKeyModel
 
@@ -224,6 +257,8 @@ def _create_oauth_key(
         api_formats=api_formats,
         is_active=True,
     )
+    if proxy:
+        new_key.proxy = proxy
     db.add(new_key)
     if flush_only:
         db.flush()
@@ -929,7 +964,10 @@ async def complete_provider_oauth(
         data = form
         json_body = None
 
-    proxy_config = getattr(provider, "proxy", None)
+    # 解析代理：前端指定 proxy_node_id 时优先使用，否则回退到 Provider 级代理
+    proxy_config, key_proxy = _resolve_proxy_for_oauth(
+        getattr(provider, "proxy", None), payload.proxy_node_id
+    )
 
     resp = await post_oauth_token(
         provider_type=provider_type,
@@ -991,6 +1029,7 @@ async def complete_provider_oauth(
         access_token=access_token,
         auth_config=auth_config,
         api_formats=_get_provider_api_formats(provider),
+        proxy=key_proxy,
     )
 
     return ProviderCompleteOAuthResponse(
@@ -1096,6 +1135,10 @@ def _parse_kiro_import_input(raw_input: str) -> list[dict[str, Any]]:
 class ImportRefreshTokenRequest(BaseModel):
     refresh_token: str = Field(..., min_length=1, description="Refresh Token")
     name: str | None = Field(None, max_length=100, description="账号名称（可选）")
+    proxy_node_id: str | None = Field(
+        None,
+        description="代理节点 ID（可选）。设置后导入验证及后续所有操作均走该代理",
+    )
 
 
 class BatchImportRequest(BaseModel):
@@ -1107,6 +1150,10 @@ class BatchImportRequest(BaseModel):
         max_length=500_000,
         description="凭据数据，支持多种格式：JSON 对象、JSON 数组、纯 Token（一行一个）",
     )
+    proxy_node_id: str | None = Field(
+        None,
+        description="代理节点 ID（可选）。设置后批量导入验证及后续所有操作均走该代理",
+    )
 
 
 class BatchImportResultItem(BaseModel):
@@ -1148,6 +1195,11 @@ async def import_refresh_token(
         raise NotFoundException("Provider 不存在", "provider")
     provider_type = _require_fixed_provider(provider)
 
+    # 解析代理：前端指定 proxy_node_id 时优先使用，否则回退到 Provider 级代理
+    proxy_config, key_proxy = _resolve_proxy_for_oauth(
+        getattr(provider, "proxy", None), payload.proxy_node_id
+    )
+
     if provider_type == ProviderType.KIRO.value:
         raw_import = payload.refresh_token.strip()
         if not raw_import:
@@ -1173,7 +1225,6 @@ async def import_refresh_token(
         cfg = KiroAuthConfig.from_dict(raw_cfg)
         cfg.provider_type = ProviderType.KIRO.value
 
-        proxy_config = getattr(provider, "proxy", None)
         try:
             access_token, new_cfg = await refresh_access_token(cfg, proxy_config=proxy_config)
         except Exception as e:
@@ -1192,6 +1243,7 @@ async def import_refresh_token(
             access_token=access_token,
             auth_config=new_cfg.to_dict(),
             api_formats=_get_provider_api_formats(provider),
+            proxy=key_proxy,
         )
 
         return ProviderCompleteOAuthResponse(
@@ -1243,7 +1295,7 @@ async def import_refresh_token(
         data = form
         json_body = None
 
-    proxy_config = getattr(provider, "proxy", None)
+    # proxy_config 和 key_proxy 已在上方 Kiro 分支之前统一解析
 
     resp = await post_oauth_token(
         provider_type=provider_type,
@@ -1312,6 +1364,7 @@ async def import_refresh_token(
         access_token=access_token,
         auth_config=auth_config,
         api_formats=_get_provider_api_formats(provider),
+        proxy=key_proxy,
     )
 
     return ProviderCompleteOAuthResponse(
@@ -1355,6 +1408,11 @@ async def batch_import_oauth(
 
     provider_type = _require_fixed_provider(provider)
 
+    # 解析代理：前端指定 proxy_node_id 时优先使用，否则回退到 Provider 级代理
+    proxy_config, key_proxy = _resolve_proxy_for_oauth(
+        getattr(provider, "proxy", None), payload.proxy_node_id
+    )
+
     # Kiro 使用专用逻辑
     if provider_type == ProviderType.KIRO.value:
         return await _batch_import_kiro_internal(
@@ -1362,6 +1420,8 @@ async def batch_import_oauth(
             provider=provider,
             raw_credentials=payload.credentials,
             db=db,
+            proxy_config=proxy_config,
+            key_proxy=key_proxy,
         )
 
     # 标准 OAuth Provider（Codex、Antigravity、GeminiCli、ClaudeCode）
@@ -1378,8 +1438,6 @@ async def batch_import_oauth(
         raise InvalidRequestException("未找到有效的 Token 数据")
 
     api_formats = _get_provider_api_formats(provider)
-
-    proxy_config = getattr(provider, "proxy", None)
     token_url = template.oauth.token_url
     is_json = "anthropic.com" in token_url
     scope_str = " ".join(template.oauth.scopes) if template.oauth.scopes else ""
@@ -1550,6 +1608,7 @@ async def batch_import_oauth(
                 auth_config=auth_config,
                 api_formats=api_formats,
                 flush_only=True,
+                proxy=key_proxy,
             )
 
             results.append(
@@ -1599,8 +1658,15 @@ async def _batch_import_kiro_internal(
     provider: Provider,
     raw_credentials: str,
     db: Session,
+    proxy_config: dict[str, Any] | None = None,
+    key_proxy: dict[str, Any] | None = None,
 ) -> BatchImportResponse:
-    """Kiro 批量导入内部实现（供通用端点调用）。"""
+    """Kiro 批量导入内部实现（供通用端点调用）。
+
+    Args:
+        proxy_config: 本次操作使用的代理配置（已由调用方解析）
+        key_proxy: 需要保存到 Key 上的代理配置
+    """
     from src.services.provider.adapters.kiro.models.credentials import KiroAuthConfig
     from src.services.provider.adapters.kiro.token_manager import refresh_access_token
 
@@ -1611,8 +1677,6 @@ async def _batch_import_kiro_internal(
 
     api_formats = _get_provider_api_formats(provider)
 
-    proxy_config = getattr(provider, "proxy", None)
-
     results: list[BatchImportResultItem] = []
     success_count = 0
     failed_count = 0
@@ -1675,6 +1739,7 @@ async def _batch_import_kiro_internal(
                 auth_config=new_cfg.to_dict(),
                 api_formats=api_formats,
                 flush_only=True,
+                proxy=key_proxy,
             )
 
             results.append(

src/api/handlers/base/request_builder.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 
 import copy
 import json
+import re
 import time
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
@@ -33,6 +34,7 @@ from src.core.api_format import (
 from src.core.crypto import crypto_service
 from src.core.logger import logger
 from src.core.provider_oauth_utils import enrich_auth_config, post_oauth_token
+from src.models.endpoint_models import parse_re_flags
 
 if TYPE_CHECKING:
     from src.models.database import ProviderAPIKey, ProviderEndpoint
@@ -215,55 +217,103 @@ def build_test_request_body(
 # ==============================================================================
 
 
-def _parse_path(path: str) -> list[str]:
+# 路径段类型：str 表示 dict key，int 表示数组索引
+PathSegment = str | int
+
+
+def _parse_path(path: str) -> list[PathSegment]:
     """
-    解析点号路径，支持转义（用 \\.
-    表示字面量点号）。
+    解析路径，支持点号分隔、转义和数组索引。
 
     Examples:
-        "metadata.user.name" -> ["metadata", "user", "name"]
-        "config\\.v1.enabled" -> ["config.v1", "enabled"]
+        "metadata.user.name"       -> ["metadata", "user", "name"]
+        "config\\.v1.enabled"      -> ["config.v1", "enabled"]
+        "messages[0].content"      -> ["messages", 0, "content"]
+        "data[0].items[2].name"    -> ["data", 0, "items", 2, "name"]
+        "messages[-1]"             -> ["messages", -1]
+        "matrix[0][1]"             -> ["matrix", 0, 1]
 
     约束：
         - 不允许空段（例如：".a" / "a." / "a..b"），遇到则返回空列表表示无效路径。
         - 仅对 "\\." 做特殊处理；其他反斜杠组合按字面量保留。
+        - 数组索引必须是整数（支持负数索引）。
     """
     raw = (path or "").strip()
     if not raw:
         return []
 
-    parts: list[str] = []
+    parts: list[PathSegment] = []
     current: list[str] = []
+    expect_key = True  # 是否期望下一个片段是 dict key
 
     i = 0
     while i < len(raw):
         ch = raw[i]
+
+        # 转义点号：\\.
         if ch == "\\" and i + 1 < len(raw) and raw[i + 1] == ".":
             current.append(".")
+            expect_key = False
             i += 2
             continue
 
+        # 点号分隔符
         if ch == ".":
-            if not current:
+            if current:
+                parts.append("".join(current))
+                current = []
+            elif expect_key:
+                # 空段（如 ".a" 或 "a..b"）
                 return []
-            parts.append("".join(current))
-            current = []
+            expect_key = True
             i += 1
             continue
 
+        # 数组索引：[N]
+        if ch == "[":
+            # 先将当前累积的 key 入栈
+            if current:
+                parts.append("".join(current))
+                current = []
+
+            # 查找闭合括号
+            j = i + 1
+            while j < len(raw) and raw[j] != "]":
+                j += 1
+            if j >= len(raw):
+                return []  # 未闭合的括号
+
+            index_str = raw[i + 1 : j].strip()
+            if not index_str:
+                return []  # 空索引
+
+            try:
+                idx = int(index_str)
+            except ValueError:
+                return []  # 非整数索引
+
+            parts.append(idx)
+            expect_key = False
+            i = j + 1
+            continue
+
         current.append(ch)
+        expect_key = False
         i += 1
 
-    if not current:
+    # 收尾：将剩余的 key 入栈
+    if current:
+        parts.append("".join(current))
+    elif expect_key:
+        # 尾部悬挂的点号（如 "a."）
         return []
 
-    parts.append("".join(current))
-    return parts
+    return parts if parts else []
 
 
-def _get_nested_value(obj: dict[str, Any], path: str) -> tuple[bool, Any]:
+def _get_nested_value(obj: Any, path: str) -> tuple[bool, Any]:
     """
-    获取嵌套值
+    获取嵌套值，支持 dict 和 list 混合遍历
 
     Returns:
         (found, value) - found 为 True 时 value 有效
@@ -273,43 +323,92 @@ def _get_nested_value(obj: dict[str, Any], path: str) -> tuple[bool, Any]:
         return False, None
 
     current: Any = obj
-    for key in parts:
-        if isinstance(current, dict) and key in current:
-            current = current[key]
+    for segment in parts:
+        if isinstance(segment, int):
+            if isinstance(current, list):
+                try:
+                    current = current[segment]
+                except IndexError:
+                    return False, None
+            else:
+                return False, None
         else:
-            return False, None
+            if isinstance(current, dict) and segment in current:
+                current = current[segment]
+            else:
+                return False, None
     return True, current
 
 
 def _set_nested_value(obj: dict[str, Any], path: str, value: Any) -> bool:
     """
-    设置嵌套值，自动创建中间层级。
+    设置嵌套值，支持 dict 和 list 混合遍历。
 
-    当中间层存在但不是 dict 时，会覆盖为 dict 后继续写入（覆写语义）。
+    - dict 中间层：下一段为 str key 时自动创建（覆写语义）；下一段为 int 时要求已存在 list。
+    - list 中间层：必须已存在且索引有效。
+    - list 元素赋值：要求索引在范围内。
 
     Returns:
         True: 写入成功
-        False: 路径无效（空/含空段等）
+        False: 路径无效或结构不匹配
     """
     parts = _parse_path(path)
     if not parts:
         return False
 
-    current: dict[str, Any] = obj
-    for key in parts[:-1]:
-        next_val = current.get(key)
-        if not isinstance(next_val, dict):
-            next_val = {}
-            current[key] = next_val
-        current = next_val
+    current: Any = obj
+    for i in range(len(parts) - 1):
+        segment = parts[i]
+        next_segment = parts[i + 1]
 
-    current[parts[-1]] = value
-    return True
+        if isinstance(segment, int):
+            # 遍历数组元素
+            if not isinstance(current, list):
+                return False
+            try:
+                current = current[segment]
+            except IndexError:
+                return False
+        else:
+            # 遍历 dict key
+            if not isinstance(current, dict):
+                return False
+            child = current.get(segment)
+
+            if isinstance(next_segment, int):
+                # 下一段是数组索引 → child 必须已经是 list
+                if not isinstance(child, list):
+                    return False
+                current = child
+            else:
+                # 下一段是 dict key → 自动创建 dict（覆写语义）
+                if not isinstance(child, dict):
+                    child = {}
+                    current[segment] = child
+                current = child
+
+    # 写入最终值
+    last = parts[-1]
+    if isinstance(last, int):
+        if not isinstance(current, list):
+            return False
+        try:
+            current[last] = value
+            return True
+        except IndexError:
+            return False
+    else:
+        if not isinstance(current, dict):
+            return False
+        current[last] = value
+        return True
 
 
 def _delete_nested_value(obj: dict[str, Any], path: str) -> bool:
     """
-    删除嵌套值
+    删除嵌套值，支持 dict 和 list 混合遍历
+
+    对于 list 元素，使用 del 删除（会移动后续元素的索引）。
 
     Returns:
         True: 删除成功
@@ -320,23 +419,40 @@ def _delete_nested_value(obj: dict[str, Any], path: str) -> bool:
         return False
 
     current: Any = obj
-    for key in parts[:-1]:
-        if isinstance(current, dict) and key in current:
-            current = current[key]
+    for segment in parts[:-1]:
+        if isinstance(segment, int):
+            if isinstance(current, list):
+                try:
+                    current = current[segment]
+                except IndexError:
+                    return False
+            else:
+                return False
         else:
-            return False
-        if not isinstance(current, dict):
-            return False
+            if isinstance(current, dict) and segment in current:
+                current = current[segment]
+            else:
+                return False
 
-    if isinstance(current, dict) and parts[-1] in current:
-        del current[parts[-1]]
-        return True
-    return False
+    last = parts[-1]
+    if isinstance(last, int):
+        if isinstance(current, list):
+            try:
+                del current[last]
+                return True
+            except IndexError:
+                return False
+        return False
+    else:
+        if isinstance(current, dict) and last in current:
+            del current[last]
+            return True
+        return False
 
 
 def _rename_nested_value(obj: dict[str, Any], from_path: str, to_path: str) -> bool:
     """
-    重命名嵌套值（移动到新路径）
+    重命名嵌套值（移动到新路径），支持 dict 和 list 混合遍历
 
     Returns:
         True: 重命名成功
@@ -354,11 +470,39 @@ def _rename_nested_value(obj: dict[str, Any], from_path: str, to_path: str) -> b
     if not found:
         return False
 
+    # 先 set 再 delete，避免 set 失败时源值已被删除导致数据丢失
+    if not _set_nested_value(obj, dst, value):
+        return False
     _delete_nested_value(obj, src)
-    _set_nested_value(obj, dst, value)
     return True
 
 
+def _is_protected_path(parts: list[PathSegment], protected_lower: frozenset[str]) -> bool:
+    """检查路径的顶层 key 是否为受保护字段（int 索引不可能是受保护字段）"""
+    if not parts:
+        return False
+    first = parts[0]
+    return isinstance(first, str) and first.lower() in protected_lower
+
+
+def _extract_path(
+    rule: dict[str, Any],
+    protected_lower: frozenset[str],
+    key: str = "path",
+) -> str | None:
+    """从规则中提取并校验 path 字段，返回 strip 后的路径或 None（无效/受保护时）。"""
+    raw = rule.get(key, "")
+    if not isinstance(raw, str):
+        return None
+    path = raw.strip()
+    parts = _parse_path(path)
+    if not parts:
+        return None
+    if _is_protected_path(parts, protected_lower):
+        return None
+    return path
+
+
 def apply_body_rules(
     body: dict[str, Any],
     rules: list[dict[str, Any]],
@@ -370,11 +514,19 @@ def apply_body_rules(
     路径语法：
     - 使用点号分隔层级：metadata.user.name
     - 转义字面量点号：config\\.v1.enabled -> key "config.v1" 下的 "enabled"
+    - 使用方括号访问数组元素：messages[0].content
+    - 支持多层嵌套：data[0].items[2].name
+    - 支持负数索引：messages[-1]
+    - 支持连续数组索引：matrix[0][1]
 
     支持的规则类型：
     - set: 设置/覆盖字段 {"action": "set", "path": "metadata.user_id", "value": 123}
     - drop: 删除字段 {"action": "drop", "path": "unwanted_field"}
     - rename: 重命名字段 {"action": "rename", "from": "old.key", "to": "new.key"}
+    - append: 向数组追加元素 {"action": "append", "path": "messages", "value": {...}}
+    - insert: 在数组指定位置插入元素 {"action": "insert", "path": "messages", "index": 0, "value": {...}}
+    - regex_replace: 正则替换字符串值 {"action": "regex_replace", "path": "messages[0].content",
+        "pattern": "\\bfoo\\b", "replacement": "bar", "flags": "i", "count": 0}
 
     Args:
         body: 原始请求体
@@ -387,7 +539,7 @@ def apply_body_rules(
     if not rules:
         return body
 
-    # 深拷贝，避免修改原始数据（尤其是嵌套 dict）
+    # 深拷贝，避免修改原始数据（尤其是嵌套 dict/list）
     result = copy.deepcopy(body)
     protected = protected_keys or PROTECTED_BODY_FIELDS
     protected_lower = frozenset(str(k).lower() for k in protected)
@@ -402,27 +554,14 @@ def apply_body_rules(
         action = action.strip().lower()
 
         if action == "set":
-            raw_path = rule.get("path", "")
-            if not isinstance(raw_path, str):
+            path = _extract_path(rule, protected_lower)
+            if not path:
                 continue
-            path = raw_path.strip()
-            value = rule.get("value")
-            parts = _parse_path(path)
-            if not parts:
-                continue
-            if parts[0].lower() in protected_lower:
-                continue
-            _set_nested_value(result, path, value)
+            _set_nested_value(result, path, rule.get("value"))
 
         elif action == "drop":
-            raw_path = rule.get("path", "")
-            if not isinstance(raw_path, str):
-                continue
-            path = raw_path.strip()
-            parts = _parse_path(path)
-            if not parts:
-                continue
-            if parts[0].lower() in protected_lower:
+            path = _extract_path(rule, protected_lower)
+            if not path:
                 continue
             _delete_nested_value(result, path)
 
@@ -441,11 +580,62 @@ def apply_body_rules(
                 continue
 
             # 受保护字段只检查顶层 key
-            if from_parts[0].lower() in protected_lower or to_parts[0].lower() in protected_lower:
+            if _is_protected_path(from_parts, protected_lower) or _is_protected_path(
+                to_parts, protected_lower
+            ):
                 continue
 
             _rename_nested_value(result, from_path, to_path)
 
+        elif action == "append":
+            path = _extract_path(rule, protected_lower)
+            if not path:
+                continue
+            found, target = _get_nested_value(result, path)
+            if not found or not isinstance(target, list):
+                continue
+            target.append(rule.get("value"))
+
+        elif action == "insert":
+            path = _extract_path(rule, protected_lower)
+            if not path:
+                continue
+            index = rule.get("index")
+            if not isinstance(index, int):
+                continue
+            found, target = _get_nested_value(result, path)
+            if not found or not isinstance(target, list):
+                continue
+            target.insert(index, rule.get("value"))
+
+        elif action == "regex_replace":
+            path = _extract_path(rule, protected_lower)
+            if not path:
+                continue
+            pattern = rule.get("pattern")
+            replacement = rule.get("replacement", "")
+            if not isinstance(pattern, str) or not isinstance(replacement, str):
+                continue
+            if not pattern:
+                continue
+
+            flags_raw = rule.get("flags", "")
+            re_flags = parse_re_flags(flags_raw if isinstance(flags_raw, str) else "")
+
+            count = rule.get("count", 0)
+            if not isinstance(count, int) or count < 0:
+                count = 0
+
+            found, current_val = _get_nested_value(result, path)
+            if not found or not isinstance(current_val, str):
+                continue
+
+            try:
+                new_val = re.compile(pattern, re_flags).sub(replacement, current_val, count=count)
+                _set_nested_value(result, path, new_val)
+            except re.error:
+                continue  # 正则表达式无效，跳过
+
     return result
 
 

src/models/endpoint_models.py

@@ -22,13 +22,123 @@ HeaderRule = dict[str, Any]
 
 
 # ========== Body Rule 类型定义 ==========
-# 请求体规则支持三种操作：
+# 请求体规则支持六种操作：
 # - set: 设置/覆盖字段 {"action": "set", "path": "metadata", "value": {"custom": "val"}}
 # - drop: 删除字段 {"action": "drop", "path": "unwanted_field"}
 # - rename: 重命名字段 {"action": "rename", "from": "old_key", "to": "new_key"}
-# 实际验证在 request_builder.py 的 apply_body_rules 中处理
+# - append: 向数组追加元素 {"action": "append", "path": "messages", "value": {...}}
+# - insert: 在数组指定位置插入 {"action": "insert", "path": "messages", "index": 0, "value": {...}}
+# - regex_replace: 正则替换字符串值 {"action": "regex_replace", "path": "...", "pattern": "...", "replacement": "..."}
+# 路径语法支持数组索引：messages[0].content, data[-1], matrix[0][1]
+# 运行时处理在 request_builder.py 的 apply_body_rules 中；结构校验见 _validate_body_rules
 BodyRule = dict[str, Any]
 
+# body_rules 允许的 action 集合
+_BODY_RULE_ACTIONS: frozenset[str] = frozenset(
+    {"set", "drop", "rename", "append", "insert", "regex_replace"}
+)
+
+# regex_replace 允许的 flags 字符
+_REGEX_FLAG_CHARS: frozenset[str] = frozenset({"i", "m", "s"})
+
+
+def parse_re_flags(flags_str: str) -> int:
+    """将 flags 字符串（i/m/s）转换为 re 标志位。
+
+    供 endpoint_models 校验和 request_builder 运行时共用。
+    """
+    result = 0
+    for f in flags_str:
+        if f == "i":
+            result |= re.IGNORECASE
+        elif f == "m":
+            result |= re.MULTILINE
+        elif f == "s":
+            result |= re.DOTALL
+    return result
+
+
+def _validate_body_rules(rules: list[BodyRule]) -> list[BodyRule]:
+    """校验 body_rules 列表的结构和正则合法性。
+
+    校验项：
+    - 每条规则必须是 dict 且包含合法 action
+    - 需要 path 的 action（set/drop/append/insert/regex_replace）必须提供非空 path 字符串
+    - rename 必须提供非空 from / to 字符串
+    - insert 的 index 必须为整数
+    - regex_replace 的 pattern 必须能通过 re.compile 编译，flags 仅允许 i/m/s
+    """
+    for idx, rule in enumerate(rules):
+        if not isinstance(rule, dict):
+            raise ValueError(f"body_rules[{idx}]: 规则必须是 JSON 对象")
+
+        action = rule.get("action")
+        if not isinstance(action, str) or action.strip().lower() not in _BODY_RULE_ACTIONS:
+            raise ValueError(
+                f"body_rules[{idx}]: action 必须是 {sorted(_BODY_RULE_ACTIONS)} 之一，"
+                f"当前值: {action!r}"
+            )
+        action = action.strip().lower()
+
+        # ---------- path 校验 ----------
+        if action in {"set", "drop", "append", "insert", "regex_replace"}:
+            path = rule.get("path")
+            if not isinstance(path, str) or not path.strip():
+                raise ValueError(f"body_rules[{idx}]: action={action!r} 必须提供非空 path")
+
+        # ---------- rename 校验 ----------
+        if action == "rename":
+            from_val = rule.get("from")
+            to_val = rule.get("to")
+            if not isinstance(from_val, str) or not from_val.strip():
+                raise ValueError(f"body_rules[{idx}]: rename 必须提供非空 from")
+            if not isinstance(to_val, str) or not to_val.strip():
+                raise ValueError(f"body_rules[{idx}]: rename 必须提供非空 to")
+
+        # ---------- insert 校验 ----------
+        if action == "insert":
+            index = rule.get("index")
+            if not isinstance(index, int) or isinstance(index, bool):
+                raise ValueError(f"body_rules[{idx}]: insert 的 index 必须为整数")
+
+        # ---------- regex_replace 校验 ----------
+        if action == "regex_replace":
+            pattern = rule.get("pattern")
+            if not isinstance(pattern, str) or not pattern:
+                raise ValueError(f"body_rules[{idx}]: regex_replace 必须提供非空 pattern 字符串")
+
+            replacement = rule.get("replacement", "")
+            if not isinstance(replacement, str):
+                raise ValueError(f"body_rules[{idx}]: regex_replace 的 replacement 必须为字符串")
+
+            # 校验 flags
+            flags_str = rule.get("flags", "")
+            re_flags = 0
+            if isinstance(flags_str, str) and flags_str:
+                invalid_flags = set(flags_str) - _REGEX_FLAG_CHARS
+                if invalid_flags:
+                    raise ValueError(
+                        f"body_rules[{idx}]: regex_replace 的 flags 仅允许 "
+                        f"{''.join(sorted(_REGEX_FLAG_CHARS))}，"
+                        f"非法字符: {''.join(sorted(invalid_flags))}"
+                    )
+                re_flags = parse_re_flags(flags_str)
+
+            # 尝试编译正则，捕获语法错误
+            try:
+                re.compile(pattern, re_flags)
+            except re.error as e:
+                raise ValueError(
+                    f"body_rules[{idx}]: regex_replace 的 pattern 不是合法正则表达式: {e}"
+                )
+
+            # 校验 count
+            count = rule.get("count", 0)
+            if not isinstance(count, int) or count < 0:
+                raise ValueError(f"body_rules[{idx}]: regex_replace 的 count 必须为非负整数")
+
+    return rules
+
 
 # ========== ProviderEndpoint CRUD ==========
 
@@ -55,7 +165,7 @@ class ProviderEndpointCreate(BaseModel):
     # 请求体配置
     body_rules: list[BodyRule] | None = Field(
         default=None,
-        description="请求体规则列表，支持 set/drop/rename 操作",
+        description="请求体规则列表，支持 set/drop/rename/append/insert/regex_replace 操作",
     )
 
     max_retries: int = Field(default=2, ge=0, le=10, description="最大重试次数")
@@ -93,6 +203,14 @@ class ProviderEndpointCreate(BaseModel):
 
         return v.rstrip("/")  # 移除末尾斜杠
 
+    @field_validator("body_rules")
+    @classmethod
+    def validate_body_rules(cls, v: list[BodyRule] | None) -> list[BodyRule] | None:
+        """校验 body_rules 结构和正则合法性"""
+        if v is None:
+            return v
+        return _validate_body_rules(v)
+
 
 class ProviderEndpointUpdate(BaseModel):
     """更新 Endpoint 请求"""
@@ -111,7 +229,7 @@ class ProviderEndpointUpdate(BaseModel):
     # 请求体配置
     body_rules: list[BodyRule] | None = Field(
         default=None,
-        description="请求体规则列表，支持 set/drop/rename 操作",
+        description="请求体规则列表，支持 set/drop/rename/append/insert/regex_replace 操作",
     )
 
     max_retries: int | None = Field(default=None, ge=0, le=10, description="最大重试次数")
@@ -137,6 +255,14 @@ class ProviderEndpointUpdate(BaseModel):
 
         return v.rstrip("/")  # 移除末尾斜杠
 
+    @field_validator("body_rules")
+    @classmethod
+    def validate_body_rules(cls, v: list[BodyRule] | None) -> list[BodyRule] | None:
+        """校验 body_rules 结构和正则合法性"""
+        if v is None:
+            return v
+        return _validate_body_rules(v)
+
 
 class ProviderEndpointResponse(BaseModel):
     """Endpoint 响应"""

src/services/provider/adapters/codex/plugin.py

@@ -88,8 +88,8 @@ def build_codex_url(
 async def enrich_codex(
     auth_config: dict[str, Any],
     token_response: dict[str, Any],
-    access_token: str,
-    proxy_config: dict[str, Any] | None,
+    access_token: str,  # noqa: ARG001
+    proxy_config: dict[str, Any] | None,  # noqa: ARG001
 ) -> dict[str, Any]:
     """Codex auth_config enrichment: parse id_token -> email/account_id/plan_type/user_id."""
     from src.core.provider_oauth_utils import parse_codex_id_token
@@ -100,17 +100,11 @@ async def enrich_codex(
         bool(id_token),
         list(token_response.keys()),
     )
-    codex_info = parse_codex_id_token(str(id_token) if id_token else None)
+    # parse_codex_id_token 仅返回非空有效字段，直接 update 即可
+    codex_info = parse_codex_id_token(id_token)
     if codex_info:
         logger.debug("Codex parsed id_token fields: {}", list(codex_info.keys()))
-    if codex_info.get("email"):
-        auth_config["email"] = codex_info["email"]
-    if codex_info.get("account_id"):
-        auth_config["account_id"] = codex_info["account_id"]
-    if codex_info.get("plan_type"):
-        auth_config["plan_type"] = codex_info["plan_type"]
-    if codex_info.get("user_id"):
-        auth_config["user_id"] = codex_info["user_id"]
+        auth_config.update(codex_info)
     return auth_config