fix(media): 视频任务 - 统一轮询超时并支持失败任务人工恢复
This commit is contained in:
1 parent
b763f62ec6
commit
96f99ea9d6
14 files changed
+510
-75
No files matched your search
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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"})
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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 个参考音频")
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
Reference in new issue
Block a user