diff --git a/backend/internal/handler/auth.go b/backend/internal/handler/auth.go index 5e4378a99..aa3806674 100644 --- a/backend/internal/handler/auth.go +++ b/backend/internal/handler/auth.go @@ -608,7 +608,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, // 同步代理与后台任务必须共享渠道槽位，否则两条入口会共同超过供应商并发上限。 releaseChannel, concurrencyLimit, err := svc.AcquireChannelSlot(c.Request.Context(), channel.ID, "", 36*time.Minute) if err != nil { - log := apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, model.ApiCallStatusFailed, 0, time.Since(startedAt), err.Error(), concurrencyLimit) + log := apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, c.GetHeader("Content-Type"), model.ApiCallStatusFailed, 0, time.Since(startedAt), err.Error(), concurrencyLimit) log.ErrorCode, log.Error = service.ChannelSlotFailureDetails(err) logSystemProxyCall(svc, log, nil) fail(c, http.StatusServiceUnavailable, err) @@ -661,7 +661,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, status = model.ApiCallStatusFailed errorText = err.Error() _ = svc.MarkBillingUncertain(billingOrderID, "系统渠道连接中断，费用状态待核对") - logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil) + logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, c.GetHeader("Content-Type"), status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil) fail(c, http.StatusBadGateway, errors.New("系统渠道连接失败")) return } @@ -676,7 +676,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, status = model.ApiCallStatusFailed errorText = readErr.Error() _ = svc.MarkBillingUncertain(billingOrderID, "系统渠道响应读取失败，费用状态待核对") - logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil) + logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, c.GetHeader("Content-Type"), status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil) fail(c, http.StatusBadGateway, errors.New("系统渠道响应读取失败")) return } @@ -694,7 +694,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, } else { _ = svc.RefundBilling(billingOrderID, "上游明确返回失败") } - logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), responseBody) + logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, c.Request.Method, path, target, body, c.GetHeader("Content-Type"), status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), responseBody) for _, key := range []string{"Content-Type", "Cache-Control", "Content-Disposition"} { if value := resp.Header.Get(key); value != "" { c.Header(key, value) @@ -704,7 +704,7 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User, c.Data(resp.StatusCode, resp.Header.Get("Content-Type"), responseBody) } -func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID string, method string, path string, target string, body []byte, status model.ApiCallStatus, statusCode int, duration time.Duration, errorText string, concurrencyLimit int) model.ApiCallLog { +func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID string, method string, path string, target string, body []byte, contentType string, status model.ApiCallStatus, statusCode int, duration time.Duration, errorText string, concurrencyLimit int) model.ApiCallLog { capability := "text" switch channel.InterfaceType { case model.ChannelInterfaceOpenAIImage: @@ -737,10 +737,12 @@ func apiCallLog(user *model.User, channel *model.ModelChannel, billingOrderID st Error: errorText, ConcurrencyLimit: concurrencyLimit, UpstreamURL: target, + RequestBody: service.SanitizeAPICallPayload(body, contentType), } } func logSystemProxyCall(svc *service.Service, log model.ApiCallLog, responseBody []byte) { + log.ResponseBody = service.SanitizeAPICallPayload(responseBody, "") svc.EnrichAPICallLog(&log, responseBody) _ = svc.LogAPICall(log) } diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index 5d2fde0ae..2249ab809 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -228,6 +228,8 @@ type ChannelModel struct { type ApiCallLog struct { ID string `json:"id" gorm:"primaryKey;size:36"` UserID string `json:"userId" gorm:"index;size:36;index:idx_api_logs_user_created,priority:1"` + UserDisplayName string `json:"userDisplayName,omitempty" gorm:"-"` + UserAccount string `json:"userAccount,omitempty" gorm:"-"` ChannelID string `json:"channelId" gorm:"index;size:36;index:idx_api_logs_channel_created,priority:1"` ChannelName string `json:"channelName" gorm:"-"` TaskID string `json:"taskId,omitempty" gorm:"index;size:36"` @@ -244,6 +246,8 @@ type ApiCallLog struct { Status ApiCallStatus `json:"status" gorm:"index;size:24;index:idx_api_logs_status_created,priority:1"` StatusCode int `json:"statusCode"` DurationMs int64 `json:"durationMs"` + PollCount int `json:"pollCount"` + ProviderStatus string `json:"providerStatus,omitempty" gorm:"size:32"` InputTokens int64 `json:"inputTokens"` OutputTokens int64 `json:"outputTokens"` CachedTokens int64 `json:"cachedTokens"` @@ -258,6 +262,9 @@ type ApiCallLog struct { Error string `json:"error"` ConcurrencyLimit int `json:"concurrencyLimit"` UpstreamURL string `json:"upstreamUrl"` + RequestBody string `json:"requestBody,omitempty" gorm:"type:text"` + ResponseBody string `json:"responseBody,omitempty" gorm:"type:text"` + StartedAt time.Time `json:"startedAt"` CreatedAt time.Time `json:"createdAt" gorm:"index;index:idx_api_logs_user_created,priority:2;index:idx_api_logs_channel_created,priority:2;index:idx_api_logs_model_created,priority:2;index:idx_api_logs_status_created,priority:2"` } diff --git a/backend/internal/repository/analytics.go b/backend/internal/repository/analytics.go index 34a1db38a..1f723275b 100644 --- a/backend/internal/repository/analytics.go +++ b/backend/internal/repository/analytics.go @@ -94,7 +94,7 @@ func (r *Repository) AnalyticsTasks(filter AnalyticsFilter) ([]model.Task, error func (r *Repository) AnalyticsAPICallLogs(filter AnalyticsFilter) ([]model.ApiCallLog, error) { var logs []model.ApiCallLog query := r.apiCallLogQuery(filter) - return logs, query.Find(&logs).Error + return logs, query.Omit("RequestBody", "ResponseBody").Find(&logs).Error } func (r *Repository) AnalyticsActivities(filter AnalyticsFilter) ([]model.UserDailyActivity, error) { @@ -113,7 +113,7 @@ func (r *Repository) QueryAPICallLogs(filter APICallLogFilter) ([]model.ApiCallL if filter.Limit <= 0 || filter.Limit > 200 { filter.Limit = 50 } - query := r.apiCallLogQuery(filter.AnalyticsFilter) + query := visibleAPICallLogQuery(r.apiCallLogQuery(filter.AnalyticsFilter)) if value := strings.TrimSpace(filter.Keyword); value != "" { pattern := "%" + strings.ToLower(value) + "%" query = query. @@ -132,7 +132,7 @@ func (r *Repository) QueryAPICallLogs(filter APICallLogFilter) ([]model.ApiCallL return nil, 0, err } var logs []model.ApiCallLog - err := query.Select("api_call_logs.*").Order("api_call_logs.created_at desc").Offset((filter.Page - 1) * filter.Limit).Limit(filter.Limit).Find(&logs).Error + err := query.Omit("RequestBody", "ResponseBody").Order("api_call_logs.created_at desc").Offset((filter.Page - 1) * filter.Limit).Limit(filter.Limit).Find(&logs).Error return logs, total, err } @@ -140,7 +140,7 @@ func (r *Repository) ExportAPICallLogs(filter APICallLogFilter, limit int) ([]mo if limit <= 0 || limit > 10_000 { limit = 10_000 } - query := r.apiCallLogQuery(filter.AnalyticsFilter) + query := visibleAPICallLogQuery(r.apiCallLogQuery(filter.AnalyticsFilter)) if len(filter.IDs) > 0 { query = query.Where("api_call_logs.id IN ?", filter.IDs) } @@ -155,10 +155,29 @@ func (r *Repository) ExportAPICallLogs(filter APICallLogFilter, limit int) ([]mo query = query.Where("api_call_logs.status = ?", filter.Status) } var logs []model.ApiCallLog - err := query.Select("api_call_logs.*").Order("api_call_logs.created_at desc").Limit(limit).Find(&logs).Error + err := query.Omit("RequestBody", "ResponseBody").Order("api_call_logs.created_at desc").Limit(limit).Find(&logs).Error return logs, err } +func visibleAPICallLogQuery(query *gorm.DB) *gorm.DB { + // 视频轮询属于一次生成调用的内部阶段，管理端只展示聚合后的创建主记录。 + return query.Where("NOT (api_call_logs.capability = ? AND api_call_logs.request_kind IN ?)", "video", []string{"poll", "download"}) +} + +func (r *Repository) VideoAPICallRoot(log model.ApiCallLog) (*model.ApiCallLog, error) { + var root model.ApiCallLog + query := r.db.Where("capability = ? AND request_kind = ?", "video", "create") + if log.TaskID != "" { + query = query.Where("task_id = ?", log.TaskID) + } else { + query = query.Where("user_id = ? AND channel_id = ? AND provider_request_id = ?", log.UserID, log.ChannelID, log.ProviderRequestID) + } + if err := query.Order("created_at desc").First(&root).Error; err != nil { + return nil, err + } + return &root, nil +} + func (r *Repository) apiCallLogQuery(filter AnalyticsFilter) *gorm.DB { query := r.db.Where("api_call_logs.created_at >= ? AND api_call_logs.created_at < ?", filter.From, filter.To) if filter.UserID != "" { diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index 58af41bb5..9366af946 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -53,7 +53,7 @@ func (r *Repository) UserStorageUsage(userID string) (UserStorageUsage, error) { (SELECT COALESCE(SUM(length(CAST(COALESCE(prompt, '') AS BLOB)) + length(CAST(COALESCE(input_json, '') AS BLOB)) + length(CAST(COALESCE(result_json, '') AS BLOB)) + length(CAST(COALESCE(error, '') AS BLOB))), 0) FROM tasks WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(message, '') AS BLOB)) + length(CAST(COALESCE(payload, '') AS BLOB))), 0) FROM task_logs WHERE user_id = ?) + (SELECT COALESCE(SUM(length(CAST(COALESCE(url, '') AS BLOB)) + length(CAST(COALESCE(payload, '') AS BLOB))), 0) FROM results WHERE user_id = ?) - + (SELECT COALESCE(SUM(length(CAST(COALESCE(path, '') AS BLOB)) + length(CAST(COALESCE(model, '') AS BLOB)) + length(CAST(COALESCE(provider_request_id, '') AS BLOB)) + length(CAST(COALESCE(error_code, '') AS BLOB)) + length(CAST(COALESCE(error, '') AS BLOB)) + length(CAST(COALESCE(upstream_url, '') AS BLOB))), 0) FROM api_call_logs WHERE user_id = ?) AS task_bytes, + + (SELECT COALESCE(SUM(length(CAST(COALESCE(path, '') AS BLOB)) + length(CAST(COALESCE(model, '') AS BLOB)) + length(CAST(COALESCE(provider_request_id, '') AS BLOB)) + length(CAST(COALESCE(error_code, '') AS BLOB)) + length(CAST(COALESCE(error, '') AS BLOB)) + length(CAST(COALESCE(upstream_url, '') AS BLOB)) + length(CAST(COALESCE(request_body, '') AS BLOB)) + length(CAST(COALESCE(response_body, '') AS BLOB))), 0) FROM api_call_logs WHERE user_id = ?) AS task_bytes, (SELECT COUNT(*) FROM api_call_logs WHERE user_id = ?) AS api_call_count ` if r.Dialect() == "postgres" { @@ -545,7 +545,7 @@ func (r *Repository) ApiCallLogs(userID string, admin bool, limit int) ([]model. if !admin { query = query.Where("user_id = ?", userID) } - err := query.Find(&logs).Error + err := query.Omit("RequestBody", "ResponseBody").Find(&logs).Error return logs, err } diff --git a/backend/internal/service/admin.go b/backend/internal/service/admin.go index ba3c6e334..49fc79ec2 100644 --- a/backend/internal/service/admin.go +++ b/backend/internal/service/admin.go @@ -472,6 +472,9 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error { if log.CreatedAt.IsZero() { log.CreatedAt = time.Now() } + if log.StartedAt.IsZero() { + log.StartedAt = log.CreatedAt.Add(-time.Duration(log.DurationMs) * time.Millisecond) + } s.estimateCallCost(&log) if log.BillingOrderID != "" && log.ProviderRequestID != "" { if err := s.repo.UpdateBillingProviderRequestID(log.BillingOrderID, log.ProviderRequestID); err != nil { @@ -493,6 +496,11 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error { return err } } + if merged, err := s.mergeVideoAPICallLog(log); err != nil { + return err + } else if merged { + return nil + } policy, err := s.RuntimePolicy() if err != nil { return err @@ -503,13 +511,65 @@ func (s *Service) LogAPICall(log model.ApiCallLog) error { if err != nil { return err } - incomingBytes := int64(len(log.Path) + len(log.Model) + len(log.ProviderRequestID) + len(log.ErrorCode) + len(log.Error) + len(log.UpstreamURL)) + incomingBytes := int64(len(log.Path) + len(log.Model) + len(log.ProviderRequestID) + len(log.ErrorCode) + len(log.Error) + len(log.UpstreamURL) + len(log.RequestBody) + len(log.ResponseBody)) if err := validateAPICallLogQuotaWithPolicy(usage, incomingBytes, policy.Resource); err != nil { return err } return s.repo.Create(&log) } +func (s *Service) mergeVideoAPICallLog(log model.ApiCallLog) (bool, error) { + if log.Capability != "video" || (log.RequestKind != "poll" && log.RequestKind != "download") { + return false, nil + } + if log.TaskID == "" && log.ProviderRequestID == "" { + return false, nil + } + root, err := s.repo.VideoAPICallRoot(log) + if errors.Is(err, gorm.ErrRecordNotFound) { + return false, nil + } + if err != nil { + return false, err + } + if log.RequestKind == "poll" { + root.PollCount++ + if log.ResponseBody != "" { + root.ResponseBody = log.ResponseBody + } + } + if log.ProviderRequestID != "" { + root.ProviderRequestID = log.ProviderRequestID + } + if log.ProviderStatus != "" { + root.ProviderStatus = log.ProviderStatus + } + startedAt := root.StartedAt + if startedAt.IsZero() { + startedAt = root.CreatedAt.Add(-time.Duration(root.DurationMs) * time.Millisecond) + root.StartedAt = startedAt + } + root.DurationMs = max(root.DurationMs, log.CreatedAt.Sub(startedAt).Milliseconds()) + root.StatusCode = log.StatusCode + root.ConcurrencyLimit = log.ConcurrencyLimit + if log.Status == model.ApiCallStatusFailed { + root.Status = log.Status + root.ErrorCode = log.ErrorCode + root.Error = log.Error + } else { + root.Status = model.ApiCallStatusSucceeded + root.ErrorCode = "" + root.Error = "" + } + if log.UsageAvailable { + root.UsageAvailable = true + root.InputTokens = log.InputTokens + root.OutputTokens = log.OutputTokens + root.CachedTokens = log.CachedTokens + } + return true, s.repo.Save(root) +} + func (s *Service) APICallLogs(actor *model.User, limit int) ([]model.ApiCallLog, error) { if actor == nil { return nil, Unauthorized("请先登录") diff --git a/backend/internal/service/analytics.go b/backend/internal/service/analytics.go index 18c089130..6940ff982 100644 --- a/backend/internal/service/analytics.go +++ b/backend/internal/service/analytics.go @@ -192,23 +192,13 @@ func (s *Service) AdminAPICallLogs(actor *model.User, query APICallLogQuery) (*A if err != nil { return nil, err } - // 历史日志允许读取逻辑删除渠道的名称，但不读取或返回渠道密钥。 - channels, err := s.repo.HistoricalSystemChannelReferences() - if err != nil { + if err := s.decorateAPICallLogs(logs); err != nil { return nil, err } - channelNames := make(map[string]string, len(channels)) - for _, channel := range channels { - channelNames[channel.ID] = channel.Name - } for index := range logs { - if logs[index].ChannelID == "" { - logs[index].ChannelName = "自定义渠道" - } else if name := channelNames[logs[index].ChannelID]; name != "" { - logs[index].ChannelName = name - } else { - logs[index].ChannelName = "已删除渠道" - } + // 原始报文只允许通过详情接口按单条读取。 + logs[index].RequestBody = "" + logs[index].ResponseBody = "" } page, limit := query.Page, query.Limit if page <= 0 { @@ -220,11 +210,56 @@ func (s *Service) AdminAPICallLogs(actor *model.User, query APICallLogQuery) (*A return &APICallLogPage{Logs: logs, Total: total, Page: page, Limit: limit}, nil } +func (s *Service) decorateAPICallLogs(logs []model.ApiCallLog) error { + // 历史日志允许读取逻辑删除渠道的名称，但不读取或返回渠道密钥。 + channels, err := s.repo.HistoricalSystemChannelReferences() + if err != nil { + return nil, err + } + channelNames := make(map[string]string, len(channels)) + for _, channel := range channels { + channelNames[channel.ID] = channel.Name + } + users, err := s.repo.Users() + if err != nil { + return err + } + userByID := make(map[string]model.User, len(users)) + for _, user := range users { + userByID[user.ID] = user + } + for index := range logs { + if logs[index].StartedAt.IsZero() { + logs[index].StartedAt = logs[index].CreatedAt + } + if logs[index].ChannelID == "" { + logs[index].ChannelName = "自定义渠道" + } else if name := channelNames[logs[index].ChannelID]; name != "" { + logs[index].ChannelName = name + } else { + logs[index].ChannelName = "已删除渠道" + } + if user, exists := userByID[logs[index].UserID]; exists { + logs[index].UserDisplayName = user.DisplayName + logs[index].UserAccount = user.Username + } + } + return nil +} + func (s *Service) AdminAPICallLog(actor *model.User, id string) (*model.ApiCallLog, error) { if err := s.RequireAdmin(actor); err != nil { return nil, err } - return s.repo.APICallLog(strings.TrimSpace(id)) + log, err := s.repo.APICallLog(strings.TrimSpace(id)) + if err != nil { + return nil, err + } + logs := []model.ApiCallLog{*log} + if err := s.decorateAPICallLogs(logs); err != nil { + return nil, err + } + return &logs[0], nil } func (s *Service) AdminAPICallLogsCSV(actor *model.User, query APICallLogQuery) ([]byte, error) { @@ -243,12 +278,19 @@ func (s *Service) AdminAPICallLogsCSV(actor *model.User, query APICallLogQuery) if err != nil { return nil, err } + if err := s.decorateAPICallLogs(logs); err != nil { + return nil, err + } var buffer bytes.Buffer buffer.WriteString("\xEF\xBB\xBF") writer := csv.NewWriter(&buffer) - _ = writer.Write([]string{"时间", "用户ID", "渠道ID", "任务ID", "计费单ID", "能力", "请求阶段", "模型", "状态", "HTTP状态", "耗时毫秒", "渠道并发上限", "供应商任务ID", "错误码", "错误"}) + _ = writer.Write([]string{"时间", "用户", "用户账号", "渠道", "模型", "能力", "状态", "轮询次数", "耗时毫秒", "输入Token", "输出Token", "缓存Token", "计费", "币种", "错误码", "错误"}) for _, log := range logs { - _ = writer.Write([]string{log.CreatedAt.UTC().Format(time.RFC3339), log.UserID, log.ChannelID, log.TaskID, log.BillingOrderID, log.Capability, log.RequestKind, log.Model, string(log.Status), strconv.Itoa(log.StatusCode), strconv.FormatInt(log.DurationMs, 10), strconv.Itoa(log.ConcurrencyLimit), log.ProviderRequestID, log.ErrorCode, log.Error}) + startedAt := log.StartedAt + if startedAt.IsZero() { + startedAt = log.CreatedAt + } + _ = writer.Write([]string{startedAt.UTC().Format(time.RFC3339), log.UserDisplayName, log.UserAccount, log.ChannelName, log.Model, log.Capability, string(log.Status), strconv.Itoa(log.PollCount), strconv.FormatInt(log.DurationMs, 10), strconv.FormatInt(log.InputTokens, 10), strconv.FormatInt(log.OutputTokens, 10), strconv.FormatInt(log.CachedTokens, 10), strconv.FormatInt(log.EstimatedCostMicros, 10), log.Currency, log.ErrorCode, log.Error}) } writer.Flush() if err := writer.Error(); err != nil { @@ -804,7 +846,13 @@ func (s *Service) estimateCallCost(log *model.ApiCallLog) { } func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { - if log == nil || len(responseBody) == 0 || !json.Valid(responseBody) { + if log == nil { + return + } + if log.ProviderRequestID == "" { + log.ProviderRequestID = providerRequestIDFromPath(log.Path) + } + if len(responseBody) == 0 || !json.Valid(responseBody) { return } var payload map[string]any @@ -843,7 +891,14 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { log.OutputTokens = firstInt64(usageMetadata, "candidatesTokenCount") log.CachedTokens = firstInt64(usageMetadata, "cachedContentTokenCount") } - log.ProviderRequestID = firstNonEmpty(stringField(payload, "task_id"), stringField(payload, "id"), stringField(payload, "request_id")) + log.ProviderRequestID = firstNonEmpty(stringField(payload, "task_id"), stringField(payload, "id"), stringField(payload, "request_id"), log.ProviderRequestID) + log.ProviderStatus = strings.ToLower(firstNonEmpty(stringField(payload, "status"), log.ProviderStatus)) + if log.ProviderStatus == "failed" || log.ProviderStatus == "cancelled" || log.ProviderStatus == "expired" { + log.Status = model.ApiCallStatusFailed + errorCode, errorMessage := providerFailureDetails(payload) + log.ErrorCode = firstNonEmpty(errorCode, log.ErrorCode) + log.Error = firstNonEmpty(errorMessage, log.Error) + } if log.Capability == "image" { if data, ok := payload["data"].([]any); ok { log.MediaCount = len(data) @@ -853,6 +908,21 @@ func (s *Service) EnrichAPICallLog(log *model.ApiCallLog, responseBody []byte) { } } +func providerRequestIDFromPath(path string) string { + parts := strings.Split(strings.Trim(strings.TrimSpace(path), "/"), "/") + for index := len(parts) - 1; index >= 0; index-- { + part := strings.TrimSpace(parts[index]) + if part == "" || part == "content" || part == "download" { + continue + } + if index > 0 && (parts[index-1] == "videos" || parts[index-1] == "tasks") { + return part + } + break + } + return "" +} + func firstInt64(values map[string]any, keys ...string) int64 { for _, key := range keys { switch value := values[key].(type) { diff --git a/backend/internal/service/api_call_payload.go b/backend/internal/service/api_call_payload.go new file mode 100644 index 000000000..5a2d1fba0 --- /dev/null +++ b/backend/internal/service/api_call_payload.go @@ -0,0 +1,154 @@ +package service + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "mime" + "mime/multipart" + "net/http" + "net/url" + "strings" +) + +const maxAPICallPayloadBytes = 128 << 10 +const maxAPICallPayloadSourceBytes = 1 << 20 + +// SanitizeAPICallPayload 保留排障所需报文，同时阻止密钥和大段内嵌媒体进入日志库。 +func SanitizeAPICallPayload(data []byte, contentType string) string { + if len(data) == 0 { + return "" + } + mediaType, params, _ := mime.ParseMediaType(contentType) + if mediaType == "" { + mediaType, _, _ = mime.ParseMediaType(http.DetectContentType(data)) + } + if mediaType == "multipart/form-data" && params["boundary"] != "" { + return sanitizeMultipartPayload(bytes.NewReader(data), params["boundary"]) + } + if len(data) > maxAPICallPayloadSourceBytes { + return fmt.Sprintf("[报文过大，已省略，共 %d 字节]", len(data)) + } + if json.Valid(data) { + var payload any + if json.Unmarshal(data, &payload) == nil { + formatted, err := json.MarshalIndent(sanitizeAPICallJSON(payload, ""), "", " ") + if err == nil { + return truncateAPICallPayload(string(formatted)) + } + } + } + if strings.HasPrefix(mediaType, "image/") || strings.HasPrefix(mediaType, "video/") || strings.HasPrefix(mediaType, "audio/") || mediaType == "application/octet-stream" { + return fmt.Sprintf("[%s 二进制报文，共 %d 字节]", defaultString(mediaType, "未知类型"), len(data)) + } + return truncateAPICallPayload(string(data)) +} + +func requestPayloadForLog(req *http.Request) string { + if req == nil || req.GetBody == nil { + return "" + } + body, err := req.GetBody() + if err != nil { + return "" + } + defer body.Close() + mediaType, params, _ := mime.ParseMediaType(req.Header.Get("Content-Type")) + if mediaType == "multipart/form-data" && params["boundary"] != "" { + return sanitizeMultipartPayload(body, params["boundary"]) + } + data, err := io.ReadAll(io.LimitReader(body, maxAPICallPayloadSourceBytes+1)) + if err != nil { + return "" + } + if len(data) > maxAPICallPayloadSourceBytes { + return fmt.Sprintf("[请求报文过大，已省略，超过 %d 字节]", maxAPICallPayloadSourceBytes) + } + return SanitizeAPICallPayload(data, req.Header.Get("Content-Type")) +} + +func sanitizeMultipartPayload(source io.Reader, boundary string) string { + reader := multipart.NewReader(source, boundary) + fields := make(map[string]any) + for partIndex := 0; partIndex < 100; partIndex++ { + part, err := reader.NextPart() + if err == io.EOF { + break + } + if err != nil { + return "[multipart 请求报文无法解析]" + } + name := part.FormName() + if part.FileName() != "" { + size, _ := io.Copy(io.Discard, part) + fields[name] = map[string]any{"fileName": part.FileName(), "contentType": part.Header.Get("Content-Type"), "size": size} + } else { + content, _ := io.ReadAll(io.LimitReader(part, maxAPICallPayloadBytes+1)) + fields[name] = sanitizeAPICallJSON(string(content), name) + } + part.Close() + } + formatted, _ := json.MarshalIndent(fields, "", " ") + return truncateAPICallPayload(string(formatted)) +} + +func sanitizeAPICallJSON(value any, key string) any { + normalizedKey := strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(key, "_", ""), "-", "")) + for _, secretKey := range []string{"apikey", "accesstoken", "authorization", "password", "secret"} { + if strings.Contains(normalizedKey, secretKey) { + return "[REDACTED]" + } + } + switch typed := value.(type) { + case map[string]any: + result := make(map[string]any, len(typed)) + for childKey, childValue := range typed { + result[childKey] = sanitizeAPICallJSON(childValue, childKey) + } + return result + case []any: + result := make([]any, len(typed)) + for index, childValue := range typed { + result[index] = sanitizeAPICallJSON(childValue, key) + } + return result + case string: + if strings.HasPrefix(typed, "data:") { + mediaType := strings.TrimPrefix(strings.SplitN(typed, ";", 2)[0], "data:") + return fmt.Sprintf("[内嵌媒体 %s，共 %d 字符]", defaultString(mediaType, "未知类型"), len(typed)) + } + if strings.Contains(normalizedKey, "base64") || strings.Contains(normalizedKey, "b64") { + return fmt.Sprintf("[内嵌编码数据，共 %d 字符]", len(typed)) + } + if sanitizedURL, ok := sanitizeAPICallURL(typed); ok { + return sanitizedURL + } + return typed + default: + return value + } +} + +func sanitizeAPICallURL(value string) (string, bool) { + parsed, err := url.Parse(value) + if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" { + return "", false + } + query := parsed.Query() + for key := range query { + normalized := strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(key, "_", ""), "-", "")) + if strings.Contains(normalized, "token") || strings.Contains(normalized, "signature") || strings.Contains(normalized, "apikey") || normalized == "key" { + query.Set(key, "[REDACTED]") + } + } + parsed.RawQuery = query.Encode() + return parsed.String(), true +} + +func truncateAPICallPayload(value string) string { + if len(value) <= maxAPICallPayloadBytes { + return value + } + return value[:maxAPICallPayloadBytes] + fmt.Sprintf("\n[报文已截断，原始长度 %d 字节]", len(value)) +} diff --git a/backend/internal/service/provider.go b/backend/internal/service/provider.go index 3191eb950..45ecbecab 100644 --- a/backend/internal/service/provider.go +++ b/backend/internal/service/provider.go @@ -906,7 +906,7 @@ func runNewAPIChannel1VideoTask(ctx context.Context, input canvasGenerationInput if videoURL == "" { return nil, fmt.Errorf("NewAPI 渠道 1 任务 %s 已完成但没有返回视频 URL", id) } - data, mimeType, err := getExternalBinary(ctx, videoURL) + data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) if err != nil { return nil, fmt.Errorf("NewAPI 渠道 1 视频结果下载失败（任务 %s）：%w", id, err) } @@ -1108,7 +1108,7 @@ func runSeedanceVideosTask(ctx context.Context, input canvasGenerationInput) (ma if status == "completed" || status == "succeeded" { videoURL := stringField(state, "video_url") if videoURL != "" { - data, mimeType, err := getExternalBinary(ctx, videoURL) + data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) if err != nil { return nil, fmt.Errorf("视频结果下载失败：%w", err) } @@ -1173,7 +1173,7 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn if videoURL == "" { return nil, errors.New("Seedance 任务成功但没有返回视频 URL") } - data, mimeType, err := getExternalBinary(ctx, videoURL) + data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) if err != nil { return nil, fmt.Errorf("视频结果下载失败：%w", err) } @@ -1396,6 +1396,7 @@ func recordProviderRequest(req *http.Request, startedAt time.Time, statusCode in APIFormat: "openai", Method: req.Method, Path: req.URL.Path, Model: metadata.Model, Status: status, StatusCode: statusCode, DurationMs: time.Since(startedAt).Milliseconds(), Error: errorText, ConcurrencyLimit: metadata.ConcurrencyLimit, UpstreamURL: req.URL.Scheme + "://" + req.URL.Host + req.URL.Path, + ProviderRequestID: metadata.ProviderRequestID, RequestBody: requestPayloadForLog(req), ResponseBody: SanitizeAPICallPayload(responseBody, ""), } channelSlotFailure := false if code, message := ChannelSlotFailureDetails(requestErr); code != "" { diff --git a/web/src/pages/admin/components/api-log-detail-drawer.tsx b/web/src/pages/admin/components/api-log-detail-drawer.tsx index 04cdd3242..d7a2b11e7 100644 --- a/web/src/pages/admin/components/api-log-detail-drawer.tsx +++ b/web/src/pages/admin/components/api-log-detail-drawer.tsx @@ -1,5 +1,5 @@ import { useEffect, useState } from "react"; -import { App, Descriptions, Drawer, Empty, Skeleton, Tag } from "antd"; +import { App, Descriptions, Drawer, Empty, Skeleton, Tabs, Tag, Typography } from "antd"; import { getAdminApiLog, type ApiCallLog } from "@/services/api/auth"; @@ -16,33 +16,54 @@ export function ApiLogDetailDrawer({ logId, onClose }: { logId: string | null; o .then((result) => active && setLog(result.log)) .catch((error) => active && message.error(error instanceof Error ? error.message : "读取请求详情失败")) .finally(() => active && setLoading(false)); - return () => { - active = false; - }; + return () => { active = false; }; }, [logId, message]); - const items = log - ? [ - ["时间", new Date(log.createdAt).toLocaleString("zh-CN", { hour12: false })], - ["状态", {log.status === "succeeded" ? "成功" : "失败"}], - ["用户 ID", log.userId], - ["任务 ID", log.taskId || "--"], - ["渠道", log.channelName || log.channelId || "--"], - ["模型", log.model || "--"], - ["请求阶段", log.requestKind || "--"], - ["供应商任务 ID", log.providerRequestId || "--"], - ["方法与路径", `${log.method} ${log.path}`], - ["HTTP 状态", String(log.statusCode || "--")], - ["耗时", `${log.durationMs} ms`], - ["渠道并发上限", log.concurrencyLimit ? String(log.concurrencyLimit) : "--"], - ["Token", log.usageAvailable ? `${log.inputTokens} 输入 / ${log.outputTokens} 输出 / ${log.cachedTokens} 缓存` : "未返回"], - ["错误码", log.errorCode || "--"], - ["错误详情", log.error || "--"], - ["上游地址", log.upstreamUrl || "--"], - ].map(([label, children], index) => ({ key: String(index), label, children })) - : []; + return ( - - {loading ? : log ? : } + + {loading ? : log ? : } ); } + +function LogDetail({ log }: { log: ApiCallLog }) { + const providerStatus = log.providerStatus?.toLowerCase(); + const processing = ["queued", "pending", "processing", "running", "in_progress"].includes(providerStatus || ""); + const failed = log.status === "failed" || ["failed", "cancelled", "expired"].includes(providerStatus || ""); + const items = [ + ["时间", new Date(log.startedAt || log.createdAt).toLocaleString("zh-CN", { hour12: false })], + ["状态", {failed ? "失败" : processing ? "处理中" : "成功"}], + ["用户", {log.userDisplayName || log.userAccount || "未知用户"}{log.userAccount ? @{log.userAccount} : null}], + ["渠道 / 模型", `${log.channelName || "未记录渠道"} / ${log.model || "未识别模型"}`], + ["能力", capabilityText(log.capability)], + ["总耗时", formatDuration(log.durationMs)], + ["视频轮询", log.capability === "video" ? `${log.pollCount || 0} 次` : "--"], + ["Token", log.usageAvailable ? `${log.inputTokens} 输入 / ${log.outputTokens} 输出 / ${log.cachedTokens} 缓存` : "未返回"], + ["计费", log.costAvailable ? `${log.currency || "USD"} ${(log.estimatedCostMicros / 1_000_000).toFixed(6)}` : "未配置价格"], + ["错误信息", [log.errorCode, log.error].filter(Boolean).join(" · ") || "--"], + ["方法与路径", `${log.method} ${log.path}`], + ["HTTP 状态", String(log.statusCode || "--")], + ["任务 ID", log.taskId || "--"], + ["供应商任务 ID", log.providerRequestId || "--"], + ["上游地址", log.upstreamUrl || "--"], + ].map(([label, children], index) => ({ key: String(index), label, children })); + + return
+ +
+
原始报文
+ }, + { key: "response", label: "响应报文", children: }, + ]} /> +
+
; +} + +function PayloadPanel({ value, empty }: { value?: string; empty: string }) { + if (!value) return ; + return
复制报文
{value}
; +} + +function capabilityText(value: string) { return ({ text: "文本", image: "图片", video: "视频", audio: "音频" } as Record)[value] || "未知"; } +function formatDuration(value: number) { if (value < 1_000) return `${value} ms`; if (value < 60_000) return `${(value / 1_000).toFixed(1)} 秒`; return `${Math.floor(value / 60_000)} 分 ${Math.round((value % 60_000) / 1_000)} 秒`; } diff --git a/web/src/pages/admin/logs/logs-page.tsx b/web/src/pages/admin/logs/logs-page.tsx index 4cc2cca25..5178553c4 100644 --- a/web/src/pages/admin/logs/logs-page.tsx +++ b/web/src/pages/admin/logs/logs-page.tsx @@ -1,20 +1,18 @@ import { App, Button, Input, Select, Table, Tag } from "antd"; import type { ColumnsType } from "antd/es/table"; import { Eye, Search } from "lucide-react"; -import { useEffect, useMemo, useRef, useState } from "react"; +import { useEffect, useRef, useState } from "react"; import { useSearchParams } from "react-router"; import { ListToolbar, TableSurface } from "@/components/layout/workspace-page"; import { useDebouncedValue } from "@/hooks/use-debounced-value"; import { exportAdminApiLogs, listAdminApiLogs, type ApiCallLog } from "@/services/api/auth"; -import { useAdminContext } from "../admin-context"; import { ApiLogDetailDrawer } from "../components/api-log-detail-drawer"; import { AdminPageFrame } from "../components/admin-shell"; import { AdminBatchBar, AdminExportButton, AdminTableEmpty, AdminTableSkeleton } from "../components/admin-ui"; export default function LogsPage() { const { message } = App.useApp(); - const { references } = useAdminContext(); const [searchParams, setSearchParams] = useSearchParams(); const keyword = searchParams.get("filter") || ""; const status = normalizeStatus(searchParams.get("status")); @@ -28,7 +26,6 @@ export default function LogsPage() { const [detailLogId, setDetailLogId] = useState(null); const requestSequence = useRef(0); const hasFilters = Boolean(keyword || status !== "all"); - const userNameById = useMemo(() => new Map(references.users.map((user) => [user.id, user.displayName || user.username])), [references.users]); const updateUrl = (patch: Record, replace = false) => { const next = new URLSearchParams(searchParams); @@ -56,16 +53,15 @@ export default function LogsPage() { }, [debouncedKeyword, status, page, pageSize]); const columns: ColumnsType = [ - { title: "时间", dataIndex: "createdAt", width: 170, render: formatTime }, - { title: "用户", dataIndex: "userId", width: 160, render: (id) => userNameById.get(id) || id }, - { title: "渠道", dataIndex: "channelName", width: 170, render: (name, log) => name || log.channelId || 未记录 }, - { title: "模型", dataIndex: "model", width: 180, render: (model) => model || 未识别 }, - { title: "能力 / 阶段", width: 125, render: (_, log) => `${capabilityText(log.capability)} / ${requestKindText(log.requestKind)}` }, - { title: "状态", dataIndex: "status", width: 110, render: (value, log) => {value === "succeeded" ? "成功" : `失败 ${log.statusCode || ""}`} }, - { title: "错误码", dataIndex: "errorCode", width: 160, ellipsis: true, render: (value) => value || "--" }, - { title: "耗时", dataIndex: "durationMs", width: 100, render: (value) => `${value}ms` }, - { title: "Token", width: 145, render: (_, log) => log.usageAvailable ? `${log.inputTokens} / ${log.outputTokens}` : "--" }, - { title: "费用", width: 110, render: (_, log) => log.costAvailable ? `${log.currency || "USD"} ${(log.estimatedCostMicros / 1_000_000).toFixed(6)}` : "--" }, + { title: "时间", width: 168, render: (_, log) => formatTime(log.startedAt || log.createdAt) }, + { title: "用户", width: 180, render: (_, log) =>
{log.userDisplayName || log.userAccount || "未知用户"}
{log.userAccount ? `@${log.userAccount}` : "账号未记录"}
}, + { title: "渠道 / 模型", width: 230, render: (_, log) =>
{log.channelName || "未记录渠道"}
{log.model || "未识别模型"}
}, + { title: "能力", dataIndex: "capability", width: 88, render: capabilityText }, + { title: "调用状态", width: 160, render: (_, log) => }, + { title: "错误信息", width: 260, render: (_, log) => log.status === "failed" || log.error || log.errorCode ?
{log.errorCode || `HTTP ${log.statusCode || "失败"}`}
{log.error || "上游未返回错误详情"}
: -- }, + { title: "耗时", dataIndex: "durationMs", width: 112, render: (value) => {formatDuration(value)} }, + { title: "计费", width: 130, render: (_, log) => log.costAvailable ?
{formatCost(log)}
估算费用
: 未配置 }, + { title: "Tokens", width: 166, render: (_, log) => log.usageAvailable ?
输入 {log.inputTokens.toLocaleString()}
输出 {log.outputTokens.toLocaleString()}
{log.cachedTokens > 0 ?
缓存 {log.cachedTokens.toLocaleString()}
: null}
: 未返回 }, { title: "操作", width: 90, fixed: "right", render: (_, log) => }, ]; @@ -77,7 +73,7 @@ export default function LogsPage() { setSelectedIds([])}> exportAdminApiLogs({ ids: selectedIds })} fileName={() => `请求明细-已选${selectedIds.length}条.csv`} label="导出已选" successMessage={`已导出选中的 ${selectedIds.length} 条请求明细`} errorMessage="导出请求明细失败" /> - {loading && logs.length === 0 ? : setSelectedIds(keys.map(String)) }} columns={columns} dataSource={logs} locale={{ emptyText: }} pagination={{ current: page, pageSize, total, showSizeChanger: true, pageSizeOptions: [20, 50, 100], showTotal: (value, range) => `${range[0]}-${range[1]} / 共 ${value} 条`, onChange: (nextPage, nextSize) => updateUrl({ page: nextSize !== pageSize ? 1 : nextPage, pageSize: nextSize }) }} scroll={{ x: 1280 }} />} + {loading && logs.length === 0 ? :
setSelectedIds(keys.map(String)) }} columns={columns} dataSource={logs} locale={{ emptyText: }} pagination={{ current: page, pageSize, total, showSizeChanger: true, pageSizeOptions: [20, 50, 100], showTotal: (value, range) => `${range[0]}-${range[1]} / 共 ${value} 条`, onChange: (nextPage, nextSize) => updateUrl({ page: nextSize !== pageSize ? 1 : nextPage, pageSize: nextSize }) }} scroll={{ x: 1580 }} />} setDetailLogId(null)} /> @@ -89,4 +85,12 @@ function normalizePageSize(value: string | null) { const parsed = positiveInt(va function normalizeStatus(value: string | null): "all" | "succeeded" | "failed" { return value === "succeeded" || value === "failed" ? value : "all"; } function formatTime(value?: string) { return value ? new Date(value).toLocaleString("zh-CN", { hour12: false }) : "--"; } function capabilityText(value: string) { return ({ text: "文本", image: "图片", video: "视频", audio: "音频" } as Record)[value] || "未知"; } -function requestKindText(value: string) { return ({ create: "创建", poll: "轮询", download: "下载", repair: "修复" } as Record)[value] || "请求"; } +function formatDuration(value: number) { if (value < 1_000) return `${value} ms`; if (value < 60_000) return `${(value / 1_000).toFixed(value < 10_000 ? 1 : 0)} 秒`; const minutes = Math.floor(value / 60_000); const seconds = Math.round((value % 60_000) / 1_000); return `${minutes} 分 ${seconds} 秒`; } +function formatCost(log: ApiCallLog) { return `${log.currency || "USD"} ${(log.estimatedCostMicros / 1_000_000).toFixed(6)}`; } + +function CallStatus({ log }: { log: ApiCallLog }) { + const providerStatus = log.providerStatus?.toLowerCase(); + const processing = ["queued", "pending", "processing", "running", "in_progress"].includes(providerStatus || ""); + const failed = log.status === "failed" || ["failed", "cancelled", "expired"].includes(providerStatus || ""); + return
{failed ? "失败" : processing ? "处理中" : "成功"}{log.capability === "video" ?
已轮询 {log.pollCount || 0} 次
: null}
; +} diff --git a/web/src/services/api/auth.ts b/web/src/services/api/auth.ts index 0ce8eeb48..f6979c48f 100644 --- a/web/src/services/api/auth.ts +++ b/web/src/services/api/auth.ts @@ -41,6 +41,8 @@ export type RuntimeLimits = { export type ApiCallLog = { id: string; userId: string; + userDisplayName?: string; + userAccount?: string; channelId: string; channelName: string; taskId?: string; @@ -56,6 +58,8 @@ export type ApiCallLog = { status: "succeeded" | "failed"; statusCode: number; durationMs: number; + pollCount: number; + providerStatus?: string; inputTokens: number; outputTokens: number; cachedTokens: number; @@ -70,6 +74,9 @@ export type ApiCallLog = { error?: string; concurrencyLimit: number; upstreamUrl: string; + requestBody?: string; + responseBody?: string; + startedAt?: string; createdAt: string; };