427 lines
17 KiB
Go
427 lines
17 KiB
Go
package plugin
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"net/url"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"billing/internal/collection"
|
|
"billing/internal/web"
|
|
)
|
|
|
|
const (
|
|
managementBase = "/v0/management/plugins/" + PluginName
|
|
resourceBase = "/v0/resource/plugins/" + PluginName
|
|
routeUsage = "/usage"
|
|
routeUsageSummary = "/usage-summary"
|
|
routePrices = "/prices"
|
|
routePriceImport = "/prices/import"
|
|
routeCatalog = "/price-catalog"
|
|
routeCatalogRefresh = "/price-catalog/refresh"
|
|
routeCatalogApply = "/price-catalog/apply"
|
|
routeKeys = "/keys"
|
|
routeKeyStats = "/key-stats"
|
|
routeUpstreams = "/upstreams"
|
|
routeModels = "/model-suggestions"
|
|
routeBillingReset = "/billing-reset"
|
|
routeBillingLedger = "/billing-ledger"
|
|
routeEvents = "/events"
|
|
resourceUI = "/ui"
|
|
)
|
|
|
|
var resourceAssets = []string{
|
|
"/ui-config.js",
|
|
"/app/main.js",
|
|
"/app/core/runtime.js",
|
|
"/app/core/shared.js",
|
|
"/app/features/keys.js",
|
|
"/app/features/pricing.js",
|
|
"/app/features/usage.js",
|
|
"/app/features/logs.js",
|
|
"/styles/base.css",
|
|
"/styles/keys.css",
|
|
"/styles/layout.css",
|
|
"/styles/pricing.css",
|
|
"/styles/responsive.css",
|
|
"/styles/usage.css",
|
|
}
|
|
|
|
func managementRegistration() ManagementRegistrationResponse {
|
|
registration := ManagementRegistrationResponse{
|
|
Routes: []ManagementRoute{
|
|
{Method: http.MethodGet, Path: managementBase + routeUsage, Description: "查看最近的用量记录。"},
|
|
{Method: http.MethodGet, Path: managementBase + routeUsageSummary, Description: "查看用户与每日用量汇总。"},
|
|
{Method: http.MethodGet, Path: managementBase + routePrices, Description: "查看模型价格。"},
|
|
{Method: http.MethodPut, Path: managementBase + routePrices, Description: "保存模型价格。"},
|
|
{Method: http.MethodDelete, Path: managementBase + routePrices, Description: "删除模型价格。"},
|
|
{Method: http.MethodPost, Path: managementBase + routePriceImport, Description: "从 models.dev 导入参考价格。"},
|
|
{Method: http.MethodGet, Path: managementBase + routeCatalog, Description: "搜索 models.dev 参考价格。"},
|
|
{Method: http.MethodPost, Path: managementBase + routeCatalogRefresh, Description: "刷新 models.dev 价格目录并预览变化。"},
|
|
{Method: http.MethodPost, Path: managementBase + routeCatalogApply, Description: "确认应用参考价格变化。"},
|
|
{Method: http.MethodGet, Path: managementBase + routeKeys, Description: "查看下游 Key。"},
|
|
{Method: http.MethodPost, Path: managementBase + routeKeys, Description: "创建下游 Key。"},
|
|
{Method: http.MethodPatch, Path: managementBase + routeKeys, Description: "更新下游 Key。"},
|
|
{Method: http.MethodDelete, Path: managementBase + routeKeys, Description: "归档下游 Key。"},
|
|
{Method: http.MethodGet, Path: managementBase + routeKeyStats, Description: "查看 Key 统计。"},
|
|
{Method: http.MethodGet, Path: managementBase + routeUpstreams, Description: "同步并查看上游账号。"},
|
|
{Method: http.MethodGet, Path: managementBase + routeModels, Description: "查看模型建议。"},
|
|
{Method: http.MethodPost, Path: managementBase + routeBillingReset, Description: "立即重置 Key 额度。"},
|
|
{Method: http.MethodGet, Path: managementBase + routeBillingLedger, Description: "查看 Key 额度账目。"},
|
|
{Method: http.MethodGet, Path: managementBase + routeEvents, Description: "查看业务事件日志。"},
|
|
},
|
|
Resources: []ResourceRoute{
|
|
{Path: resourceBase + resourceUI, Menu: "用量记录", Description: "查看 CPA 最近收到的请求用量。"},
|
|
},
|
|
}
|
|
for _, path := range resourceAssets {
|
|
registration.Resources = append(registration.Resources, ResourceRoute{Path: resourceBase + path})
|
|
}
|
|
return registration
|
|
}
|
|
|
|
func (a *App) handleManagement(raw []byte) ([]byte, error) {
|
|
var req ManagementRequest
|
|
if err := json.Unmarshal(raw, &req); err != nil {
|
|
return nil, fmt.Errorf("解析管理请求: %w", err)
|
|
}
|
|
path := strings.TrimRight(req.Path, "/")
|
|
if req.Method == http.MethodGet && path == resourceBase+resourceUI {
|
|
if strings.TrimSpace(req.Query.Get("view")) != "" {
|
|
return OKEnvelope(a.readOnlyResponse(req.Query))
|
|
}
|
|
return OKEnvelope(ManagementResponse{
|
|
StatusCode: http.StatusOK,
|
|
Headers: http.Header{
|
|
"Content-Type": []string{"text/html; charset=utf-8"},
|
|
"Cache-Control": []string{"no-store"},
|
|
},
|
|
Body: web.UI(),
|
|
})
|
|
}
|
|
if req.Method == http.MethodGet {
|
|
name := strings.TrimPrefix(path, resourceBase+"/")
|
|
if body, contentType, found := web.Asset(name); found {
|
|
return OKEnvelope(staticResourceResponse(contentType, body))
|
|
}
|
|
}
|
|
if req.Method == http.MethodGet && path == managementBase+routeUsage {
|
|
return OKEnvelope(a.usageResponse(req.Query))
|
|
}
|
|
if req.Method == http.MethodGet && path == managementBase+routeUsageSummary {
|
|
return OKEnvelope(a.usageDashboardResponse())
|
|
}
|
|
if path == managementBase+routePrices {
|
|
switch req.Method {
|
|
case http.MethodGet:
|
|
return OKEnvelope(a.listPrices())
|
|
case http.MethodPut:
|
|
return OKEnvelope(a.auditFailedManagement(a.putPrice(req.Body), priceEvent(req.Body, "保存")))
|
|
case http.MethodDelete:
|
|
return OKEnvelope(a.auditFailedManagement(a.deletePrice(req.Body), priceEvent(req.Body, "删除")))
|
|
}
|
|
}
|
|
if req.Method == http.MethodPost && path == managementBase+routePriceImport {
|
|
return OKEnvelope(a.auditFailedManagement(a.importCatalogPrice(req.Body), priceEvent(req.Body, "导入")))
|
|
}
|
|
if req.Method == http.MethodGet && path == managementBase+routeCatalog {
|
|
return OKEnvelope(a.searchPriceCatalog(req.Query))
|
|
}
|
|
if req.Method == http.MethodPost && path == managementBase+routeCatalogRefresh {
|
|
return OKEnvelope(a.refreshPriceCatalog())
|
|
}
|
|
if req.Method == http.MethodPost && path == managementBase+routeCatalogApply {
|
|
return OKEnvelope(a.auditFailedManagement(a.applyCatalogChanges(req.Body), "管理员批量更新模型价格"))
|
|
}
|
|
if path == managementBase+routeKeys {
|
|
switch req.Method {
|
|
case http.MethodGet:
|
|
return OKEnvelope(a.listManagedKeys(req.Query.Get("include_archived") == "1"))
|
|
case http.MethodPost:
|
|
return OKEnvelope(a.auditFailedManagement(a.createManagedKey(req.Body), keyEvent(req.Body, "创建")))
|
|
case http.MethodPatch:
|
|
return OKEnvelope(a.auditFailedManagement(a.updateManagedKey(req.Body), keyEvent(req.Body, "修改")))
|
|
case http.MethodDelete:
|
|
return OKEnvelope(a.auditFailedManagement(a.archiveManagedKey(req.Body), keyEvent(req.Body, "归档")))
|
|
}
|
|
}
|
|
if req.Method == http.MethodGet && path == managementBase+routeKeyStats {
|
|
return OKEnvelope(a.managedKeyStats(strings.TrimSpace(req.Query.Get("id"))))
|
|
}
|
|
if req.Method == http.MethodGet && path == managementBase+routeUpstreams {
|
|
return OKEnvelope(a.upstreamAccounts())
|
|
}
|
|
if req.Method == http.MethodGet && path == managementBase+routeModels {
|
|
return OKEnvelope(a.modelSuggestions())
|
|
}
|
|
if req.Method == http.MethodPost && path == managementBase+routeBillingReset {
|
|
return OKEnvelope(a.auditFailedManagement(a.resetManagedKeyBilling(req.Body), billingResetEvent(req.Body)))
|
|
}
|
|
if req.Method == http.MethodGet && path == managementBase+routeBillingLedger {
|
|
return OKEnvelope(a.managedKeyLedger(req.Query))
|
|
}
|
|
if req.Method == http.MethodGet && path == managementBase+routeEvents {
|
|
return OKEnvelope(a.businessEvents(req.Query))
|
|
}
|
|
return OKEnvelope(jsonManagementResponse(http.StatusNotFound, map[string]any{
|
|
"error": map[string]string{
|
|
"code": "not_found",
|
|
"message": "管理路由不存在: " + req.Method + " " + req.Path,
|
|
},
|
|
}))
|
|
}
|
|
|
|
func staticResourceResponse(contentType string, body []byte) ManagementResponse {
|
|
return ManagementResponse{
|
|
StatusCode: http.StatusOK,
|
|
Headers: http.Header{
|
|
"Content-Type": []string{contentType},
|
|
"Cache-Control": []string{"no-store"},
|
|
},
|
|
Body: body,
|
|
}
|
|
}
|
|
|
|
type usageListResponse struct {
|
|
Records []usageListItem `json:"records"`
|
|
Pagination usagePaginationResult `json:"pagination"`
|
|
}
|
|
|
|
type usagePaginationResult struct {
|
|
Page int `json:"page"`
|
|
PageSize int `json:"page_size"`
|
|
Total int64 `json:"total"`
|
|
TotalPages int `json:"total_pages"`
|
|
PreviousCursor string `json:"previous_cursor,omitempty"`
|
|
NextCursor string `json:"next_cursor,omitempty"`
|
|
}
|
|
|
|
// usageListItem 是请求明细表的稳定接口。当前回调拿不到的字段保留为空值。
|
|
type usageListItem struct {
|
|
RequestID string `json:"request_id"`
|
|
ExecutionID string `json:"execution_id"`
|
|
TraceID string `json:"trace_id"`
|
|
RequestedAt time.Time `json:"requested_at"`
|
|
APIKey string `json:"api_key"`
|
|
KeyAlias string `json:"key_alias"`
|
|
AuthID string `json:"auth_id"`
|
|
AuthIndex string `json:"auth_index"`
|
|
AuthType string `json:"auth_type"`
|
|
Model string `json:"model"`
|
|
ReasoningEffort string `json:"reasoning_effort"`
|
|
ServiceTier string `json:"service_tier"`
|
|
Speed string `json:"speed"`
|
|
Failed bool `json:"failed"`
|
|
Outcome string `json:"outcome"`
|
|
StatusCode int `json:"status_code"`
|
|
Error string `json:"error"`
|
|
ExecutorType string `json:"executor_type"`
|
|
RequestType string `json:"request_type"`
|
|
Endpoint string `json:"endpoint"`
|
|
TTFTMilliseconds int64 `json:"ttft_ms"`
|
|
SpeedTPS *float64 `json:"speed_tps"`
|
|
InputTokens int64 `json:"input_tokens"`
|
|
OutputTokens int64 `json:"output_tokens"`
|
|
ReasoningTokens int64 `json:"reasoning_tokens"`
|
|
CacheReadTokens int64 `json:"cache_read_tokens"`
|
|
CacheWriteTokens int64 `json:"cache_write_tokens"`
|
|
CacheRate *float64 `json:"cache_rate"`
|
|
TotalTokens int64 `json:"total_tokens"`
|
|
CostUSD *float64 `json:"cost_usd"`
|
|
CostAvailable bool `json:"cost_available"`
|
|
PriceTier string `json:"price_tier"`
|
|
FastRequested bool `json:"fast_requested"`
|
|
FastPricingApplied bool `json:"fast_pricing_applied"`
|
|
ClientIP string `json:"client_ip"`
|
|
}
|
|
|
|
func (a *App) usageResponse(values url.Values) ManagementResponse {
|
|
query, err := parseUsageQuery(values)
|
|
if err != nil {
|
|
return managementError(http.StatusBadRequest, "invalid_query", err.Error())
|
|
}
|
|
a.mu.RLock()
|
|
defer a.mu.RUnlock()
|
|
if a.usage == nil {
|
|
return jsonManagementResponse(http.StatusServiceUnavailable, map[string]any{
|
|
"error": map[string]string{"code": "database_unavailable", "message": "用量数据库尚未初始化"},
|
|
})
|
|
}
|
|
page, err := a.usage.Query(context.Background(), query)
|
|
if err != nil {
|
|
status := http.StatusInternalServerError
|
|
code := "database_error"
|
|
if strings.Contains(err.Error(), "cursor") || strings.Contains(err.Error(), "page_size") || strings.Contains(err.Error(), "from") || strings.Contains(err.Error(), "result") {
|
|
status = http.StatusBadRequest
|
|
code = "invalid_query"
|
|
}
|
|
return jsonManagementResponse(status, map[string]any{
|
|
"error": map[string]string{"code": code, "message": err.Error()},
|
|
})
|
|
}
|
|
items := make([]usageListItem, 0, len(page.Records))
|
|
for _, record := range page.Records {
|
|
items = append(items, usageItem(record))
|
|
}
|
|
return jsonManagementResponse(http.StatusOK, usageListResponse{Records: items, Pagination: usagePaginationResult{
|
|
Page: page.Page, PageSize: page.PageSize, Total: page.Total, TotalPages: page.TotalPages,
|
|
PreviousCursor: page.PreviousCursor, NextCursor: page.NextCursor,
|
|
}})
|
|
}
|
|
|
|
func parseUsageQuery(values url.Values) (collection.UsageQuery, error) {
|
|
query := collection.UsageQuery{Page: 1, PageSize: 100, Cursor: strings.TrimSpace(values.Get("cursor")), KeyID: values.Get("key_id"), Model: values.Get("model"), Result: values.Get("result"), AuthID: values.Get("auth_id"), Endpoint: values.Get("endpoint"), RequestID: values.Get("request_id")}
|
|
if raw := strings.TrimSpace(values.Get("page")); raw != "" {
|
|
value, err := strconv.Atoi(raw)
|
|
if err != nil || value < 1 {
|
|
return query, fmt.Errorf("page 必须是正整数")
|
|
}
|
|
query.Page = value
|
|
}
|
|
if raw := strings.TrimSpace(values.Get("page_size")); raw != "" {
|
|
value, err := strconv.Atoi(raw)
|
|
if err != nil || value < 1 || value > 100 {
|
|
return query, fmt.Errorf("page_size 必须在 1 到 100 之间")
|
|
}
|
|
query.PageSize = value
|
|
}
|
|
for name, target := range map[string]**time.Time{"from": &query.From, "to": &query.To} {
|
|
raw := strings.TrimSpace(values.Get(name))
|
|
if raw == "" {
|
|
continue
|
|
}
|
|
parsed, err := time.Parse(time.RFC3339Nano, raw)
|
|
if err != nil {
|
|
return query, fmt.Errorf("%s 必须是 RFC3339 时间", name)
|
|
}
|
|
*target = &parsed
|
|
}
|
|
return query, nil
|
|
}
|
|
|
|
type usageSummaryDTO struct {
|
|
Requests int64 `json:"requests"`
|
|
InputTokens int64 `json:"input_tokens"`
|
|
OutputTokens int64 `json:"output_tokens"`
|
|
TotalTokens int64 `json:"total_tokens"`
|
|
CostUSD float64 `json:"cost_usd"`
|
|
}
|
|
|
|
func usageSummaryItem(summary collection.UsageSummary) usageSummaryDTO {
|
|
return usageSummaryDTO{Requests: summary.Requests, InputTokens: summary.InputTokens, OutputTokens: summary.OutputTokens, TotalTokens: summary.TotalTokens, CostUSD: float64(summary.CostMicros) / 1_000_000}
|
|
}
|
|
|
|
func (a *App) usageDashboardResponse() ManagementResponse {
|
|
a.mu.RLock()
|
|
defer a.mu.RUnlock()
|
|
if a.usage == nil {
|
|
return managementError(http.StatusServiceUnavailable, "database_unavailable", "用量数据库尚未初始化")
|
|
}
|
|
location, _ := time.LoadLocation("Asia/Shanghai")
|
|
dashboard, err := a.usage.Dashboard(context.Background(), time.Now().In(location), 7)
|
|
if err != nil {
|
|
return managementError(http.StatusInternalServerError, "database_error", err.Error())
|
|
}
|
|
users := make([]map[string]any, 0, len(dashboard.Users))
|
|
for _, user := range dashboard.Users {
|
|
userDays := make([]map[string]any, 0, len(user.Days))
|
|
for _, day := range user.Days {
|
|
userDays = append(userDays, map[string]any{"date": day.Date.Format("2006-01-02"), "total_tokens": day.TotalTokens, "cost_usd": float64(day.CostMicros) / 1_000_000})
|
|
}
|
|
users = append(users, map[string]any{"key_id": user.KeyID, "key_alias": user.KeyAlias, "today": usageSummaryItem(user.Today), "days": userDays, "last_used_at": user.LastUsedAt})
|
|
}
|
|
days := make([]map[string]any, 0, len(dashboard.Days))
|
|
for _, day := range dashboard.Days {
|
|
days = append(days, map[string]any{"date": day.Date.Format("2006-01-02"), "total_tokens": day.TotalTokens, "cost_usd": float64(day.CostMicros) / 1_000_000})
|
|
}
|
|
return jsonManagementResponse(http.StatusOK, map[string]any{"today": usageSummaryItem(dashboard.Today), "users": users, "days": days})
|
|
}
|
|
|
|
func usageItem(record collection.Record) usageListItem {
|
|
ttftMilliseconds := record.TTFT.Milliseconds()
|
|
speedTPS := usageSpeed(record.OutputTokens, record.TTFT, record.Latency)
|
|
if isCompactEndpoint(record.Endpoint) {
|
|
// Compact 返回的是一次性 JSON，首字延迟和生成速度没有可比较的含义。
|
|
ttftMilliseconds = 0
|
|
speedTPS = nil
|
|
}
|
|
return usageListItem{
|
|
RequestID: record.RequestID,
|
|
ExecutionID: record.ExecutionID,
|
|
TraceID: record.TraceID,
|
|
RequestedAt: record.RequestedAt,
|
|
APIKey: record.APIKey,
|
|
KeyAlias: record.KeyAlias,
|
|
AuthID: record.AuthID,
|
|
AuthIndex: record.AuthIndex,
|
|
AuthType: record.AuthType,
|
|
Model: record.Model,
|
|
ReasoningEffort: record.ReasoningEffort,
|
|
ServiceTier: record.ServiceTier,
|
|
Speed: record.Speed,
|
|
Failed: record.Failed,
|
|
Outcome: record.Outcome,
|
|
StatusCode: record.StatusCode,
|
|
Error: record.Error,
|
|
ExecutorType: record.ExecutorType,
|
|
RequestType: record.RequestType,
|
|
Endpoint: record.Endpoint,
|
|
TTFTMilliseconds: ttftMilliseconds,
|
|
SpeedTPS: speedTPS,
|
|
InputTokens: record.InputTokens,
|
|
OutputTokens: record.OutputTokens,
|
|
ReasoningTokens: record.ReasoningTokens,
|
|
CacheReadTokens: record.CacheReadTokens,
|
|
CacheWriteTokens: record.CacheWriteTokens,
|
|
CacheRate: usageCacheRate(record.CacheReadTokens, record.InputTokens),
|
|
TotalTokens: record.TotalTokens,
|
|
CostUSD: microsToUSD(record.CostMicros),
|
|
CostAvailable: record.CostMicros != nil,
|
|
PriceTier: record.PriceTier,
|
|
FastRequested: record.FastRequested,
|
|
FastPricingApplied: record.FastPricingApplied,
|
|
ClientIP: record.ClientIP,
|
|
}
|
|
}
|
|
|
|
func microsToUSD(micros *int64) *float64 {
|
|
if micros == nil {
|
|
return nil
|
|
}
|
|
value := float64(*micros) / 1_000_000
|
|
return &value
|
|
}
|
|
|
|
func usageSpeed(outputTokens int64, ttft, latency time.Duration) *float64 {
|
|
if outputTokens <= 0 || ttft <= 0 || latency <= ttft {
|
|
return nil
|
|
}
|
|
value := float64(outputTokens) / (latency - ttft).Seconds()
|
|
return &value
|
|
}
|
|
|
|
func usageCacheRate(cacheReadTokens, inputTokens int64) *float64 {
|
|
if inputTokens <= 0 {
|
|
return nil
|
|
}
|
|
value := float64(cacheReadTokens) / float64(inputTokens) * 100
|
|
return &value
|
|
}
|
|
|
|
func jsonManagementResponse(status int, value any) ManagementResponse {
|
|
body, err := json.Marshal(value)
|
|
if err != nil {
|
|
body = []byte(`{"error":{"code":"marshal_error","message":"无法生成响应"}}`)
|
|
status = http.StatusInternalServerError
|
|
}
|
|
return ManagementResponse{
|
|
StatusCode: status,
|
|
Headers: http.Header{"Content-Type": []string{"application/json; charset=utf-8"}},
|
|
Body: body,
|
|
}
|
|
}
|