feat(nox-api): implement batch creation rate limit resilience and pending backfilling

backend/app/routers/upstreams.py

@@ -24,7 +24,7 @@ from app.schemas.upstream import (
     UpstreamCreate, UpstreamUpdate, UpstreamResponse, SnapshotResponse, TestResult,
     UpstreamBatchActionItem, UpstreamBatchActionSummary, UpstreamBatchActionResponse,
 )
-from app.services.upstream_client import UpstreamClient, UpstreamError, build_snapshot, build_new_api_token_name, mask_secret, _extract_key_value
+from app.services.upstream_client import UpstreamClient, UpstreamError, _PendingKeyError, build_snapshot, build_new_api_token_name, mask_secret, _extract_key_value
 from app.services.auth_config import MASK, mask_auth_config, normalize_auth_config
 from app.services.snapshot_service import diff_snapshots
 from app.services import scheduler as sched_svc
@@ -415,7 +415,8 @@ def _ensure_group_key(
             )
             if row and row.key_name:
                 # 本地已有记录（无论 key_id 是否存在），优先用本地已存的 key_name 查远端
-                # 这样旧格式 key_name 也能找回远端对应 token，不重复创建
+                # 这样旧格式 key_name 也能找回远端对应 token，不重复创建。
+                # created_pending_key 记录也走同一路径，尝试回填 key/id。
                 lookup_name = row.key_name
                 found_by_name = lookup_name  # 记录实际命中的远端名称
                 try:
