From ef2845af3acc10fd6588d0f2e20694496519c172 Mon Sep 17 00:00:00 2001 From: qian Date: Tue, 19 May 2026 00:33:02 +0800 Subject: [PATCH 1/4] =?UTF-8?q?feat(model):=E6=96=B0=E5=A2=9E=E8=A7=86?= =?UTF-8?q?=E9=A2=91=E7=B1=BB=E5=9E=8B=E7=9A=84=E5=A4=A7=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/admin/v1/model.go | 6 +-- internal/handler/relay/task.go | 53 ++++++++++++++++--- internal/logic/admin/model.go | 3 +- internal/logic/admin/tenant_model.go | 26 ++++++--- internal/logic/task/async_provider.go | 49 +++++++++++++++-- relay/handler/task_handler.go | 13 +++++ relay/taskchannel/volcengine/adaptor.go | 10 +++- web/admin/src/views/ai/ModelsPage.vue | 3 ++ .../src/components/common/ApiKeyEditModal.vue | 2 + 9 files changed, 140 insertions(+), 25 deletions(-) diff --git a/api/admin/v1/model.go b/api/admin/v1/model.go index d233192..1cf2cb3 100644 --- a/api/admin/v1/model.go +++ b/api/admin/v1/model.go @@ -7,7 +7,7 @@ type ModelListReq struct { g.Meta `path:"/models" method:"get" mime:"json" tags:"管理后台-模型" summary:"模型列表"` Page int `json:"page" d:"1" v:"min:1" dc:"页码"` PageSize int `json:"page_size" d:"20" v:"min:1|max:100" dc:"每页数量"` - Category string `json:"category" dc:"模型分类筛选：chat/embedding/image/audio/rerank"` + Category string `json:"category" dc:"模型分类筛选：chat/embedding/image/audio/rerank/video"` Status string `json:"status" dc:"状态筛选：active/deprecated/offline"` Search string `json:"search" dc:"搜索关键词（模型名或显示名）"` } @@ -44,7 +44,7 @@ type ModelCreateReq struct { g.Meta `path:"/models" method:"post" mime:"json" tags:"管理后台-模型" summary:"创建模型"` ModelId string `json:"model_id" v:"required|length:1,100#请输入模型标识|模型标识长度1-100" dc:"模型唯一标识"` ModelName string `json:"model_name" dc:"模型显示名称"` - Category string `json:"category" v:"required|in:chat,embedding,image,audio,rerank#请选择分类|分类必须是 chat/embedding/image/audio/rerank" dc:"模型分类"` + Category string `json:"category" v:"required|in:chat,embedding,image,audio,rerank,video#请选择分类|分类必须是 chat/embedding/image/audio/rerank/video" dc:"模型分类"` MaxContext int `json:"max_context_tokens" dc:"最大上下文 token 数"` MaxOutput int `json:"max_output_tokens" dc:"最大输出 token 数"` Capabilities map[string]bool `json:"capabilities" dc:"模型能力特性"` @@ -62,7 +62,7 @@ type ModelUpdateReq struct { g.Meta `path:"/models/{id}" method:"put" mime:"json" tags:"管理后台-模型" summary:"更新模型"` ID int64 `json:"id" in:"path" v:"required" dc:"模型ID"` ModelName string `json:"model_name" dc:"模型显示名称"` - Category string `json:"category" v:"in:chat,embedding,image,audio,rerank" dc:"模型分类"` + Category string `json:"category" v:"in:chat,embedding,image,audio,rerank,video" dc:"模型分类"` MaxContext int `json:"max_context_tokens" dc:"最大上下文 token 数"` MaxOutput int `json:"max_output_tokens" dc:"最大输出 token 数"` Capabilities map[string]bool `json:"capabilities" dc:"模型能力特性"` diff --git a/internal/handler/relay/task.go b/internal/handler/relay/task.go index 1caffe9..6bef02e 100644 --- a/internal/handler/relay/task.go +++ b/internal/handler/relay/task.go @@ -1,12 +1,14 @@ package relay import ( + "encoding/json" "strings" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/net/ghttp" "github.com/qianfree/team-api/internal/logic/billing" + "github.com/qianfree/team-api/internal/logic/relay" "github.com/qianfree/team-api/internal/logic/task" "github.com/qianfree/team-api/internal/middleware" "github.com/qianfree/team-api/relay/common" @@ -16,6 +18,7 @@ import ( var ( taskDataProvider = task.DefaultAsyncProvider taskBillingProvider = billing.NewTaskBillingProvider() + relayDataProvider = relay.NewDataProvider() ) // HandleTaskSubmit 处理异步任务提交（POST /v1/video/generations, POST /suno/submit/:action） @@ -41,8 +44,22 @@ func HandleTaskSubmit(r *ghttp.Request) { // 选择渠道 channelMeta, err := selectTaskChannel(r, body) if err != nil { - r.Response.WriteStatus(503, g.Map{ - "error": g.Map{"type": "server_error", "message": err.Error()}, + statusCode := 503 + errType := "server_error" + errMsg := err.Error() + + if err == common.ErrTenantModelNotEnabled { + statusCode = 403 + errType = "permission_denied" + errMsg = "当前租户未启用该模型，请联系管理员" + } else if err == common.ErrChannelUnavailable { + statusCode = 503 + errType = "server_error" + errMsg = "该模型暂无可用的渠道，请稍后重试或联系管理员" + } + + r.Response.WriteStatus(statusCode, g.Map{ + "error": g.Map{"type": errType, "message": errMsg}, }) return } @@ -109,14 +126,34 @@ func extractTaskID(r *ghttp.Request) string { return r.Get("task_id").String() } -// selectTaskChannel 为异步任务选择渠道 +// selectTaskChannel 为异步任务选择渠道（复用实时请求的渠道调度逻辑） func selectTaskChannel(r *ghttp.Request, body []byte) (*common.ChannelMeta, error) { - // 复用现有的渠道调度逻辑 - // 暂时返回一个占位符，后续集成到 scheduler + // 从请求体提取模型名 + var req struct { + Model string `json:"model"` + } + if err := json.Unmarshal(body, &req); err != nil || req.Model == "" { + return nil, common.ErrChannelUnavailable + } + + ctx := r.Context() + tenantID := middleware.GetTenantID(ctx) + userID := middleware.GetUserID(ctx) + + selection, err := relayDataProvider.GetChannelForModel(ctx, tenantID, userID, req.Model, nil) + if err != nil { + return nil, err + } + return &common.ChannelMeta{ - ChannelID: 1, - BaseURL: "https://api.openai.com", - ApiKey: "", + ChannelID: selection.ChannelID, + ChannelType: selection.ChannelType, + ChannelName: selection.ChannelName, + BaseURL: selection.BaseURL, + ApiKey: selection.ApiKey, + UpstreamModelName: selection.UpstreamModelName, + IsModelMapped: selection.IsModelMapped, + Settings: selection.Settings, }, nil } diff --git a/internal/logic/admin/model.go b/internal/logic/admin/model.go index 11986db..36d6019 100644 --- a/internal/logic/admin/model.go +++ b/internal/logic/admin/model.go @@ -252,9 +252,10 @@ func (s *sAdmin) UpdateModel(ctx context.Context, req *v1.ModelUpdateReq) (*v1.M return nil, nil } -// DeleteModel 删除模型（同时删除定价记录） +// DeleteModel 删除模型（同时删除定价记录和租户分配记录） func (s *sAdmin) DeleteModel(ctx context.Context, req *v1.ModelDeleteReq) (*v1.ModelDeleteRes, error) { _, _ = dao.MdlPricing.Ctx(ctx).Where("model_id", req.ID).Delete() + _, _ = dao.MdlTenantModels.Ctx(ctx).Where("model_id", req.ID).Delete() _, err := dao.MdlModels.Ctx(ctx).Where("id", req.ID).Delete() if err != nil { return nil, err diff --git a/internal/logic/admin/tenant_model.go b/internal/logic/admin/tenant_model.go index ceee8c5..c98f6d5 100644 --- a/internal/logic/admin/tenant_model.go +++ b/internal/logic/admin/tenant_model.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" + "github.com/gogf/gf/v2/frame/g" v1 "github.com/qianfree/team-api/api/admin/v1" "github.com/qianfree/team-api/internal/dao" "github.com/qianfree/team-api/internal/logic/billing" @@ -83,22 +84,28 @@ func (s *sAdmin) BatchAssignModels(ctx context.Context, req *v1.TenantModelBatch assigned := 0 for _, a := range req.Assignments { - var model *struct { + g.Log().Debugf(ctx, "BatchAssignModels: tenant_id=%d, model_id=%d, enabled=%v", req.TenantID, a.ModelID, a.Enabled) + + var model struct { ID int64 `json:"id"` Status string `json:"status"` } err := dao.MdlModels.Ctx(ctx). Where("id", a.ModelID). Scan(&model) - if err != nil || model == nil || model.Status != "active" { + g.Log().Debugf(ctx, "BatchAssignModels: model scan result id=%d, status=%q, err=%v", model.ID, model.Status, err) + if err != nil || model.ID == 0 || model.Status != "active" { + g.Log().Warningf(ctx, "BatchAssignModels: skip model_id=%d, err=%v, id=%d, status=%q", a.ModelID, err, model.ID, model.Status) continue } - count, _ := dao.MdlTenantModels.Ctx(ctx). + count, countErr := dao.MdlTenantModels.Ctx(ctx). Where("tenant_id", req.TenantID). Where("model_id", a.ModelID). Count() + g.Log().Debugf(ctx, "BatchAssignModels: existing count=%d, countErr=%v", count, countErr) if count > 0 { + g.Log().Debugf(ctx, "BatchAssignModels: skip already assigned, model_id=%d", a.ModelID) continue } @@ -108,13 +115,13 @@ func (s *sAdmin) BatchAssignModels(ctx context.Context, req *v1.TenantModelBatch maxConc = &concurrency } - var tiersJSON string + var tiersJSON any if len(a.CustomPricingTiers) > 0 { b, _ := json.Marshal(a.CustomPricingTiers) tiersJSON = string(b) } - _, err = dao.MdlTenantModels.Ctx(ctx).Insert(do.MdlTenantModels{ + insertData := do.MdlTenantModels{ TenantId: req.TenantID, ModelId: a.ModelID, Enabled: a.Enabled, @@ -129,9 +136,14 @@ func (s *sAdmin) BatchAssignModels(ctx context.Context, req *v1.TenantModelBatch CustomCacheCreationPrice: a.CustomCacheCreationPrice, CustomPricingTiers: tiersJSON, Multiplier: 1.0, - }) - if err == nil { + } + g.Log().Debugf(ctx, "BatchAssignModels: inserting data tenant_id=%v, model_id=%v, enabled=%v, max_concurrency=%v", insertData.TenantId, insertData.ModelId, insertData.Enabled, insertData.MaxConcurrency) + _, err = dao.MdlTenantModels.Ctx(ctx).Insert(insertData) + if err != nil { + g.Log().Errorf(ctx, "BatchAssignModels: insert failed, model_id=%d, err=%v", a.ModelID, err) + } else { assigned++ + g.Log().Debugf(ctx, "BatchAssignModels: insert success, model_id=%d, assigned=%d", a.ModelID, assigned) } } diff --git a/internal/logic/task/async_provider.go b/internal/logic/task/async_provider.go index f25170d..57b0150 100644 --- a/internal/logic/task/async_provider.go +++ b/internal/logic/task/async_provider.go @@ -7,8 +7,12 @@ import ( "time" "github.com/gogf/gf/v2/errors/gerror" + "github.com/gogf/gf/v2/os/gtime" "github.com/qianfree/team-api/internal/dao" + "github.com/qianfree/team-api/internal/logic/relay" + do "github.com/qianfree/team-api/internal/model/do" + "github.com/qianfree/team-api/internal/utility/crypto" "github.com/qianfree/team-api/relay/common" ) @@ -331,19 +335,18 @@ func (p *AsyncProvider) GetTimedOutTasks(ctx context.Context, cutoffUnix int64, return tasks, nil } -// GetChannelByID 获取渠道基本信息 +// GetChannelByID 获取渠道基本信息（含从 chn_channel_keys 解密的 API Key） func (p *AsyncProvider) GetChannelByID(ctx context.Context, channelID int64) (*common.ChannelBasicInfo, error) { var row struct { ID int64 `json:"id"` Type int `json:"type"` Name string `json:"name"` BaseURL string `json:"base_url"` - ApiKey string `json:"api_key"` Settings json.RawMessage `json:"settings"` } err := dao.ChnChannels.Ctx(ctx). Where("id", channelID). - Fields("id, type, name, base_url, api_key, settings"). + Fields("id, type, name, base_url, settings"). Scan(&row) if err != nil { return nil, gerror.Wrapf(err, "query channel failed: id=%d", channelID) @@ -351,12 +354,50 @@ func (p *AsyncProvider) GetChannelByID(ctx context.Context, channelID int64) (*c if row.ID == 0 { return nil, nil } + + // 从 chn_channel_keys 获取解密后的 API Key + apiKey, keyErr := getChannelApiKey(ctx, channelID) + if keyErr != nil { + return nil, gerror.Wrapf(keyErr, "get channel key failed: channelID=%d", channelID) + } + return &common.ChannelBasicInfo{ ID: row.ID, Type: row.Type, Name: row.Name, BaseURL: row.BaseURL, - ApiKey: row.ApiKey, + ApiKey: apiKey, Settings: row.Settings, }, nil } + +// getChannelApiKey 从 chn_channel_keys 获取并解密渠道 API Key +func getChannelApiKey(ctx context.Context, channelID int64) (string, error) { + type keyRow struct { + ID int64 `json:"id"` + EncryptedKey string `json:"encrypted_key"` + } + + var key *keyRow + err := dao.ChnChannelKeys.Ctx(ctx). + Where("channel_id", channelID). + Where("status", "active"). + Fields("id, encrypted_key"). + Scan(&key) + if err != nil || key == nil { + return "", fmt.Errorf("no active key found for channel %d", channelID) + } + + // 更新最后使用时间 + dao.ChnChannelKeys.Ctx(ctx). + Where("id", key.ID). + Data(do.ChnChannelKeys{LastUsedAt: gtime.Now()}). + Update() + + encKey := relay.GetEncryptionKey() + decrypted, err := crypto.DecryptString(encKey, key.EncryptedKey) + if err != nil { + return "", fmt.Errorf("decrypt key failed: %w", err) + } + return decrypted, nil +} diff --git a/relay/handler/task_handler.go b/relay/handler/task_handler.go index 7660833..942eb73 100644 --- a/relay/handler/task_handler.go +++ b/relay/handler/task_handler.go @@ -10,6 +10,7 @@ import ( "net/http" "time" + "github.com/gogf/gf/v2/frame/g" "github.com/qianfree/team-api/relay/common" "github.com/qianfree/team-api/relay/constant" "github.com/qianfree/team-api/relay/taskchannel" @@ -65,6 +66,7 @@ func HandleTaskSubmit( providerType := constant.ProviderType(channelMeta.ChannelType) platform, ok := constant.ProviderTypeToTaskPlatform(providerType) if !ok { + g.Log().Warningf(ctx, "HandleTaskSubmit: unsupported task platform, channelType=%d, modelName=%s", channelMeta.ChannelType, modelName) writeTaskError(rc.Writer, http.StatusBadRequest, "unsupported task platform", "") return } @@ -76,6 +78,8 @@ func HandleTaskSubmit( return } + g.Log().Debugf(ctx, "HandleTaskSubmit: modelName=%s, platform=%s, channelID=%d, channelType=%d, baseURL=%s, upstreamModel=%s", modelName, platform, channelMeta.ChannelID, channelMeta.ChannelType, channelMeta.BaseURL, channelMeta.UpstreamModelName) + // 4. 构建 RelayInfo info := &common.RelayInfo{ Context: ctx, @@ -94,6 +98,7 @@ func HandleTaskSubmit( // 5. 校验请求 if taskErr := adaptor.ValidateRequest(ctx, info, body); taskErr != nil { + g.Log().Warningf(ctx, "HandleTaskSubmit: validate failed, model=%s, err=%s", modelName, taskErr.Message) writeTaskError(rc.Writer, taskErr.StatusCode, taskErr.Message, taskErr.ErrCode) return } @@ -102,12 +107,15 @@ func HandleTaskSubmit( ratios := adaptor.EstimateBilling(ctx, info, body) estimatedCost, err := billingProvider.EstimateTaskCost(ctx, rc.TenantID, modelName, ratios) if err != nil { + g.Log().Errorf(ctx, "HandleTaskSubmit: estimate cost failed, model=%s, err=%v", modelName, err) writeTaskError(rc.Writer, http.StatusInternalServerError, "estimate cost failed: "+err.Error(), "") return } + g.Log().Debugf(ctx, "HandleTaskSubmit: estimatedCost=%.4f", estimatedCost) preDeductAmount, err := billingProvider.PreDeductTask(ctx, rc.TenantID, rc.RequestID, estimatedCost) if err != nil { + g.Log().Warningf(ctx, "HandleTaskSubmit: pre-deduct failed, model=%s, err=%v", modelName, err) writeTaskError(rc.Writer, http.StatusPaymentRequired, "insufficient balance", "") return } @@ -116,6 +124,7 @@ func HandleTaskSubmit( requestBody, err := adaptor.BuildRequestBody(ctx, info, body) if err != nil { billingProvider.SettleTaskFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount) + g.Log().Errorf(ctx, "HandleTaskSubmit: build request failed, model=%s, err=%v", modelName, err) writeTaskError(rc.Writer, http.StatusInternalServerError, "build request failed: "+err.Error(), "") return } @@ -123,15 +132,19 @@ func HandleTaskSubmit( resp, err := adaptor.DoRequest(ctx, info, requestBody) if err != nil { billingProvider.SettleTaskFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount) + g.Log().Errorf(ctx, "HandleTaskSubmit: upstream request failed, model=%s, err=%v", modelName, err) writeTaskError(rc.Writer, http.StatusBadGateway, "upstream request failed: "+err.Error(), "") return } defer resp.Body.Close() + g.Log().Debugf(ctx, "HandleTaskSubmit: upstream responded status=%d", resp.StatusCode) + // 8. 解析响应 upstreamTaskID, taskData, taskErr := adaptor.DoResponse(ctx, resp, info) if taskErr != nil { billingProvider.SettleTaskFailed(ctx, rc.TenantID, rc.RequestID, preDeductAmount) + g.Log().Warningf(ctx, "HandleTaskSubmit: upstream response error, model=%s, status=%d, message=%q, body=%s", modelName, taskErr.StatusCode, taskErr.Message, string(taskData)) writeTaskError(rc.Writer, taskErr.StatusCode, taskErr.Message, taskErr.ErrCode) return } diff --git a/relay/taskchannel/volcengine/adaptor.go b/relay/taskchannel/volcengine/adaptor.go index dd693f5..b13e01e 100644 --- a/relay/taskchannel/volcengine/adaptor.go +++ b/relay/taskchannel/volcengine/adaptor.go @@ -115,6 +115,8 @@ func (a *VolcengineVideoAdaptor) AdjustBillingOnSubmit(_ *common.RelayInfo, _ [] func (a *VolcengineVideoAdaptor) BuildRequestURL(info *common.RelayInfo) (string, error) { baseURL := strings.TrimRight(info.ChannelMeta.BaseURL, "/") + // 兼容 base URL 带 /api 后缀的情况 + baseURL = strings.TrimSuffix(baseURL, "/api") return fmt.Sprintf("%s/api/v3/contents/generations/tasks", baseURL), nil } @@ -277,9 +279,13 @@ func (a *VolcengineVideoAdaptor) DoResponse(_ context.Context, resp *http.Respon // 非成功状态码 if resp.StatusCode != http.StatusOK { + msg := string(body) + if msg == "" { + msg = fmt.Sprintf("upstream returned status %d", resp.StatusCode) + } return "", body, &common.TaskError{ StatusCode: resp.StatusCode, - Message: string(body), + Message: msg, } } @@ -301,7 +307,7 @@ func (a *VolcengineVideoAdaptor) FetchTask(baseURL, apiKey string, taskData []by return nil, fmt.Errorf("volcengine: invalid task data: %w", err) } - url := fmt.Sprintf("%s/api/v3/contents/generations/tasks/%s", strings.TrimRight(baseURL, "/"), data.TaskID) + url := fmt.Sprintf("%s/api/v3/contents/generations/tasks/%s", strings.TrimSuffix(strings.TrimRight(baseURL, "/"), "/api"), data.TaskID) req, err := http.NewRequest(http.MethodGet, url, nil) if err != nil { diff --git a/web/admin/src/views/ai/ModelsPage.vue b/web/admin/src/views/ai/ModelsPage.vue index a04a189..8a37bdf 100644 --- a/web/admin/src/views/ai/ModelsPage.vue +++ b/web/admin/src/views/ai/ModelsPage.vue @@ -28,6 +28,7 @@ const categoryOptions = [ { label: '向量', value: 'embedding' }, { label: '图像', value: 'image' }, { label: '音频', value: 'audio' }, + { label: '视频', value: 'video' }, { label: '重排', value: 'rerank' }, ] @@ -111,6 +112,7 @@ const categoryTagColor: Record = { embedding: 'green', image: 'orangered', audio: 'red', + video: 'purple', rerank: 'arcoblue', } @@ -125,6 +127,7 @@ const categoryTagLabel: Record = { embedding: '向量', image: '图像', audio: '音频', + video: '视频', rerank: '重排', } diff --git a/web/tenant/src/components/common/ApiKeyEditModal.vue b/web/tenant/src/components/common/ApiKeyEditModal.vue index d680b26..db42dd8 100644 --- a/web/tenant/src/components/common/ApiKeyEditModal.vue +++ b/web/tenant/src/components/common/ApiKeyEditModal.vue @@ -54,6 +54,7 @@ const categoryLabel: Record = { embedding: '嵌入', image: '图像', audio: '语音', + video: '视频', rerank: '重排', } const categoryBadgeClass: Record = { @@ -61,6 +62,7 @@ const categoryBadgeClass: Record = { embedding: 'badge-purple', image: 'badge-warning', audio: 'badge-success', + video: 'badge-purple', rerank: 'badge-gray', } From 1b6134a890973a5b0bd68c3c24bd04053716b840 Mon Sep 17 00:00:00 2001 From: qian Date: Tue, 19 May 2026 00:45:02 +0800 Subject: [PATCH 2/4] =?UTF-8?q?feat(model):=E8=A7=86=E9=A2=91=E5=A4=A7?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=E6=B7=BB=E5=8A=A0=E8=AF=B7=E6=B1=82=E5=AE=A1?= =?UTF-8?q?=E8=AE=A1=E5=92=8C=E7=94=A8=E9=87=8F=E8=AE=B0=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/handler/relay/task.go | 31 ++++++++++++++++++++++++ internal/logic/task/async_polling.go | 35 ++++++++++++++++++++++++++++ 2 files changed, 66 insertions(+) diff --git a/internal/handler/relay/task.go b/internal/handler/relay/task.go index 6bef02e..6c19f61 100644 --- a/internal/handler/relay/task.go +++ b/internal/handler/relay/task.go @@ -1,6 +1,7 @@ package relay import ( + "context" "encoding/json" "strings" @@ -64,10 +65,40 @@ func HandleTaskSubmit(r *ghttp.Request) { return } + // 记录任务提交审计日志（使用 capture 模式以支持异步写入） + capture := NewResponseCaptureWriter(r.Response.Writer) + rc.Writer = capture + relay_handler.HandleTaskSubmit( r.Context(), body, r.URL.Path, r.Header, rc, taskDataProvider, taskBillingProvider, channelMeta, ) + + // 异步记录审计日志 + tenantID := rc.TenantID + userID := rc.UserID + apiKeyID := rc.ApiKeyID + requestID := rc.RequestID + path := r.URL.Path + clientIP := rc.ClientIP + userAgent := r.Header.Get("User-Agent") + statusCode := capture.StatusCode() + responseBody := capture.Body() + go func() { + relayDataProvider.RecordAudit(context.Background(), &common.AuditRecord{ + TenantID: tenantID, + UserID: userID, + ApiKeyID: apiKeyID, + RequestID: requestID, + Method: "POST", + Path: path, + StatusCode: statusCode, + ClientIP: clientIP, + UserAgent: userAgent, + RequestBody: string(body), + ResponseBody: responseBody, + }) + }() } // HandleTaskFetch 处理异步任务查询（GET /v1/video/generations/:id, GET /suno/fetch/:id） diff --git a/internal/logic/task/async_polling.go b/internal/logic/task/async_polling.go index 43d6045..a07c409 100644 --- a/internal/logic/task/async_polling.go +++ b/internal/logic/task/async_polling.go @@ -9,6 +9,7 @@ import ( "github.com/gogf/gf/v2/frame/g" "github.com/qianfree/team-api/internal/logic/billing" + "github.com/qianfree/team-api/internal/logic/relay" "github.com/qianfree/team-api/relay/common" "github.com/qianfree/team-api/relay/constant" "github.com/qianfree/team-api/relay/taskchannel" @@ -227,6 +228,9 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co } g.Log().Infof(ctx, "poll: task %s completed", task.PublicTaskID) + // 记录用量日志 + recordTaskUsage(task, true, "") + } else if taskInfo.Status == common.TaskStatusFailure { // 退还预扣费用 if task.PreDeductAmount > 0 && !task.BillingSettled { @@ -239,5 +243,36 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co } } g.Log().Infof(ctx, "poll: task %s failed: %s", task.PublicTaskID, task.FailReason) + + // 记录用量日志 + recordTaskUsage(task, false, task.FailReason) } } + +// recordTaskUsage 异步记录视频/音乐等异步任务的用量日志 +func recordTaskUsage(task *common.AsyncTask, success bool, errMsg string) { + latencyMs := 0 + if task.SubmitTime != nil && task.FinishTime != nil { + latencyMs = int(task.FinishTime.Sub(*task.SubmitTime).Milliseconds()) + } + + status := "success" + if !success { + status = "error" + } + + relay.NewDataProvider().RecordUsage(context.Background(), &common.UsageRecord{ + TenantID: task.TenantID, + UserID: task.UserID, + ApiKeyID: task.ApiKeyID, + ChannelID: task.ChannelID, + ModelName: task.ModelName, + RelayMode: int(constant.RelayModeVideoGenerations), + LatencyMs: float64(latencyMs), + IsStream: false, + Success: success, + RequestID: task.PublicTaskID, + Status: status, + ErrorMessage: errMsg, + }) +} From 5084406687721087948d1b9e7d651dad4048ac87 Mon Sep 17 00:00:00 2001 From: qian Date: Tue, 19 May 2026 23:28:54 +0800 Subject: [PATCH 3/4] =?UTF-8?q?feat(model):=E6=B7=BB=E5=8A=A0=E5=AF=B9?= =?UTF-8?q?=E7=81=AB=E5=B1=B1=E8=A7=86=E9=A2=91=E6=A8=A1=E5=9E=8B=E7=9A=84?= =?UTF-8?q?=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/admin/v1/audit.go | 1 + api/admin/v1/task_management.go | 11 +- api/tenant/tenant.go | 2 + api/tenant/v1/audit_tenant.go | 1 + api/tenant/v1/task_tenant.go | 49 +++ .../tenant/tenant_v1_tenant_task_detail.go | 12 + .../tenant/tenant_v1_tenant_task_list.go | 12 + internal/dao/aud_content_filter_logs.go | 22 + .../dao/internal/aud_content_filter_logs.go | 105 +++++ internal/dao/internal/aud_request_logs.go | 110 ++--- internal/dao/internal/bil_usage_logs.go | 4 +- internal/dao/internal/tsk_model_tasks.go | 2 + internal/handler/relay/task.go | 72 ++-- internal/logic/admin/audit.go | 4 + internal/logic/admin/dashboard.go | 2 +- internal/logic/admin/task_management.go | 3 + internal/logic/relay/provider.go | 67 +++ internal/logic/task/async_polling.go | 77 +++- internal/logic/task/async_provider.go | 9 + internal/logic/tenant/audit_config.go | 8 +- internal/logic/tenant/personal_dashboard.go | 4 +- internal/logic/tenant/task.go | 168 ++++++++ internal/model/do/aud_content_filter_logs.go | 29 ++ internal/model/do/aud_request_logs.go | 57 +-- internal/model/do/bil_usage_logs.go | 3 +- internal/model/do/tsk_model_tasks.go | 1 + .../model/entity/aud_content_filter_logs.go | 27 ++ internal/model/entity/aud_request_logs.go | 55 +-- internal/model/entity/bil_usage_logs.go | 113 ++--- internal/model/entity/tsk_model_tasks.go | 49 +-- internal/service/admin.go | 2 +- internal/service/tenant.go | 6 + migrations/000003_async_task_support.sql | 59 +++ relay/common/provider.go | 17 +- relay/common/task_adaptor.go | 5 + relay/common/task_provider.go | 5 + relay/handler/task_handler.go | 6 + relay/taskchannel/volcengine/adaptor.go | 20 +- web/admin/src/views/ai/TaskLogsPage.vue | 15 + .../views/security/RequestAuditLogsPage.vue | 88 +++- web/admin/src/views/tenant/UsageLogsPage.vue | 25 +- web/tenant/src/router/tenant.ts | 6 + .../views/service/RequestAuditLogsPage.vue | 62 +++ web/tenant/src/views/service/TaskLogsPage.vue | 391 ++++++++++++++++++ .../src/views/service/UsageLogsPage.vue | 24 +- 45 files changed, 1553 insertions(+), 257 deletions(-) create mode 100644 api/tenant/v1/task_tenant.go create mode 100644 internal/controller/tenant/tenant_v1_tenant_task_detail.go create mode 100644 internal/controller/tenant/tenant_v1_tenant_task_list.go create mode 100644 internal/dao/aud_content_filter_logs.go create mode 100644 internal/dao/internal/aud_content_filter_logs.go create mode 100644 internal/logic/tenant/task.go create mode 100644 internal/model/do/aud_content_filter_logs.go create mode 100644 internal/model/entity/aud_content_filter_logs.go create mode 100644 migrations/000003_async_task_support.sql create mode 100644 web/tenant/src/views/service/TaskLogsPage.vue diff --git a/api/admin/v1/audit.go b/api/admin/v1/audit.go index d41737b..0bf93bc 100644 --- a/api/admin/v1/audit.go +++ b/api/admin/v1/audit.go @@ -60,6 +60,7 @@ type RequestAuditLogListReq struct { ApiKeyID int `json:"api_key_id" dc:"API Key ID"` Username string `json:"username" dc:"用户名（模糊匹配）"` RequestId string `json:"request_id" dc:"Request ID（精确匹配）"` + TaskId string `json:"task_id" dc:"异步任务ID（精确匹配）"` Method string `json:"method" dc:"HTTP 方法"` Path string `json:"path" dc:"请求路径"` StatusCode int `json:"status_code" dc:"状态码"` diff --git a/api/admin/v1/task_management.go b/api/admin/v1/task_management.go index 153c2dd..005729f 100644 --- a/api/admin/v1/task_management.go +++ b/api/admin/v1/task_management.go @@ -5,11 +5,12 @@ import "github.com/gogf/gf/v2/frame/g" // === 大模型异步任务管理 === type TaskListReq struct { - g.Meta `path:"/tasks" method:"get" mime:"json" tags:"管理后台-任务管理" summary:"大模型异步任务列表"` - Page int `json:"page" in:"query" d:"1" v:"min:1" dc:"页码"` - PageSize int `json:"page_size" in:"query" d:"20" v:"min:1|max:100" dc:"每页数量"` - Status string `json:"status" in:"query" dc:"筛选状态"` - Platform string `json:"platform" in:"query" dc:"筛选平台(sora/kling/midjourney/suno)"` + g.Meta `path:"/tasks" method:"get" mime:"json" tags:"管理后台-任务管理" summary:"大模型异步任务列表"` + Page int `json:"page" in:"query" d:"1" v:"min:1" dc:"页码"` + PageSize int `json:"page_size" in:"query" d:"20" v:"min:1|max:100" dc:"每页数量"` + Status string `json:"status" in:"query" dc:"筛选状态"` + Platform string `json:"platform" in:"query" dc:"筛选平台(sora/kling/midjourney/suno)"` + PublicTaskID string `json:"public_task_id" in:"query" dc:"任务ID（精确匹配）"` } type ModelTaskItem struct { diff --git a/api/tenant/tenant.go b/api/tenant/tenant.go index 17a7dae..a3d90c2 100644 --- a/api/tenant/tenant.go +++ b/api/tenant/tenant.go @@ -154,6 +154,8 @@ type ITenantV1 interface { Tenant2FAVerify(ctx context.Context, req *v1.Tenant2FAVerifyReq) (res *v1.Tenant2FAVerifyRes, err error) Tenant2FAConfirm(ctx context.Context, req *v1.Tenant2FAConfirmReq) (res *v1.Tenant2FAConfirmRes, err error) TenantLoginHistory(ctx context.Context, req *v1.TenantLoginHistoryReq) (res *v1.TenantLoginHistoryRes, err error) + TenantTaskList(ctx context.Context, req *v1.TenantTaskListReq) (res *v1.TenantTaskListRes, err error) + TenantTaskDetail(ctx context.Context, req *v1.TenantTaskDetailReq) (res *v1.TenantTaskDetailRes, err error) TenantTicketCreate(ctx context.Context, req *v1.TenantTicketCreateReq) (res *v1.TenantTicketCreateRes, err error) TenantTicketList(ctx context.Context, req *v1.TenantTicketListReq) (res *v1.TenantTicketListRes, err error) TenantTicketGet(ctx context.Context, req *v1.TenantTicketGetReq) (res *v1.TenantTicketGetRes, err error) diff --git a/api/tenant/v1/audit_tenant.go b/api/tenant/v1/audit_tenant.go index 9abd4fd..b2c7263 100644 --- a/api/tenant/v1/audit_tenant.go +++ b/api/tenant/v1/audit_tenant.go @@ -40,6 +40,7 @@ type TenantRequestAuditLogsReq struct { PageSize int `json:"page_size" in:"query" d:"20" v:"min:1|max:100"` Username string `json:"username" in:"query" dc:"用户名（模糊匹配）"` RequestId string `json:"request_id" in:"query" dc:"Request ID（精确匹配）"` + TaskId string `json:"task_id" in:"query" dc:"异步任务ID（精确匹配）"` Path string `json:"path" in:"query" dc:"请求路径（模糊匹配）"` StatusCode int `json:"status_code" in:"query" dc:"HTTP 状态码"` StartDate string `json:"start_date" in:"query" dc:"开始日期"` diff --git a/api/tenant/v1/task_tenant.go b/api/tenant/v1/task_tenant.go new file mode 100644 index 0000000..340b85a --- /dev/null +++ b/api/tenant/v1/task_tenant.go @@ -0,0 +1,49 @@ +package v1 + +import "github.com/gogf/gf/v2/frame/g" + +// === 租户端异步任务查询 === + +type TenantTaskListReq struct { + g.Meta `path:"/tasks" method:"get" mime:"json" tags:"租户控制台-任务管理" summary:"租户异步任务列表"` + Page int `json:"page" in:"query" d:"1" v:"min:1" dc:"页码"` + PageSize int `json:"page_size" in:"query" d:"20" v:"min:1|max:100" dc:"每页数量"` + Status string `json:"status" in:"query" dc:"筛选状态"` + Platform string `json:"platform" in:"query" dc:"筛选平台(sora/kling/midjourney/suno)"` + PublicTaskID string `json:"public_task_id" in:"query" dc:"任务ID（精确匹配）"` +} + +type TenantTaskItem struct { + ID int64 `json:"id"` + PublicTaskID string `json:"public_task_id"` + Platform string `json:"platform"` + Action string `json:"action"` + Status string `json:"status"` + Progress string `json:"progress"` + ModelName string `json:"model_name"` + FailReason string `json:"fail_reason,omitempty"` + PreDeductAmount float64 `json:"pre_deduct_amount"` + ActualCost float64 `json:"actual_cost"` + BillingSettled bool `json:"billing_settled"` + ResultURL string `json:"result_url,omitempty"` + Username string `json:"username,omitempty"` + SubmitTime string `json:"submit_time,omitempty"` + FinishTime string `json:"finish_time,omitempty"` + CreatedAt string `json:"created_at"` +} + +type TenantTaskListRes struct { + List []TenantTaskItem `json:"list"` + Total int `json:"total"` + Page int `json:"page"` + PageSize int `json:"page_size"` +} + +type TenantTaskDetailReq struct { + g.Meta `path:"/tasks/{id}" method:"get" mime:"json" tags:"租户控制台-任务管理" summary:"租户异步任务详情"` + ID int64 `json:"id" in:"path" v:"required" dc:"任务ID"` +} + +type TenantTaskDetailRes struct { + Task TenantTaskItem `json:"task"` +} diff --git a/internal/controller/tenant/tenant_v1_tenant_task_detail.go b/internal/controller/tenant/tenant_v1_tenant_task_detail.go new file mode 100644 index 0000000..d9a2d52 --- /dev/null +++ b/internal/controller/tenant/tenant_v1_tenant_task_detail.go @@ -0,0 +1,12 @@ +package tenant + +import ( + "context" + + "github.com/qianfree/team-api/api/tenant/v1" + "github.com/qianfree/team-api/internal/service" +) + +func (c *ControllerV1) TenantTaskDetail(ctx context.Context, req *v1.TenantTaskDetailReq) (res *v1.TenantTaskDetailRes, err error) { + return service.Tenant().TenantTaskDetail(ctx, req) +} diff --git a/internal/controller/tenant/tenant_v1_tenant_task_list.go b/internal/controller/tenant/tenant_v1_tenant_task_list.go new file mode 100644 index 0000000..b2e9ab8 --- /dev/null +++ b/internal/controller/tenant/tenant_v1_tenant_task_list.go @@ -0,0 +1,12 @@ +package tenant + +import ( + "context" + + "github.com/qianfree/team-api/api/tenant/v1" + "github.com/qianfree/team-api/internal/service" +) + +func (c *ControllerV1) TenantTaskList(ctx context.Context, req *v1.TenantTaskListReq) (res *v1.TenantTaskListRes, err error) { + return service.Tenant().TenantTaskList(ctx, req) +} diff --git a/internal/dao/aud_content_filter_logs.go b/internal/dao/aud_content_filter_logs.go new file mode 100644 index 0000000..73611f0 --- /dev/null +++ b/internal/dao/aud_content_filter_logs.go @@ -0,0 +1,22 @@ +// ================================================================================= +// This file is auto-generated by the GoFrame CLI tool. You may modify it as needed. +// ================================================================================= + +package dao + +import ( + "github.com/qianfree/team-api/internal/dao/internal" +) + +// audContentFilterLogsDao is the data access object for the table aud_content_filter_logs. +// You can define custom methods on it to extend its functionality as needed. +type audContentFilterLogsDao struct { + *internal.AudContentFilterLogsDao +} + +var ( + // AudContentFilterLogs is a globally accessible object for table aud_content_filter_logs operations. + AudContentFilterLogs = audContentFilterLogsDao{internal.NewAudContentFilterLogsDao()} +) + +// Add your custom methods and functionality below. diff --git a/internal/dao/internal/aud_content_filter_logs.go b/internal/dao/internal/aud_content_filter_logs.go new file mode 100644 index 0000000..ea69ed5 --- /dev/null +++ b/internal/dao/internal/aud_content_filter_logs.go @@ -0,0 +1,105 @@ +// ========================================================================== +// Code generated and maintained by GoFrame CLI tool. DO NOT EDIT. +// ========================================================================== + +package internal + +import ( + "context" + + "github.com/gogf/gf/v2/database/gdb" + "github.com/gogf/gf/v2/frame/g" +) + +// AudContentFilterLogsDao is the data access object for the table aud_content_filter_logs. +type AudContentFilterLogsDao struct { + table string // table is the underlying table name of the DAO. + group string // group is the database configuration group name of the current DAO. + columns AudContentFilterLogsColumns // columns contains all the column names of Table for convenient usage. + handlers []gdb.ModelHandler // handlers for customized model modification. +} + +// AudContentFilterLogsColumns defines and stores column names for the table aud_content_filter_logs. +type AudContentFilterLogsColumns struct { + Id string // 主键ID + TenantId string // 租户ID + UserId string // 用户ID + ApiKeyId string // API Key ID + RequestId string // 请求唯一ID + Method string // HTTP 方法 + Path string // 请求路径 + ClientIp string // 客户端 IP + FilterMode string // 过滤模式：log / replace / block + MatchedWords string // 命中的敏感词列表（JSONB 数组） + OriginalSnippet string // 原始请求体片段（截断存储，仅 replace 模式） + Blocked string // 是否被拦截（mode=block 时为 true） + CreatedAt string // 创建时间 + ProjectId string // 项目ID +} + +// audContentFilterLogsColumns holds the columns for the table aud_content_filter_logs. +var audContentFilterLogsColumns = AudContentFilterLogsColumns{ + Id: "id", + TenantId: "tenant_id", + UserId: "user_id", + ApiKeyId: "api_key_id", + RequestId: "request_id", + Method: "method", + Path: "path", + ClientIp: "client_ip", + FilterMode: "filter_mode", + MatchedWords: "matched_words", + OriginalSnippet: "original_snippet", + Blocked: "blocked", + CreatedAt: "created_at", + ProjectId: "project_id", +} + +// NewAudContentFilterLogsDao creates and returns a new DAO object for table data access. +func NewAudContentFilterLogsDao(handlers ...gdb.ModelHandler) *AudContentFilterLogsDao { + return &AudContentFilterLogsDao{ + group: "default", + table: "aud_content_filter_logs", + columns: audContentFilterLogsColumns, + handlers: handlers, + } +} + +// DB retrieves and returns the underlying raw database management object of the current DAO. +func (dao *AudContentFilterLogsDao) DB() gdb.DB { + return g.DB(dao.group) +} + +// Table returns the table name of the current DAO. +func (dao *AudContentFilterLogsDao) Table() string { + return dao.table +} + +// Columns returns all column names of the current DAO. +func (dao *AudContentFilterLogsDao) Columns() AudContentFilterLogsColumns { + return dao.columns +} + +// Group returns the database configuration group name of the current DAO. +func (dao *AudContentFilterLogsDao) Group() string { + return dao.group +} + +// Ctx creates and returns a Model for the current DAO. It automatically sets the context for the current operation. +func (dao *AudContentFilterLogsDao) Ctx(ctx context.Context) *gdb.Model { + model := dao.DB().Model(dao.table) + for _, handler := range dao.handlers { + model = handler(model) + } + return model.Safe().Ctx(ctx) +} + +// Transaction wraps the transaction logic using function f. +// It rolls back the transaction and returns the error if function f returns a non-nil error. +// It commits the transaction and returns nil if function f returns nil. +// +// Note: Do not commit or roll back the transaction in function f, +// as it is automatically handled by this function. +func (dao *AudContentFilterLogsDao) Transaction(ctx context.Context, f func(ctx context.Context, tx gdb.TX) error) (err error) { + return dao.Ctx(ctx).Transaction(ctx, f) +} diff --git a/internal/dao/internal/aud_request_logs.go b/internal/dao/internal/aud_request_logs.go index 58602b2..f62e8d1 100644 --- a/internal/dao/internal/aud_request_logs.go +++ b/internal/dao/internal/aud_request_logs.go @@ -21,60 +21,70 @@ type AudRequestLogsDao struct { // AudRequestLogsColumns defines and stores column names for the table aud_request_logs. type AudRequestLogsColumns struct { - Id string // 主键ID - TenantId string // 租户ID - UserId string // 用户ID - ApiKeyId string // 使用的 API Key ID - RequestId string // 请求唯一ID（关联全链路追踪） - Method string // HTTP 方法（GET/POST/PUT/DELETE） - Path string // 请求路径 - QueryParams string // 查询参数（URL Query String） - StatusCode string // HTTP 响应状态码 - ClientIp string // 客户端 IP - UserAgent string // 客户端 User-Agent - RequestBody string // 请求体（敏感字段脱敏后存储） - ResponseBody string // 响应体（截断后存储） - LatencyMs string // 请求延迟（毫秒） - AuditLevel string // 审计级别：full（完整记录）/ masked（脱敏记录）/ question_only（仅记录提问）/ none（不记录） - CreatedAt string // 创建时间 - UpdatedAt string // 更新时间 - TenantRequestBody string // 租户级请求体（按租户审计级别处理） - TenantResponseBody string // 租户级响应体（按租户审计级别处理） - TenantAuditLevel string // 租户审计级别：full/full_text/masked/question_only/none - ProjectId string // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） - FirstTokenMs string // 首个 Token 出现的用时（毫秒），仅流式请求有值 - RequestHeaders string // 请求头信息（仅审计级别为 all 时记录，管理后台调试用） - ResponseHeaders string // 响应头信息（仅审计级别为 all 时记录，管理后台调试用） - ForwardingTrace string // 请求转发路径追踪（仅管理员可见） + Id string // 主键ID + TenantId string // 租户ID + UserId string // 用户ID + ApiKeyId string // 使用的 API Key ID + RequestId string // 请求唯一ID（关联全链路追踪） + Method string // HTTP 方法（GET/POST/PUT/DELETE） + Path string // 请求路径 + QueryParams string // 查询参数（URL Query String） + StatusCode string // HTTP 响应状态码 + ClientIp string // 客户端 IP + UserAgent string // 客户端 User-Agent + RequestBody string // 请求体（敏感字段脱敏后存储） + ResponseBody string // 响应体（截断后存储） + LatencyMs string // 请求延迟（毫秒） + AuditLevel string // 审计级别：full（完整记录）/ masked（脱敏记录）/ question_only（仅记录提问）/ none（不记录） + CreatedAt string // 创建时间 + UpdatedAt string // 更新时间 + TenantRequestBody string // 租户级请求体（按租户审计级别处理） + TenantResponseBody string // 租户级响应体（按租户审计级别处理） + TenantAuditLevel string // 租户审计级别：full/full_text/masked/question_only/none + ProjectId string // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） + FirstTokenMs string // 首个 Token 出现的用时（毫秒），仅流式请求有值 + RequestHeaders string // 请求头信息（仅审计级别为 all 时记录，管理后台调试用） + ResponseHeaders string // 响应头信息（仅审计级别为 all 时记录，管理后台调试用） + ForwardingTrace string // 请求转发路径追踪（仅管理员可见） + TaskId string // 异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id + TaskStatus string // 异步任务终态：SUCCESS / FAILURE + TaskResult string // 异步任务完成时上游返回的原始响应体 + TaskUpstreamHeaders string // 异步任务完成时上游返回的响应头（仅审计级别为 full 时记录） + TaskCompletedAt string // 异步任务达到终态的时间 } // audRequestLogsColumns holds the columns for the table aud_request_logs. var audRequestLogsColumns = AudRequestLogsColumns{ - Id: "id", - TenantId: "tenant_id", - UserId: "user_id", - ApiKeyId: "api_key_id", - RequestId: "request_id", - Method: "method", - Path: "path", - QueryParams: "query_params", - StatusCode: "status_code", - ClientIp: "client_ip", - UserAgent: "user_agent", - RequestBody: "request_body", - ResponseBody: "response_body", - LatencyMs: "latency_ms", - AuditLevel: "audit_level", - CreatedAt: "created_at", - UpdatedAt: "updated_at", - TenantRequestBody: "tenant_request_body", - TenantResponseBody: "tenant_response_body", - TenantAuditLevel: "tenant_audit_level", - ProjectId: "project_id", - FirstTokenMs: "first_token_ms", - RequestHeaders: "request_headers", - ResponseHeaders: "response_headers", - ForwardingTrace: "forwarding_trace", + Id: "id", + TenantId: "tenant_id", + UserId: "user_id", + ApiKeyId: "api_key_id", + RequestId: "request_id", + Method: "method", + Path: "path", + QueryParams: "query_params", + StatusCode: "status_code", + ClientIp: "client_ip", + UserAgent: "user_agent", + RequestBody: "request_body", + ResponseBody: "response_body", + LatencyMs: "latency_ms", + AuditLevel: "audit_level", + CreatedAt: "created_at", + UpdatedAt: "updated_at", + TenantRequestBody: "tenant_request_body", + TenantResponseBody: "tenant_response_body", + TenantAuditLevel: "tenant_audit_level", + ProjectId: "project_id", + FirstTokenMs: "first_token_ms", + RequestHeaders: "request_headers", + ResponseHeaders: "response_headers", + ForwardingTrace: "forwarding_trace", + TaskId: "task_id", + TaskStatus: "task_status", + TaskResult: "task_result", + TaskUpstreamHeaders: "task_upstream_headers", + TaskCompletedAt: "task_completed_at", } // NewAudRequestLogsDao creates and returns a new DAO object for table data access. diff --git a/internal/dao/internal/bil_usage_logs.go b/internal/dao/internal/bil_usage_logs.go index 2f89668..3c5cfdb 100644 --- a/internal/dao/internal/bil_usage_logs.go +++ b/internal/dao/internal/bil_usage_logs.go @@ -48,7 +48,7 @@ type BilUsageLogsColumns struct { ActualCost string // 实际扣除费用（含折扣后） RequestedModel string // 用户请求的模型名 UpstreamModel string // 上游实际模型名（模型映射后） - RequestType string // 请求类型: 1=sync, 2=stream, 3=websocket + RequestType string // 请求类型: 1=sync, 2=stream, 3=async, 4=websocket UserAgent string // 客户端 User-Agent FirstTokenMs string // 首 token 延迟（毫秒） ServiceTier string // 服务等级 (default/flex等) @@ -77,6 +77,7 @@ type BilUsageLogsColumns struct { UpstreamEndpoint string // 上游实际请求路径 BillingSnapshot string // 完整计费计算过程快照（JSONB） ProjectId string // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） + TaskId string // 异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id，普通请求为空 } // bilUsageLogsColumns holds the columns for the table bil_usage_logs. @@ -137,6 +138,7 @@ var bilUsageLogsColumns = BilUsageLogsColumns{ UpstreamEndpoint: "upstream_endpoint", BillingSnapshot: "billing_snapshot", ProjectId: "project_id", + TaskId: "task_id", } // NewBilUsageLogsDao creates and returns a new DAO object for table data access. diff --git a/internal/dao/internal/tsk_model_tasks.go b/internal/dao/internal/tsk_model_tasks.go index 8752f84..936eb1a 100644 --- a/internal/dao/internal/tsk_model_tasks.go +++ b/internal/dao/internal/tsk_model_tasks.go @@ -23,6 +23,7 @@ type TskModelTasksDao struct { type TskModelTasksColumns struct { Id string // PublicTaskId string // + RequestId string // Platform string // Action string // Status string // @@ -51,6 +52,7 @@ type TskModelTasksColumns struct { var tskModelTasksColumns = TskModelTasksColumns{ Id: "id", PublicTaskId: "public_task_id", + RequestId: "request_id", Platform: "platform", Action: "action", Status: "status", diff --git a/internal/handler/relay/task.go b/internal/handler/relay/task.go index 6c19f61..fcc16f5 100644 --- a/internal/handler/relay/task.go +++ b/internal/handler/relay/task.go @@ -12,7 +12,7 @@ import ( "github.com/qianfree/team-api/internal/logic/relay" "github.com/qianfree/team-api/internal/logic/task" "github.com/qianfree/team-api/internal/middleware" - "github.com/qianfree/team-api/relay/common" + relay_common "github.com/qianfree/team-api/relay/common" relay_handler "github.com/qianfree/team-api/relay/handler" ) @@ -36,6 +36,7 @@ func HandleTaskSubmit(r *ghttp.Request) { TenantID: middleware.GetTenantID(r.Context()), UserID: middleware.GetUserID(r.Context()), ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), RequestID: r.GetCtxVar("RequestId").String(), Writer: r.Response.Writer, Scope: r.GetCtxVar("ApiKeyScope").String(), @@ -49,11 +50,11 @@ func HandleTaskSubmit(r *ghttp.Request) { errType := "server_error" errMsg := err.Error() - if err == common.ErrTenantModelNotEnabled { + if err == relay_common.ErrTenantModelNotEnabled { statusCode = 403 errType = "permission_denied" errMsg = "当前租户未启用该模型，请联系管理员" - } else if err == common.ErrChannelUnavailable { + } else if err == relay_common.ErrChannelUnavailable { statusCode = 503 errType = "server_error" errMsg = "该模型暂无可用的渠道，请稍后重试或联系管理员" @@ -75,30 +76,7 @@ func HandleTaskSubmit(r *ghttp.Request) { ) // 异步记录审计日志 - tenantID := rc.TenantID - userID := rc.UserID - apiKeyID := rc.ApiKeyID - requestID := rc.RequestID - path := r.URL.Path - clientIP := rc.ClientIP - userAgent := r.Header.Get("User-Agent") - statusCode := capture.StatusCode() - responseBody := capture.Body() - go func() { - relayDataProvider.RecordAudit(context.Background(), &common.AuditRecord{ - TenantID: tenantID, - UserID: userID, - ApiKeyID: apiKeyID, - RequestID: requestID, - Method: "POST", - Path: path, - StatusCode: statusCode, - ClientIP: clientIP, - UserAgent: userAgent, - RequestBody: string(body), - ResponseBody: responseBody, - }) - }() + go recordTaskSubmitAudit(r, rc, capture, body) } // HandleTaskFetch 处理异步任务查询（GET /v1/video/generations/:id, GET /suno/fetch/:id） @@ -115,6 +93,7 @@ func HandleTaskFetch(r *ghttp.Request) { TenantID: middleware.GetTenantID(r.Context()), UserID: middleware.GetUserID(r.Context()), ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), RequestID: r.GetCtxVar("RequestId").String(), Writer: r.Response.Writer, } @@ -132,6 +111,7 @@ func HandleSunoFetchBatch(r *ghttp.Request) { TenantID: middleware.GetTenantID(r.Context()), UserID: middleware.GetUserID(r.Context()), ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), RequestID: r.GetCtxVar("RequestId").String(), Writer: r.Response.Writer, } @@ -158,13 +138,13 @@ func extractTaskID(r *ghttp.Request) string { } // selectTaskChannel 为异步任务选择渠道（复用实时请求的渠道调度逻辑） -func selectTaskChannel(r *ghttp.Request, body []byte) (*common.ChannelMeta, error) { +func selectTaskChannel(r *ghttp.Request, body []byte) (*relay_common.ChannelMeta, error) { // 从请求体提取模型名 var req struct { Model string `json:"model"` } if err := json.Unmarshal(body, &req); err != nil || req.Model == "" { - return nil, common.ErrChannelUnavailable + return nil, relay_common.ErrChannelUnavailable } ctx := r.Context() @@ -176,7 +156,7 @@ func selectTaskChannel(r *ghttp.Request, body []byte) (*common.ChannelMeta, erro return nil, err } - return &common.ChannelMeta{ + return &relay_common.ChannelMeta{ ChannelID: selection.ChannelID, ChannelType: selection.ChannelType, ChannelName: selection.ChannelName, @@ -202,6 +182,7 @@ func HandleMjSubmit(r *ghttp.Request) { TenantID: middleware.GetTenantID(r.Context()), UserID: middleware.GetUserID(r.Context()), ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), RequestID: r.GetCtxVar("RequestId").String(), Writer: r.Response.Writer, Scope: r.GetCtxVar("ApiKeyScope").String(), @@ -216,10 +197,15 @@ func HandleMjSubmit(r *ghttp.Request) { return } + capture := NewResponseCaptureWriter(r.Response.Writer) + rc.Writer = capture + relay_handler.HandleTaskSubmit( r.Context(), body, r.URL.Path, r.Header, rc, taskDataProvider, taskBillingProvider, channelMeta, ) + + go recordTaskSubmitAudit(r, rc, capture, body) } // HandleMjFetch 处理 Midjourney 任务查询（GET /mj/task/:id/fetch） @@ -239,6 +225,7 @@ func HandleMjFetch(r *ghttp.Request) { TenantID: middleware.GetTenantID(r.Context()), UserID: middleware.GetUserID(r.Context()), ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), RequestID: r.GetCtxVar("RequestId").String(), Writer: r.Response.Writer, } @@ -263,6 +250,7 @@ func HandleMjImage(r *ghttp.Request) { TenantID: middleware.GetTenantID(r.Context()), UserID: middleware.GetUserID(r.Context()), ApiKeyID: middleware.GetApiKeyID(r.Context()), + ProjectID: middleware.GetProjectID(r.Context()), RequestID: r.GetCtxVar("RequestId").String(), Writer: r.Response.Writer, } @@ -281,3 +269,27 @@ func extractMjTaskID(r *ghttp.Request) string { } return "" } + +// recordTaskSubmitAudit 记录异步任务提交的审计日志 +func recordTaskSubmitAudit(r *ghttp.Request, rc *relay_handler.TaskRelayContext, capture *ResponseCaptureWriter, body []byte) { + relayDataProvider.RecordAudit(context.Background(), &relay_common.AuditRecord{ + TenantID: rc.TenantID, + UserID: rc.UserID, + ApiKeyID: rc.ApiKeyID, + ProjectID: rc.ProjectID, + RequestID: rc.RequestID, + Method: "POST", + Path: r.URL.Path, + QueryParams: r.URL.RawQuery, + StatusCode: capture.StatusCode(), + ClientIP: rc.ClientIP, + UserAgent: r.Header.Get("User-Agent"), + RequestBody: string(body), + ResponseBody: capture.Body(), + IsStream: false, + RequestHeaders: captureRequestHeaders(r), + ResponseHeaders: capture.ResponseHeaders(), + TaskID: rc.TaskID, + TaskStatus: "SUBMITTED", + }) +} diff --git a/internal/logic/admin/audit.go b/internal/logic/admin/audit.go index 44242a5..7a009be 100644 --- a/internal/logic/admin/audit.go +++ b/internal/logic/admin/audit.go @@ -251,6 +251,10 @@ func (s *sAdmin) ListRequestAuditLogs(ctx context.Context, req *v1.RequestAuditL conditions = append(conditions, "a.request_id = ?") args = append(args, req.RequestId) } + if req.TaskId != "" { + conditions = append(conditions, "a.task_id = ?") + args = append(args, req.TaskId) + } if req.Method != "" { conditions = append(conditions, "a.method = ?") args = append(args, req.Method) diff --git a/internal/logic/admin/dashboard.go b/internal/logic/admin/dashboard.go index a785fe7..04df3d9 100644 --- a/internal/logic/admin/dashboard.go +++ b/internal/logic/admin/dashboard.go @@ -245,7 +245,7 @@ func (s *sAdmin) GetAllUsageLogs(ctx context.Context, req *v1.AdminUsageLogListR } dataSQL := fmt.Sprintf( - `SELECT u.id, u.tenant_id, COALESCE(tn.name, '') AS tenant_name, u.user_id, COALESCE(t.username, '') AS username, u.project_id, COALESCE(p.name, '') AS project_name, u.api_key_id, COALESCE(ak.name, '') AS api_key_name, u.channel_id, u.channel_name, u.channel_type, u.model_name, u.requested_model, u.upstream_model, u.relay_mode, u.request_type, u.input_tokens, u.output_tokens, u.cache_creation_tokens, u.cache_read_tokens, u.cache_creation_5m_tokens, u.cache_creation_1h_tokens, u.reasoning_tokens, u.audio_input_tokens, u.audio_output_tokens, u.image_output_tokens, u.input_cost, u.output_cost, u.cache_creation_cost, u.cache_read_cost, u.total_cost, u.actual_cost, u.currency, u.billing_mode, u.billing_source, u.rate_multiplier, u.latency_ms, u.first_token_ms, u.status, u.error_message, u.retry_index, u.client_ip, u.user_agent, u.service_tier, u.reasoning_effort, u.stream_end_reason, u.image_count, u.image_size, u.pre_deduct_amount, u.refund_amount, u.supplement_amount, u.billing_summary, u.billing_snapshot, u.inbound_endpoint, u.request_id, u.created_at + `SELECT u.id, u.tenant_id, COALESCE(tn.name, '') AS tenant_name, u.user_id, COALESCE(t.username, '') AS username, u.project_id, COALESCE(p.name, '') AS project_name, u.api_key_id, COALESCE(ak.name, '') AS api_key_name, u.channel_id, u.channel_name, u.channel_type, u.model_name, u.requested_model, u.upstream_model, u.relay_mode, u.request_type, u.input_tokens, u.output_tokens, u.cache_creation_tokens, u.cache_read_tokens, u.cache_creation_5m_tokens, u.cache_creation_1h_tokens, u.reasoning_tokens, u.audio_input_tokens, u.audio_output_tokens, u.image_output_tokens, u.input_cost, u.output_cost, u.cache_creation_cost, u.cache_read_cost, u.total_cost, u.actual_cost, u.currency, u.billing_mode, u.billing_source, u.rate_multiplier, u.latency_ms, u.first_token_ms, u.status, u.error_message, u.retry_index, u.client_ip, u.user_agent, u.service_tier, u.reasoning_effort, u.stream_end_reason, u.image_count, u.image_size, u.pre_deduct_amount, u.refund_amount, u.supplement_amount, u.billing_summary, u.billing_snapshot, u.inbound_endpoint, u.request_id, u.task_id, u.created_at FROM %s%s ORDER BY u.created_at DESC LIMIT %d OFFSET %d`, fromClause, where, pageSize, (page-1)*pageSize, ) diff --git a/internal/logic/admin/task_management.go b/internal/logic/admin/task_management.go index f4e5cfb..420299d 100644 --- a/internal/logic/admin/task_management.go +++ b/internal/logic/admin/task_management.go @@ -23,6 +23,9 @@ func (s *sAdmin) TaskList(ctx context.Context, req *v1.TaskListReq) (*v1.TaskLis if req.Platform != "" { m = m.Where("platform", req.Platform) } + if req.PublicTaskID != "" { + m = m.Where("public_task_id", req.PublicTaskID) + } var tasks []struct { Id int64 `json:"id"` diff --git a/internal/logic/relay/provider.go b/internal/logic/relay/provider.go index c93b12a..a5ebb55 100644 --- a/internal/logic/relay/provider.go +++ b/internal/logic/relay/provider.go @@ -366,6 +366,7 @@ func buildUsageLogDO(record *common.UsageRecord) do.BilUsageLogs { BillingSnapshot: jsonNullIfEmpty(record.BillingSnapshot), BillingSummary: record.BillingSummary, + TaskId: record.TaskID, } } @@ -448,6 +449,15 @@ func (p *DataProviderImpl) RecordAudit(ctx context.Context, record *common.Audit "response_headers": responseHeadersJSON, "forwarding_trace": forwardingTraceJSON, } + + // 异步任务字段 + if record.TaskID != "" { + insertData["task_id"] = record.TaskID + } + if record.TaskStatus != "" { + insertData["task_status"] = record.TaskStatus + } + _, insertErr := dao.AudRequestLogs.Ctx(bgCtx).Data(insertData).Insert() if insertErr != nil { g.Log().Errorf(bgCtx, @@ -457,6 +467,63 @@ func (p *DataProviderImpl) RecordAudit(ctx context.Context, record *common.Audit }() } +// UpdateTaskAudit implements DataProvider.UpdateTaskAudit +// 异步任务完成时更新提交阶段写入的审计记录 +func (p *DataProviderImpl) UpdateTaskAudit(ctx context.Context, record *common.AuditRecord) { + go func() { + defer func() { + if r := recover(); r != nil { + g.Log().Errorf(context.Background(), + "update task audit panic: task_id=%s panic=%v", + record.TaskID, r) + } + }() + + if record.TaskID == "" { + return + } + + bgCtx := context.Background() + globalLevel, tenantLevel := lcommon.GetAuditLevels(bgCtx, record.TenantID) + + // 按审计级别处理任务结果 + taskResult := record.TaskResult + if globalLevel == lcommon.AuditLevelNone && tenantLevel == lcommon.AuditLevelNone { + taskResult = "" + } else if globalLevel == lcommon.AuditLevelMasked { + taskResult = lcommon.MaskSensitiveData(taskResult) + } + taskResult = truncateBody(taskResult, 65536) + + // 上游响应头仅在 full 级别记录 + upstreamHeadersJSON := "null" + if globalLevel == lcommon.AuditLevelFull && record.TaskUpstreamHeaders != nil { + b, _ := json.Marshal(record.TaskUpstreamHeaders) + upstreamHeadersJSON = string(b) + } + + updateData := g.Map{ + "task_status": record.TaskStatus, + "task_result": taskResult, + "task_upstream_headers": upstreamHeadersJSON, + "updated_at": time.Now(), + } + if record.TaskCompletedAt != nil { + updateData["task_completed_at"] = record.TaskCompletedAt + } + + _, err := dao.AudRequestLogs.Ctx(bgCtx). + Where("task_id", record.TaskID). + Data(updateData). + Update() + if err != nil { + g.Log().Errorf(bgCtx, + "update task audit failed: task_id=%s err=%v", + record.TaskID, err) + } + }() +} + func truncateBody(s string, maxLen int) string { if len(s) <= maxLen { return s diff --git a/internal/logic/task/async_polling.go b/internal/logic/task/async_polling.go index a07c409..efa7170 100644 --- a/internal/logic/task/async_polling.go +++ b/internal/logic/task/async_polling.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "io" + "net/http" "time" "github.com/gogf/gf/v2/frame/g" @@ -103,6 +104,9 @@ func handleTimedOutTasks(ctx context.Context) { } } g.Log().Infof(ctx, "poll: task %s timed out", t.PublicTaskID) + + // 更新审计记录 + recordTaskCompletionAudit(t, "TIMEOUT", "", nil) } } @@ -195,6 +199,12 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co if task.StartTime == nil { task.StartTime = &now } + // 保留上游返回的 token 用量 + if taskInfo.TotalTokens > 0 { + task.CompletionTokens = taskInfo.CompletionTokens + task.PromptTokens = taskInfo.PromptTokens + task.TotalTokens = taskInfo.TotalTokens + } } if err := DefaultAsyncProvider.UpdateTaskCAS(ctx, task, oldStatus); err != nil { @@ -231,6 +241,9 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co // 记录用量日志 recordTaskUsage(task, true, "") + // 更新审计记录 + recordTaskCompletionAudit(task, "SUCCESS", string(body), upstreamRespHeaders(resp)) + } else if taskInfo.Status == common.TaskStatusFailure { // 退还预扣费用 if task.PreDeductAmount > 0 && !task.BillingSettled { @@ -246,6 +259,9 @@ func pollSingleTask(ctx context.Context, adaptor common.TaskAdaptor, channel *co // 记录用量日志 recordTaskUsage(task, false, task.FailReason) + + // 更新审计记录 + recordTaskCompletionAudit(task, "FAILURE", string(body), upstreamRespHeaders(resp)) } } @@ -262,17 +278,54 @@ func recordTaskUsage(task *common.AsyncTask, success bool, errMsg string) { } relay.NewDataProvider().RecordUsage(context.Background(), &common.UsageRecord{ - TenantID: task.TenantID, - UserID: task.UserID, - ApiKeyID: task.ApiKeyID, - ChannelID: task.ChannelID, - ModelName: task.ModelName, - RelayMode: int(constant.RelayModeVideoGenerations), - LatencyMs: float64(latencyMs), - IsStream: false, - Success: success, - RequestID: task.PublicTaskID, - Status: status, - ErrorMessage: errMsg, + TenantID: task.TenantID, + UserID: task.UserID, + ApiKeyID: task.ApiKeyID, + ChannelID: task.ChannelID, + ModelName: task.ModelName, + RelayMode: int(constant.RelayModeVideoGenerations), + RequestType: 3, // async + LatencyMs: float64(latencyMs), + IsStream: false, + Success: success, + RequestID: task.RequestID, + Status: status, + ErrorMessage: errMsg, + PromptTokens: task.PromptTokens, + CompletionTokens: task.CompletionTokens, + TotalTokens: task.TotalTokens, + TotalCost: task.ActualCost, + ActualCost: task.ActualCost, + PreDeductAmount: task.PreDeductAmount, + BillingMode: "per_request", + BillingSource: "task", + TaskID: task.PublicTaskID, }) } + +// recordTaskCompletionAudit 更新提交阶段写入的审计记录，补充异步任务最终结果 +func recordTaskCompletionAudit(task *common.AsyncTask, status string, resultBody string, upstreamHeaders map[string]string) { + now := time.Now() + relay.NewDataProvider().UpdateTaskAudit(context.Background(), &common.AuditRecord{ + TenantID: task.TenantID, + TaskID: task.PublicTaskID, + TaskStatus: status, + TaskResult: resultBody, + TaskUpstreamHeaders: upstreamHeaders, + TaskCompletedAt: &now, + }) +} + +// upstreamRespHeaders 从上游 HTTP 响应中提取响应头 +func upstreamRespHeaders(resp *http.Response) map[string]string { + if resp == nil || resp.Header == nil { + return nil + } + headers := make(map[string]string) + for k, vals := range resp.Header { + if len(vals) > 0 { + headers[k] = vals[0] + } + } + return headers +} diff --git a/internal/logic/task/async_provider.go b/internal/logic/task/async_provider.go index 57b0150..a40e42a 100644 --- a/internal/logic/task/async_provider.go +++ b/internal/logic/task/async_provider.go @@ -25,6 +25,7 @@ var DefaultAsyncProvider = &AsyncProvider{} func (p *AsyncProvider) CreateTask(ctx context.Context, task *common.AsyncTask) error { _, err := dao.TskModelTasks.Ctx(ctx).Insert(map[string]any{ "public_task_id": task.PublicTaskID, + "request_id": task.RequestID, "platform": task.Platform, "action": task.Action, "status": task.Status, @@ -105,6 +106,7 @@ func (p *AsyncProvider) GetTaskByPublicID(ctx context.Context, publicTaskID stri var row struct { ID int64 `json:"id"` PublicTaskID string `json:"public_task_id"` + RequestId string `json:"request_id"` Platform string `json:"platform"` Action string `json:"action"` Status string `json:"status"` @@ -140,6 +142,7 @@ func (p *AsyncProvider) GetTaskByPublicID(ctx context.Context, publicTaskID stri return &common.AsyncTask{ ID: row.ID, PublicTaskID: row.PublicTaskID, + RequestID: row.RequestId, Platform: row.Platform, Action: row.Action, Status: row.Status, @@ -170,6 +173,7 @@ func (p *AsyncProvider) GetTaskByPublicIDAndUser(ctx context.Context, publicTask var row struct { ID int64 `json:"id"` PublicTaskID string `json:"public_task_id"` + RequestId string `json:"request_id"` Platform string `json:"platform"` Action string `json:"action"` Status string `json:"status"` @@ -205,6 +209,7 @@ func (p *AsyncProvider) GetTaskByPublicIDAndUser(ctx context.Context, publicTask return &common.AsyncTask{ ID: row.ID, PublicTaskID: row.PublicTaskID, + RequestID: row.RequestId, Platform: row.Platform, Action: row.Action, Status: row.Status, @@ -234,6 +239,7 @@ func (p *AsyncProvider) GetNonTerminalTasks(ctx context.Context, limit int) ([]* var rows []struct { ID int64 `json:"id"` PublicTaskID string `json:"public_task_id"` + RequestId string `json:"request_id"` Platform string `json:"platform"` Action string `json:"action"` Status string `json:"status"` @@ -266,6 +272,7 @@ func (p *AsyncProvider) GetNonTerminalTasks(ctx context.Context, limit int) ([]* tasks = append(tasks, &common.AsyncTask{ ID: r.ID, PublicTaskID: r.PublicTaskID, + RequestID: r.RequestId, Platform: r.Platform, Action: r.Action, Status: r.Status, @@ -294,6 +301,7 @@ func (p *AsyncProvider) GetTimedOutTasks(ctx context.Context, cutoffUnix int64, var rows []struct { ID int64 `json:"id"` PublicTaskID string `json:"public_task_id"` + RequestId string `json:"request_id"` Platform string `json:"platform"` Status string `json:"status"` TenantID int64 `json:"tenant_id"` @@ -320,6 +328,7 @@ func (p *AsyncProvider) GetTimedOutTasks(ctx context.Context, cutoffUnix int64, tasks = append(tasks, &common.AsyncTask{ ID: r.ID, PublicTaskID: r.PublicTaskID, + RequestID: r.RequestId, Platform: r.Platform, Status: r.Status, TenantID: r.TenantID, diff --git a/internal/logic/tenant/audit_config.go b/internal/logic/tenant/audit_config.go index 7485132..1e2b66e 100644 --- a/internal/logic/tenant/audit_config.go +++ b/internal/logic/tenant/audit_config.go @@ -173,6 +173,10 @@ func (s *sTenant) TenantRequestAuditLogs(ctx context.Context, req *v1.TenantRequ conditions = append(conditions, "a.request_id = ?") args = append(args, req.RequestId) } + if req.TaskId != "" { + conditions = append(conditions, "a.task_id = ?") + args = append(args, req.TaskId) + } if req.Path != "" { conditions = append(conditions, "a.path LIKE ?") args = append(args, "%"+req.Path+"%") @@ -204,7 +208,7 @@ func (s *sTenant) TenantRequestAuditLogs(ctx context.Context, req *v1.TenantRequ } dataSQL := fmt.Sprintf( - `SELECT a.id, a.request_id, COALESCE(t.username, '') AS username, a.project_id, COALESCE(p.name, '') AS project_name, a.user_id, a.method, a.path, a.query_params, a.status_code, a.client_ip, a.user_agent, a.latency_ms, a.first_token_ms, a.tenant_audit_level AS audit_level, a.created_at + `SELECT a.id, a.request_id, COALESCE(t.username, '') AS username, a.project_id, COALESCE(p.name, '') AS project_name, a.user_id, a.method, a.path, a.query_params, a.status_code, a.client_ip, a.user_agent, a.latency_ms, a.first_token_ms, a.tenant_audit_level AS audit_level, a.task_id, a.task_status, a.task_completed_at, a.created_at FROM %s WHERE %s ORDER BY a.created_at DESC LIMIT %d OFFSET %d`, fromClause, where, pageSize, (page-1)*pageSize, ) @@ -263,5 +267,7 @@ func (s *sTenant) TenantRequestAuditLogDetail(ctx context.Context, req *v1.Tenan // 请求头/响应头仅管理后台可见，租户不可见 delete(detail, "request_headers") delete(detail, "response_headers") + // 异步任务上游响应头仅管理后台可见 + delete(detail, "task_upstream_headers") return &v1.TenantRequestAuditLogDetailRes{Data: detail}, nil } diff --git a/internal/logic/tenant/personal_dashboard.go b/internal/logic/tenant/personal_dashboard.go index 987c22c..815adec 100644 --- a/internal/logic/tenant/personal_dashboard.go +++ b/internal/logic/tenant/personal_dashboard.go @@ -76,8 +76,8 @@ func (s *sTenant) PersonalDashboard(ctx context.Context, req *v1.PersonalDashboa errorRate := v1.PersonalErrorRate{} reqTypeItems := []v1.PersonalReqTypeItem{} - reqTypeLabels := map[int]string{1: "同步", 2: "流式", 3: "WebSocket"} - reqTypeNames := map[int]string{1: "sync", 2: "stream", 3: "websocket"} + reqTypeLabels := map[int]string{1: "同步", 2: "流式", 3: "异步", 4: "WebSocket"} + reqTypeNames := map[int]string{1: "sync", 2: "stream", 3: "async", 4: "websocket"} totalReqs := 0 for _, r := range reqTypeRows { errorRate.Total += r.Total diff --git a/internal/logic/tenant/task.go b/internal/logic/tenant/task.go new file mode 100644 index 0000000..9295905 --- /dev/null +++ b/internal/logic/tenant/task.go @@ -0,0 +1,168 @@ +package tenant + +import ( + "context" + + v1 "github.com/qianfree/team-api/api/tenant/v1" + "github.com/qianfree/team-api/internal/dao" + "github.com/qianfree/team-api/internal/logic/common" +) + +// TenantTaskList 租户异步任务列表 +// owner/admin 可查看租户所有任务，member 只能查看自己的任务 +func (s *sTenant) TenantTaskList(ctx context.Context, req *v1.TenantTaskListReq) (*v1.TenantTaskListRes, error) { + tenantID := ctxTenantID(ctx) + role := ctxUserRole(ctx) + userID := ctxUserID(ctx) + + m := dao.TskModelTasks.Ctx(ctx). + LeftJoin("tnt_users u", "u.id = tsk_model_tasks.user_id AND u.tenant_id = tsk_model_tasks.tenant_id"). + Fields("tsk_model_tasks.*, COALESCE(u.username, '') as username"). + Where("tsk_model_tasks.tenant_id", tenantID). + OrderDesc("tsk_model_tasks.id") + + // member 只能看自己的任务 + if role == "member" { + m = m.Where("tsk_model_tasks.user_id", userID) + } + + if req.Status != "" { + m = m.Where("tsk_model_tasks.status", req.Status) + } + if req.Platform != "" { + m = m.Where("tsk_model_tasks.platform", req.Platform) + } + if req.PublicTaskID != "" { + m = m.Where("tsk_model_tasks.public_task_id", req.PublicTaskID) + } + + var tasks []struct { + Id int64 `json:"id"` + PublicTaskId string `json:"public_task_id"` + Platform string `json:"platform"` + Action string `json:"action"` + Status string `json:"status"` + Progress string `json:"progress"` + ModelName string `json:"model_name"` + FailReason string `json:"fail_reason"` + PreDeductAmount float64 `json:"pre_deduct_amount"` + ActualCost float64 `json:"actual_cost"` + BillingSettled bool `json:"billing_settled"` + ResultUrl string `json:"result_url"` + Username string `json:"username"` + SubmitTime *string `json:"submit_time"` + FinishTime *string `json:"finish_time"` + CreatedAt *string `json:"created_at"` + } + + var total int + err := m.Page(req.Page, req.PageSize).ScanAndCount(&tasks, &total, false) + if err != nil { + return nil, err + } + + list := make([]v1.TenantTaskItem, 0, len(tasks)) + for _, t := range tasks { + item := v1.TenantTaskItem{ + ID: t.Id, + PublicTaskID: t.PublicTaskId, + Platform: t.Platform, + Action: t.Action, + Status: t.Status, + Progress: t.Progress, + ModelName: t.ModelName, + FailReason: t.FailReason, + PreDeductAmount: t.PreDeductAmount, + ActualCost: t.ActualCost, + BillingSettled: t.BillingSettled, + ResultURL: t.ResultUrl, + Username: t.Username, + } + if t.SubmitTime != nil { + item.SubmitTime = *t.SubmitTime + } + if t.FinishTime != nil { + item.FinishTime = *t.FinishTime + } + if t.CreatedAt != nil { + item.CreatedAt = *t.CreatedAt + } + list = append(list, item) + } + + return &v1.TenantTaskListRes{ + List: list, + Total: total, + Page: req.Page, + PageSize: req.PageSize, + }, nil +} + +// TenantTaskDetail 租户异步任务详情 +// owner/admin 可查看租户所有任务，member 只能查看自己的任务 +func (s *sTenant) TenantTaskDetail(ctx context.Context, req *v1.TenantTaskDetailReq) (*v1.TenantTaskDetailRes, error) { + tenantID := ctxTenantID(ctx) + role := ctxUserRole(ctx) + userID := ctxUserID(ctx) + + var task struct { + Id int64 `json:"id"` + PublicTaskId string `json:"public_task_id"` + Platform string `json:"platform"` + Action string `json:"action"` + Status string `json:"status"` + Progress string `json:"progress"` + ModelName string `json:"model_name"` + FailReason string `json:"fail_reason"` + PreDeductAmount float64 `json:"pre_deduct_amount"` + ActualCost float64 `json:"actual_cost"` + BillingSettled bool `json:"billing_settled"` + ResultUrl string `json:"result_url"` + UserId int64 `json:"user_id"` + SubmitTime *string `json:"submit_time"` + FinishTime *string `json:"finish_time"` + CreatedAt *string `json:"created_at"` + } + + m := dao.TskModelTasks.Ctx(ctx).Where("id", req.ID).Where("tenant_id", tenantID) + // member 只能看自己的任务 + if role == "member" { + m = m.Where("user_id", userID) + } + + err := m.Scan(&task) + if err != nil { + return nil, err + } + if task.Id == 0 { + return nil, common.NewNotFoundError("任务") + } + + item := v1.TenantTaskItem{ + ID: task.Id, + PublicTaskID: task.PublicTaskId, + Platform: task.Platform, + Action: task.Action, + Status: task.Status, + Progress: task.Progress, + ModelName: task.ModelName, + FailReason: task.FailReason, + PreDeductAmount: task.PreDeductAmount, + ActualCost: task.ActualCost, + BillingSettled: task.BillingSettled, + ResultURL: task.ResultUrl, + } + if task.SubmitTime != nil { + item.SubmitTime = *task.SubmitTime + } + if task.FinishTime != nil { + item.FinishTime = *task.FinishTime + } + if task.CreatedAt != nil { + item.CreatedAt = *task.CreatedAt + } + + return &v1.TenantTaskDetailRes{ + Task: item, + }, nil +} diff --git a/internal/model/do/aud_content_filter_logs.go b/internal/model/do/aud_content_filter_logs.go new file mode 100644 index 0000000..6204d32 --- /dev/null +++ b/internal/model/do/aud_content_filter_logs.go @@ -0,0 +1,29 @@ +// ================================================================================= +// Code generated and maintained by GoFrame CLI tool. DO NOT EDIT. +// ================================================================================= + +package do + +import ( + "github.com/gogf/gf/v2/frame/g" + "github.com/gogf/gf/v2/os/gtime" +) + +// AudContentFilterLogs is the golang structure of table aud_content_filter_logs for DAO operations like Where/Data. +type AudContentFilterLogs struct { + g.Meta `orm:"table:aud_content_filter_logs, do:true"` + Id any // 主键ID + TenantId any // 租户ID + UserId any // 用户ID + ApiKeyId any // API Key ID + RequestId any // 请求唯一ID + Method any // HTTP 方法 + Path any // 请求路径 + ClientIp any // 客户端 IP + FilterMode any // 过滤模式：log / replace / block + MatchedWords any // 命中的敏感词列表（JSONB 数组） + OriginalSnippet any // 原始请求体片段（截断存储，仅 replace 模式） + Blocked any // 是否被拦截（mode=block 时为 true） + CreatedAt *gtime.Time // 创建时间 + ProjectId any // 项目ID +} diff --git a/internal/model/do/aud_request_logs.go b/internal/model/do/aud_request_logs.go index b0cd83e..3cd909a 100644 --- a/internal/model/do/aud_request_logs.go +++ b/internal/model/do/aud_request_logs.go @@ -11,30 +11,35 @@ import ( // AudRequestLogs is the golang structure of table aud_request_logs for DAO operations like Where/Data. type AudRequestLogs struct { - g.Meta `orm:"table:aud_request_logs, do:true"` - Id any // 主键ID - TenantId any // 租户ID - UserId any // 用户ID - ApiKeyId any // 使用的 API Key ID - RequestId any // 请求唯一ID（关联全链路追踪） - Method any // HTTP 方法（GET/POST/PUT/DELETE） - Path any // 请求路径 - QueryParams any // 查询参数（URL Query String） - StatusCode any // HTTP 响应状态码 - ClientIp any // 客户端 IP - UserAgent any // 客户端 User-Agent - RequestBody any // 请求体（敏感字段脱敏后存储） - ResponseBody any // 响应体（截断后存储） - LatencyMs any // 请求延迟（毫秒） - AuditLevel any // 审计级别：full（完整记录）/ masked（脱敏记录）/ question_only（仅记录提问）/ none（不记录） - CreatedAt *gtime.Time // 创建时间 - UpdatedAt *gtime.Time // 更新时间 - TenantRequestBody any // 租户级请求体（按租户审计级别处理） - TenantResponseBody any // 租户级响应体（按租户审计级别处理） - TenantAuditLevel any // 租户审计级别：full/full_text/masked/question_only/none - ProjectId any // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） - FirstTokenMs any // 首个 Token 出现的用时（毫秒），仅流式请求有值 - RequestHeaders any // 请求头信息（仅审计级别为 all 时记录，管理后台调试用） - ResponseHeaders any // 响应头信息（仅审计级别为 all 时记录，管理后台调试用） - ForwardingTrace any // 请求转发路径追踪（仅管理员可见） + g.Meta `orm:"table:aud_request_logs, do:true"` + Id any // 主键ID + TenantId any // 租户ID + UserId any // 用户ID + ApiKeyId any // 使用的 API Key ID + RequestId any // 请求唯一ID（关联全链路追踪） + Method any // HTTP 方法（GET/POST/PUT/DELETE） + Path any // 请求路径 + QueryParams any // 查询参数（URL Query String） + StatusCode any // HTTP 响应状态码 + ClientIp any // 客户端 IP + UserAgent any // 客户端 User-Agent + RequestBody any // 请求体（敏感字段脱敏后存储） + ResponseBody any // 响应体（截断后存储） + LatencyMs any // 请求延迟（毫秒） + AuditLevel any // 审计级别：full（完整记录）/ masked（脱敏记录）/ question_only（仅记录提问）/ none（不记录） + CreatedAt *gtime.Time // 创建时间 + UpdatedAt *gtime.Time // 更新时间 + TenantRequestBody any // 租户级请求体（按租户审计级别处理） + TenantResponseBody any // 租户级响应体（按租户审计级别处理） + TenantAuditLevel any // 租户审计级别：full/full_text/masked/question_only/none + ProjectId any // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） + FirstTokenMs any // 首个 Token 出现的用时（毫秒），仅流式请求有值 + RequestHeaders any // 请求头信息（仅审计级别为 all 时记录，管理后台调试用） + ResponseHeaders any // 响应头信息（仅审计级别为 all 时记录，管理后台调试用） + ForwardingTrace any // 请求转发路径追踪（仅管理员可见） + TaskId any // 异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id + TaskStatus any // 异步任务终态：SUCCESS / FAILURE + TaskResult any // 异步任务完成时上游返回的原始响应体 + TaskUpstreamHeaders any // 异步任务完成时上游返回的响应头（仅审计级别为 full 时记录） + TaskCompletedAt *gtime.Time // 异步任务达到终态的时间 } diff --git a/internal/model/do/bil_usage_logs.go b/internal/model/do/bil_usage_logs.go index b6d3503..d45a98b 100644 --- a/internal/model/do/bil_usage_logs.go +++ b/internal/model/do/bil_usage_logs.go @@ -39,7 +39,7 @@ type BilUsageLogs struct { ActualCost any // 实际扣除费用（含折扣后） RequestedModel any // 用户请求的模型名 UpstreamModel any // 上游实际模型名（模型映射后） - RequestType any // 请求类型: 1=sync, 2=stream, 3=websocket + RequestType any // 请求类型: 1=sync, 2=stream, 3=async, 4=websocket UserAgent any // 客户端 User-Agent FirstTokenMs any // 首 token 延迟（毫秒） ServiceTier any // 服务等级 (default/flex等) @@ -68,4 +68,5 @@ type BilUsageLogs struct { UpstreamEndpoint any // 上游实际请求路径 BillingSnapshot any // 完整计费计算过程快照（JSONB） ProjectId any // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） + TaskId any // 异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id，普通请求为空 } diff --git a/internal/model/do/tsk_model_tasks.go b/internal/model/do/tsk_model_tasks.go index d4f7055..824cc50 100644 --- a/internal/model/do/tsk_model_tasks.go +++ b/internal/model/do/tsk_model_tasks.go @@ -14,6 +14,7 @@ type TskModelTasks struct { g.Meta `orm:"table:tsk_model_tasks, do:true"` Id any // PublicTaskId any // + RequestId any // 任务提交时的原始请求 ID Platform any // Action any // Status any // diff --git a/internal/model/entity/aud_content_filter_logs.go b/internal/model/entity/aud_content_filter_logs.go new file mode 100644 index 0000000..d5db34d --- /dev/null +++ b/internal/model/entity/aud_content_filter_logs.go @@ -0,0 +1,27 @@ +// ================================================================================= +// Code generated and maintained by GoFrame CLI tool. DO NOT EDIT. +// ================================================================================= + +package entity + +import ( + "github.com/gogf/gf/v2/os/gtime" +) + +// AudContentFilterLogs is the golang structure for table aud_content_filter_logs. +type AudContentFilterLogs struct { + Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID + TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID"` // 租户ID + UserId int64 `json:"user_id" orm:"user_id" description:"用户ID"` // 用户ID + ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:"API Key ID"` // API Key ID + RequestId string `json:"request_id" orm:"request_id" description:"请求唯一ID"` // 请求唯一ID + Method string `json:"method" orm:"method" description:"HTTP 方法"` // HTTP 方法 + Path string `json:"path" orm:"path" description:"请求路径"` // 请求路径 + ClientIp string `json:"client_ip" orm:"client_ip" description:"客户端 IP"` // 客户端 IP + FilterMode string `json:"filter_mode" orm:"filter_mode" description:"过滤模式：log / replace / block"` // 过滤模式：log / replace / block + MatchedWords string `json:"matched_words" orm:"matched_words" description:"命中的敏感词列表（JSONB 数组）"` // 命中的敏感词列表（JSONB 数组） + OriginalSnippet string `json:"original_snippet" orm:"original_snippet" description:"原始请求体片段（截断存储，仅 replace 模式）"` // 原始请求体片段（截断存储，仅 replace 模式） + Blocked bool `json:"blocked" orm:"blocked" description:"是否被拦截（mode=block 时为 true）"` // 是否被拦截（mode=block 时为 true） + CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 + ProjectId int64 `json:"project_id" orm:"project_id" description:"项目ID"` // 项目ID +} diff --git a/internal/model/entity/aud_request_logs.go b/internal/model/entity/aud_request_logs.go index 7ba392a..dc48e93 100644 --- a/internal/model/entity/aud_request_logs.go +++ b/internal/model/entity/aud_request_logs.go @@ -10,29 +10,34 @@ import ( // AudRequestLogs is the golang structure for table aud_request_logs. type AudRequestLogs struct { - Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID - TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID"` // 租户ID - UserId int64 `json:"user_id" orm:"user_id" description:"用户ID"` // 用户ID - ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:"使用的 API Key ID"` // 使用的 API Key ID - RequestId string `json:"request_id" orm:"request_id" description:"请求唯一ID（关联全链路追踪）"` // 请求唯一ID（关联全链路追踪） - Method string `json:"method" orm:"method" description:"HTTP 方法（GET/POST/PUT/DELETE）"` // HTTP 方法（GET/POST/PUT/DELETE） - Path string `json:"path" orm:"path" description:"请求路径"` // 请求路径 - QueryParams string `json:"query_params" orm:"query_params" description:"查询参数（URL Query String）"` // 查询参数（URL Query String） - StatusCode int `json:"status_code" orm:"status_code" description:"HTTP 响应状态码"` // HTTP 响应状态码 - ClientIp string `json:"client_ip" orm:"client_ip" description:"客户端 IP"` // 客户端 IP - UserAgent string `json:"user_agent" orm:"user_agent" description:"客户端 User-Agent"` // 客户端 User-Agent - RequestBody string `json:"request_body" orm:"request_body" description:"请求体（敏感字段脱敏后存储）"` // 请求体（敏感字段脱敏后存储） - ResponseBody string `json:"response_body" orm:"response_body" description:"响应体（截断后存储）"` // 响应体（截断后存储） - LatencyMs int `json:"latency_ms" orm:"latency_ms" description:"请求延迟（毫秒）"` // 请求延迟（毫秒） - AuditLevel string `json:"audit_level" orm:"audit_level" description:"审计级别：full（完整记录）/ masked（脱敏记录）/ question_only（仅记录提问）/ none（不记录）"` // 审计级别：full（完整记录）/ masked（脱敏记录）/ question_only（仅记录提问）/ none（不记录） - CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 - UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 - TenantRequestBody string `json:"tenant_request_body" orm:"tenant_request_body" description:"租户级请求体（按租户审计级别处理）"` // 租户级请求体（按租户审计级别处理） - TenantResponseBody string `json:"tenant_response_body" orm:"tenant_response_body" description:"租户级响应体（按租户审计级别处理）"` // 租户级响应体（按租户审计级别处理） - TenantAuditLevel string `json:"tenant_audit_level" orm:"tenant_audit_level" description:"租户审计级别：full/full_text/masked/question_only/none"` // 租户审计级别：full/full_text/masked/question_only/none - ProjectId int64 `json:"project_id" orm:"project_id" description:"关联项目ID（通过API Key关联，NULL表示个人密钥无项目）"` // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） - FirstTokenMs int `json:"first_token_ms" orm:"first_token_ms" description:"首个 Token 出现的用时（毫秒），仅流式请求有值"` // 首个 Token 出现的用时（毫秒），仅流式请求有值 - RequestHeaders string `json:"request_headers" orm:"request_headers" description:"请求头信息（仅审计级别为 all 时记录，管理后台调试用）"` // 请求头信息（仅审计级别为 all 时记录，管理后台调试用） - ResponseHeaders string `json:"response_headers" orm:"response_headers" description:"响应头信息（仅审计级别为 all 时记录，管理后台调试用）"` // 响应头信息（仅审计级别为 all 时记录，管理后台调试用） - ForwardingTrace string `json:"forwarding_trace" orm:"forwarding_trace" description:"请求转发路径追踪（仅管理员可见）"` // 请求转发路径追踪（仅管理员可见） + Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID + TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID"` // 租户ID + UserId int64 `json:"user_id" orm:"user_id" description:"用户ID"` // 用户ID + ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:"使用的 API Key ID"` // 使用的 API Key ID + RequestId string `json:"request_id" orm:"request_id" description:"请求唯一ID（关联全链路追踪）"` // 请求唯一ID（关联全链路追踪） + Method string `json:"method" orm:"method" description:"HTTP 方法（GET/POST/PUT/DELETE）"` // HTTP 方法（GET/POST/PUT/DELETE） + Path string `json:"path" orm:"path" description:"请求路径"` // 请求路径 + QueryParams string `json:"query_params" orm:"query_params" description:"查询参数（URL Query String）"` // 查询参数（URL Query String） + StatusCode int `json:"status_code" orm:"status_code" description:"HTTP 响应状态码"` // HTTP 响应状态码 + ClientIp string `json:"client_ip" orm:"client_ip" description:"客户端 IP"` // 客户端 IP + UserAgent string `json:"user_agent" orm:"user_agent" description:"客户端 User-Agent"` // 客户端 User-Agent + RequestBody string `json:"request_body" orm:"request_body" description:"请求体（敏感字段脱敏后存储）"` // 请求体（敏感字段脱敏后存储） + ResponseBody string `json:"response_body" orm:"response_body" description:"响应体（截断后存储）"` // 响应体（截断后存储） + LatencyMs int `json:"latency_ms" orm:"latency_ms" description:"请求延迟（毫秒）"` // 请求延迟（毫秒） + AuditLevel string `json:"audit_level" orm:"audit_level" description:"审计级别：full（完整记录）/ masked（脱敏记录）/ question_only（仅记录提问）/ none（不记录）"` // 审计级别：full（完整记录）/ masked（脱敏记录）/ question_only（仅记录提问）/ none（不记录） + CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 + UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 + TenantRequestBody string `json:"tenant_request_body" orm:"tenant_request_body" description:"租户级请求体（按租户审计级别处理）"` // 租户级请求体（按租户审计级别处理） + TenantResponseBody string `json:"tenant_response_body" orm:"tenant_response_body" description:"租户级响应体（按租户审计级别处理）"` // 租户级响应体（按租户审计级别处理） + TenantAuditLevel string `json:"tenant_audit_level" orm:"tenant_audit_level" description:"租户审计级别：full/full_text/masked/question_only/none"` // 租户审计级别：full/full_text/masked/question_only/none + ProjectId int64 `json:"project_id" orm:"project_id" description:"关联项目ID（通过API Key关联，NULL表示个人密钥无项目）"` // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） + FirstTokenMs int `json:"first_token_ms" orm:"first_token_ms" description:"首个 Token 出现的用时（毫秒），仅流式请求有值"` // 首个 Token 出现的用时（毫秒），仅流式请求有值 + RequestHeaders string `json:"request_headers" orm:"request_headers" description:"请求头信息（仅审计级别为 all 时记录，管理后台调试用）"` // 请求头信息（仅审计级别为 all 时记录，管理后台调试用） + ResponseHeaders string `json:"response_headers" orm:"response_headers" description:"响应头信息（仅审计级别为 all 时记录，管理后台调试用）"` // 响应头信息（仅审计级别为 all 时记录，管理后台调试用） + ForwardingTrace string `json:"forwarding_trace" orm:"forwarding_trace" description:"请求转发路径追踪（仅管理员可见）"` // 请求转发路径追踪（仅管理员可见） + TaskId string `json:"task_id" orm:"task_id" description:"异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id"` // 异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id + TaskStatus string `json:"task_status" orm:"task_status" description:"异步任务终态：SUCCESS / FAILURE"` // 异步任务终态：SUCCESS / FAILURE + TaskResult string `json:"task_result" orm:"task_result" description:"异步任务完成时上游返回的原始响应体"` // 异步任务完成时上游返回的原始响应体 + TaskUpstreamHeaders string `json:"task_upstream_headers" orm:"task_upstream_headers" description:"异步任务完成时上游返回的响应头（仅审计级别为 full 时记录）"` // 异步任务完成时上游返回的响应头（仅审计级别为 full 时记录） + TaskCompletedAt *gtime.Time `json:"task_completed_at" orm:"task_completed_at" description:"异步任务达到终态的时间"` // 异步任务达到终态的时间 } diff --git a/internal/model/entity/bil_usage_logs.go b/internal/model/entity/bil_usage_logs.go index 21abc42..1852fbc 100644 --- a/internal/model/entity/bil_usage_logs.go +++ b/internal/model/entity/bil_usage_logs.go @@ -10,60 +10,61 @@ import ( // BilUsageLogs is the golang structure for table bil_usage_logs. type BilUsageLogs struct { - Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID - TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID"` // 租户ID - UserId int64 `json:"user_id" orm:"user_id" description:"用户ID"` // 用户ID - ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:"使用的 API Key ID"` // 使用的 API Key ID - ChannelId int64 `json:"channel_id" orm:"channel_id" description:"使用的渠道ID"` // 使用的渠道ID - ModelName string `json:"model_name" orm:"model_name" description:"调用的模型名"` // 调用的模型名 - RequestId string `json:"request_id" orm:"request_id" description:"请求唯一ID"` // 请求唯一ID - RelayMode string `json:"relay_mode" orm:"relay_mode" description:"代理模式：chat_completions / embeddings / images_generations 等"` // 代理模式：chat_completions / embeddings / images_generations 等 - InputTokens int `json:"input_tokens" orm:"input_tokens" description:"输入 token 数"` // 输入 token 数 - OutputTokens int `json:"output_tokens" orm:"output_tokens" description:"输出 token 数"` // 输出 token 数 - TotalCost float64 `json:"total_cost" orm:"total_cost" description:"本次调用费用"` // 本次调用费用 - Currency string `json:"currency" orm:"currency" description:"货币（USD）"` // 货币（USD） - LatencyMs int `json:"latency_ms" orm:"latency_ms" description:"请求延迟（毫秒）"` // 请求延迟（毫秒） - Status string `json:"status" orm:"status" description:"状态：success（成功）/ error（错误）/ timeout（超时）/ cancelled（取消）"` // 状态：success（成功）/ error（错误）/ timeout（超时）/ cancelled（取消） - ErrorMessage string `json:"error_message" orm:"error_message" description:"错误信息（成功时为 NULL）"` // 错误信息（成功时为 NULL） - ClientIp string `json:"client_ip" orm:"client_ip" description:"客户端 IP"` // 客户端 IP - CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 - UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 - CacheCreationTokens int `json:"cache_creation_tokens" orm:"cache_creation_tokens" description:"缓存创建 token 数 (Claude)"` // 缓存创建 token 数 (Claude) - CacheReadTokens int `json:"cache_read_tokens" orm:"cache_read_tokens" description:"缓存读取 token 数 (Claude/OpenAI)"` // 缓存读取 token 数 (Claude/OpenAI) - InputCost float64 `json:"input_cost" orm:"input_cost" description:"输入 token 费用"` // 输入 token 费用 - OutputCost float64 `json:"output_cost" orm:"output_cost" description:"输出 token 费用"` // 输出 token 费用 - CacheCreationCost float64 `json:"cache_creation_cost" orm:"cache_creation_cost" description:"缓存创建费用"` // 缓存创建费用 - CacheReadCost float64 `json:"cache_read_cost" orm:"cache_read_cost" description:"缓存读取费用"` // 缓存读取费用 - ActualCost float64 `json:"actual_cost" orm:"actual_cost" description:"实际扣除费用（含折扣后）"` // 实际扣除费用（含折扣后） - RequestedModel string `json:"requested_model" orm:"requested_model" description:"用户请求的模型名"` // 用户请求的模型名 - UpstreamModel string `json:"upstream_model" orm:"upstream_model" description:"上游实际模型名（模型映射后）"` // 上游实际模型名（模型映射后） - RequestType int `json:"request_type" orm:"request_type" description:"请求类型: 1=sync, 2=stream, 3=websocket"` // 请求类型: 1=sync, 2=stream, 3=websocket - UserAgent string `json:"user_agent" orm:"user_agent" description:"客户端 User-Agent"` // 客户端 User-Agent - FirstTokenMs int `json:"first_token_ms" orm:"first_token_ms" description:"首 token 延迟（毫秒）"` // 首 token 延迟（毫秒） - ServiceTier string `json:"service_tier" orm:"service_tier" description:"服务等级 (default/flex等)"` // 服务等级 (default/flex等) - ReasoningEffort string `json:"reasoning_effort" orm:"reasoning_effort" description:"推理力度 (low/medium/high)"` // 推理力度 (low/medium/high) - ChannelName string `json:"channel_name" orm:"channel_name" description:"渠道名称"` // 渠道名称 - ChannelType int `json:"channel_type" orm:"channel_type" description:"渠道类型 (ProviderType)"` // 渠道类型 (ProviderType) - BillingMode string `json:"billing_mode" orm:"billing_mode" description:"计费模式 (token/per_request/tiered)"` // 计费模式 (token/per_request/tiered) - BillingSource string `json:"billing_source" orm:"billing_source" description:"定价来源 (base/tenant/custom)"` // 定价来源 (base/tenant/custom) - RateMultiplier float64 `json:"rate_multiplier" orm:"rate_multiplier" description:"费率倍率/折扣"` // 费率倍率/折扣 - PreDeductAmount float64 `json:"pre_deduct_amount" orm:"pre_deduct_amount" description:"预扣金额"` // 预扣金额 - RefundAmount float64 `json:"refund_amount" orm:"refund_amount" description:"退款金额"` // 退款金额 - SupplementAmount float64 `json:"supplement_amount" orm:"supplement_amount" description:"补扣金额"` // 补扣金额 - ImageCount int `json:"image_count" orm:"image_count" description:"生成图片数量"` // 生成图片数量 - ImageSize string `json:"image_size" orm:"image_size" description:"图片尺寸"` // 图片尺寸 - StreamEndReason string `json:"stream_end_reason" orm:"stream_end_reason" description:"流结束原因 (done/timeout/client_gone/error/panic)"` // 流结束原因 (done/timeout/client_gone/error/panic) - RetryIndex int `json:"retry_index" orm:"retry_index" description:"重试次数（0=首次成功）"` // 重试次数（0=首次成功） - BillingSummary string `json:"billing_summary" orm:"billing_summary" description:"计费快照文本（人类可读的计费过程描述）"` // 计费快照文本（人类可读的计费过程描述） - CacheCreation5MTokens int `json:"cache_creation_5m_tokens" orm:"cache_creation_5m_tokens" description:"Claude 5分钟缓存创建 token 数"` // Claude 5分钟缓存创建 token 数 - CacheCreation1HTokens int `json:"cache_creation_1h_tokens" orm:"cache_creation_1h_tokens" description:"Claude 1小时缓存创建 token 数"` // Claude 1小时缓存创建 token 数 - AudioInputTokens int `json:"audio_input_tokens" orm:"audio_input_tokens" description:"音频输入 token 数"` // 音频输入 token 数 - AudioOutputTokens int `json:"audio_output_tokens" orm:"audio_output_tokens" description:"音频输出 token 数"` // 音频输出 token 数 - ImageOutputTokens int `json:"image_output_tokens" orm:"image_output_tokens" description:"图像输出 token 数（DALL-E 等）"` // 图像输出 token 数（DALL-E 等） - ReasoningTokens int `json:"reasoning_tokens" orm:"reasoning_tokens" description:"推理 token 数（O1/o3 等）"` // 推理 token 数（O1/o3 等） - AccountCost float64 `json:"account_cost" orm:"account_cost" description:"上游账户成本（用于利润分析）"` // 上游账户成本（用于利润分析） - InboundEndpoint string `json:"inbound_endpoint" orm:"inbound_endpoint" description:"客户端请求路径（如 /v1/chat/completions）"` // 客户端请求路径（如 /v1/chat/completions） - UpstreamEndpoint string `json:"upstream_endpoint" orm:"upstream_endpoint" description:"上游实际请求路径"` // 上游实际请求路径 - BillingSnapshot string `json:"billing_snapshot" orm:"billing_snapshot" description:"完整计费计算过程快照（JSONB）"` // 完整计费计算过程快照（JSONB） - ProjectId int64 `json:"project_id" orm:"project_id" description:"关联项目ID（通过API Key关联，NULL表示个人密钥无项目）"` // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） + Id int64 `json:"id" orm:"id" description:"主键ID"` // 主键ID + TenantId int64 `json:"tenant_id" orm:"tenant_id" description:"租户ID"` // 租户ID + UserId int64 `json:"user_id" orm:"user_id" description:"用户ID"` // 用户ID + ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:"使用的 API Key ID"` // 使用的 API Key ID + ChannelId int64 `json:"channel_id" orm:"channel_id" description:"使用的渠道ID"` // 使用的渠道ID + ModelName string `json:"model_name" orm:"model_name" description:"调用的模型名"` // 调用的模型名 + RequestId string `json:"request_id" orm:"request_id" description:"请求唯一ID"` // 请求唯一ID + RelayMode string `json:"relay_mode" orm:"relay_mode" description:"代理模式：chat_completions / embeddings / images_generations 等"` // 代理模式：chat_completions / embeddings / images_generations 等 + InputTokens int `json:"input_tokens" orm:"input_tokens" description:"输入 token 数"` // 输入 token 数 + OutputTokens int `json:"output_tokens" orm:"output_tokens" description:"输出 token 数"` // 输出 token 数 + TotalCost float64 `json:"total_cost" orm:"total_cost" description:"本次调用费用"` // 本次调用费用 + Currency string `json:"currency" orm:"currency" description:"货币（USD）"` // 货币（USD） + LatencyMs int `json:"latency_ms" orm:"latency_ms" description:"请求延迟（毫秒）"` // 请求延迟（毫秒） + Status string `json:"status" orm:"status" description:"状态：success（成功）/ error（错误）/ timeout（超时）/ cancelled（取消）"` // 状态：success（成功）/ error（错误）/ timeout（超时）/ cancelled（取消） + ErrorMessage string `json:"error_message" orm:"error_message" description:"错误信息（成功时为 NULL）"` // 错误信息（成功时为 NULL） + ClientIp string `json:"client_ip" orm:"client_ip" description:"客户端 IP"` // 客户端 IP + CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:"创建时间"` // 创建时间 + UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:"更新时间"` // 更新时间 + CacheCreationTokens int `json:"cache_creation_tokens" orm:"cache_creation_tokens" description:"缓存创建 token 数 (Claude)"` // 缓存创建 token 数 (Claude) + CacheReadTokens int `json:"cache_read_tokens" orm:"cache_read_tokens" description:"缓存读取 token 数 (Claude/OpenAI)"` // 缓存读取 token 数 (Claude/OpenAI) + InputCost float64 `json:"input_cost" orm:"input_cost" description:"输入 token 费用"` // 输入 token 费用 + OutputCost float64 `json:"output_cost" orm:"output_cost" description:"输出 token 费用"` // 输出 token 费用 + CacheCreationCost float64 `json:"cache_creation_cost" orm:"cache_creation_cost" description:"缓存创建费用"` // 缓存创建费用 + CacheReadCost float64 `json:"cache_read_cost" orm:"cache_read_cost" description:"缓存读取费用"` // 缓存读取费用 + ActualCost float64 `json:"actual_cost" orm:"actual_cost" description:"实际扣除费用（含折扣后）"` // 实际扣除费用（含折扣后） + RequestedModel string `json:"requested_model" orm:"requested_model" description:"用户请求的模型名"` // 用户请求的模型名 + UpstreamModel string `json:"upstream_model" orm:"upstream_model" description:"上游实际模型名（模型映射后）"` // 上游实际模型名（模型映射后） + RequestType int `json:"request_type" orm:"request_type" description:"请求类型: 1=sync, 2=stream, 3=async, 4=websocket"` // 请求类型: 1=sync, 2=stream, 3=async, 4=websocket + UserAgent string `json:"user_agent" orm:"user_agent" description:"客户端 User-Agent"` // 客户端 User-Agent + FirstTokenMs int `json:"first_token_ms" orm:"first_token_ms" description:"首 token 延迟（毫秒）"` // 首 token 延迟（毫秒） + ServiceTier string `json:"service_tier" orm:"service_tier" description:"服务等级 (default/flex等)"` // 服务等级 (default/flex等) + ReasoningEffort string `json:"reasoning_effort" orm:"reasoning_effort" description:"推理力度 (low/medium/high)"` // 推理力度 (low/medium/high) + ChannelName string `json:"channel_name" orm:"channel_name" description:"渠道名称"` // 渠道名称 + ChannelType int `json:"channel_type" orm:"channel_type" description:"渠道类型 (ProviderType)"` // 渠道类型 (ProviderType) + BillingMode string `json:"billing_mode" orm:"billing_mode" description:"计费模式 (token/per_request/tiered)"` // 计费模式 (token/per_request/tiered) + BillingSource string `json:"billing_source" orm:"billing_source" description:"定价来源 (base/tenant/custom)"` // 定价来源 (base/tenant/custom) + RateMultiplier float64 `json:"rate_multiplier" orm:"rate_multiplier" description:"费率倍率/折扣"` // 费率倍率/折扣 + PreDeductAmount float64 `json:"pre_deduct_amount" orm:"pre_deduct_amount" description:"预扣金额"` // 预扣金额 + RefundAmount float64 `json:"refund_amount" orm:"refund_amount" description:"退款金额"` // 退款金额 + SupplementAmount float64 `json:"supplement_amount" orm:"supplement_amount" description:"补扣金额"` // 补扣金额 + ImageCount int `json:"image_count" orm:"image_count" description:"生成图片数量"` // 生成图片数量 + ImageSize string `json:"image_size" orm:"image_size" description:"图片尺寸"` // 图片尺寸 + StreamEndReason string `json:"stream_end_reason" orm:"stream_end_reason" description:"流结束原因 (done/timeout/client_gone/error/panic)"` // 流结束原因 (done/timeout/client_gone/error/panic) + RetryIndex int `json:"retry_index" orm:"retry_index" description:"重试次数（0=首次成功）"` // 重试次数（0=首次成功） + BillingSummary string `json:"billing_summary" orm:"billing_summary" description:"计费快照文本（人类可读的计费过程描述）"` // 计费快照文本（人类可读的计费过程描述） + CacheCreation5MTokens int `json:"cache_creation_5m_tokens" orm:"cache_creation_5m_tokens" description:"Claude 5分钟缓存创建 token 数"` // Claude 5分钟缓存创建 token 数 + CacheCreation1HTokens int `json:"cache_creation_1h_tokens" orm:"cache_creation_1h_tokens" description:"Claude 1小时缓存创建 token 数"` // Claude 1小时缓存创建 token 数 + AudioInputTokens int `json:"audio_input_tokens" orm:"audio_input_tokens" description:"音频输入 token 数"` // 音频输入 token 数 + AudioOutputTokens int `json:"audio_output_tokens" orm:"audio_output_tokens" description:"音频输出 token 数"` // 音频输出 token 数 + ImageOutputTokens int `json:"image_output_tokens" orm:"image_output_tokens" description:"图像输出 token 数（DALL-E 等）"` // 图像输出 token 数（DALL-E 等） + ReasoningTokens int `json:"reasoning_tokens" orm:"reasoning_tokens" description:"推理 token 数（O1/o3 等）"` // 推理 token 数（O1/o3 等） + AccountCost float64 `json:"account_cost" orm:"account_cost" description:"上游账户成本（用于利润分析）"` // 上游账户成本（用于利润分析） + InboundEndpoint string `json:"inbound_endpoint" orm:"inbound_endpoint" description:"客户端请求路径（如 /v1/chat/completions）"` // 客户端请求路径（如 /v1/chat/completions） + UpstreamEndpoint string `json:"upstream_endpoint" orm:"upstream_endpoint" description:"上游实际请求路径"` // 上游实际请求路径 + BillingSnapshot string `json:"billing_snapshot" orm:"billing_snapshot" description:"完整计费计算过程快照（JSONB）"` // 完整计费计算过程快照（JSONB） + ProjectId int64 `json:"project_id" orm:"project_id" description:"关联项目ID（通过API Key关联，NULL表示个人密钥无项目）"` // 关联项目ID（通过API Key关联，NULL表示个人密钥无项目） + TaskId string `json:"task_id" orm:"task_id" description:"异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id，普通请求为空"` // 异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id，普通请求为空 } diff --git a/internal/model/entity/tsk_model_tasks.go b/internal/model/entity/tsk_model_tasks.go index 2124e86..867d802 100644 --- a/internal/model/entity/tsk_model_tasks.go +++ b/internal/model/entity/tsk_model_tasks.go @@ -10,28 +10,29 @@ import ( // TskModelTasks is the golang structure for table tsk_model_tasks. type TskModelTasks struct { - Id int64 `json:"id" orm:"id" description:""` // - PublicTaskId string `json:"public_task_id" orm:"public_task_id" description:""` // - Platform string `json:"platform" orm:"platform" description:""` // - Action string `json:"action" orm:"action" description:""` // - Status string `json:"status" orm:"status" description:""` // - Progress string `json:"progress" orm:"progress" description:""` // - FailReason string `json:"fail_reason" orm:"fail_reason" description:""` // - TenantId int64 `json:"tenant_id" orm:"tenant_id" description:""` // - UserId int64 `json:"user_id" orm:"user_id" description:""` // - ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:""` // - ChannelId int64 `json:"channel_id" orm:"channel_id" description:""` // - ModelName string `json:"model_name" orm:"model_name" description:""` // - UpstreamModel string `json:"upstream_model" orm:"upstream_model" description:""` // - PreDeductAmount float64 `json:"pre_deduct_amount" orm:"pre_deduct_amount" description:""` // - ActualCost float64 `json:"actual_cost" orm:"actual_cost" description:""` // - BillingSettled bool `json:"billing_settled" orm:"billing_settled" description:""` // - ResultUrl string `json:"result_url" orm:"result_url" description:""` // - Data string `json:"data" orm:"data" description:""` // - PrivateData string `json:"private_data" orm:"private_data" description:""` // - SubmitTime *gtime.Time `json:"submit_time" orm:"submit_time" description:""` // - StartTime *gtime.Time `json:"start_time" orm:"start_time" description:""` // - FinishTime *gtime.Time `json:"finish_time" orm:"finish_time" description:""` // - CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:""` // - UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:""` // + Id int64 `json:"id" orm:"id" description:""` // + PublicTaskId string `json:"public_task_id" orm:"public_task_id" description:""` // + RequestId string `json:"request_id" orm:"request_id" description:"任务提交时的原始请求 ID"` // + Platform string `json:"platform" orm:"platform" description:""` // + Action string `json:"action" orm:"action" description:""` // + Status string `json:"status" orm:"status" description:""` // + Progress string `json:"progress" orm:"progress" description:""` // + FailReason string `json:"fail_reason" orm:"fail_reason" description:""` // + TenantId int64 `json:"tenant_id" orm:"tenant_id" description:""` // + UserId int64 `json:"user_id" orm:"user_id" description:""` // + ApiKeyId int64 `json:"api_key_id" orm:"api_key_id" description:""` // + ChannelId int64 `json:"channel_id" orm:"channel_id" description:""` // + ModelName string `json:"model_name" orm:"model_name" description:""` // + UpstreamModel string `json:"upstream_model" orm:"upstream_model" description:""` // + PreDeductAmount float64 `json:"pre_deduct_amount" orm:"pre_deduct_amount" description:""` // + ActualCost float64 `json:"actual_cost" orm:"actual_cost" description:""` // + BillingSettled bool `json:"billing_settled" orm:"billing_settled" description:""` // + ResultUrl string `json:"result_url" orm:"result_url" description:""` // + Data string `json:"data" orm:"data" description:""` // + PrivateData string `json:"private_data" orm:"private_data" description:""` // + SubmitTime *gtime.Time `json:"submit_time" orm:"submit_time" description:""` // + StartTime *gtime.Time `json:"start_time" orm:"start_time" description:""` // + FinishTime *gtime.Time `json:"finish_time" orm:"finish_time" description:""` // + CreatedAt *gtime.Time `json:"created_at" orm:"created_at" description:""` // + UpdatedAt *gtime.Time `json:"updated_at" orm:"updated_at" description:""` // } diff --git a/internal/service/admin.go b/internal/service/admin.go index ca37559..3a099d2 100644 --- a/internal/service/admin.go +++ b/internal/service/admin.go @@ -202,7 +202,7 @@ type ( CreateModel(ctx context.Context, req *v1.ModelCreateReq) (*v1.ModelCreateRes, error) // UpdateModel 更新模型（含弃用状态管理） UpdateModel(ctx context.Context, req *v1.ModelUpdateReq) (*v1.ModelUpdateRes, error) - // DeleteModel 删除模型（同时删除定价记录） + // DeleteModel 删除模型（同时删除定价记录和租户分配记录） DeleteModel(ctx context.Context, req *v1.ModelDeleteReq) (*v1.ModelDeleteRes, error) // ListModelPricing 模型定价列表（模型定价页面专用） ListModelPricing(ctx context.Context, req *v1.PricingListReq) (*v1.PricingListRes, error) diff --git a/internal/service/tenant.go b/internal/service/tenant.go index 0542299..99751c2 100644 --- a/internal/service/tenant.go +++ b/internal/service/tenant.go @@ -286,6 +286,12 @@ type ( GetIPWhitelist(ctx context.Context, _ *v1.TenantIPWhitelistGetReq) (*v1.TenantIPWhitelistGetRes, error) // UpdateIPWhitelist updates the tenant's IP whitelist configuration. UpdateIPWhitelist(ctx context.Context, req *v1.TenantIPWhitelistUpdateReq) (*v1.TenantIPWhitelistUpdateRes, error) + // TenantTaskList 租户异步任务列表 + // owner/admin 可查看租户所有任务，member 只能查看自己的任务 + TenantTaskList(ctx context.Context, req *v1.TenantTaskListReq) (*v1.TenantTaskListRes, error) + // TenantTaskDetail 租户异步任务详情 + // owner/admin 可查看租户所有任务，member 只能查看自己的任务 + TenantTaskDetail(ctx context.Context, req *v1.TenantTaskDetailReq) (*v1.TenantTaskDetailRes, error) // TicketCreate 创建工单 TicketCreate(ctx context.Context, req *v1.TenantTicketCreateReq) (*v1.TenantTicketCreateRes, error) // TicketList 获取租户工单列表 diff --git a/migrations/000003_async_task_support.sql b/migrations/000003_async_task_support.sql new file mode 100644 index 0000000..54074d8 --- /dev/null +++ b/migrations/000003_async_task_support.sql @@ -0,0 +1,59 @@ +-- +goose Up +-- 异步任务支持：审计日志任务字段 + 用量日志任务字段 + 任务表request_id + 请求类型注释更新 + +-- 1. aud_request_logs 新增任务结果相关字段 +ALTER TABLE aud_request_logs + ADD COLUMN IF NOT EXISTS task_id VARCHAR(64), + ADD COLUMN IF NOT EXISTS task_status VARCHAR(20), + ADD COLUMN IF NOT EXISTS task_result TEXT, + ADD COLUMN IF NOT EXISTS task_upstream_headers JSONB, + ADD COLUMN IF NOT EXISTS task_completed_at TIMESTAMPTZ; + +CREATE INDEX IF NOT EXISTS idx_aud_request_logs_task_id + ON aud_request_logs (task_id) WHERE task_id IS NOT NULL; + +COMMENT ON COLUMN aud_request_logs.task_id IS '异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id'; +COMMENT ON COLUMN aud_request_logs.task_status IS '异步任务终态：SUCCESS / FAILURE'; +COMMENT ON COLUMN aud_request_logs.task_result IS '异步任务完成时上游返回的原始响应体'; +COMMENT ON COLUMN aud_request_logs.task_upstream_headers IS '异步任务完成时上游返回的响应头（仅审计级别为 full 时记录）'; +COMMENT ON COLUMN aud_request_logs.task_completed_at IS '异步任务达到终态的时间'; + +-- 2. bil_usage_logs 新增 task_id 字段 +ALTER TABLE bil_usage_logs + ADD COLUMN IF NOT EXISTS task_id VARCHAR(64); + +CREATE INDEX IF NOT EXISTS idx_bil_usage_logs_task_id + ON bil_usage_logs (task_id) WHERE task_id IS NOT NULL; + +COMMENT ON COLUMN bil_usage_logs.task_id IS '异步任务公开ID（task_xxxxx），关联 tsk_model_tasks.public_task_id，普通请求为空'; + +-- 3. bil_usage_logs.request_type 注释更新（新增 async=3） +COMMENT ON COLUMN bil_usage_logs.request_type IS '请求类型: 1=sync, 2=stream, 3=async, 4=websocket'; + +-- 4. tsk_model_tasks 新增 request_id 字段，存储任务提交时的原始请求 ID +ALTER TABLE tsk_model_tasks + ADD COLUMN IF NOT EXISTS request_id VARCHAR(64); + +CREATE INDEX IF NOT EXISTS idx_tsk_model_tasks_request_id + ON tsk_model_tasks (request_id) WHERE request_id IS NOT NULL; + +COMMENT ON COLUMN tsk_model_tasks.request_id IS '任务提交时的原始请求 ID（req_xxxxx），关联 aud_request_logs.request_id'; + +-- +goose Down +DROP INDEX IF EXISTS idx_tsk_model_tasks_request_id; +ALTER TABLE tsk_model_tasks + DROP COLUMN IF EXISTS request_id; + +DROP INDEX IF EXISTS idx_aud_request_logs_task_id; +ALTER TABLE aud_request_logs + DROP COLUMN IF EXISTS task_id, + DROP COLUMN IF EXISTS task_status, + DROP COLUMN IF EXISTS task_result, + DROP COLUMN IF EXISTS task_upstream_headers, + DROP COLUMN IF EXISTS task_completed_at; + +DROP INDEX IF EXISTS idx_bil_usage_logs_task_id; +ALTER TABLE bil_usage_logs + DROP COLUMN IF EXISTS task_id; + +COMMENT ON COLUMN bil_usage_logs.request_type IS '请求类型: 1=sync, 2=stream, 3=websocket'; diff --git a/relay/common/provider.go b/relay/common/provider.go index 5aab1b2..c301a6c 100644 --- a/relay/common/provider.go +++ b/relay/common/provider.go @@ -3,6 +3,7 @@ package common import ( "context" "errors" + "time" ) // ErrChannelUnavailable 没有可用渠道 @@ -47,6 +48,10 @@ type DataProvider interface { // RecordAudit 异步记录请求审计日志。 RecordAudit(ctx context.Context, record *AuditRecord) + // UpdateTaskAudit 更新异步任务的审计记录（任务完成时调用）。 + // 通过 task_id 查找提交阶段写入的审计记录，补充最终结果。 + UpdateTaskAudit(ctx context.Context, record *AuditRecord) + // UpdateChannelHealth 更新渠道健康度（请求成功/失败后调用）。 UpdateChannelHealth(ctx context.Context, channelID int64, success bool, latencyMs float64) @@ -145,7 +150,7 @@ type UsageRecord struct { // 请求元数据 RequestedModel string // 用户请求的模型名 UpstreamModel string // 上游实际模型名 - RequestType int // 1=sync, 2=stream, 3=websocket + RequestType int // 1=sync, 2=stream, 3=async, 4=websocket UserAgent string ClientIP string FirstTokenMs int @@ -186,6 +191,9 @@ type UsageRecord struct { // 计费快照 BillingSnapshot string // JSON 字符串 BillingSummary string // 人类可读中文文本 + + // 异步任务关联 + TaskID string // 异步任务公开ID（task_xxxxx），普通请求为空 } // AuditRecord 请求审计日志记录 @@ -209,6 +217,13 @@ type AuditRecord struct { RequestHeaders map[string]string // 请求头（仅审计级别为 all 时记录） ResponseHeaders map[string]string // 响应头（仅审计级别为 all 时记录） ForwardingTrace *ForwardingTrace // 转发路径追踪（仅管理员可见） + + // 异步任务完成字段（仅 UpdateTaskAudit 时填充） + TaskID string // 异步任务公开ID，关联 tsk_model_tasks.public_task_id + TaskStatus string // SUCCESS / FAILURE + TaskResult string // 上游最终响应体 + TaskUpstreamHeaders map[string]string // 上游最终轮询的响应头 + TaskCompletedAt *time.Time // 任务到达终态的时间 } // ModelInfo 模型信息（用于 /v1/models 端点） diff --git a/relay/common/task_adaptor.go b/relay/common/task_adaptor.go index d958cc5..387a869 100644 --- a/relay/common/task_adaptor.go +++ b/relay/common/task_adaptor.go @@ -59,6 +59,11 @@ type TaskInfo struct { Data []byte // 上游原始响应 SubTasks []SubTask // 子任务结果（如多图/多视频场景） ActualCost float64 // 上游返回的实际费用（0 表示未提供，使用预扣金额） + + // Token 用量（部分异步任务上游会返回，如火山引擎视频模型） + PromptTokens int + CompletionTokens int + TotalTokens int } // TaskStatusEnum 任务状态枚举 diff --git a/relay/common/task_provider.go b/relay/common/task_provider.go index 85d7e56..2ed2a17 100644 --- a/relay/common/task_provider.go +++ b/relay/common/task_provider.go @@ -37,6 +37,7 @@ type TaskDataProvider interface { type AsyncTask struct { ID int64 `json:"id"` PublicTaskID string `json:"public_task_id"` // task_xxxxx + RequestID string `json:"request_id"` // 原始请求 ID（req_xxxxx） Platform string `json:"platform"` // sora, kling, suno Action string `json:"action"` // generate, music, lyrics Status string `json:"status"` @@ -55,6 +56,10 @@ type AsyncTask struct { ActualCost float64 `json:"actual_cost"` BillingSettled bool `json:"billing_settled"` + PromptTokens int `json:"prompt_tokens,omitempty"` + CompletionTokens int `json:"completion_tokens,omitempty"` + TotalTokens int `json:"total_tokens,omitempty"` + ResultURL string `json:"result_url,omitempty"` Data json.RawMessage `json:"data,omitempty"` // 上游原始响应 PrivateData json.RawMessage `json:"private_data,omitempty"` // 上游任务 ID、API Key 等敏感数据 diff --git a/relay/handler/task_handler.go b/relay/handler/task_handler.go index 942eb73..2b308e1 100644 --- a/relay/handler/task_handler.go +++ b/relay/handler/task_handler.go @@ -26,6 +26,8 @@ type TaskRelayContext struct { TenantID int64 UserID int64 ApiKeyID int64 + ProjectID int64 + TaskID string // 由 HandleTaskSubmit 在创建任务后设置 RequestID string Writer http.ResponseWriter Scope string @@ -167,6 +169,7 @@ func HandleTaskSubmit( task := &common.AsyncTask{ PublicTaskID: publicTaskID, + RequestID: rc.RequestID, Platform: string(platform), Action: "generate", Status: "SUBMITTED", @@ -189,6 +192,9 @@ func HandleTaskSubmit( return } + // 设置 TaskID 供外层审计使用 + rc.TaskID = publicTaskID + // 11. 返回响应 respBody := map[string]any{ "id": publicTaskID, diff --git a/relay/taskchannel/volcengine/adaptor.go b/relay/taskchannel/volcengine/adaptor.go index b13e01e..966cda0 100644 --- a/relay/taskchannel/volcengine/adaptor.go +++ b/relay/taskchannel/volcengine/adaptor.go @@ -53,6 +53,12 @@ type submitResponse struct { ID string `json:"id"` } +// taskUsage 火山引擎任务 usage +type taskUsage struct { + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` +} + // taskResponse 查询任务响应 type taskResponse struct { ID string `json:"id"` @@ -61,9 +67,10 @@ type taskResponse struct { Content struct { VideoURL string `json:"video_url"` } `json:"content"` - Seed int `json:"seed"` - Duration int `json:"duration"` - Ratio string `json:"ratio"` + Usage taskUsage `json:"usage"` + Seed int `json:"seed"` + Duration int `json:"duration"` + Ratio string `json:"ratio"` Error struct { Code string `json:"code"` Message string `json:"message"` @@ -330,6 +337,13 @@ func (a *VolcengineVideoAdaptor) ParseTaskResult(body []byte) (*common.TaskInfo, info := &common.TaskInfo{Data: body} + // 提取 token 用量 + if resp.Usage.TotalTokens > 0 { + info.CompletionTokens = resp.Usage.CompletionTokens + info.TotalTokens = resp.Usage.TotalTokens + info.PromptTokens = resp.Usage.TotalTokens - resp.Usage.CompletionTokens + } + switch resp.Status { case "pending", "queued": info.Status = common.TaskStatusQueued diff --git a/web/admin/src/views/ai/TaskLogsPage.vue b/web/admin/src/views/ai/TaskLogsPage.vue index 6d8ece8..3c03fab 100644 --- a/web/admin/src/views/ai/TaskLogsPage.vue +++ b/web/admin/src/views/ai/TaskLogsPage.vue @@ -1,5 +1,6 @@ @@ -216,6 +224,13 @@ onMounted(() => { + import { ref, reactive, onMounted, h } from 'vue' -import { useRoute } from 'vue-router' +import { useRoute, useRouter } from 'vue-router' import { Tag, Button, } from '@arco-design/web-vue' @@ -8,6 +8,8 @@ import type { TableColumnData } from '@arco-design/web-vue' import PageHeader from '@/components/PageHeader.vue' import request from '@/utils/request' +const router = useRouter() + const loading = ref(false) const data = ref([]) const pagination = reactive({ @@ -22,6 +24,7 @@ const filter = reactive({ api_key_id: '', username: '', request_id: '', + task_id: '', method: '', status_code: '', date_range: [] as string[], @@ -59,6 +62,24 @@ const auditLevelLabel: Record = { none: '关闭', } +const taskStatusLabel: Record = { + SUBMITTED: '已提交', + QUEUED: '排队中', + IN_PROGRESS: '处理中', + SUCCESS: '成功', + FAILURE: '失败', + TIMEOUT: '超时', +} + +const taskStatusColor: Record = { + SUBMITTED: 'arcoblue', + QUEUED: 'cyan', + IN_PROGRESS: 'orange', + SUCCESS: 'green', + FAILURE: 'red', + TIMEOUT: 'orangered', +} + function formatJson(str: string): string { if (!str) return '' try { @@ -82,6 +103,11 @@ function formatMs(ms: number): string { return `${(ms / 1000).toFixed(2)}s` } +function goToTask(taskId: string) { + showDetail.value = false + router.push({ path: '/admin/task-logs', query: { public_task_id: taskId } }) +} + const columns: TableColumnData[] = [ { title: 'ID', dataIndex: 'id', width: 70 }, { title: '租户', dataIndex: 'tenant_name', width: 120, ellipsis: true, render({ record }) { @@ -159,6 +185,7 @@ async function fetchData() { if (filter.username) params.username = filter.username if (filter.method) params.method = filter.method if (filter.request_id) params.request_id = filter.request_id + if (filter.task_id) params.task_id = filter.task_id if (filter.status_code) params.status_code = parseInt(filter.status_code) if (filter.date_range.length === 2) { params.start_date = filter.date_range[0] @@ -181,11 +208,14 @@ async function fetchDetail(id: number) { showDetail.value = true try { const res: any = await request.get(`/admin/audit/request-logs/${id}`) - const data = res.data?.data?.data || null - if (data?.forwarding_trace) { - data.forwarding_trace = parseJson(data.forwarding_trace) + const detail = res.data?.data?.data || null + if (detail?.forwarding_trace) { + detail.forwarding_trace = parseJson(detail.forwarding_trace) } - detailRecord.value = data + if (detail?.task_upstream_headers) { + detail.task_upstream_headers = parseJson(detail.task_upstream_headers) + } + detailRecord.value = detail } catch { detailRecord.value = null } finally { @@ -203,6 +233,7 @@ function handleReset() { filter.api_key_id = '' filter.username = '' filter.request_id = '' + filter.task_id = '' filter.method = '' filter.status_code = '' filter.date_range = [] @@ -215,6 +246,9 @@ onMounted(() => { if (route.query.request_id) { filter.request_id = String(route.query.request_id) } + if (route.query.task_id) { + filter.task_id = String(route.query.task_id) + } fetchData() }) @@ -253,6 +287,13 @@ onMounted(() => { style="width: 200px" @keydown.enter="handleFilter" /> + { - +