feat(compute): /v1中继鉴权重写与平台用户余额视图

- relay 鉴权重写：JWT→引擎 PAT 逐用户归属，拒绝匿名调用，余额<=0 预检 403 - compute_client 增 list_engine_users/get_engine_user/sync_user_mirror 等 - 运营端新增 /admin/compute/platform-users 平台用户视图(引擎真实余额/折扣/用量)与充值镜像回写 Co-Authored-By: Claude <noreply@anthropic.com>
2026-08-30 22:29:44 +08:00
parent f1147dced3
commit e44fea96ca
3 changed files with 156 additions and 21 deletions
@@ -521,11 +521,11 @@ async def compute_user_balance(
    db: Database = Depends(get_db),
    actor: dict = Depends(require_roles("operator")),
 ):
-    """给引擎用户充值/扣减/覆盖余额（配额单位）。
+    """给引擎用户充值/扣减/覆盖余额。

-    value 为配额单位：/api/status 的 quota_per_unit（默认 500000 = $1）。
+    value 为微元（人民币实际金额，1 元 = 1_000_000，compute-service meter.py 口径）。
    mode：add 充值 / subtract 扣减 / override 直接设为该值。
-    引擎按用户余额计量，额度耗尽时调用 /v1 返回 403。
+    引擎按用户余额实际扣费，额度耗尽时调用 /v1 返回 403。
    """
    if req.mode not in ("add", "subtract", "override"):
        raise HTTPException(status_code=400, detail="mode 需为 add/subtract/override")
@@ -535,6 +535,8 @@ async def compute_user_balance(
        result = await compute_client.adjust_user_quota(req.engine_user_id, req.value, req.mode)
    except compute_client.ComputeError as exc:
        raise HTTPException(status_code=502, detail=f"算力引擎对接失败: {exc}") from exc
+    # 充值后即时回写平台侧余额镜像，避免与引擎实际余额漂移（best-effort）。
+    await compute_client.sync_user_mirror(db, req.engine_user_id)
    await write_audit(db, action="compute.balance", resource="compute",
                resource_id=str(req.engine_user_id),
                detail=f"{req.mode} {req.value} quota" + (f" ({req.reason})" if req.reason else ""),
@@ -582,6 +584,57 @@ async def sync_compute_users(
    return {"total": len(users), "synced": created, "pat_issued": pats}


+@router.get("/compute/platform-users", summary="平台用户 × 引擎真实余额（充值以引擎为准）")
+async def compute_platform_users(
+    db: Database = Depends(get_db),
+    _u: dict = Depends(require_roles("operator")),
+):
+    """按「平台用户」聚合展示算力账户，余额取**引擎真实值**（compute.db users.quota，扣费真源）。
+
+    - 映射键：用户名字符串（platform.username == engine.username）。
+    - 引擎侧字段（engine_id / quota / used_quota / discount / group）逐页全量拉取后 join；
+      引擎缺失的用户标记 provisioned=False（余额回退镜像值并置 0）。
+    - 镜像（compute_quota）仅作参考列返回，**不作为充值/展示真源**。
+    """
+    users = await db.users.list()
+    engine: dict[str, dict] = {}
+    page = 1
+    while page <= 20:  # 上限 20 页 × 100 = 2000 引擎用户，足够且防失控
+        try:
+            items = await compute_client.list_engine_users(page=page, page_size=100)
+        except compute_client.ComputeError as exc:
+            raise HTTPException(status_code=502, detail=f"算力引擎对接失败: {exc}") from exc
+        for it in items:
+            uname = str(it.get("username") or "")
+            if uname:
+                engine[uname] = it
+        if len(items) < 100:
+            break
+        page += 1
+    items_out = []
+    for u in users:
+        uname = u.get("username") or ""
+        e = engine.get(uname) or {}
+        items_out.append({
+            "id": u["id"],
+            "username": uname,
+            "nickname": u.get("nickname") or "",
+            "role": u.get("role") or "",
+            "provisioned": bool(e),
+            "engine_id": e.get("id"),
+            # 真实余额（引擎为准）；引擎缺失时回退镜像（0）
+            "quota": int(e.get("quota") if e else (u.get("compute_quota") or 0)) or 0,
+            "used_quota": int(e.get("used_quota") if e else (u.get("compute_used_quota") or 0)) or 0,
+            "discount": int(e.get("discount") or 100) if e else 100,
+            "group": e.get("group") or "",
+            "engine_display_name": e.get("display_name") or "",
+            "request_count": int(e.get("request_count") or 0) if e else 0,
+            "last_login_at": e.get("last_login_at"),
+            "mirror_quota": int(u.get("compute_quota") or 0),  # 参考列
+        })
+    return {"items": items_out, "total": len(items_out), "engine_total": len(engine)}
+
+
@router.api_route(
    "/compute/proxy/{path:path}",
    methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
@@ -17,28 +17,66 @@ from fastapi.responses import JSONResponse, StreamingResponse
 from starlette.background import BackgroundTask

 from ... import config
-from ...services import compute_catalog
+from ...services import compute_catalog, compute_client

 router = APIRouter(prefix="/v1", tags=["relay"])


-def _auth_headers(request: Request) -> dict[str, str]:
+def _is_jwt(token: str) -> bool:
+    """平台 JWT 形如 header.payload.signature（两处 '.'）；引擎 PAT 不含 '.'。"""
+    return token.count(".") == 2
+
+
+async def _user_engine_pat(token: str) -> str:
+    """把平台 JWT 解析为该用户的引擎消费令牌（PAT）。
+
+    取该用户引擎名下第一枚有效令牌；没有则现场签发一枚，保证「登录用户=引擎用户」。
+    """
+    from ...jwt import decode_access_token
+
+    claims = decode_access_token(token)
+    username = (claims or {}).get("username") or ""
+    if not username:
+        raise HTTPException(status_code=401, detail="登录态无效，请重新登录")
+    await compute_client.ensure_user(username)
+    items = await compute_client.list_user_tokens(username)
+    for it in items or []:
+        key = it.get("key") or it.get("token")
+        if key:
+            return key
+    issued = await compute_client.issue_user_token(username, name="平台中继令牌")
+    key = issued.get("key") or issued.get("token") or ""
+    if not key:
+        raise HTTPException(status_code=502, detail="算力令牌签发失败，请联系运营方")
+    return key
+
+
+async def _auth_headers(request: Request) -> dict[str, str]:
    """模型调用鉴权：把请求归到「真实用户令牌」，由引擎按该用户余额计量。