@@ -442,13 +443,15 @@ def _ensure_group_key(
                     if found_by_name != lookup_name:
                         row.key_name = found_by_name
                     row.raw_json = json.dumps(existing, ensure_ascii=False)
+                    # pending 行回填成功后升级为 exists
                     row.status = "exists"
+                    row.error = None
                     row.updated_at = datetime.now(timezone.utc)
                     db.add(row)
                     db.commit()
                     db.refresh(row)
-                    return _key_response(row, include_value=False)
-                # 远端不存在，需要重新创建
+                    return _key_response(row, include_value=bool(key_value))
+                # 远端不存在，需要重新创建（包括 pending 行远端真的丢了）
                 row.status = "replaced"
 
             # 2. 查远端是否有同名 Key
@@ -507,16 +510,43 @@ def _ensure_group_key(
                 return _key_response(row, include_value=False)
 
             # 3. 远端不存在，创建新 Key（使用新短名）
-            created = client.create_api_key(
-                stable_name,
-                gid,
-                quota=body.quota,
-                expires_in_days=body.expires_in_days,
-                rate_limit_5h=body.rate_limit_5h,
-                rate_limit_1d=body.rate_limit_1d,
-                rate_limit_7d=body.rate_limit_7d,
-                endpoint=body.endpoint,
-            )
+            try:
+                created = client.create_api_key(
+                    stable_name,
+                    gid,
+                    quota=body.quota,
+                    expires_in_days=body.expires_in_days,
+                    rate_limit_5h=body.rate_limit_5h,
+                    rate_limit_1d=body.rate_limit_1d,
+                    rate_limit_7d=body.rate_limit_7d,
+                    endpoint=body.endpoint,
+                )
+            except _PendingKeyError as pke:
+                # POST 已成功但无法取得明文 key；落库为 pending，避免重复创建
+                pending_err = str(pke)
+                if row:
+                    row.key_id = row.key_id  # 保留原有（可能为空）
+                    row.key_name = stable_name
+                    row.key_value = ""
+                    row.managed_prefix = prefix
+                    row.status = "created_pending_key"
+                    row.error = pending_err
+                else:
+                    row = UpstreamGeneratedKey(
+                        upstream_id=upstream.id,
+                        group_id=gid,
+                        group_name=gname,
+                        key_name=stable_name,
+                        key_value="",
+                        managed_prefix=prefix,
+                        status="created_pending_key",
+                        error=pending_err,
+                    )
+                    db.add(row)
+                db.commit()
+                db.refresh(row)
+                return _key_response(row, include_value=False)
+
             if row:
                 # 复用旧行
                 row.key_id = created.get("id") or None
@@ -593,6 +623,16 @@ def generate_keys_by_groups(
         except Exception as exc:
             raise HTTPException(502, str(exc))
 
+        # New-API/Nox 上游：批量前预热 token 列表缓存（一次 /api/token/ 拉取），
+        # 避免每个分组都调 /api/token/search 触发 Nox 搜索限流（10 次/60 秒）。
+        if _is_new_api_user_upstream(u) or _is_nox_api_upstream(u):
+            try:
+                client.warm_token_list_cache()
+            except Exception as exc:
+                logger.warning(
+                    "generate_keys_by_groups: warm_token_list_cache failed for %s: %s", uid, exc
+                )
+
         for group in groups:
             gid = _group_id(group)
             if not gid or (selected and gid not in selected):
@@ -602,13 +642,17 @@ def generate_keys_by_groups(
 
     created = len([item for item in results if item.status == "created"])
     existed = len([item for item in results if item.status == "exists"])
+    pending = len([item for item in results if item.status == "created_pending_key"])
     total = len(results)
     msg_parts = []
     if created:
         msg_parts.append(f"新创建 {created}")
     if existed:
         msg_parts.append(f"已存在 {existed}")
+    if pending:
+        msg_parts.append(f"待回填 {pending}")
     msg = "、".join(msg_parts) + f" / 共 {total} 个分组" if msg_parts else f"共处理 {total} 个分组"
+    # created_pending_key 视为非失败（POST 已成功，只是 key 待回填）
     return GenerateKeysByGroupsResponse(
         success=total > 0 and all(item.status != "failed" for item in results),
         message=msg,

backend/app/services/upstream_client.py

@@ -16,6 +16,14 @@ class UpstreamError(RuntimeError):
     pass
 
 
+class _PendingKeyError(UpstreamError):
+    """POST /api/token/ 成功但最终无法取得明文 key 时抛出。
+
+    路由层捕获后将记录落库为 created_pending_key，而不是 failed。
+    """
+    pass
+
+
 NEW_API_DEFAULT_QUOTA_PER_UNIT = 500000
 
 # New-API/Nox-API token name 上限（UTF-8 字节）
@@ -367,6 +375,47 @@ def build_snapshot(upstream_id: int, base_url: str, api_prefix: str,
     }
 
 
+# 429 重试配置（秒）
+_RATE_LIMIT_BACKOFFS = (2, 5, 10, 20)
+_RATE_LIMIT_MAX_WAIT = 60
+
+
+def _retry_on_429(fn: Callable[[], Any], backoffs: tuple[int, ...] = _RATE_LIMIT_BACKOFFS) -> Any:
+    """执行 fn()，若遇 429 则按退避序列重试，超限后抛 UpstreamError。
+
+    fn 应是 HTTP 调用；429 响应会被 httpx.HTTPStatusError 捕获（raise_for_status 之后）
+    或者 fn 内部已经解析为 UpstreamError 且 message 含 429。
+    """
+    total_waited = 0
+    for attempt, wait in enumerate(backoffs):
+        try:
+            return fn()
+        except Exception as exc:
+            status = None
+            retry_after: int | None = None
+            # httpx HTTPStatusError
+            if hasattr(exc, "response") and hasattr(exc.response, "status_code"):
+                status = exc.response.status_code
+                try:
+                    retry_after = int(exc.response.headers.get("Retry-After", 0))
+                except (TypeError, ValueError):
+                    retry_after = None
+            # UpstreamError wrapping a 429 text
+            if status != 429 and "429" in str(exc):
+                status = 429
+            if status != 429:
+                raise
+            sleep_time = retry_after if (retry_after and retry_after > 0) else wait
+            if total_waited + sleep_time > _RATE_LIMIT_MAX_WAIT:
+                raise UpstreamError(
+                    f"rate limited (429) after {total_waited}s total wait; giving up"
+                ) from exc
+            time.sleep(sleep_time)
+            total_waited += sleep_time
+    # 最后一次尝试，不捕获
+    return fn()
+
+
 class UpstreamClient:
     """Sync HTTP client that handles all auth types."""
 
@@ -389,6 +438,8 @@ class UpstreamClient:
         self._cookies: dict[str, str] = {}
         self._new_api_user: str = ""
         self._client = httpx.Client(timeout=timeout)
+        # 批量生成期间的 token 列表缓存（name -> record），避免重复调用 /api/token/search
+        self._token_list_cache: dict[str, dict[str, Any]] | None = None
 
     def close(self) -> None:
         self._client.close()
@@ -727,6 +778,30 @@ class UpstreamClient:
             raise UpstreamError("New-API token key response did not include key")
         return key_value
 
+    def warm_token_list_cache(self) -> None:
+        """预热 token 列表缓存；批量生成前调用一次，后续 find_smartup_group_key 走缓存。
+
+        缓存以 token name 为 key；同名取最新（id 最大）。
+        """
+        all_tokens = self._list_all_new_api_tokens()
+        cache: dict[str, dict[str, Any]] = {}
+        for t in all_tokens:
+            name = str(t.get("name") or t.get("key_name") or "")
+            if not name:
+                continue
+            existing = cache.get(name)
+            if existing is None or (t.get("id") or 0) > (existing.get("id") or 0):
+                cache[name] = t
+        self._token_list_cache = cache
+
+    def _cache_add_token(self, record: dict[str, Any]) -> None:
+        """创建成功后把新 token 追加进缓存，让同批次后续查询能感知它。"""
+        if self._token_list_cache is None:
+            return
+        name = str(record.get("name") or record.get("key_name") or "")
+        if name:
+            self._token_list_cache[name] = record
+
     def _create_new_api_token(
         self,
         name: str,
@@ -734,6 +809,12 @@ class UpstreamClient:
         quota: float = 0,
         expires_in_days: int | None = None,
     ) -> dict[str, Any]:
+        """创建 Nox/New-API token。
+
+        主路径：POST 响应里直接有明文 key → 立即返回，不调用 search。
+        Fallback：响应无 key → 按名称查 token 列表再按 id 取明文 key；查询带 429 退避重试。
+        极端情况：POST 已成功但最终无法取得 key → 抛 _PendingKeyError，调用方应落库为 pending。
+        """
         unlimited = quota <= 0
         body: dict[str, Any] = {
             "name": name,
@@ -749,14 +830,62 @@ class UpstreamClient:
         payload = self._request("POST", "/api/token/", body)
         self._ensure_api_success(payload, "create New-API token")
 
-        matches = self._list_new_api_tokens(search=name, group_id=group_id)
-        token = next((i for i in matches if str(i.get("name") or "").strip() == name.strip()), None)
+        # ── 主路径：从创建响应直接取明文 key（Nox-API 当前行为：data="sk-..."）──
+        create_data = _unwrap_data(payload)
+        key_from_response = _extract_key_value(create_data) if create_data else ""
+        token_id_from_response = _extract_id(create_data) if isinstance(create_data, dict) else ""
+
+        if key_from_response and "*" not in key_from_response:
+            record = {
+                "name": name,
+                "group": str(group_id),
+                "group_id": str(group_id),
+                "key": key_from_response,
+                "id": token_id_from_response or None,
+            }
+            if isinstance(create_data, dict):
+                record.update(create_data)
+            self._cache_add_token(record)
+            return {
+                "id": token_id_from_response or "",
+                "key": key_from_response,
+                "masked_key": mask_secret(key_from_response),
+                "raw": record,
+            }
+
+        # ── Fallback：按名称找 token，再按 id 取明文 key；带 429 退避重试 ──
+        try:
+            matches = _retry_on_429(
+                lambda: self._list_new_api_tokens(search=name, group_id=group_id)
+            )
+        except Exception as exc:
+            raise _PendingKeyError(
+                f"POST /api/token/ 成功，但查询 token 列表失败（{exc}）；"
+                "key 待下次回填"
+            ) from exc
+
+        token = next(
+            (i for i in matches if str(i.get("name") or "").strip() == name.strip()), None
+        )
         if not token:
-            raise UpstreamError("New-API token was created but could not be found by name")
+            raise _PendingKeyError(
+                "POST /api/token/ 成功，但列表中找不到该 token；key 待下次回填"
+            )
+
         token_id = token.get("id")
         if token_id is None:
-            raise UpstreamError("New-API token list response did not include id")
-        key_value = self._get_new_api_token_key(token_id)
+            raise _PendingKeyError(
+                "POST /api/token/ 成功，但 token 无 id 无法取明文 key；key 待下次回填"
+            )
+
+        try:
+            key_value = _retry_on_429(lambda: self._get_new_api_token_key(token_id))
+        except Exception as exc:
+            raise _PendingKeyError(
+                f"POST /api/token/ 成功，但取明文 key 失败（{exc}）；key 待下次回填"
+            ) from exc
+
+        self._cache_add_token({**self._normalize_key_record(token), "key": key_value})
         return {
             "id": str(token_id),
             "key": key_value,
@@ -892,9 +1021,21 @@ class UpstreamClient:
     ) -> dict[str, Any] | None:
         """查找同一上游分组下是否已存在 SmartUp 前缀的 Key。
 
-        匹配规则：key_name 等于 expected_name，且以 prefix 开头。
-        返回匹配到的第一个 Key，或 None。
+        优先走本地 token 列表缓存（warm_token_list_cache 预热后有效），
+        避免在批量生成时高频调用 /api/token/search 触发 Nox 限流。
+        缓存未命中时回退到远端搜索。
         """
+        # ── 缓存命中路径 ──
+        if self._token_list_cache is not None:
+            record = self._token_list_cache.get(expected_name)
+            if record is not None:
+                rec_group = str(record.get("group_id") or record.get("group") or "")
+                if not rec_group or rec_group == str(group_id):
+                    return self._hydrate_new_api_token_key(record)
+            # 缓存中无此名 → 远端也不存在（批次内强一致性假设）
+            return None
+
+        # ── 无缓存：走远端搜索（原有逻辑）──
         gid = int(group_id) if str(group_id).isdigit() else group_id
         keys = self.list_api_keys(search=prefix, group_id=gid, status="active")
         for k in keys: