From e44fea96ca4728356623a00b66570cd42bea66af Mon Sep 17 00:00:00 2001 From: Pine Date: Sun, 30 Aug 2026 22:29:44 +0800 Subject: [PATCH] =?UTF-8?q?feat(compute):=20/v1=E4=B8=AD=E7=BB=A7=E9=89=B4?= =?UTF-8?q?=E6=9D=83=E9=87=8D=E5=86=99=E4=B8=8E=E5=B9=B3=E5=8F=B0=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E4=BD=99=E9=A2=9D=E8=A7=86=E5=9B=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - relay 鉴权重写：JWT→引擎 PAT 逐用户归属，拒绝匿名调用，余额<=0 预检 403 - compute_client 增 list_engine_users/get_engine_user/sync_user_mirror 等 - 运营端新增 /admin/compute/platform-users 平台用户视图(引擎真实余额/折扣/用量)与充值镜像回写 Co-Authored-By: Claude --- app/api/routers/rbac_operator.py | 59 +++++++++++++++++++++++++-- app/api/routers/relay.py | 70 ++++++++++++++++++++++++-------- app/services/compute_client.py | 48 +++++++++++++++++++++- 3 files changed, 156 insertions(+), 21 deletions(-) diff --git a/app/api/routers/rbac_operator.py b/app/api/routers/rbac_operator.py index bcfed99..a6a1acb 100644 --- a/app/api/routers/rbac_operator.py +++ b/app/api/routers/rbac_operator.py @@ -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"], diff --git a/app/api/routers/relay.py b/app/api/routers/relay.py index 3b05234..2e4ece1 100644 --- a/app/api/routers/relay.py +++ b/app/api/routers/relay.py @@ -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 ）调用", - ) + client_auth = (request.headers.get("Authorization", "") or "").strip() + if not client_auth: + raise HTTPException( + status_code=401, + detail="缺少算力令牌：请登录后调用（或携带用户算力令牌 Bearer ）", + ) + 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) diff --git a/app/services/compute_client.py b/app/services/compute_client.py index a9a6681..cd92ccf 100644 --- a/app/services/compute_client.py +++ b/app/services/compute_client.py @@ -188,8 +188,8 @@ async def adjust_user_quota(engine_user_id: int, value: int, mode: str = "add") - mode="add" 充值（增加余额） - mode="subtract" 扣减 - mode="override" 直接设为该值 - ``value`` 为配额单位（见 /api/status 的 quota_per_unit：500000 = $1）。 - 被充值用户的其他 PAT/会话会实时扣除，额度耗尽时引擎返回 403。 + ``value`` 为微元（人民币实际金额，1 元 = 1_000_000，见 compute-service meter.py）。 + 额度耗尽时引擎 /v1 返回 403。 """ return await _request( method="POST", path="/api/user/manage", headers=_admin_headers(), @@ -197,6 +197,50 @@ async def adjust_user_quota(engine_user_id: int, value: int, mode: str = "add") ) +async def list_engine_users(page: int = 1, page_size: int = 100, status: int = 1) -> list[dict]: + """分页拉取引擎用户（GET /api/user），返回 items 列表。""" + data = await _request( + method="GET", + path=f"/api/user?status={int(status)}&type=-1&p={int(page)}&page_size={int(page_size)}", + headers=_admin_headers(), + ) + return (data.get("data") or {}).get("items") or [] + + +async def get_engine_user(engine_user_id: int) -> dict | None: + """按 id 取引擎用户（GET /api/user?id=），无则 None。""" + data = await _request( + method="GET", path=f"/api/user?id={int(engine_user_id)}&p=1&page_size=1", + headers=_admin_headers(), + ) + items = (data.get("data") or {}).get("items") or [] + return items[0] if items else None + + +async def sync_user_mirror(db, engine_user_id: int) -> bool: + """把引擎用户余额回写到平台镜像（compute_quota/compute_used_quota）。 + + best-effort：引擎用户不存在或平台用户未映射时静默返回 False。 + """ + try: + engine_user = await get_engine_user(engine_user_id) + username = str((engine_user or {}).get("username") or "") + if not username: + return False + balance = await user_balance(username) + platform_user = await db.users.get_by_username(username) + if platform_user is None: + return False + await db.users.set_compute_mirror( + platform_user["id"], provisioned=True, username=username, + quota=int(balance.get("quota") or 0), + used_quota=int(balance.get("used_quota") or 0), + ) + return True + except Exception: # noqa: BLE001 + return False + + async def set_user_group_by_username(username: str, group: str) -> dict: """把引擎用户划到某个用户组（企业折扣凭 User.Group 触发组倍率）。