From 96f99ea9d6aadb7c4574758eacfc5f5b7e1405e0 Mon Sep 17 00:00:00 2001 From: ddCat <2695092045@qq.com> Date: Fri, 31 Jul 2026 23:24:03 +0800 Subject: [PATCH] =?UTF-8?q?fix(media):=20=E8=A7=86=E9=A2=91=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=20-=20=E7=BB=9F=E4=B8=80=E8=BD=AE=E8=AF=A2=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E5=B9=B6=E6=94=AF=E6=8C=81=E5=A4=B1=E8=B4=A5=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E4=BA=BA=E5=B7=A5=E6=81=A2=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/handler/auth.go | 13 ++ backend/internal/handler/routes.go | 13 ++ backend/internal/model/models.go | 1 + backend/internal/repository/analytics.go | 11 +- backend/internal/repository/repository.go | 32 ++++ backend/internal/service/analytics.go | 1 + backend/internal/service/provider.go | 133 +++++++++++--- .../service/provider_task_recovery.go | 166 ++++++++++++++++++ backend/internal/service/service.go | 120 +++++++++---- .../components/api-log-detail-drawer.tsx | 37 +++- web/src/pages/admin/logs/logs-page.tsx | 2 +- web/src/pages/tasks/index.tsx | 31 +++- web/src/services/api/auth.ts | 13 ++ web/src/services/api/task-center.ts | 12 ++ 14 files changed, 510 insertions(+), 75 deletions(-) create mode 100644 backend/internal/service/provider_task_recovery.go diff --git a/backend/internal/handler/auth.go b/backend/internal/handler/auth.go index 805673f7f..89916ec34 100644 --- a/backend/internal/handler/auth.go +++ b/backend/internal/handler/auth.go @@ -577,6 +577,19 @@ func RegisterAdminRoutes(r *gin.RouterGroup, svc *service.Service) { } ok(c, gin.H{"log": log}) }) + r.POST("/admin/api-logs/:id/query-task", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + result, err := svc.AdminQueryFailedVideoTask(c.Request.Context(), user, c.Param("id")) + if err != nil { + failService(c, err) + return + } + ok(c, result) + }) r.GET("/admin/api-logs-export.csv", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { diff --git a/backend/internal/handler/routes.go b/backend/internal/handler/routes.go index 6841910be..5472bde9f 100644 --- a/backend/internal/handler/routes.go +++ b/backend/internal/handler/routes.go @@ -78,6 +78,19 @@ func RegisterTaskRoutes(r *gin.RouterGroup, svc *service.Service) { } ok(c, task) }) + r.POST("/tasks/:id/query-provider", func(c *gin.Context) { + user, err := currentUser(c, svc) + if err != nil { + failService(c, err) + return + } + result, err := svc.QueryFailedVideoTask(c.Request.Context(), user.ID, c.Param("id")) + if err != nil { + failService(c, err) + return + } + ok(c, result) + }) r.POST("/tasks/:id/cancel", func(c *gin.Context) { user, err := currentUser(c, svc) if err != nil { diff --git a/backend/internal/model/models.go b/backend/internal/model/models.go index 923c6b6f6..097a71227 100644 --- a/backend/internal/model/models.go +++ b/backend/internal/model/models.go @@ -241,6 +241,7 @@ type ApiCallLog struct { ChannelID string `json:"channelId" gorm:"index;size:36;index:idx_api_logs_channel_created,priority:1"` ChannelName string `json:"channelName" gorm:"-"` TaskID string `json:"taskId,omitempty" gorm:"index;size:36"` + TaskStatus TaskStatus `json:"taskStatus,omitempty" gorm:"-"` BillingOrderID string `json:"billingOrderId,omitempty" gorm:"index;size:36"` Source string `json:"source" gorm:"index;size:64"` Capability string `json:"capability" gorm:"index;size:32"` diff --git a/backend/internal/repository/analytics.go b/backend/internal/repository/analytics.go index 1f4ecaa54..065fddae6 100644 --- a/backend/internal/repository/analytics.go +++ b/backend/internal/repository/analytics.go @@ -159,10 +159,19 @@ func (r *Repository) APICallLogTasks(ids []string) ([]model.Task, error) { return []model.Task{}, nil } var tasks []model.Task - err := r.db.Select("id", "user_id", "type", "result_json").Where("id IN ?", ids).Find(&tasks).Error + err := r.db.Select("id", "user_id", "type", "status", "result_json").Where("id IN ?", ids).Find(&tasks).Error return tasks, err } +func (r *Repository) LatestProviderRequestIDForTask(taskID string) (string, error) { + var log model.ApiCallLog + err := r.db.Select("provider_request_id"). + Where("task_id = ? AND provider_request_id <> ''", taskID). + Order("created_at desc"). + First(&log).Error + return strings.TrimSpace(log.ProviderRequestID), err +} + func visibleAPICallLogQuery(query *gorm.DB) *gorm.DB { // 视频轮询属于一次生成调用的内部阶段，管理端只展示聚合后的创建主记录。 return query.Where("NOT (api_call_logs.capability = ? AND api_call_logs.request_kind IN ?)", "video", []string{"poll", "download"}) diff --git a/backend/internal/repository/repository.go b/backend/internal/repository/repository.go index e6a186107..7ce216c5e 100644 --- a/backend/internal/repository/repository.go +++ b/backend/internal/repository/repository.go @@ -14,6 +14,8 @@ import ( var ErrDailyUploadLimitExceeded = errors.New("daily upload limit exceeded") +var ErrTaskProviderRecoveryConflict = errors.New("task provider recovery is already running") + type Repository struct { db *gorm.DB } @@ -344,6 +346,36 @@ func (r *Repository) UpdateTaskProviderState(id string, providerRequestID string return r.db.Model(&model.Task{}).Where("id = ?", id).Updates(updates).Error } +// 人工恢复仅锁定失败任务；旧 worker 的租约可覆盖，但未过期的人工恢复租约不能并发抢占。 +func (r *Repository) ClaimFailedTaskProviderRecovery(id string, userID string, owner string, leaseDuration time.Duration) error { + now := time.Now() + query := r.db.Model(&model.Task{}).Where( + "id = ? AND status = ? AND (lease_owner = '' OR lease_owner NOT LIKE ? OR lease_expires_at IS NULL OR lease_expires_at <= ?)", + id, model.TaskStatusFailed, "manual-recovery:%", now, + ) + if strings.TrimSpace(userID) != "" { + query = query.Where("user_id = ?", userID) + } + result := query.Updates(map[string]any{ + "lease_owner": owner, + "lease_expires_at": now.Add(leaseDuration), + "updated_at": now, + }) + if result.Error != nil { + return result.Error + } + if result.RowsAffected != 1 { + return ErrTaskProviderRecoveryConflict + } + return nil +} + +func (r *Repository) ReleaseTaskProviderRecovery(id string, owner string) error { + return r.db.Model(&model.Task{}). + Where("id = ? AND lease_owner = ?", id, owner). + Updates(map[string]any{"lease_owner": "", "lease_expires_at": nil, "updated_at": time.Now()}).Error +} + func (r *Repository) UpdateTaskProgress(id string, stage string, progress int) error { return r.db.Model(&model.Task{}).Where("id = ?", id).Updates(map[string]any{ "stage": stage, "progress": progress, "updated_at": time.Now(), diff --git a/backend/internal/service/analytics.go b/backend/internal/service/analytics.go index d56e87b24..d3a662e4b 100644 --- a/backend/internal/service/analytics.go +++ b/backend/internal/service/analytics.go @@ -264,6 +264,7 @@ func (s *Service) decorateAPICallLogs(logs []model.ApiCallLog) error { logs[index].UserAccount = user.Username } if task, exists := taskByID[logs[index].TaskID]; exists && task.UserID == logs[index].UserID { + logs[index].TaskStatus = task.Status previewURL, previewKind := taskMediaPreview(task.ResultJSON, task.Type) if canvasResourceID(previewURL) != "" { logs[index].MediaPreviewURL = "/api/admin/api-logs/" + logs[index].ID + "/media" diff --git a/backend/internal/service/provider.go b/backend/internal/service/provider.go index 1e7699dbf..1b9196b98 100644 --- a/backend/internal/service/provider.go +++ b/backend/internal/service/provider.go @@ -11,6 +11,7 @@ import ( "math" "mime" "mime/multipart" + "net" "net/http" "net/textproto" "net/url" @@ -987,10 +988,20 @@ func nestedNewAPIVideoResultURL(payload map[string]interface{}, allowResultURL b const newAPIChannel1VideoPollInterval = 20 * time.Second const ( - newAPIChannel2VideoPollInterval = 5 * time.Second - newAPIChannel2VideoPollTimeout = 5 * time.Minute + newAPIChannel2VideoPollInterval = 10 * time.Second + newAPIChannel2VideoRetryInterval = time.Minute + newAPIChannel2VideoMaxQueryRetries = 3 ) +type newAPIChannel2ResponseError struct { + Code string + Message string +} + +func (e newAPIChannel2ResponseError) Error() string { + return fmt.Sprintf("NewAPI Video Generations 任务查询失败（%s）：%s", e.Code, defaultString(e.Message, "上游查询失败")) +} + func runNewAPIChannel2VideoTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) { id := resumedProviderRequestID(ctx) var created map[string]interface{} @@ -1013,34 +1024,23 @@ func runNewAPIChannel2VideoTask(ctx context.Context, input canvasGenerationInput return nil, errors.New("NewAPI Video Generations 没有返回任务 ID") } - for deadline := time.Now().Add(newAPIChannel2VideoPollTimeout); time.Now().Before(deadline); { - var state map[string]interface{} - if err := getJSON(ctx, input.Config, "/video/generations/"+id, &state); err != nil { - return nil, err - } - if data, ok := state["data"].(map[string]interface{}); ok { - state = data - } - status := strings.ToUpper(strings.TrimSpace(stringField(state, "status"))) - switch status { - case "SUCCESS": - videoURL := strings.TrimSpace(stringField(state, "result_url")) - if videoURL == "" { - return nil, fmt.Errorf("NewAPI Video Generations 任务 %s 已成功但没有返回视频地址", id) + consecutiveQueryFailures := 0 + for deadline := providerPollingDeadline(ctx); time.Now().Before(deadline); { + result, _, err := queryNewAPIChannel2VideoTask(ctx, input, id) + if err != nil { + if !isTransientNewAPIChannel2QueryError(err) || consecutiveQueryFailures >= newAPIChannel2VideoMaxQueryRetries { + return nil, err } - data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) - if err != nil { - return nil, fmt.Errorf("NewAPI Video Generations 视频结果下载失败（任务 %s）：%w", id, err) + consecutiveQueryFailures++ + logNewAPIChannel2QueryRetry(ctx, id, consecutiveQueryFailures, err) + if err := sleepContext(ctx, newAPIChannel2VideoRetryInterval); err != nil { + return nil, err } - mimeType = normalizedMediaMimeType(mimeType, data) - return map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}, nil - case "FAILURE": - reason := strings.TrimSpace(stringField(state, "fail_reason")) - return nil, fmt.Errorf("NewAPI Video Generations 视频生成失败（任务 %s）：%s", id, defaultString(reason, "上游返回失败")) - case "SUBMITTED", "QUEUED", "IN_PROGRESS", "NOT_START", "": - // 按上游协议继续轮询；空状态也可能出现在任务刚写入队列时。 - default: - return nil, fmt.Errorf("NewAPI Video Generations 任务 %s 返回未知状态：%s", id, status) + continue + } + consecutiveQueryFailures = 0 + if result != nil { + return result, nil } if err := sleepContext(ctx, newAPIChannel2VideoPollInterval); err != nil { return nil, err @@ -1049,6 +1049,83 @@ func runNewAPIChannel2VideoTask(ctx context.Context, input canvasGenerationInput return nil, fmt.Errorf("NewAPI Video Generations 视频生成超时（任务 %s）", id) } +// 单次查询只读取既有上游任务，不创建新任务；自动轮询和人工恢复共用这条安全边界。 +func queryNewAPIChannel2VideoTask(ctx context.Context, input canvasGenerationInput, id string) (map[string]interface{}, string, error) { + var payload map[string]interface{} + if err := getJSON(ctx, input.Config, "/video/generations/"+id, &payload); err != nil { + return nil, "", err + } + if err := newAPIChannel2PayloadError(payload); err != nil { + return nil, "", err + } + state := payload + if data, ok := payload["data"].(map[string]interface{}); ok { + state = data + } + status := strings.ToUpper(strings.TrimSpace(stringField(state, "status"))) + switch status { + case "SUCCESS": + videoURL := strings.TrimSpace(stringField(state, "result_url")) + if videoURL == "" { + return nil, status, fmt.Errorf("NewAPI Video Generations 任务 %s 已成功但没有返回视频地址", id) + } + data, mimeType, err := getExternalBinary(withProviderRequestKind(ctx, "download"), videoURL) + if err != nil { + return nil, status, fmt.Errorf("NewAPI Video Generations 视频结果下载失败（任务 %s）：%w", id, err) + } + mimeType = normalizedMediaMimeType(mimeType, data) + return map[string]interface{}{"mode": "video", "video": map[string]interface{}{"dataUrl": dataURL(mimeType, data), "mimeType": mimeType}}, status, nil + case "FAILURE": + reason := strings.TrimSpace(stringField(state, "fail_reason")) + return nil, status, fmt.Errorf("NewAPI Video Generations 视频生成失败（任务 %s）：%s", id, defaultString(reason, "上游返回失败")) + case "SUBMITTED", "QUEUED", "IN_PROGRESS", "NOT_START", "": + return nil, status, nil + default: + return nil, status, fmt.Errorf("NewAPI Video Generations 任务 %s 返回未知状态：%s", id, status) + } +} + +func newAPIChannel2PayloadError(payload map[string]interface{}) error { + code := strings.ToLower(strings.TrimSpace(stringField(payload, "code"))) + if code == "" || code == "0" || code == "ok" || code == "success" { + return nil + } + return newAPIChannel2ResponseError{Code: code, Message: firstNonEmptyString(stringField(payload, "message"), stringField(payload, "msg"))} +} + +func isTransientNewAPIChannel2QueryError(err error) bool { + if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + var responseErr newAPIChannel2ResponseError + if errors.As(err, &responseErr) { + return responseErr.Code == "do_request_failed" || strings.Contains(strings.ToLower(responseErr.Message), "do request failed") + } + var httpErr providerHTTPError + if errors.As(err, &httpErr) { + body := strings.ToLower(httpErr.Body) + return httpErr.StatusCode == http.StatusRequestTimeout || httpErr.StatusCode == http.StatusTooManyRequests || httpErr.StatusCode >= http.StatusInternalServerError || strings.Contains(body, "do_request_failed") || strings.Contains(body, "do request failed") + } + if errors.Is(err, io.ErrUnexpectedEOF) { + return true + } + var networkErr net.Error + if errors.As(err, &networkErr) { + return true + } + message := strings.ToLower(err.Error()) + return strings.Contains(message, "do_request_failed") || strings.Contains(message, "do request failed") +} + +func logNewAPIChannel2QueryRetry(ctx context.Context, providerTaskID string, retry int, err error) { + metadata, ok := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext) + if !ok || metadata.Service == nil || metadata.TaskID == "" { + return + } + payload := fmt.Sprintf("供应商任务 %s，第 %d/%d 次重试：%s", providerTaskID, retry, newAPIChannel2VideoMaxQueryRetries, safeProviderLogError(err)) + _ = metadata.Service.log(metadata.UserID, metadata.TaskID, "warn", "上游任务查询失败，1 分钟后重试", payload) +} + func newAPIChannel2VideoBody(input canvasGenerationInput) (map[string]interface{}, error) { if len(input.ReferenceImages) > 9 || len(input.ReferenceVideos) > 3 || len(input.ReferenceAudios) > 3 { return nil, errors.New("NewAPI Video Generations 最多支持 9 张参考图、3 个参考视频和 3 个参考音频") diff --git a/backend/internal/service/provider_task_recovery.go b/backend/internal/service/provider_task_recovery.go new file mode 100644 index 000000000..36b1df6b5 --- /dev/null +++ b/backend/internal/service/provider_task_recovery.go @@ -0,0 +1,166 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + "infinite-canvas/backend/internal/model" + "infinite-canvas/backend/internal/repository" +) + +const providerTaskRecoveryLeaseDuration = 10 * time.Minute + +type ProviderTaskQueryResult struct { + Task *model.Task `json:"task"` + ProviderStatus string `json:"providerStatus"` + Recovered bool `json:"recovered"` + BillingSettled bool `json:"billingSettled"` +} + +func (s *Service) QueryFailedVideoTask(ctx context.Context, userID string, taskID string) (*ProviderTaskQueryResult, error) { + task, err := s.repo.TaskForUser(strings.TrimSpace(userID), strings.TrimSpace(taskID)) + if err != nil { + return nil, err + } + return s.queryFailedVideoTask(ctx, task, strings.TrimSpace(userID)) +} + +func (s *Service) AdminQueryFailedVideoTask(ctx context.Context, actor *model.User, logID string) (*ProviderTaskQueryResult, error) { + if err := s.RequireAdmin(actor); err != nil { + return nil, err + } + log, err := s.repo.APICallLog(strings.TrimSpace(logID)) + if err != nil { + return nil, err + } + if log.Capability != "video" || strings.TrimSpace(log.TaskID) == "" { + return nil, BadAuthRequest("该请求没有可查询的视频任务") + } + task, err := s.repo.Task(log.TaskID) + if err != nil { + return nil, err + } + if task.UserID != log.UserID { + return nil, BadAuthRequest("请求与任务归属不一致") + } + if task.ProviderRequestID == "" { + task.ProviderRequestID = strings.TrimSpace(log.ProviderRequestID) + } + result, err := s.queryFailedVideoTask(ctx, task, "") + if err != nil { + return nil, err + } + if err := s.appendAdminAudit(actor, "api_log.query_provider_task", "task", task.ID, "人工查询失败视频任务", map[string]any{ + "apiCallLogId": log.ID, "providerRequestId": task.ProviderRequestID, "providerStatus": result.ProviderStatus, "recovered": result.Recovered, + }); err != nil { + return nil, err + } + return result, nil +} + +func (s *Service) queryFailedVideoTask(ctx context.Context, task *model.Task, claimUserID string) (*ProviderTaskQueryResult, error) { + if task == nil || task.ID == "" { + return nil, BadAuthRequest("任务不存在") + } + if task.Status != model.TaskStatusFailed { + return nil, BadAuthRequest("只能人工查询状态为失败的任务") + } + if !strings.HasPrefix(task.Type, "canvas_video") && !strings.HasPrefix(task.Type, "video_") { + return nil, BadAuthRequest("该任务不是视频生成任务") + } + + s.hydrateTaskProviderRequestID(task) + providerRequestID := strings.TrimSpace(task.ProviderRequestID) + if providerRequestID == "" { + return nil, BadAuthRequest("该任务没有可恢复的上游任务 ID") + } + if task.BillingOrderID != "" { + order, err := s.repo.BillingOrder(task.BillingOrderID) + if err != nil { + return nil, err + } + if order.UserID != task.UserID || order.TaskID != task.ID { + return nil, BadAuthRequest("任务与计费订单归属不一致") + } + if order.Status == model.BillingStatusRefunded { + return nil, BadAuthRequest("该任务已退款，不能恢复并重新扣费") + } + } + + decryptedInput, err := s.decryptTaskInputJSON(task.InputJSON) + if err != nil { + return nil, fmt.Errorf("读取任务配置失败：%w", err) + } + var input canvasGenerationInput + if err := json.Unmarshal([]byte(decryptedInput), &input); err != nil { + return nil, fmt.Errorf("任务输入解析失败：%w", err) + } + config, err := s.resolveProviderConfig(input.Config) + if err != nil { + return nil, err + } + if config.InterfaceType != string(model.ChannelInterfaceNewAPIChannel2) { + return nil, BadAuthRequest("该任务不使用 NewAPI Video Generations 协议") + } + input.Config = config + task.InputJSON = decryptedInput + task.ProviderRequestID = providerRequestID + if err := s.repo.UpdateTaskProviderState(task.ID, providerRequestID, task.PollStage, task.NextPollAt); err != nil { + return nil, err + } + + owner := "manual-recovery:" + newID() + if err := s.repo.ClaimFailedTaskProviderRecovery(task.ID, claimUserID, owner, providerTaskRecoveryLeaseDuration); err != nil { + if errors.Is(err, repository.ErrTaskProviderRecoveryConflict) { + return nil, &AuthError{Status: 409, Message: "该任务正在查询上游状态，请稍后再试"} + } + return nil, err + } + task.LeaseOwner = owner + defer func() { _ = s.repo.ReleaseTaskProviderRecovery(task.ID, owner) }() + + queryCtx := withProviderAnalytics(ctx, s, *task) + result, providerStatus, err := queryNewAPIChannel2VideoTask(queryCtx, input, providerRequestID) + if err != nil { + _ = s.log(task.UserID, task.ID, "error", "人工查询上游视频任务失败", err.Error()) + return nil, err + } + if result == nil { + _ = s.log(task.UserID, task.ID, "info", "人工查询完成，上游任务仍在处理", providerStatus) + return &ProviderTaskQueryResult{Task: taskForOutput(*task), ProviderStatus: providerStatus, Recovered: false}, nil + } + + result, err = s.persistGeneratedMediaResult(task.UserID, result) + if err != nil { + _ = s.log(task.UserID, task.ID, "error", "人工查询已取得视频，但结果保存失败", err.Error()) + return nil, err + } + resultJSON, err := json.Marshal(result) + if err != nil { + return nil, err + } + task.Error = "" + task.PollStage = strings.ToLower(providerStatus) + task.NextPollAt = nil + if err := s.saveTaskCompletionWithinStorageQuota(task, resultJSON, nil, false); err != nil { + _ = s.MarkBillingUncertain(task.BillingOrderID, "人工查询确认上游成功，但任务结果未保存："+err.Error()) + _ = s.log(task.UserID, task.ID, "error", "人工查询已取得视频，但任务恢复失败", err.Error()) + return nil, err + } + if err := s.RegisterTaskOutputFromTask(*task); err != nil { + _ = s.log(task.UserID, task.ID, "error", "任务恢复成功但项目产物登记失败", err.Error()) + } + billingSettled := true + if err := s.SettleBilling(task.BillingOrderID, providerRequestID); err != nil { + billingSettled = false + _ = s.MarkBillingUncertain(task.BillingOrderID, "人工查询确认生成成功，但积分结算失败："+err.Error()) + _ = s.log(task.UserID, task.ID, "error", "任务恢复成功但积分结算失败，已进入待核对", err.Error()) + } else { + _ = s.log(task.UserID, task.ID, "info", "人工查询确认生成成功，任务已恢复并完成结算", providerStatus) + } + return &ProviderTaskQueryResult{Task: taskForOutput(*task), ProviderStatus: providerStatus, Recovered: true, BillingSettled: billingSettled}, nil +} diff --git a/backend/internal/service/service.go b/backend/internal/service/service.go index e2c171eba..07c1a42ff 100644 --- a/backend/internal/service/service.go +++ b/backend/internal/service/service.go @@ -68,26 +68,27 @@ type SessionDetail struct { } type TaskSummary struct { - ID string `json:"id"` - SessionID string `json:"sessionId,omitempty"` - ProjectID string `json:"projectId,omitempty"` - Type string `json:"type"` - Status model.TaskStatus `json:"status"` - Stage string `json:"stage"` - Progress int `json:"progress"` - Prompt string `json:"prompt"` - Operation string `json:"operation,omitempty"` - Provider string `json:"provider,omitempty"` - Model string `json:"model,omitempty"` - ErrorCode string `json:"errorCode,omitempty"` - PreviewURL string `json:"previewUrl,omitempty"` - PreviewKind string `json:"previewKind,omitempty"` - Attempts int `json:"attempts"` - StartedAt *time.Time `json:"startedAt"` - CompletedAt *time.Time `json:"completedAt"` - CreatedAt time.Time `json:"createdAt"` - UpdatedAt time.Time `json:"updatedAt"` - Billing *TaskBillingSummary `json:"billing,omitempty"` + ID string `json:"id"` + SessionID string `json:"sessionId,omitempty"` + ProjectID string `json:"projectId,omitempty"` + Type string `json:"type"` + Status model.TaskStatus `json:"status"` + Stage string `json:"stage"` + Progress int `json:"progress"` + Prompt string `json:"prompt"` + Operation string `json:"operation,omitempty"` + Provider string `json:"provider,omitempty"` + Model string `json:"model,omitempty"` + ProviderRequestID string `json:"providerRequestId,omitempty"` + ErrorCode string `json:"errorCode,omitempty"` + PreviewURL string `json:"previewUrl,omitempty"` + PreviewKind string `json:"previewKind,omitempty"` + Attempts int `json:"attempts"` + StartedAt *time.Time `json:"startedAt"` + CompletedAt *time.Time `json:"completedAt"` + CreatedAt time.Time `json:"createdAt"` + UpdatedAt time.Time `json:"updatedAt"` + Billing *TaskBillingSummary `json:"billing,omitempty"` } type TaskBillingSummary struct { @@ -386,9 +387,43 @@ func (s *Service) Task(userID string, id string) (*model.Task, error) { if err != nil { return nil, err } + s.hydrateTaskProviderRequestID(task) return taskForOutput(*task), nil } +func (s *Service) hydrateTaskProviderRequestID(task *model.Task) { + if task == nil || task.ProviderRequestID != "" { + return + } + if task.BillingOrderID != "" { + if order, err := s.repo.BillingOrder(task.BillingOrderID); err == nil { + task.ProviderRequestID = strings.TrimSpace(order.ProviderRequestID) + } + } + if task.ProviderRequestID == "" { + if providerRequestID, err := s.repo.LatestProviderRequestIDForTask(task.ID); err == nil { + task.ProviderRequestID = providerRequestID + } + } +} + +// 上游请求日志会在任务执行期间更新 provider 状态，终态保存前必须重新合并，避免旧任务对象覆盖可恢复 ID。 +func (s *Service) refreshTaskProviderState(task *model.Task) error { + if task == nil || task.ID == "" { + return errors.New("任务状态无效") + } + latest, err := s.repo.Task(task.ID) + if err != nil { + return fmt.Errorf("刷新任务上游状态失败：%w", err) + } + if latest.ProviderRequestID != "" { + task.ProviderRequestID = latest.ProviderRequestID + } + task.PollStage = latest.PollStage + task.NextPollAt = latest.NextPollAt + return nil +} + func (s *Service) RetryTask(userID string, id string) (*model.Task, error) { task, err := s.repo.TaskForUser(userID, id) if err != nil { @@ -521,6 +556,9 @@ func taskSummariesForOutputWithBilling(tasks []model.Task, orders map[string]mod summary := taskSummaryForOutput(task) if order, ok := orders[task.ID]; ok { summary.Billing = &TaskBillingSummary{AmountMicrocredits: order.AmountMicrocredits, Status: order.Status} + if summary.ProviderRequestID == "" { + summary.ProviderRequestID = order.ProviderRequestID + } } result = append(result, summary) } @@ -550,25 +588,26 @@ func taskSummaryForOutput(task model.Task) TaskSummary { } previewURL, previewKind := taskMediaPreview(task.ResultJSON, task.Type) return TaskSummary{ - ID: task.ID, - SessionID: task.SessionID, - ProjectID: task.ProjectID, - Type: task.Type, - Status: task.Status, - Stage: task.Stage, - Progress: task.Progress, - Prompt: truncateRunes(task.Prompt, 500), - Operation: task.Operation, - Provider: task.Provider, - Model: task.Model, - ErrorCode: errorCode, - PreviewURL: previewURL, - PreviewKind: previewKind, - Attempts: task.Attempts, - StartedAt: task.StartedAt, - CompletedAt: task.CompletedAt, - CreatedAt: task.CreatedAt, - UpdatedAt: task.UpdatedAt, + ID: task.ID, + SessionID: task.SessionID, + ProjectID: task.ProjectID, + Type: task.Type, + Status: task.Status, + Stage: task.Stage, + Progress: task.Progress, + Prompt: truncateRunes(task.Prompt, 500), + Operation: task.Operation, + Provider: task.Provider, + Model: task.Model, + ProviderRequestID: task.ProviderRequestID, + ErrorCode: errorCode, + PreviewURL: previewURL, + PreviewKind: previewKind, + Attempts: task.Attempts, + StartedAt: task.StartedAt, + CompletedAt: task.CompletedAt, + CreatedAt: task.CreatedAt, + UpdatedAt: task.UpdatedAt, } } @@ -778,6 +817,9 @@ func (s *Service) processClaimedTask(task *model.Task) error { return err } result, canvasOps, err := s.processTask(ctx, *task) + if stateErr := s.refreshTaskProviderState(task); stateErr != nil { + return stateErr + } providerSucceeded := err == nil if err == nil { result, err = s.persistGeneratedMediaResult(task.UserID, result) diff --git a/web/src/pages/admin/components/api-log-detail-drawer.tsx b/web/src/pages/admin/components/api-log-detail-drawer.tsx index cd47f11f6..4a976b03b 100644 --- a/web/src/pages/admin/components/api-log-detail-drawer.tsx +++ b/web/src/pages/admin/components/api-log-detail-drawer.tsx @@ -1,12 +1,14 @@ import { useEffect, useState } from "react"; -import { App, Descriptions, Drawer, Empty, Skeleton, Tabs, Tag, Typography } from "antd"; +import { App, Button, Descriptions, Drawer, Empty, Skeleton, Tabs, Tag, Typography } from "antd"; +import { RefreshCw } from "lucide-react"; -import { getAdminApiLog, type ApiCallLog } from "@/services/api/auth"; +import { getAdminApiLog, queryAdminApiLogTask, type ApiCallLog } from "@/services/api/auth"; -export function ApiLogDetailDrawer({ logId, onClose }: { logId: string | null; onClose: () => void }) { +export function ApiLogDetailDrawer({ logId, onClose, onLogUpdated }: { logId: string | null; onClose: () => void; onLogUpdated?: (log: ApiCallLog) => void }) { const { message } = App.useApp(); const [log, setLog] = useState(null); const [loading, setLoading] = useState(false); + const [querying, setQuerying] = useState(false); useEffect(() => { if (!logId) return; let active = true; @@ -19,14 +21,36 @@ export function ApiLogDetailDrawer({ logId, onClose }: { logId: string | null; o return () => { active = false; }; }, [logId, message]); + const queryProviderTask = async () => { + if (!log) return; + setQuerying(true); + try { + const result = await queryAdminApiLogTask(log.id); + const refreshed = await getAdminApiLog(log.id); + setLog(refreshed.log); + onLogUpdated?.(refreshed.log); + if (result.recovered) { + window.dispatchEvent(new CustomEvent("wallet:updated")); + if (result.billingSettled) message.success("已获取上游视频，任务已恢复并完成结算"); + else message.warning("已获取上游视频，任务已恢复，计费状态待核对"); + } else { + message.info(`上游任务仍在处理中${result.providerStatus ? `（${result.providerStatus}）` : ""}`); + } + } catch (error) { + message.error(error instanceof Error ? error.message : "查询上游任务失败"); + } finally { + setQuerying(false); + } + }; + return ( - {loading ? : log ? : } + {loading ? : log ? : } ); } -function LogDetail({ log }: { log: ApiCallLog }) { +function LogDetail({ log, querying, onQueryProviderTask }: { log: ApiCallLog; querying: boolean; onQueryProviderTask: () => void }) { const providerStatus = log.providerStatus?.toLowerCase(); const processing = ["queued", "pending", "processing", "running", "in_progress"].includes(providerStatus || ""); const failed = log.status === "failed" || ["failed", "cancelled", "expired"].includes(providerStatus || ""); @@ -49,7 +73,10 @@ function LogDetail({ log }: { log: ApiCallLog }) { ["上游地址", log.upstreamUrl || "--"], ].map(([label, children], index) => ({ key: String(index), label, children })); + const canQueryProviderTask = log.capability === "video" && log.taskStatus === "failed" && Boolean(log.taskId && log.providerRequestId); + return
+ {canQueryProviderTask ?
: null}
原始报文
diff --git a/web/src/pages/admin/logs/logs-page.tsx b/web/src/pages/admin/logs/logs-page.tsx index a097ce3cc..0995545ef 100644 --- a/web/src/pages/admin/logs/logs-page.tsx +++ b/web/src/pages/admin/logs/logs-page.tsx @@ -78,7 +78,7 @@ export default function LogsPage() { {loading && logs.length === 0 ? : setSelectedIds(keys.map(String)) }} columns={columns} dataSource={logs} locale={{ emptyText: }} pagination={{ current: page, pageSize, total, showSizeChanger: true, pageSizeOptions: [20, 50, 100], showTotal: (value, range) => `${range[0]}-${range[1]} / 共 ${value} 条`, onChange: (nextPage, nextSize) => updateUrl({ page: nextSize !== pageSize ? 1 : nextPage, pageSize: nextSize }) }} scroll={{ x: 1700 }} />} - setDetailLogId(null)} /> + setDetailLogId(null)} onLogUpdated={(next) => setLogs((items) => items.map((item) => item.id === next.id ? next : item))} /> setMediaPreview(null)} footer={mediaPreview ? : null} destroyOnHidden> {mediaPreview?.kind === "video" ? diff --git a/web/src/pages/tasks/index.tsx b/web/src/pages/tasks/index.tsx index 1cf313bd9..ca31bd97e 100644 --- a/web/src/pages/tasks/index.tsx +++ b/web/src/pages/tasks/index.tsx @@ -9,7 +9,7 @@ import { WorkspaceState } from "@/components/layout/workspace-state"; import { CONTENT_MODERATION_ERROR_CODE, generationErrorMessage, isContentModerationError } from "@/lib/generation-error"; import { formatTaskKind, operationOptions, statusLabel } from "@/lib/generation-task-display"; -import { cancelGenerationTask, createAgentSession, createGenerationTask, listGenerationTasks, listTaskLogs, queryGenerationTask, retryGenerationTask, type CreateTaskInput, type GenerationTask, type TaskLog, type TaskStatus } from "@/services/api/task-center"; +import { cancelGenerationTask, createAgentSession, createGenerationTask, listGenerationTasks, listTaskLogs, queryFailedVideoProviderTask, queryGenerationTask, retryGenerationTask, type CreateTaskInput, type GenerationTask, type TaskLog, type TaskStatus } from "@/services/api/task-center"; import { syncGenerationTaskToCanvasStore } from "@/lib/canvas/canvas-generation-task-sync"; import { useCanvasStore } from "@/stores/canvas/use-canvas-store"; import { resolveModelRequestConfig, useConfigStore, useEffectiveConfig, type AiConfig } from "@/stores/use-config-store"; @@ -191,6 +191,30 @@ export default function TasksPage() { } }; + const queryProviderTask = async (task: GenerationTask) => { + setActingId(task.id); + try { + const result = await queryFailedVideoProviderTask(task.id); + if (!result.recovered) { + setTaskLogs(await listTaskLogs(task.id)); + message.info(`上游任务仍在处理中${result.providerStatus ? `（${result.providerStatus}）` : ""}`); + return; + } + setDetailTask(result.task); + setTasks((items) => items.map((item) => (item.id === task.id ? { ...item, ...result.task } : item))); + setTaskLogs(await listTaskLogs(task.id)); + await syncGenerationTaskToCanvasStore(result.task); + window.dispatchEvent(new CustomEvent("wallet:updated")); + void loadTasks(false); + if (result.billingSettled) message.success("已获取上游视频，任务已恢复并完成结算"); + else message.warning("已获取上游视频，任务已恢复，计费状态待管理员核对"); + } catch (error) { + message.error(error instanceof Error ? error.message : "查询上游任务失败"); + } finally { + setActingId(""); + } + }; + const submitTask = async () => { const values = await form.validateFields(); setCreating(true); @@ -422,6 +446,7 @@ export default function TasksPage() { + {canQueryProviderTask(detailTask) ?
: null} {detailTask.error ? 
{generationErrorMessage(detailTask.error)}
: null} @@ -453,6 +478,10 @@ export default function TasksPage() { ); } +function canQueryProviderTask(task: GenerationTask) { + return task.status === "failed" && (task.type.startsWith("canvas_video") || task.type.startsWith("video_")) && Boolean(task.providerRequestId); +} + function reconcileTaskSummaries(current: GenerationTask[], next: GenerationTask[]) { if (current.length !== next.length) return next; const currentById = new Map(current.map((task) => [task.id, task])); diff --git a/web/src/services/api/auth.ts b/web/src/services/api/auth.ts index e3d3e6bd3..55383c41a 100644 --- a/web/src/services/api/auth.ts +++ b/web/src/services/api/auth.ts @@ -2,6 +2,7 @@ import axios from "axios"; import type { ModelChannel } from "@/stores/use-config-store"; import type { CreditLedgerEntry } from "@/services/api/wallet"; +import type { GenerationTask, TaskStatus } from "@/services/api/task-center"; const api = axios.create({ baseURL: import.meta.env.VITE_CANVAS_BACKEND_URL || "/api", withCredentials: true }); @@ -46,6 +47,7 @@ export type ApiCallLog = { channelId: string; channelName: string; taskId?: string; + taskStatus?: TaskStatus; source: string; capability: "text" | "image" | "video" | "audio" | ""; operation?: string; @@ -83,6 +85,13 @@ export type ApiCallLog = { createdAt: string; }; +export type AdminProviderTaskQueryResult = { + task: GenerationTask; + providerStatus: string; + recovered: boolean; + billingSettled: boolean; +}; + export type AdminAuditEvent = { id: string; actorUserId: string; @@ -430,6 +439,10 @@ export function getAdminApiLog(id: string) { return request<{ log: ApiCallLog }>(api.get(`/admin/api-logs/${encodeURIComponent(id)}`)); } +export function queryAdminApiLogTask(id: string) { + return request(api.post(`/admin/api-logs/${encodeURIComponent(id)}/query-task`)); +} + export async function exportAdminApiLogs(params: AdminListParams & { ids?: string[] } = {}) { const response = await api.get("/admin/api-logs-export.csv", { params: { ...params, ids: params.ids?.join(",") }, responseType: "blob" }); return response.data; diff --git a/web/src/services/api/task-center.ts b/web/src/services/api/task-center.ts index e917ed6be..05139237c 100644 --- a/web/src/services/api/task-center.ts +++ b/web/src/services/api/task-center.ts @@ -24,6 +24,7 @@ export type GenerationTask = { operation?: string; provider?: string; model?: string; + providerRequestId?: string; errorCode?: string; previewUrl?: string; previewKind?: "image" | "video"; @@ -43,6 +44,13 @@ export type GenerationTask = { updated_at?: string; }; +export type ProviderTaskQueryResult = { + task: GenerationTask; + providerStatus: string; + recovered: boolean; + billingSettled: boolean; +}; + export type AgentSession = { id: string; projectId?: string; @@ -185,6 +193,10 @@ export function retryGenerationTask(id: string) { return request(api.post(`/tasks/${encodeURIComponent(id)}/retry`)); } +export function queryFailedVideoProviderTask(id: string) { + return request(api.post(`/tasks/${encodeURIComponent(id)}/query-provider`)); +} + export function cancelGenerationTask(id: string) { return request(api.post(`/tasks/${encodeURIComponent(id)}/cancel`)); }