From edca0dcfa7eb56a5ad91aa9c31a18b20ef77932c Mon Sep 17 00:00:00 2001 From: Pine Date: Sun, 13 Sep 2026 17:41:02 +0800 Subject: [PATCH] =?UTF-8?q?P1/P2=20=E8=AE=A1=E8=B4=B9=E4=BD=93=E7=B3=BB?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E7=BB=9F=E4=B8=80=E6=89=A3=E8=B4=B9?= =?UTF-8?q?=E5=9B=9E=E8=B0=83+=E8=AF=B7=E6=B1=82=E5=89=8D=E9=A2=84?= =?UTF-8?q?=E6=A3=80+=E5=AF=B9=E8=B4=A6=E6=8E=A5=E5=8F=A3+=E6=8A=98?= =?UTF-8?q?=E6=89=A3=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../0079_compute_usage_engine_log_id.py | 39 ++++++++ app/api/routers/compute_internal.py | 61 ++++++++++++ app/api/routers/rbac_compute_pricing.py | 99 +++++++++++++++++++ app/api/routers/relay.py | 81 +++++++-------- app/infrastructure/models.py | 1 + app/main.py | 4 +- app/services/compute_client.py | 24 +++++ app/services/compute_pricing_service.py | 95 ++++++++++++++++++ 8 files changed, 360 insertions(+), 44 deletions(-) create mode 100644 alembic/versions/0079_compute_usage_engine_log_id.py create mode 100644 app/api/routers/compute_internal.py diff --git a/alembic/versions/0079_compute_usage_engine_log_id.py b/alembic/versions/0079_compute_usage_engine_log_id.py new file mode 100644 index 0000000..87de5c8 --- /dev/null +++ b/alembic/versions/0079_compute_usage_engine_log_id.py @@ -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") diff --git a/app/api/routers/compute_internal.py b/app/api/routers/compute_internal.py new file mode 100644 index 0000000..e801699 --- /dev/null +++ b/app/api/routers/compute_internal.py @@ -0,0 +1,61 @@ +# -*- coding: utf-8 -*- +"""compute 引擎 → server-core 内部回调接口。 + +compute 扣费完成后异步回调此接口，按引擎实际费用（微元）扣平台来源账本。 +鉴权：Authorization: Bearer （与引擎侧 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 diff --git a/app/api/routers/rbac_compute_pricing.py b/app/api/routers/rbac_compute_pricing.py index 485ce15..4e45704 100644 --- a/app/api/routers/rbac_compute_pricing.py +++ b/app/api/routers/rbac_compute_pricing.py @@ -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, + } diff --git a/app/api/routers/relay.py b/app/api/routers/relay.py index 506cb81..05a9368 100644 --- a/app/api/routers/relay.py +++ b/app/api/routers/relay.py @@ -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, diff --git a/app/infrastructure/models.py b/app/infrastructure/models.py index 3d31504..43d6daa 100644 --- a/app/infrastructure/models.py +++ b/app/infrastructure/models.py @@ -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="") diff --git a/app/main.py b/app/main.py index 80e9de4..42d13cd 100644 --- a/app/main.py +++ b/app/main.py @@ -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) diff --git a/app/services/compute_client.py b/app/services/compute_client.py index f06887d..a18788a 100644 --- a/app/services/compute_client.py +++ b/app/services/compute_client.py @@ -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( diff --git a/app/services/compute_pricing_service.py b/app/services/compute_pricing_service.py index 6cf1e8b..ac10c09 100644 --- a/app/services/compute_pricing_service.py +++ b/app/services/compute_pricing_service.py @@ -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