vent_agent_2.0/api/usage_routes.py

"""用量与计费路由。"""

from __future__ import annotations

import asyncio

from fastapi import APIRouter

from api.models import QuotaUpdateRequest
from billing import guard, pricing
from db import billing_store, db_conn
from obs import get_logger

logger = get_logger(__name__)

router = APIRouter(tags=["用量"])


@router.get("/usage/today")
async def usage_today(user_name: str | None = None):
    """当日用量与额度。

    计费库不可用时不报 5xx——这个接口是给用户看「我还剩多少额度」的，
    因为一个次要故障就返回错误会让用户以为整个服务挂了。所以返回
    ``available=False`` 并附原因。
    """
    user = (user_name or "default").strip() or "default"
    snapshot = await asyncio.to_thread(guard.snapshot, user, None)
    return snapshot


@router.get("/usage/recent")
async def usage_recent(user_name: str | None = None, limit: int = 20):
    """最近若干条用量流水。"""
    user = (user_name or "default").strip() or "default"
    items = await asyncio.to_thread(billing_store.recent_usage, user, min(limit, 200))
    return {"user_name": user, "items": items, "count": len(items)}


@router.put("/usage/quota")
async def update_quota(payload: QuotaUpdateRequest):
    """调整某用户当日额度（0 = 不限）。"""
    user = (payload.user_name or "default").strip() or "default"
    day = db_conn.day_key()
    quota = await asyncio.to_thread(
        billing_store.set_quota_limit, user, day, payload.limit_tokens
    )
    logger.info("额度调整 user=%s day=%s limit=%d", user, day, payload.limit_tokens)
    return quota


@router.get("/usage/pricing")
async def get_pricing():
    """模型单价表。"""
    items = await asyncio.to_thread(pricing.pricing_table)
    return {"items": items, "count": len(items), 'enabled': pricing.enabled()}


@router.get("/health/billing")
async def billing_health():
    """计费子系统自检：单价表是否覆盖当前配置的模型。"""
    import config

    if not pricing.enabled():
        return {'ok': True, 'enabled': False, 'missing': [], 'hint': '计费已关闭，仅记录Token用量'}

    models = {config.MODEL}
    table = await asyncio.to_thread(pricing.pricing_table)
    priced = {row["model"] for row in table}
    missing = sorted(m for m in models if m and m not in priced)
    return {
        "ok": not missing,
        'enabled': True,
        "configured_models": sorted(m for m in models if m),
        "priced_models": sorted(priced),
        "missing": missing,
        "hint": "缺单价的模型会在请求开始时 fail-closed；补 billing_pricing 或设 BILLING_ALLOW_UNPRICED=1",
    }