-    - 优先透传客户端 Authorization（用户算力 PAT → compute 归户计量，引擎按其余额判定）。
-    - 无用户令牌：仅回落受限消费令牌 COMPUTE_RELAY_TOKEN（引擎按该受限用户余额计量）。
-    - 二者皆无：**拒绝匿名**——绝不回落 COMPUTE_ADMIN_TOKEN（root 无限额度），
-      否则「用户无余额仍能免费使用」。（引擎用户额度耗尽会返回 403，见 billing_session）。
+    - 客户端携带引擎 PAT：直接透传（compute 校验令牌与余额，无效即 401）。
+    - 客户端携带平台 JWT（桌面端/网页端登录态）：解析用户 → 归户到其引擎 PAT。
+    - 二者皆无：**拒绝匿名**——不回落 COMPUTE_RELAY_TOKEN/COMPUTE_ADMIN_TOKEN，
+      否则「用户无余额仍能免费使用」。
    """
-    client_auth = request.headers.get("Authorization", "")
-    if client_auth:
-        return {"Authorization": client_auth}
-    if config.COMPUTE_RELAY_TOKEN:
-        return {"Authorization": f"Bearer {config.COMPUTE_RELAY_TOKEN}"}
-    raise HTTPException(
-        status_code=401,
-        detail="缺少算力令牌：请携带有效的用户算力令牌（Bearer <PAT>）调用",
-    )
+    client_auth = (request.headers.get("Authorization", "") or "").strip()
+    if not client_auth:
+        raise HTTPException(
+            status_code=401,
+            detail="缺少算力令牌：请登录后调用（或携带用户算力令牌 Bearer <PAT>）",
+        )
+    token = client_auth[7:].strip() if client_auth.lower().startswith("bearer ") else client_auth
+    if not token:
+        raise HTTPException(status_code=401, detail="算力令牌为空")
+    if _is_jwt(token):
+        try:
+            pat = await _user_engine_pat(token)
+        except HTTPException:
+            raise
+        except Exception as exc:  # noqa: BLE001
+            raise HTTPException(status_code=502, detail=f"算力引擎对接失败: {exc}") from exc
+        return {"Authorization": f"Bearer {pat}"}
+    return {"Authorization": f"Bearer {token}"}


@router.post("/chat/completions")
@@ -51,7 +89,7 @@ async def relay_chat_completions(request: Request):

    chat_url = f"{config.COMPUTE_BASE_URL.rstrip('/')}/v1/chat/completions"
    stream = bool(body.get("stream", False))
-    headers = {**_auth_headers(request), "Content-Type": "application/json"}
+    headers = {**(await _auth_headers(request)), "Content-Type": "application/json"}

    client = httpx.AsyncClient(timeout=None)
    upstream_request = client.build_request("POST", chat_url, json=body, headers=headers)