feat(admin): 请求明细 - 聚合视频轮询并补全调用报文与运营字段

This commit is contained in:
ddCat
2026-07-28 20:12:24 +08:00
parent 940b716e5c
commit a8ae2d63d6
11 changed files with 422 additions and 77 deletions

View File

@@ -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)
}

View File

@@ -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"`
}

View File

@@ -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 != "" {

View File

@@ -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
}

View File

@@ -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("请先登录")

View File

@@ -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) {

View File

@@ -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))
}

View File

@@ -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 != "" {