P1/P2 计费体系修复：统一扣费回调+请求前预检+对账接口+折扣同步
P1-1 统一扣费回调： - 新增 POST /api/compute/internal/deduct（COMPUTE_ADMIN_TOKEN鉴权，engine_log_id幂等） - compute 扣费后异步回调，按引擎实际费用（微元）扣平台来源账本 - 删除 relay.py 中 _bill_usage/_bill_usage_anthropic 调用（改由compute回调记账） - ComputeUsageRecord 加 engine_log_id 字段，alembic 0079 迁移 P1-2 请求前预检： - _user_engine_pat 中查 compute 余额（5s缓存），余额<=0直接403 - 避免无余额请求仍转发到引擎 P2-1 对账接口： - GET /admin/compute/reconcile 对比引擎used_quota vs 平台累计扣费 - 支持分页、only_mismatch过滤 P2-2 折扣统一： - 用户级折扣设置/删除时异步同步等效系数到引擎users.discount - compute /api/user/manage 支持 username 定位用户 - compute_client 新增 set_user_discount_by_username
This commit is contained in:
Pine
@@ -0,0 +1,39 @@
|
||||
"""0079 compute_usage_records 加 engine_log_id 幂等键
|
||||
|
||||
compute 引擎扣费后回调 server-core 记账时，用引擎日志 id 做幂等，
|
||||
防止网络重试导致重复扣费。
|
||||
|
||||
Revision ID: 0079
|
||||
Revises: 0078
|
||||
Create Date: 2026-09-13
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
|
||||
revision = "0079"
|
||||
down_revision = "0078"
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def _column_exists(conn, table_name: str, column_name: str) -> bool:
|
||||
try:
|
||||
result = conn.execute(sa.text(f"SHOW COLUMNS FROM `{table_name}` LIKE '{column_name}'"))
|
||||
return result.fetchone() is not None
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
if not _column_exists(conn, "compute_usage_records", "engine_log_id"):
|
||||
op.add_column(
|
||||
"compute_usage_records",
|
||||
sa.Column("engine_log_id", sa.Integer, nullable=False, server_default="0", index=True),
|
||||
)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
conn = op.get_bind()
|
||||
if _column_exists(conn, "compute_usage_records", "engine_log_id"):
|
||||
op.drop_column("compute_usage_records", "engine_log_id")
|
||||
@@ -0,0 +1,61 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""compute 引擎 → server-core 内部回调接口。
|
||||
|
||||
compute 扣费完成后异步回调此接口，按引擎实际费用（微元）扣平台来源账本。
|
||||
鉴权：Authorization: Bearer <COMPUTE_ADMIN_TOKEN>（与引擎侧 PINEAGENTS_INTERNAL_TOKEN 一致）。
|
||||
幂等：engine_log_id 去重，防网络重试重复扣费。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Depends, Header, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
||||
from ... import config
|
||||
from ...infrastructure.repositories import Database
|
||||
from ..dependencies import get_db
|
||||
from ...services.compute_pricing_service import deduct_by_engine_cost
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/compute/internal", tags=["compute-internal"])
|
||||
|
||||
|
||||
class EngineDeductRequest(BaseModel):
|
||||
username: str
|
||||
actual_cost_micro: int
|
||||
engine_log_id: int
|
||||
model_name: str = ""
|
||||
token_count: int = 0
|
||||
|
||||
|
||||
def _verify_internal_token(authorization: str = Header(default="")) -> None:
|
||||
"""校验 compute 引擎内部回调令牌。"""
|
||||
token = authorization[7:].strip() if authorization.lower().startswith("bearer ") else authorization.strip()
|
||||
expected = config.COMPUTE_ADMIN_TOKEN or ""
|
||||
if not expected or token != expected:
|
||||
raise HTTPException(status_code=401, detail="invalid internal token")
|
||||
|
||||
|
||||
@router.post("/deduct", summary="compute 引擎扣费回调（按实际费用扣平台账本）")
|
||||
async def engine_deduct(
|
||||
req: EngineDeductRequest,
|
||||
db: Database = Depends(get_db),
|
||||
_auth: None = Depends(_verify_internal_token),
|
||||
):
|
||||
"""compute 扣费后回调：actual_cost_micro（微元）→ 扣企业分配/个人余额，engine_log_id 幂等。"""
|
||||
try:
|
||||
result = await deduct_by_engine_cost(
|
||||
db,
|
||||
username=req.username,
|
||||
actual_cost_micro=req.actual_cost_micro,
|
||||
engine_log_id=req.engine_log_id,
|
||||
model_name=req.model_name,
|
||||
token_count=req.token_count,
|
||||
)
|
||||
return {"ok": result.get("ok", True), **result}
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("[compute-internal] deduct callback failed username=%s engine_log_id=%s err=%s",
|
||||
req.username, req.engine_log_id, exc)
|
||||
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
||||
@@ -424,6 +424,28 @@ async def tenant_user_discounts(
|
||||
return {"items": items}
|
||||
|
||||
|
||||
async def _sync_user_discount_to_engine(db: Database, user_id: str, tenant_id: str, user_discount: int) -> None:
|
||||
"""P2-2 折扣统一：把 server-core 四级折扣等效值同步到引擎 users.discount。
|
||||
|
||||
等效系数 = (1 - tenant_discount/100) * (1 - user_discount/100) * 100。
|
||||
异步执行，失败仅记日志（不阻断主流程）。
|
||||
"""
|
||||
try:
|
||||
import asyncio
|
||||
from ...services import compute_client
|
||||
from ...services.compute_pricing_service import _get_tenant_discount
|
||||
user = await db.users.get_by_id(user_id)
|
||||
username = (user or {}).get("username", "") if user else ""
|
||||
if not username:
|
||||
return
|
||||
tenant_disc = await _get_tenant_discount(db, tenant_id)
|
||||
engine_discount = int(round((1 - tenant_disc / 100) * (1 - user_discount / 100) * 100))
|
||||
engine_discount = max(1, min(500, engine_discount))
|
||||
asyncio.create_task(compute_client.set_user_discount_by_username(username, engine_discount))
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("[compute-pricing] 同步折扣到引擎失败 user=%s: %s", user_id, exc)
|
||||
|
||||
|
||||
@router.post("/tenant/user-discounts", summary="载体端：设置账号折扣")
|
||||
async def tenant_set_user_discount(
|
||||
body: UserDiscountBody,
|
||||
@@ -465,6 +487,8 @@ async def tenant_set_user_discount(
|
||||
await write_audit(db, action="compute.user_discount_set", resource="tenant_user_discount",
|
||||
resource_id=tud.id, detail=f"user={body.user_id}, discount={body.discount}",
|
||||
user=user, request=request)
|
||||
# P2-2 折扣统一：同步等效折扣到引擎 users.discount（异步，失败仅记日志）
|
||||
await _sync_user_discount_to_engine(db, body.user_id, tenant.id, body.discount)
|
||||
return {"ok": True, "id": tud.id}
|
||||
|
||||
|
||||
@@ -487,6 +511,8 @@ async def tenant_delete_user_discount(
|
||||
await db.session.commit()
|
||||
await write_audit(db, action="compute.user_discount_delete", resource="tenant_user_discount",
|
||||
resource_id=discount_id, user=user, request=request)
|
||||
# P2-2 折扣统一：删除用户折扣后恢复为仅 tenant 折扣（异步，失败仅记日志）
|
||||
await _sync_user_discount_to_engine(db, tud.user_id, tenant.id, 0)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@@ -1071,3 +1097,76 @@ async def transfer_from_personal(
|
||||
resource_id=company_id, detail=f"user {user['id']} {total_micro} micro → {amount_fen} fen, reason={body.reason}",
|
||||
user=user, request=request)
|
||||
return {"ok": True, "amount_fen": amount_fen, "company_balance": company.compute_balance}
|
||||
|
||||
|
||||
# ═══════════════════════════════════════════════════════════════════
|
||||
# 运营端：算力对账（P2-1）
|
||||
# ═══════════════════════════════════════════════════════════════════
|
||||
|
||||
@router.get("/admin/compute/reconcile", summary="运营端：算力扣费对账（引擎used_quota vs 平台累计扣费）")
|
||||
async def admin_compute_reconcile(
|
||||
page: int = 1, page_size: int = 50,
|
||||
only_mismatch: int = 1,
|
||||
db: Database = Depends(get_db),
|
||||
user: dict = Depends(get_current_user),
|
||||
):
|
||||
"""对比 compute 引擎 used_quota 与 server-core compute_usage_records 累计扣费。
|
||||
|
||||
引擎侧 used_quota 单位微元；平台侧 actual_amount 单位分（1分=10000微元）。
|
||||
only_mismatch=1 只返回有差异的用户；=0 返回全部。
|
||||
"""
|
||||
check_role(user, ["admin", "operator"])
|
||||
from ...services import compute_client
|
||||
from sqlalchemy import func as _sa_func
|
||||
|
||||
# 分页拉取引擎用户
|
||||
engine_users = await compute_client.list_engine_users(page=page, page_size=page_size, status=-1)
|
||||
# 批量查平台累计扣费（按 username 关联）
|
||||
usernames = [u.get("username", "") for u in engine_users if u.get("username")]
|
||||
platform_totals: dict[str, int] = {}
|
||||
if usernames:
|
||||
# 先按 username 查 user_id，再汇总 usage records
|
||||
user_rows = (await db.session.execute(
|
||||
select(User.id, User.username).where(User.username.in_(usernames))
|
||||
)).all()
|
||||
uid_to_name = {uid: uname for uid, uname in user_rows}
|
||||
if uid_to_name:
|
||||
usage_rows = (await db.session.execute(
|
||||
select(ComputeUsageRecord.user_id, _sa_func.sum(ComputeUsageRecord.actual_amount))
|
||||
.where(ComputeUsageRecord.user_id.in_(list(uid_to_name.keys())))
|
||||
.group_by(ComputeUsageRecord.user_id)
|
||||
)).all()
|
||||
for uid, total_fen in usage_rows:
|
||||
uname = uid_to_name.get(uid, "")
|
||||
if uname:
|
||||
platform_totals[uname] = int(total_fen or 0) * 10000 # 分→微元
|
||||
|
||||
items = []
|
||||
mismatched = 0
|
||||
for eu in engine_users:
|
||||
uname = eu.get("username", "")
|
||||
engine_used = int(eu.get("used_quota", 0) or 0)
|
||||
platform_used = platform_totals.get(uname, 0)
|
||||
diff = engine_used - platform_used
|
||||
if only_mismatch and abs(diff) < 10000: # 容差 1 分（10000微元）内视为一致
|
||||
continue
|
||||
if abs(diff) >= 10000:
|
||||
mismatched += 1
|
||||
items.append({
|
||||
"username": uname,
|
||||
"engine_user_id": eu.get("id"),
|
||||
"engine_used_quota": engine_used,
|
||||
"engine_used_yuan": round(engine_used / 1_000_000, 4),
|
||||
"platform_total_deducted": platform_used,
|
||||
"platform_total_yuan": round(platform_used / 1_000_000, 4),
|
||||
"diff_micro": diff,
|
||||
"diff_yuan": round(diff / 1_000_000, 4),
|
||||
})
|
||||
|
||||
return {
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"engine_users_total": len(engine_users),
|
||||
"mismatched_count": mismatched,
|
||||
"items": items,
|
||||
}
|
||||
|
||||
+38
-43
@@ -53,6 +53,7 @@ async def _user_engine_pat(token: str) -> str:
|
||||
"""把平台 JWT 解析为该用户的引擎消费令牌（PAT）。
|
||||
|
||||
取该用户引擎名下第一枚有效令牌；没有则现场签发一枚，保证「登录用户=引擎用户」。
|
||||
请求前预检：引擎余额 <=0 直接 403（5s 缓存，避免无余额仍转发到引擎）。
|
||||
"""
|
||||
from ...jwt import decode_access_token
|
||||
|
||||
@@ -61,6 +62,10 @@ async def _user_engine_pat(token: str) -> str:
|
||||
if not username:
|
||||
raise HTTPException(status_code=401, detail="登录态无效，请重新登录")
|
||||
await compute_client.ensure_user(username)
|
||||
# 请求前预检：余额 <=0 直接拒绝（5s 内存缓存，查询失败时放行由引擎侧兜底）
|
||||
quota = await _get_cached_engine_balance(username)
|
||||
if quota == 0:
|
||||
raise HTTPException(status_code=403, detail="算力余额不足，请先充值")
|
||||
items = await compute_client.list_user_tokens(username)
|
||||
for it in items or []:
|
||||
key = it.get("key") or it.get("token")
|
||||
@@ -74,6 +79,27 @@ async def _user_engine_pat(token: str) -> str:
|
||||
return key
|
||||
|
||||
|
||||
# 引擎余额预检缓存：username -> (timestamp, quota)，5s TTL
|
||||
_balance_cache: dict[str, tuple[float, int]] = {}
|
||||
_BALANCE_CACHE_TTL = 5.0
|
||||
|
||||
|
||||
async def _get_cached_engine_balance(username: str) -> int:
|
||||
"""查 compute 引擎用户余额（5s 缓存）。查询失败返回 -1（放行，由引擎侧兜底）。"""
|
||||
import time as _time
|
||||
now = _time.time()
|
||||
cached = _balance_cache.get(username)
|
||||
if cached and now - cached[0] < _BALANCE_CACHE_TTL:
|
||||
return cached[1]
|
||||
try:
|
||||
bal = await compute_client.user_balance(username)
|
||||
quota = int(bal.get("quota", 0) or 0)
|
||||
except Exception: # noqa: BLE001
|
||||
quota = -1
|
||||
_balance_cache[username] = (now, quota)
|
||||
return quota
|
||||
|
||||
|
||||
async def _auth_headers(request: Request) -> dict[str, str]:
|
||||
"""模型调用鉴权：把请求归到「真实用户令牌」，由引擎按该用户余额计量。
|
||||
|
||||
@@ -157,9 +183,7 @@ async def relay_chat_completions(request: Request, db: Database = Depends(get_db
|
||||
await client.aclose()
|
||||
raise HTTPException(status_code=502, detail="Upstream relay failed") from None
|
||||
await client.aclose()
|
||||
# 按实际用量记账（best-effort，失败不影响响应）
|
||||
if billing_user_id and isinstance(payload, dict) and payload.get("usage"):
|
||||
await _bill_usage(db, billing_user_id, model, payload.get("usage"))
|
||||
# 记账由 compute 引擎扣费后异步回调 /api/compute/internal/deduct（engine_log_id 幂等）
|
||||
return JSONResponse(status_code=upstream.status_code, content=payload)
|
||||
|
||||
try:
|
||||
@@ -183,31 +207,19 @@ async def relay_chat_completions(request: Request, db: Database = Depends(get_db
|
||||
background=BackgroundTask(client.aclose),
|
||||
)
|
||||
|
||||
# 流式：透传 SSE 同时解析 usage chunk，结束后记账
|
||||
async def _stream_with_billing():
|
||||
usage: dict | None = None
|
||||
# 流式：透传 SSE；记账由 compute 引擎扣费后异步回调（engine_log_id 幂等）
|
||||
async def _stream_passthrough():
|
||||
try:
|
||||
async for line in upstream.aiter_lines():
|
||||
yield (line + "\n").encode("utf-8")
|
||||
if line.startswith("data:"):
|
||||
payload = line[5:].strip()
|
||||
if payload and payload != "[DONE]":
|
||||
try:
|
||||
obj = json.loads(payload)
|
||||
if isinstance(obj, dict) and obj.get("usage"):
|
||||
usage = obj["usage"]
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
async for raw in upstream.aiter_raw():
|
||||
yield raw
|
||||
finally:
|
||||
try:
|
||||
await client.aclose()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
if usage:
|
||||
await _bill_usage(db, billing_user_id, model, usage)
|
||||
|
||||
return StreamingResponse(
|
||||
_stream_with_billing(),
|
||||
_stream_passthrough(),
|
||||
status_code=upstream.status_code,
|
||||
media_type="text/event-stream",
|
||||
headers=resp_headers,
|
||||
@@ -246,9 +258,7 @@ async def relay_messages(request: Request, db: Database = Depends(get_db)):
|
||||
await client.aclose()
|
||||
raise HTTPException(status_code=502, detail="Upstream relay failed") from None
|
||||
await client.aclose()
|
||||
# Anthropic usage（input_tokens/output_tokens）→ 平台记账
|
||||
if billing_user_id and isinstance(payload, dict) and payload.get("usage"):
|
||||
await _bill_usage_anthropic(db, billing_user_id, model, payload.get("usage"))
|
||||
# 记账由 compute 引擎扣费后异步回调 /api/compute/internal/deduct
|
||||
return JSONResponse(status_code=upstream.status_code, content=payload)
|
||||
|
||||
try:
|
||||
@@ -272,34 +282,19 @@ async def relay_messages(request: Request, db: Database = Depends(get_db)):
|
||||
background=BackgroundTask(client.aclose),
|
||||
)
|
||||
|
||||
# 流式：透传 SSE 同时解析 usage（message_start.input_tokens + message_delta.output_tokens），结束后记账
|
||||
async def _stream_with_billing():
|
||||
usage: dict | None = None
|
||||
# 流式：透传 SSE；记账由 compute 引擎扣费后异步回调
|
||||
async def _stream_passthrough_messages():
|
||||
try:
|
||||
async for line in upstream.aiter_lines():
|
||||
yield (line + "\n").encode("utf-8")
|
||||
if line.startswith("data:"):
|
||||
payload = line[5:].strip()
|
||||
if payload:
|
||||
try:
|
||||
obj = json.loads(payload)
|
||||
if isinstance(obj, dict):
|
||||
if obj.get("type") == "message_start" and isinstance(obj.get("message"), dict):
|
||||
usage = dict(obj["message"].get("usage") or {})
|
||||
elif obj.get("type") == "message_delta" and isinstance(obj.get("usage"), dict):
|
||||
usage = {**(usage or {}), **obj["usage"]}
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
async for raw in upstream.aiter_raw():
|
||||
yield raw
|
||||
finally:
|
||||
try:
|
||||
await client.aclose()
|
||||
except Exception: # noqa: BLE001
|
||||
pass
|
||||
if usage:
|
||||
await _bill_usage_anthropic(db, billing_user_id, model, usage)
|
||||
|
||||
return StreamingResponse(
|
||||
_stream_with_billing(),
|
||||
_stream_passthrough_messages(),
|
||||
status_code=upstream.status_code,
|
||||
media_type="text/event-stream",
|
||||
headers=resp_headers,
|
||||
|
||||
@@ -1430,6 +1430,7 @@ class ComputeUsageRecord(Base):
|
||||
discount_source: Mapped[str] = mapped_column(String, default="standard") # standard/tenant/company/tenant_user
|
||||
actual_amount: Mapped[int] = mapped_column(BigInteger, default=0) # 实际扣费（分）
|
||||
balance_source: Mapped[str] = mapped_column(String, default="personal") # company/personal
|
||||
engine_log_id: Mapped[int] = mapped_column(Integer, default=0, index=True) # compute 引擎日志 id（幂等键）
|
||||
created_at: Mapped[str] = mapped_column(String, default="")
|
||||
|
||||
|
||||
|
||||
+3
-1
@@ -53,6 +53,7 @@ from app.incubator import routers as incubator_router
|
||||
from app.api.routers import invite as invite_router
|
||||
from app.api.routers import rbac_compute_pricing as compute_pricing_router
|
||||
from app.api.routers import rbac_compute_assets as compute_assets_router
|
||||
from app.api.routers import compute_internal as compute_internal_router
|
||||
from app.api.routers import agent_gate as agent_gate_router
|
||||
from app.im import router as im_router
|
||||
|
||||
@@ -220,10 +221,11 @@ app.include_router(rbac_hall_router.community_router)
|
||||
app.include_router(rbac_moderation_router.router)
|
||||
app.include_router(rbac_moderation_router.report_router)
|
||||
app.include_router(rbac_hall_router.dm_router)
|
||||
# compute_pricing 必须在 relay_router 之前注册：relay 有 /{path:path} catch-all，
|
||||
# compute_pricing / compute_internal 必须在 relay_router 之前注册：relay 有 /{path:path} catch-all，
|
||||
# 会吞掉 /api/compute/* 导致全部 404。
|
||||
app.include_router(compute_pricing_router.router)
|
||||
app.include_router(compute_assets_router.router)
|
||||
app.include_router(compute_internal_router.router)
|
||||
app.include_router(relay_router.router)
|
||||
app.include_router(pay_router.router)
|
||||
app.include_router(market_router.router)
|
||||
|
||||
@@ -232,6 +232,30 @@ async def adjust_user_quota(engine_user_id: int, value: int, mode: str = "add")
|
||||
)
|
||||
|
||||
|
||||
async def set_user_discount(engine_user_id: int, discount: int) -> dict:
|
||||
"""设置引擎用户折扣系数（1-500，100=原价）。
|
||||
|
||||
对应引擎 ``POST /api/user/manage``，action="set_discount"。
|
||||
用于 P2-2 折扣统一：server-core 四级折扣变更时同步到引擎 users.discount。
|
||||
"""
|
||||
return await _request(
|
||||
method="POST", path="/api/user/manage", headers=_admin_headers(),
|
||||
json={"id": int(engine_user_id), "action": "set_discount", "value": int(discount)},
|
||||
)
|
||||
|
||||
|
||||
async def set_user_discount_by_username(username: str, discount: int) -> dict:
|
||||
"""按 username 设置引擎用户折扣系数（1-500，100=原价）。
|
||||
|
||||
对应引擎 ``POST /api/user/manage``，action="set_discount"，按 username 定位用户。
|
||||
折扣系数换算：server-core 减免百分比 d → 引擎系数 = 100 - d（d=0→100原价，d=100→0免费）。
|
||||
"""
|
||||
return await _request(
|
||||
method="POST", path="/api/user/manage", headers=_admin_headers(),
|
||||
json={"username": username, "action": "set_discount", "value": int(discount)},
|
||||
)
|
||||
|
||||
|
||||
async def list_engine_users(page: int = 1, page_size: int = 100, status: int = 1) -> list[dict]:
|
||||
"""分页拉取引擎用户（GET /api/user），返回 items 列表。"""
|
||||
data = await _request(
|
||||
|
||||
@@ -395,6 +395,101 @@ async def check_user_balance_available(db: Database, user_id: str) -> bool:
|
||||
return (company_total + personal) > 0
|
||||
|
||||
|
||||
async def deduct_by_engine_cost(
|
||||
db: Database, username: str, actual_cost_micro: int,
|
||||
engine_log_id: int, model_name: str = "", token_count: int = 0,
|
||||
) -> dict:
|
||||
"""compute 引擎扣费后回调：按引擎实际费用（微元）扣平台来源账本。
|
||||
|
||||
幂等：engine_log_id 已存在记录则直接返回成功（防网络重试重复扣）。
|
||||
扣费优先级：企业分配余额（折扣从低到高）→ 个人余额。
|
||||
actual_cost_micro 单位微元（1元=1e6），换算为分（1元=100分）：amount_fen = micro // 10000。
|
||||
"""
|
||||
# 幂等检查
|
||||
from sqlalchemy import select as _sa_select
|
||||
existing = (await db.session.execute(
|
||||
_sa_select(ComputeUsageRecord).where(ComputeUsageRecord.engine_log_id == int(engine_log_id))
|
||||
)).scalar_one_or_none()
|
||||
if existing is not None:
|
||||
return {"ok": True, "amount": existing.actual_amount, "source": existing.balance_source,
|
||||
"idempotent": True, "reason": "already_recorded"}
|
||||
|
||||
amount_fen = max(0, int(actual_cost_micro) // MICRO_PER_FEN)
|
||||
if amount_fen <= 0:
|
||||
return {"ok": True, "amount": 0, "source": "free", "reason": "zero_cost"}
|
||||
|
||||
user = await db.users.get_by_username(username)
|
||||
if not user:
|
||||
return {"ok": False, "amount": 0, "source": "", "reason": "user_not_found"}
|
||||
user_id = user["id"]
|
||||
|
||||
# 构造 price_info（引擎已按折扣扣费，平台侧记录 actual_amount 即可，standard_price 用 actual 反推）
|
||||
price_info = {
|
||||
"standard_price": amount_fen,
|
||||
"tenant_discount": 0,
|
||||
"discount": 0,
|
||||
"discount_source": "engine",
|
||||
"actual_price": amount_fen,
|
||||
"tenant_id": "",
|
||||
"company_id": "",
|
||||
}
|
||||
|
||||
# 1. 先扣企业分配余额
|
||||
company_balances = await _get_user_company_balances(db, user_id)
|
||||
company_balances.sort(key=lambda x: x["discount"])
|
||||
remaining = amount_fen
|
||||
charged_from = ""
|
||||
for cb in company_balances:
|
||||
if remaining <= 0:
|
||||
break
|
||||
take = min(remaining, cb["balance"])
|
||||
if take > 0:
|
||||
ok = await _deduct_from_company_balance(db, cb["company_id"], user_id, take)
|
||||
if ok:
|
||||
remaining -= take
|
||||
charged_from = cb["company_id"]
|
||||
if remaining <= 0:
|
||||
await _record_usage_with_engine_id(db, user_id, model_name, token_count, price_info,
|
||||
amount_fen, "company", charged_from, int(engine_log_id))
|
||||
return {"ok": True, "amount": amount_fen, "source": "company", "company_id": charged_from}
|
||||
|
||||
# 2. 企业不足部分从个人余额扣
|
||||
if remaining > 0:
|
||||
ok = await _deduct_from_user_balance(db, user_id, remaining)
|
||||
if ok:
|
||||
await _record_usage_with_engine_id(db, user_id, model_name, token_count, price_info,
|
||||
amount_fen, "personal", "", int(engine_log_id))
|
||||
return {"ok": True, "amount": amount_fen, "source": "personal", "reason": ""}
|
||||
|
||||
# 3. 全部不足：记为欠费（引擎已放行），不阻断
|
||||
await _record_usage_with_engine_id(db, user_id, model_name, token_count, price_info,
|
||||
amount_fen, "personal", "", int(engine_log_id))
|
||||
return {"ok": False, "amount": amount_fen, "source": "personal", "reason": "insufficient_balance"}
|
||||
|
||||
|
||||
async def _record_usage_with_engine_id(
|
||||
db: Database, user_id: str, model: str, token_count: int,
|
||||
price_info: dict, amount: int, balance_source: str, company_id: str, engine_log_id: int,
|
||||
):
|
||||
"""记录算力使用（含引擎日志 id 幂等键）。"""
|
||||
record = ComputeUsageRecord(
|
||||
id=new_id(),
|
||||
user_id=user_id,
|
||||
company_id=company_id,
|
||||
model=model,
|
||||
token_count=token_count,
|
||||
standard_price=price_info["standard_price"],
|
||||
discount=price_info["discount"],
|
||||
discount_source=price_info["discount_source"],
|
||||
actual_amount=amount,
|
||||
balance_source=balance_source,
|
||||
engine_log_id=engine_log_id,
|
||||
created_at=now_str(),
|
||||
)
|
||||
db.session.add(record)
|
||||
await db.session.commit()
|
||||
|
||||
|
||||
async def _record_usage(
|
||||
db: Database, user_id: str, model: str, token_count: int,
|
||||
price_info: dict, amount: int, balance_source: str, company_id: str
|
||||
|
||||
Reference in New Issue
Block a user