becomer/likedream
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🐛 查寝各种问题修复

Browse Source
This commit is contained in:
sugarBecomer
2025-11-07 23:36:02 +08:00
parent 618d6384d9
commit 948d118a4f
10 changed files with 76 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/application/cpdaily/imageuploader/upload.go
+1 -1
View File
@@ -75,7 +75,7 @@ func UploadImageFromStorage(ctx context.Context, host string, imageId uuid.UUID,
utils.OkOrPanic(errors.NewBusinessError(503, "上传图片失败: "+utils.ReadString(resp.Body)), oops.WithContext(ctx).With("response", utils.ReadString(resp.Body)).Wrap)
}
previewUrl := GetPreviewUrlAsync(ctx, host, targetImage.Key, taskType, client)
previewUrl := GetPreviewUrlAsync(ctx, host, imageUploadPolicy.FileName, taskType, client)
return previewUrl
}
internal/application/cpdaily/task/sleepcheck.go
+1 -1
View File
@@ -585,7 +585,7 @@ func (c *CpdailySleepCheckTask) FillCustomForm(ctx context.Context, form map[str
previewUrl := imageuploader.UploadImageFromStorage(ctx, c.host, imageId, types.TaskTypeSleepCheck, c.client)
resultForm["signImg"] = gvar.New(previewUrl)
resultForm["signPhotoUrl"] = gvar.New(previewUrl)
}
// 判断是否存在二维码查寝
internal/ent/cptask/cptask.go
+2 -1
View File
@@ -160,6 +160,7 @@ const (
TaskStatusDone TaskStatus = "done"
TaskStatusFail TaskStatus = "fail"
TaskStatusPause TaskStatus = "pause"
TaskStatusQueue TaskStatus = "queue"
TaskStatusUnknown TaskStatus = "unknown"
)
@@ -170,7 +171,7 @@ func (ts TaskStatus) String() string {
// TaskStatusValidator is a validator for the "task_status" field enum values. It is called by the builders before save.
func TaskStatusValidator(ts TaskStatus) error {
switch ts {
case TaskStatusNotStart, TaskStatusDoing, TaskStatusDone, TaskStatusFail, TaskStatusPause, TaskStatusUnknown:
case TaskStatusNotStart, TaskStatusDoing, TaskStatusDone, TaskStatusFail, TaskStatusPause, TaskStatusQueue, TaskStatusUnknown:
return nil
default:
return fmt.Errorf("cptask: invalid enum value for task_status field: %q", ts)
internal/ent/internal/schema.go
+1 -1
View File
File diff suppressed because one or more lines are too long
internal/ent/migrate/schema.go
+4 -4
View File
@@ -57,7 +57,7 @@ var (
{Name: "task_type", Type: field.TypeEnum, Enums: []string{"collection", "sign", "sleep_check"}},
{Name: "title", Type: field.TypeString},
{Name: "frequency", Type: field.TypeInt, Default: 0},
{Name: "task_status", Type: field.TypeEnum, Enums: []string{"not_start", "doing", "done", "fail", "pause", "unknown"}},
{Name: "task_status", Type: field.TypeEnum, Enums: []string{"not_start", "doing", "done", "fail", "pause", "queue", "unknown"}},
{Name: "success_count", Type: field.TypeInt, Default: 0},
{Name: "form_data", Type: field.TypeString, Nullable: true, Size: 2147483647},
{Name: "step", Type: field.TypeEnum, Enums: []string{"form", "execute", "complete"}},
@@ -84,7 +84,7 @@ var (
Symbol: "cp_tasks_sys_user_cp_task",
Columns: []*schema.Column{CpTasksColumns[15]},
RefColumns: []*schema.Column{SysUserColumns[0]},
OnDelete: schema.NoAction,
OnDelete: schema.Cascade,
},
},
}
@@ -122,7 +122,7 @@ var (
Symbol: "cp_task_logs_sys_user_cp_task_logs",
Columns: []*schema.Column{CpTaskLogsColumns[8]},
RefColumns: []*schema.Column{SysUserColumns[0]},
OnDelete: schema.NoAction,
OnDelete: schema.Cascade,
},
},
}
@@ -179,7 +179,7 @@ var (
Symbol: "sys_agent_sys_user_agent",
Columns: []*schema.Column{SysAgentColumns[5]},
RefColumns: []*schema.Column{SysUserColumns[0]},
OnDelete: schema.SetNull,
OnDelete: schema.Cascade,
},
},
}
internal/ent/schema/cptask.go
+1 -1
View File
@@ -22,7 +22,7 @@ func (CpTask) Fields() []ent.Field {
field.Enum("task_type").Values("collection", "sign", "sleep_check"),
field.String("title").NotEmpty(),
field.Int("frequency").Default(0),
field.Enum("task_status").Values("not_start", "doing", "done", "fail", "pause", "unknown"),
field.Enum("task_status").Values("not_start", "doing", "done", "fail", "pause", "queue", "unknown"),
field.Int("success_count").Default(0),
field.Text("form_data").Optional(),
field.Enum("step").Values("form", "execute", "complete"),
internal/framework/schedule/cpdaily_handler.go
+13 -3
View File
@@ -14,6 +14,7 @@ import (
"likedream-goframe/internal/ent/cptask"
"likedream-goframe/internal/ent/cpuser"
"likedream-goframe/internal/framework/errors"
"likedream-goframe/internal/framework/logger"
"likedream-goframe/internal/framework/model/email"
email2 "likedream-goframe/internal/service/email"
"likedream-goframe/internal/utils"
@@ -23,6 +24,7 @@ import (
"github.com/google/uuid"
"github.com/hibiken/asynq"
"github.com/samber/oops"
"go.uber.org/zap"
)
var emailService = email2.NewV1()
@@ -132,13 +134,15 @@ func (c *CpdailyTaskHandler) ProcessTask(ctx context.Context, task *asynq.Task)
submitForm := taskExecutor.FillCustomForm(ctx, cpdailyRealTask, customFormItems, deviceID, targetTask.Edges.CpUser.Username)
taskExecutor.Submit(ctx, submitForm, targetTask.Edges.CpUser.Username, deviceID)
db.Global.CpTask.UpdateOneID(taskPayload.CpTaskID).SetTaskStatus(cptask.TaskStatusDone).AddSuccessCount(1).ExecX(ctx)
// 任务完成后重置状态为 TaskStatusDone，更新下次执行时间 以便下次继续执行
executeTime := utils.GetCpdailyNextExecuteTime(targetTask)
db.Global.CpTask.UpdateOneID(taskPayload.CpTaskID).SetTaskStatus(cptask.TaskStatusDone).SetNextExecuteTime(executeTime).AddSuccessCount(1).ExecX(ctx)
// 创建日志
db.Global.CpTaskLog.Create().
SetCpTaskID(targetTask.ID).
SetCpUserID(targetTask.Edges.CpUser.ID).
SetSystemUserID(targetTask.Edges.CpUser.Edges.Sysuser.ID).
SetSystemUserID(targetTask.Edges.SysUser.ID).
SetIsSuccess(true).
SetRealName(realTitle).
ExecX(ctx)
@@ -188,5 +192,11 @@ func (c *CpdailyTaskHandler) sendTaskNotificationEmail(ctx context.Context, task
return // 没有邮箱地址，不发送邮件
}
emailService.SendTaskNotificationEmail(ctx, emailAddr, taskName, status, details)
// 发送邮件，捕获可能的错误并忽略，避免影响任务执行主流程
e := utils.Catch(func() {
emailService.SendTaskNotificationEmail(ctx, emailAddr, taskName, status, details)
})
if e != nil {
logger.Logger.Warn("发送任务通知邮件失败，已忽略", zap.Error(e))
}
}
internal/framework/schedule/scanner.go
+8 -2
View File
@@ -118,14 +118,19 @@ func (ts *TaskScanner) scanCpdailyTask() {
queue.AddTask(ts.ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
asynq.TaskID(fmt.Sprintf("cpdaily:task:%d", task.ID)),
asynq.MaxRetry(3),
asynq.Unique(1*time.Minute),
asynq.Unique(10*time.Second),
asynq.ProcessAt(execTime),
asynq.Deadline(execTime.Add(12*time.Minute)),
asynq.Queue(cst.TaskQueueCpdaily),
)
// 更新未开始的任务状态为队列中
db.Global.CpTask.Update().Where(cptask.ID(task.ID), cptask.TaskStatusEQ(cptask.TaskStatusNotStart)).SetTaskStatus(cptask.TaskStatusQueue).Exec(ts.ctx)
})
if e != nil {
if errors.Is(e, asynq.ErrDuplicateTask) {
// 任务已在队列中，不更新下次执行时间，等任务执行完成后再处理
g.Log().Debugf(ts.ctx, "任务ID %d 已在队列中，跳过本次扫描", task.ID)
failedIDS = append(failedIDS, task.ID)
continue
} else if errors.Is(e, context.Canceled) {
return
@@ -150,8 +155,9 @@ func (ts *TaskScanner) scanCpdailyTask() {
}
// 重置30分钟之前的任务到下次执行时间
// 只处理已完成配置且状态为未开始的任务，避免重复处理正在执行或已失败的任务
expiredTasks := db.Global.CpTask.Query().
Where(cptask.NextExecuteTimeLT(minExecuteTime), cptask.NextExecuteTimeNotNil(), cptask.NextExecuteTimeGT(0)).
Where(cptask.NextExecuteTimeLT(minExecuteTime), cptask.NextExecuteTimeNotNil(), cptask.NextExecuteTimeGT(0), cptask.StepEQ(cptask.StepComplete), cptask.TaskStatusEQ(cptask.TaskStatusNotStart)).
Select(cptask.FieldID, cptask.FieldExecuteHour, cptask.FieldExecuteMinute).
AllX(ts.ctx)
internal/service/cpdaily/cptask.go
+44 -13
View File
@@ -5,7 +5,6 @@ import (
"context"
"encoding/gob"
"encoding/json"
"errors"
"fmt"
v1 "likedream-goframe/internal/api/v1"
"likedream-goframe/internal/application/cpdaily/login"
@@ -178,6 +177,10 @@ func (v CpTaskServiceV1) PostExecute(ctx context.Context, req *v1.PostCpTaskExec
if req.Frequency < 1 || consume < 0 {
utils.OkOrPanic(errors2.NewBusinessError(400, "执行次数应不低于原有次数"))
}
// 记录原始任务状态，用于判断是否需要立即扫描
originalStatus := cpTask.TaskStatus
//扣积分
basePoint := decimal.NewFromInt(cst.BasePointCpdaily)
if user.Edges.Agent != nil {
@@ -196,7 +199,27 @@ func (v CpTaskServiceV1) PostExecute(ctx context.Context, req *v1.PostCpTaskExec
SetExecuteMinute(req.ExecuteMinute).
SaveX(ctx)
nextExecuteAt := utils.GetCpdailyNextExecuteTime(cpTask)
// 计算下次执行时间
var nextExecuteAt int64
// 只有在原始状态为未开始时，才考虑5分钟窗口立即扫描
if originalStatus == cptask.TaskStatusNotStart {
currentTime := time.Now().In(cst.TimeZoneShangHai)
executeTime := time.Date(currentTime.Year(), currentTime.Month(), currentTime.Day(), req.ExecuteHour, req.ExecuteMinute, 0, 0, cst.TimeZoneShangHai)
timeDiff := executeTime.Sub(currentTime)
if timeDiff >= -5*time.Minute && timeDiff <= 5*time.Minute {
// 执行时间在当前时间前后5分钟内，设置为当前时间，让扫描器立即扫描到
nextExecuteAt = currentTime.Unix()
} else {
// 否则使用正常的计算逻辑
nextExecuteAt = utils.GetCpdailyNextExecuteTime(cpTask)
}
} else {
// 其他状态（暂停、失败等），使用正常的计算逻辑
nextExecuteAt = utils.GetCpdailyNextExecuteTime(cpTask)
}
db.Global.CpTask.UpdateOne(cpTask).SetNextExecuteTime(nextExecuteAt).ExecX(ctx)
}, &sql.TxOptions{})
utils.OkOrPanic(e)
@@ -221,14 +244,21 @@ func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq)
user := utils.GetUserInfoFromCtx(ctx)
taskExist := db.Global.CpTask.Query().
// 查询任务并检查状态
task := db.Global.CpTask.Query().
Where(cptask.ID(req.Id), cptask.UserID(user.ID)).
CountX(ctx)
Select(cptask.FieldTaskStatus).
FirstX(ctx)
if taskExist == 0 {
if task == nil {
utils.OkOrPanic(errors2.NewBusinessError(404, "任务不存在"))
}
// 检查任务是否正在执行中
if task.TaskStatus == cptask.TaskStatusDoing {
utils.OkOrPanic(errors2.NewBusinessError(400, "任务正在执行中，请稍后再试"))
}
payload := &schedule.CpdailyTaskPayload{
CpTaskID: req.Id,
}
@@ -238,22 +268,23 @@ func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq)
e := gob.NewEncoder(payloadWriter).Encode(payload)
utils.OkOrPanic(e, oops.WithContext(ctx).With("payload", payload).In("CpdailyTaskScanner.scanCpdailyTask").Wrap)
// 使用时间戳生成唯一的 TaskID，避免与自动扫描的任务冲突，也避免 Unique 限制
// 由于已经检查了任务状态，不需要使用 Unique 来防重复
taskID := fmt.Sprintf("cpdaily:task:manual:%d:%d", req.Id, time.Now().UnixNano())
e = utils.Catch(func() {
queue.AddTask(ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
asynq.TaskID(fmt.Sprintf("cpdaily:task:%d", req.Id)),
asynq.TaskID(taskID),
asynq.Unique(5*time.Second),
asynq.MaxRetry(0),
asynq.Unique(10*time.Minute),
asynq.Timeout(10*time.Minute),
asynq.Queue(cst.TaskQueueCpdaily),
)
// 更新未开始的任务状态为队列中
db.Global.CpTask.Update().Where(cptask.ID(task.ID), cptask.TaskStatusEQ(cptask.TaskStatusNotStart)).SetTaskStatus(cptask.TaskStatusQueue).Exec(ctx)
})
if e != nil {
if errors.Is(e, asynq.ErrDuplicateTask) {
utils.OkOrPanic(errors2.NewBusinessError(400, "任务已经在执行中"))
} else {
g.Log().Errorf(ctx, "立即执行任务入列失败: %+v", e)
utils.OkOrPanic(errors2.NewBusinessError(500, "任务队列中已存在此任务，立即执行任务入列失败"))
}
g.Log().Errorf(ctx, "立即执行任务入列失败: %+v", e)
utils.OkOrPanic(errors2.NewBusinessError(500, "同一个任务请勿连续执行"))
}
return utils.SuccessWithMsg("执行成功")
}
internal/utils/schedule.go
+1 -1
View File
@@ -26,7 +26,7 @@ func GetCpdailyNextExecuteTime(task *ent.CpTask) int64 {
// 如果当天执行时间已过，返回明天的执行时间
executeTime := time.Date(current.Year(), current.Month(), current.Day(), *executeHour, *executeMinute, 0, 0, cst.TimeZoneShangHai)
executeTime.AddDate(0, 0, 1)
executeTime = executeTime.AddDate(0, 0, 1)
return executeTime.Unix()
}