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" resourceUI = "/ui" ) func managementRegistration() ManagementRegistrationResponse { return 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 额度账目。"}, }, Resources: []ResourceRoute{ {Path: resourceBase + resourceUI, Menu: "用量记录", Description: "查看 CPA 最近收到的请求用量。"}, }, } } 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 && 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.putPrice(req.Body)) case http.MethodDelete: return OKEnvelope(a.deletePrice(req.Body)) } } if req.Method == http.MethodPost && path == managementBase+routePriceImport { return OKEnvelope(a.importCatalogPrice(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.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.createManagedKey(req.Body)) case http.MethodPatch: return OKEnvelope(a.updateManagedKey(req.Body)) case http.MethodDelete: return OKEnvelope(a.archiveManagedKey(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.resetManagedKeyBilling(req.Body)) } if req.Method == http.MethodGet && path == managementBase+routeBillingLedger { return OKEnvelope(a.managedKeyLedger(req.Query)) } return OKEnvelope(jsonManagementResponse(http.StatusNotFound, map[string]any{ "error": map[string]string{ "code": "not_found", "message": "管理路由不存在: " + req.Method + " " + req.Path, }, })) } 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 { users = append(users, map[string]any{"key_id": user.KeyID, "key_alias": user.KeyAlias, "today": usageSummaryItem(user.Today), "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}) } 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, } }