✨ 音乐下载

internal/api/server.go

@@ -7,6 +7,7 @@ import (
 	"likedream-goframe/internal/controller/common"
 	"likedream-goframe/internal/controller/cpdaily"
 	"likedream-goframe/internal/controller/menu"
+	"likedream-goframe/internal/controller/music"
 	"likedream-goframe/internal/controller/role"
 	"likedream-goframe/internal/controller/user"
 	"likedream-goframe/internal/controller/user_login"
@@ -53,6 +54,8 @@ func StartServer(ctx context.Context) {
 		cpdailyGroup.Bind(cpdaily.NewCpTaskCtrlV1())
 		cpdailyGroup.Bind(cpdaily.NewCpImageCtrlV1())
 		cpdailyGroup.Bind(cpdaily.NewCpTaskLogCtrlV1())
+		musicGroup := group.Group("/music").Middleware(middleware.AuthHandler, middleware.PermissionHandler)
+		musicGroup.Bind(music.NewMusicCtrlV1())
 	})
 
 	s.SetPort(port.Int())

internal/api/v1/music_req.go

@@ -0,0 +1,13 @@
+package v1
+
+import "github.com/gogf/gf/v2/frame/g"
+
+type SearchMusicReq struct {
+	g.Meta  `path:"/search" method:"get" tags:"Music" summary:"搜索音乐"`
+	Keyword string `json:"keyword"`
+}
+
+type DownloadMusicReq struct {
+	g.Meta `path:"/download" method:"get" tags:"Music" summary:"下载音乐"`
+	Path   string `json:"path"`
+}

internal/application/classifier/classifier.go

@@ -45,7 +45,7 @@ func NewImageClassifier(modelPath, classesPath string) (*ImageClassifier, error)
 	ort.SetSharedLibraryPath(dir)
 	err = ort.InitializeEnvironment()
 	if err != nil {
-		logger.Logger.Info("初始化onnx：", zap.String("绝对路径", absPath), zap.String("父文件夹", dir))
+		logger.Logger.Error("初始化onnx：", zap.String("绝对路径", absPath), zap.String("父文件夹", dir))
 		return nil, fmt.Errorf("初始化ONNX Runtime失败: %v", err)
 	}
 

internal/application/cpdaily/imageuploader/upload.go

@@ -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/collection.go

@@ -161,7 +161,7 @@ func (c *CpdailyCollectionTask) QueryFormByTitle(ctx context.Context, titleGlob
 	}
 
 	if matchTask == nil {
-		utils.OkOrPanic(errors.NewBusinessError(404, fmt.Sprintf("未找到表单信息: %s", titleGlob)), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, fmt.Sprintf("未找到表单信息: %s", titleGlob)), func(err error) error {
 			return oops.With("host", c.host).
 				In("CpdailyCollectionTask.QueryFormByTitle").
 				Wrap(err)
@@ -196,7 +196,7 @@ func (c *CpdailyCollectionTask) QueryFormByTitle(ctx context.Context, titleGlob
 	detailDatas := detailJSON["datas"].MapStrVar()
 
 	if len(detailDatas) == 0 {
-		utils.OkOrPanic(errors.NewBusinessError(404, "未找到表单数据"), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, "未找到表单数据"), func(err error) error {
 			return oops.With("host", c.host).
 				In("CpdailyCollectionTask.QueryFormByTitle").
 				With("titleGlob", titleGlob).
@@ -230,7 +230,7 @@ func (c *CpdailyCollectionTask) QueryFormByTitle(ctx context.Context, titleGlob
 	formDatas := formJSON["datas"].MapStrVar()
 
 	if len(formDatas) == 0 {
-		utils.OkOrPanic(errors.NewBusinessError(404, "未找到表单数据"), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, "未找到表单数据"), func(err error) error {
 			return oops.With("host", c.host).
 				In("CpdailyCollectionTask.QueryFormByTitle").
 				With("titleGlob", titleGlob).

internal/application/cpdaily/task/sign.go

@@ -147,8 +147,8 @@ func (c *CpdailyAutoSignTask) QueryFormByTitle(ctx context.Context, titleGlob st
 
 	findMatchTitle := func(jsonArr []any) map[string]*gvar.Var {
 		for _, rawJson := range jsonArr {
-			item, ok := rawJson.(map[string]*gvar.Var)
-			if !ok {
+			item := gvar.New(rawJson).MapStrVar()
+			if len(item) == 0 {
 				continue
 			}
 			taskName, exist := item["taskName"]
@@ -178,7 +178,7 @@ func (c *CpdailyAutoSignTask) QueryFormByTitle(ctx context.Context, titleGlob st
 			if matchTask == nil {
 				matchTask = findMatchTitle(signedTasks)
 				if matchTask == nil {
-					utils.OkOrPanic(errors.NewBusinessError(404, fmt.Sprintf("任务未找到: %s", titleGlob)), func(err error) error {
+					utils.OkOrPanic(errors.NewBusinessError(503, fmt.Sprintf("任务未找到: %s", titleGlob)), func(err error) error {
 						return oops.With("host", c.host).
 							In("CpdailyAutoSignTask.QueryFormByTitle").
 							Wrap(err)
@@ -210,7 +210,7 @@ func (c *CpdailyAutoSignTask) QueryFormByTitle(ctx context.Context, titleGlob st
 	formDatas := detailResponseJSON["datas"].MapStrVar()
 
 	if len(formDatas) == 0 {
-		utils.OkOrPanic(errors.NewBusinessError(404, "任务数据为空"), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, "任务数据为空"), func(err error) error {
 			return oops.With("host", c.host).
 				In("CpdailyAutoSignTask.QueryFormByTitle").
 				With("titleGlob", titleGlob).
@@ -293,7 +293,7 @@ func (c *CpdailyAutoSignTask) QueryFormByYearMonth(ctx context.Context, titleGlo
 			if matchTask == nil {
 				matchTask = findMatchTitle(signedTasks)
 				if matchTask == nil {
-					utils.OkOrPanic(errors.NewBusinessError(404, fmt.Sprintf("任务未找到: %s", titleGlob)), func(err error) error {
+					utils.OkOrPanic(errors.NewBusinessError(503, fmt.Sprintf("任务未找到: %s", titleGlob)), func(err error) error {
 						return oops.
 							With("host", c.host).
 							With("year", year).
@@ -332,7 +332,7 @@ func (c *CpdailyAutoSignTask) QueryFormByYearMonth(ctx context.Context, titleGlo
 	formDatas := detailResponseJSON["datas"].MapStrVar()
 
 	if len(formDatas) == 0 {
-		utils.OkOrPanic(errors.NewBusinessError(404, "任务数据为空"), func(err error) error {
+		utils.OkOrPanic(errors.NewBusinessError(503, "任务数据为空"), func(err error) error {
 			return oops.
 				With("host", c.host).
 				With("year", year).
@@ -379,12 +379,12 @@ func (c *CpdailyAutoSignTask) ParseFormJson(ctx context.Context, rawJson map[str
 	dto = &formtype.CpTaskFormItem{
 		Title:    "签到具体地址",
 		ColName:  "address",
-		Type:     formtype.ItemTypeAddress,
+		Type:     formtype.ItemTypeInput,
 		Required: true,
 	}
 	list = append(list, dto)
 
-	var current uint
+	var current uint = 1
 
 	if isPhoto := rows["isPhoto"]; isPhoto != nil && isPhoto.Int() == 1 {
 		dto = &formtype.CpTaskFormItem{
@@ -539,7 +539,7 @@ func (c *CpdailyAutoSignTask) FillCustomForm(ctx context.Context, form map[strin
 
 		previewUrl := imageuploader.UploadImageFromStorage(ctx, c.host, imageId, types.TaskTypeSign, c.client)
 
-		resultForm["signImg"] = gvar.New(previewUrl)
+		resultForm["signPhotoUrl"] = gvar.New(previewUrl)
 	}
 
 	// 处理验证码

internal/application/cpdaily/task/sleepcheck.go

@@ -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/application/httpclient/music.go

@@ -0,0 +1,15 @@
+package httpclient
+
+import (
+	"crypto/tls"
+	"time"
+
+	"resty.dev/v3"
+)
+
+var Music = resty.New().
+	SetCookieJar(nil).
+	SetTimeout(3 * time.Minute).
+	SetTLSClientConfig(&tls.Config{
+		InsecureSkipVerify: true,
+	})

internal/application/music/downloader.go

@@ -0,0 +1,73 @@
+package music
+
+import (
+	"context"
+	"likedream-goframe/internal/application/httpclient"
+	"likedream-goframe/internal/common/errors"
+	"likedream-goframe/internal/utils"
+	"net/url"
+	"regexp"
+	"strings"
+
+	"github.com/PuerkitoBio/goquery"
+	"github.com/samber/oops"
+	"github.com/tidwall/gjson"
+)
+
+var MusicDownloader *Downloader
+
+type Downloader struct {
+}
+
+func (c *Downloader) Search(ctx context.Context, keyword string) (result []MusicSearchResult) {
+	client := httpclient.Music.Clone(ctx)
+	response, err := client.R().Get("https://www.gequbao.net/s/" + url.PathEscape(keyword))
+	if err != nil {
+		utils.OkOrPanic(errors.NewBusinessError(503, "搜索出错1"), oops.WithContext(ctx).Wrap)
+	}
+	defer response.Body.Close()
+	reader, err := goquery.NewDocumentFromReader(response.Body)
+	if err != nil {
+		utils.OkOrPanic(errors.NewBusinessError(503, "搜索出错2"), oops.WithContext(ctx).Wrap)
+	}
+	card := reader.Find(".card")
+	find := card.Find(".card-text > .row:not(:first-child)")
+	find.Each(func(i int, item *goquery.Selection) {
+		selection := item.Find("a")
+		url, _ := selection.Attr("href")
+		name := strings.TrimSpace(selection.Find(".text-primary span").Text())
+		singer := strings.TrimSpace(selection.Find(".text-jade").Text())
+		result = append(result, MusicSearchResult{
+			Name:   name,
+			Singer: singer,
+			Path:   url,
+		})
+	})
+	return
+}
+
+func (c *Downloader) Download(ctx context.Context, url string) (result []byte) {
+	client := httpclient.Music.Clone(ctx)
+	response, err := client.R().Get("https://www.gequbao.net" + url)
+	if err != nil {
+		utils.OkOrPanic(errors.NewBusinessError(503, "下载出错1"), oops.WithContext(ctx).Wrap)
+	}
+	html := response.String()
+	reg := regexp.MustCompile("window\\.play_id = '(.*?)';")
+	play_id := reg.FindStringSubmatch(html)[1]
+	post, err := client.R().SetBody(map[string]string{
+		"id": play_id,
+	}).Post("https://www.gequbao.net/api/play-url")
+	if err != nil {
+		utils.OkOrPanic(errors.NewBusinessError(503, "下载出错2"), oops.WithContext(ctx).Wrap)
+	}
+	download_url := gjson.Get(post.String(), "data.url").String()
+
+	// 下载音频文件
+	downloadResp, err := client.R().Get(download_url)
+	if err != nil {
+		utils.OkOrPanic(errors.NewBusinessError(503, "下载出错3"), oops.WithContext(ctx).Wrap)
+	}
+
+	return downloadResp.Bytes()
+}

internal/application/music/model.go

@@ -0,0 +1,7 @@
+package music
+
+type MusicSearchResult struct {
+	Name   string `json:"name"`
+	Singer string `json:"singer"`
+	Path   string `json:"url"`
+}

internal/controller/agent/agent_ctrlV1.go

@@ -23,7 +23,7 @@ func (a AgentCtrlV1) List(ctx context.Context, req *v1.AgentListReq) (res *utils
 // Add 添加下级代理
 func (a AgentCtrlV1) Add(ctx context.Context, req *v1.AgentAddReq) (res *utils.Response[any], err error) {
 	defer utils.Pass(&err)
-	return service.Global.Agent.Add(ctx, req), nil
+	panic("implement me")
 }
 
 // Ban 禁用代理

internal/controller/music/i_music_ctrl.go

@@ -0,0 +1,13 @@
+package music
+
+import (
+	"context"
+	v1 "likedream-goframe/internal/api/v1"
+	"likedream-goframe/internal/application/music"
+	"likedream-goframe/internal/utils"
+)
+
+type IMusicCtrlV1 interface {
+	Search(ctx context.Context, req *v1.SearchMusicReq) (res *utils.Response[[]music.MusicSearchResult], err error)
+	Download(ctx context.Context, req *v1.DownloadMusicReq) (res *utils.Response[any], err error)
+}

internal/controller/music/music_ctrlV1.go

@@ -0,0 +1,49 @@
+package music
+
+import (
+	"context"
+	"fmt"
+	v1 "likedream-goframe/internal/api/v1"
+	"likedream-goframe/internal/application/music"
+	"likedream-goframe/internal/utils"
+
+	"github.com/gogf/gf/v2/net/ghttp"
+)
+
+type MusicCtrlV1 struct {
+}
+
+func NewMusicCtrlV1() IMusicCtrlV1 {
+	return &MusicCtrlV1{}
+}
+
+func (c *MusicCtrlV1) Search(ctx context.Context, req *v1.SearchMusicReq) (res *utils.Response[[]music.MusicSearchResult], err error) {
+	defer utils.Pass(&err)
+	return utils.SuccessWithData(music.MusicDownloader.Search(ctx, req.Keyword)), nil
+}
+
+func (c *MusicCtrlV1) Download(ctx context.Context, req *v1.DownloadMusicReq) (res *utils.Response[any], err error) {
+	defer utils.Pass(&err)
+
+	// 获取音频二进制数据
+	audioData := music.MusicDownloader.Download(ctx, req.Path)
+
+	// 从 context 获取 ghttp.Request
+	r := ghttp.RequestFromCtx(ctx)
+
+	// 构建文件名：Name - Singer.mp3
+	filename := fmt.Sprintf("music.mp3")
+
+	// 设置响应头
+	r.Response.Header().Set("Content-Type", "audio/mpeg")
+	r.Response.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=\"%s\"", filename))
+	r.Response.Header().Set("Content-Length", fmt.Sprintf("%d", len(audioData)))
+
+	// 写入二进制数据
+	r.Response.Write(audioData)
+
+	// 标记已处理，防止框架再次处理响应
+	r.ExitAll()
+
+	return nil, nil
+}

internal/cst/task.go

@@ -1,9 +1,11 @@
 package cst
 
 const (
-	TaskCpdaily = "task:cpdaily"
-	TaskEmail   = "task:email"
+	TaskCpdaily    = "task:cpdaily"
+	TaskEmail      = "task:email"
+	TaskResetDaily = "task:reset:daily"
 
 	TaskQueueCpdaily = "queue:cpdaily"
 	TaskQueueEmail   = "queue:email"
+	TaskQueueDefault = "default"
 )

internal/ent/cptask/cptask.go

@@ -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/migrate/schema.go

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

@@ -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/ent/schema/sys_user.go

@@ -46,14 +46,14 @@ func (SysUser) Edges() []ent.Edge {
 		edge.To("children", SysUser.Type).
 			From("parent").
 			Unique(),
-		edge.To("roles", SysRole.Type),                                               // 用户拥有角色
-		edge.To("agent", SysAgent.Type).Unique(),                                     // 用户拥有代理
-		edge.To("images", CpImage.Type).Annotations(entsql.OnDelete(entsql.Cascade)), // 用户拥有的图片
-		edge.To("cpuser", CpUser.Type).Annotations(entsql.OnDelete(entsql.Cascade)),  // 用户拥有的cp用户
-		edge.To("cp_task", CpTask.Type),
-		edge.To("cp_task_logs", CpTaskLog.Type), // 用户拥有的cp任务日志
-		edge.To("created_cards", SysCard.Type),  // 用户创建的卡
-		edge.To("used_cards", SysCard.Type),     // 用户使用的卡
+		edge.To("roles", SysRole.Type).Annotations(entsql.OnDelete(entsql.Cascade)),           // 用户拥有角色
+		edge.To("agent", SysAgent.Type).Unique().Annotations(entsql.OnDelete(entsql.Cascade)), // 用户拥有代理
+		edge.To("images", CpImage.Type).Annotations(entsql.OnDelete(entsql.Cascade)),          // 用户拥有的图片
+		edge.To("cpuser", CpUser.Type).Annotations(entsql.OnDelete(entsql.Cascade)),           // 用户拥有的cp用户
+		edge.To("cp_task", CpTask.Type).Annotations(entsql.OnDelete(entsql.Cascade)),
+		edge.To("cp_task_logs", CpTaskLog.Type).Annotations(entsql.OnDelete(entsql.Cascade)), // 用户拥有的cp任务日志
+		edge.To("created_cards", SysCard.Type),                                               // 用户创建的卡
+		edge.To("used_cards", SysCard.Type),                                                  // 用户使用的卡
 		edge.To("cashback_cards", SysCard.Type),
 	}
 }

internal/framework/schedule/cpdaily_handler.go

@@ -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/init.go

@@ -14,12 +14,17 @@ func Init(ctx context.Context) {
 	initServer(ctx)
 	initInspector(ctx)
 	initScanner()
+	initPeriodicTasks(ctx)
 }
 
 func Stop() {
 	globalScanner.StopScan()
 	g.Log().Info(gctx.GetInitCtx(), "task scanner stopped")
 
+	if globalPeriodicTaskManager != nil {
+		globalPeriodicTaskManager.Shutdown()
+	}
+
 	e := queue.GetClient().Close()
 	utils.OkOrPanic(e)
 	g.Log().Info(gctx.GetInitCtx(), "asynq client closed")

internal/framework/schedule/periodic.go

@@ -0,0 +1,49 @@
+package schedule
+
+import (
+	"context"
+	"likedream-goframe/internal/cst"
+	"likedream-goframe/internal/framework/redis"
+	"likedream-goframe/internal/utils"
+
+	"github.com/hibiken/asynq"
+)
+
+var globalPeriodicTaskManager *asynq.PeriodicTaskManager
+
+func initPeriodicTasks(ctx context.Context) {
+	// 创建周期性任务配置提供者
+	provider := &periodicTaskConfigProvider{}
+
+	// 创建周期性任务管理器
+	globalPeriodicTaskManager, _ = asynq.NewPeriodicTaskManager(
+		asynq.PeriodicTaskManagerOpts{
+			RedisUniversalClient:       redis.GetClient(),
+			PeriodicTaskConfigProvider: provider,
+			SyncInterval:               0, // 立即同步，使用默认值
+		},
+	)
+
+	// 启动周期性任务管理器
+	e := globalPeriodicTaskManager.Start()
+	utils.OkOrPanic(e)
+}
+
+// periodicTaskConfigProvider 提供周期性任务配置
+type periodicTaskConfigProvider struct{}
+
+// GetConfigs 返回所有周期性任务配置
+func (p *periodicTaskConfigProvider) GetConfigs() ([]*asynq.PeriodicTaskConfig, error) {
+	return []*asynq.PeriodicTaskConfig{
+		{
+			//Cronspec: "0 1 * * *", // 每天凌晨1点执行
+			Cronspec: "0 1 * * * ",
+			Task:     asynq.NewTask(cst.TaskResetDaily, nil),
+			Opts: []asynq.Option{
+				asynq.Queue(cst.TaskQueueDefault),
+				asynq.MaxRetry(3),
+				asynq.Timeout(0), // 使用默认超时
+			},
+		},
+	}, nil
+}

internal/framework/schedule/reset_handler.go

@@ -0,0 +1,28 @@
+package schedule
+
+import (
+	"context"
+	"likedream-goframe/internal/db"
+	"likedream-goframe/internal/ent/cptask"
+	"likedream-goframe/internal/utils"
+
+	"github.com/gogf/gf/v2/frame/g"
+	"github.com/hibiken/asynq"
+)
+
+type ResetTaskHandler struct {
+}
+
+// ProcessTask 处理每日重置任务状态
+func (h *ResetTaskHandler) ProcessTask(ctx context.Context, _ *asynq.Task) (err error) {
+	defer utils.Pass(&err)
+
+	g.Log().Info(ctx, "开始执行每日任务状态重置")
+
+	// 执行重置任务状态
+	db.Global.CpTask.Update().Where(cptask.TaskStatusIn(cptask.TaskStatusDone, cptask.TaskStatusFail)).SetTaskStatus(cptask.TaskStatusNotStart).ExecX(ctx)
+
+	g.Log().Info(ctx, "每日任务状态重置完成")
+
+	return
+}

internal/framework/schedule/scanner.go

@@ -116,16 +116,21 @@ func (ts *TaskScanner) scanCpdailyTask() {
 
 		e = utils.Catch(func() {
 			queue.AddTask(ts.ctx, asynq.NewTask(cst.TaskCpdaily, payloadWriter.Bytes()),
-				asynq.TaskID(fmt.Sprintf("cpdaily:task:%d", task.ID)),
+				asynq.TaskID(fmt.Sprintf("cpdaily:task:schedule:%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/framework/schedule/server.go

@@ -30,7 +30,8 @@ func initServer(ctx context.Context) {
 		ErrorHandler:             taskErrorHandler{},
 		Queues: map[string]int{
 			cst.TaskQueueEmail:   2,
-			cst.TaskQueueCpdaily: 8,
+			cst.TaskQueueCpdaily: 10,
+			cst.TaskQueueDefault: 1,
 		},
 	})
 
@@ -38,6 +39,7 @@ func initServer(ctx context.Context) {
 
 	mux.Handle(cst.TaskCpdaily, &CpdailyTaskHandler{})
 	mux.Handle(cst.TaskEmail, &EmailTaskHandler{})
+	mux.Handle(cst.TaskResetDaily, &ResetTaskHandler{})
 
 	e := globalServer.Start(mux)
 	utils.OkOrPanic(e)

internal/service/agent/agent_service.go

@@ -60,57 +60,6 @@ func (s *AgentServiceV1) List(ctx context.Context, req *v1.AgentListReq) *utils.
 	return utils.SuccessWithTotal[[]v1.AgentListRes](res, total)
 }
 
-func (s *AgentServiceV1) Add(ctx context.Context, req *v1.AgentAddReq) *utils.Response[any] {
-	// 参数校验
-	currentUser := utils.GetUserInfoFromCtx(ctx)
-	agent := currentUser.QueryAgent().OnlyX(ctx)
-	if agent.Level >= req.Level {
-		return utils.FailParam("下级代理级别不能高于当前代理级别")
-	}
-	if agent.Rate.Compare(req.Rate) <= 0 {
-		return utils.FailParam("下级代理费率比例不能低于当前代理比例")
-	}
-	tx, err := db.Global.Tx(ctx)
-	utils.OkOrPanic(err, func(err error) error {
-		return oops.
-			WithContext(ctx).
-			With("创建事务失败").
-			Wrap(err)
-	})
-	defer tx.Rollback()
-	user := tx.SysUser.Query().Where(sysuser.IDEQ(currentUser.ID)).OnlyX(ctx)
-	user.Balance = utils.CalculateBalance(user.Balance, req.Rate, req.Balance, req.Rate)
-	if user.Balance.LessThan(decimal.NewFromInt(0)) {
-		return utils.FailParam("余额不足")
-	}
-	// 扣除余额
-	_, err = tx.SysUser.UpdateOne(user).SetBalance(user.Balance).Save(ctx)
-	utils.OkOrPanic(err, func(err error) error {
-		return oops.
-			WithContext(ctx).
-			With("更新用户余额失败").
-			Wrap(err)
-	})
-	// 创建用户
-	tmp := tx.SysUser.Create().
-		SetUsername(req.Username).
-		SetPassword(utils.PasswordBcryptHash(req.Password)).
-		SetEmail(req.Email).
-		SetParent(user)
-	save := tmp.SaveX(ctx)
-	// 关联角色
-	role := tx.SysRole.Query().Where(sysrole.NameEQ(cst.KeyRoleAgent)).AllX(ctx)
-	tx.SysUser.UpdateOne(save).AddRoles(role...).ExecX(ctx)
-	// 创建代理
-	tx.SysAgent.Create().
-		SetLevel(req.Level).
-		SetRate(req.Rate).
-		SaveX(ctx)
-
-	err = tx.Commit()
-	return utils.SuccessWithMsg("")
-}
-
 func (s *AgentServiceV1) Ban(ctx context.Context, req *v1.AgentBanReq) *utils.Response[any] {
 	user := db.Global.SysUser.Query().Where(sysuser.IDEQ(req.ID)).OnlyX(ctx)
 	if !utils.IsAdminFromCtx(ctx) {
@@ -127,19 +76,22 @@ func (s *AgentServiceV1) Ban(ctx context.Context, req *v1.AgentBanReq) *utils.Re
 }
 
 func (s *AgentServiceV1) Recharge(ctx context.Context, req *v1.AgentRechargeReq) *utils.Response[any] {
-	// 参数校验
 	user := db.Global.SysUser.Query().Where(sysuser.IDEQ(req.ID)).WithAgent().OnlyX(ctx)
+
 	if !utils.IsAdminFromCtx(ctx) {
-		currentUser := db.Global.SysUser.Query().Where(sysuser.ID(utils.GetUserInfoFromCtx(ctx).ID)).WithAgent().WithRoles().OnlyX(ctx)
+		currentUser := utils.GetUserInfoFromCtx(ctx)
+		// 参数校验
+		if user.Edges.Agent == nil {
+			return utils.FailParam("用户不是代理")
+		}
 		// 判断是否是自己的下级代理
 		parent := user.QueryParent().WithAgent().FirstX(ctx)
 		if parent == nil || parent.ID != currentUser.ID {
 			return utils.FailParam("非法的下级代理参数")
 		}
-		if !utils.HasRole(currentUser.Edges.Roles, cst.KeyRoleAdmin) {
-			needBalance := utils.CalculateBalance(parent.Balance, parent.Edges.Agent.Rate, req.Amount, user.Edges.Agent.Rate)
-			balance.BalanceService.ChangeUserBalance(ctx, parent.ID, needBalance.Neg())
-		}
+		needBalance := utils.NeedBalance(parent.Edges.Agent.Rate, req.Amount, user.Edges.Agent.Rate)
+		balance.BalanceService.ChangeUserBalance(ctx, parent.ID, needBalance.Neg())
+
 	}
 	balance.BalanceService.ChangeUserBalance(ctx, user.ID, req.Amount)
 	return utils.SuccessWithMsg("操作成功")
@@ -185,11 +137,13 @@ func (s *AgentServiceV1) AgentPutInfoReq(ctx context.Context, req *v1.AgentPutIn
 		oldBalance = utils.RevertBalance(user.Balance, user.Edges.Agent.Rate)
 		tx.SysAgent.UpdateOne(user.Edges.Agent).SetLevel(req.Level).SetRate(req.Rate).SaveX(ctx)
 	}
+	err = tx.Commit()
 	//按照新费率计算余额
 	newBalance := utils.TransformBalance(oldBalance, req.Rate)
-	// 6. 更新用户余额
-	balance.BalanceService.ChangeUserBalance(ctx, user.ID, newBalance.Sub(oldBalance))
-	err = tx.Commit()
+	// 6. 更新用户余额(如果更新了费率或者第一次改为代理)
+	if user.Edges.Agent == nil || !user.Edges.Agent.Rate.Equal(req.Rate) {
+		balance.BalanceService.ChangeUserBalance(ctx, user.ID, newBalance.Sub(user.Balance))
+	}
 	utils.OkOrPanic(err, func(err error) error {
 		return oops.
 			WithContext(ctx).

internal/service/balance/balance_service.go

@@ -4,33 +4,40 @@ import (
 	"context"
 	"likedream-goframe/internal/common/errors"
 	"likedream-goframe/internal/db"
+	"likedream-goframe/internal/ent"
 	"likedream-goframe/internal/ent/sysuser"
 	"likedream-goframe/internal/utils"
-	"sync"
 
-	"github.com/samber/oops"
 	"github.com/shopspring/decimal"
 )
 
-var BalanceService = &BalanceServiceV1{
-	lock: sync.Mutex{},
-}
+var BalanceService = &BalanceServiceV1{}
 
 type BalanceServiceV1 struct {
-	lock sync.Mutex
 }
 
-func (v *BalanceServiceV1) ChangeUserBalance(ctx context.Context, userId int64, balance decimal.Decimal) {
-	v.lock.Lock()
-	defer v.lock.Unlock()
-	tx, err := db.Global.Tx(ctx)
-	utils.OkOrPanic(err, oops.WithContext(ctx).Wrap)
-	defer tx.Rollback()
-	user := tx.SysUser.Query().Where(sysuser.ID(userId)).WithRoles().OnlyX(ctx)
-	newBalance := user.Balance.Add(balance)
-	if newBalance.LessThan(decimal.Zero) {
-		utils.OkOrPanic(errors.NewBusinessError(400, "余额不足"))
-	}
-	tx.SysUser.Update().Where(sysuser.ID(userId)).SetBalance(newBalance).SaveX(ctx)
-	tx.Commit()
+func (v *BalanceServiceV1) ChangeUserBalance(ctx context.Context, userId int64, balance decimal.Decimal) error {
+	return db.WithTx(ctx, func(c context.Context, tx *ent.Tx) {
+		// 使用 SELECT ... FOR UPDATE 锁定用户记录
+		user, err := tx.SysUser.Query().
+			Where(sysuser.ID(userId)).
+			ForUpdate().
+			Only(ctx)
+		if err != nil {
+			utils.OkOrPanic(err)
+		}
+
+		newBalance := user.Balance.Add(balance)
+		if newBalance.LessThan(decimal.Zero) {
+			utils.OkOrPanic(errors.NewBusinessError(400, "余额不足"))
+		}
+
+		// 更新余额
+		tx.SysUser.Update().
+			Where(sysuser.ID(userId)).
+			SetBalance(newBalance).
+			ExecX(ctx)
+		tx.Commit()
+	}, nil)
+
 }

internal/service/cpdaily/cptask.go

@@ -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,13 +244,24 @@ func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq)
 
 	user := utils.GetUserInfoFromCtx(ctx)
 
-	taskExist := db.Global.CpTask.Query().
+	// 查询任务并检查状态
+	cpTask := db.Global.CpTask.Query().
 		Where(cptask.ID(req.Id), cptask.UserID(user.ID)).
-		CountX(ctx)
+		Select(cptask.FieldTaskStatus, cptask.FieldFrequency, cptask.FieldSuccessCount).
+		FirstX(ctx)
 
-	if taskExist == 0 {
+	if cpTask == nil {
 		utils.OkOrPanic(errors2.NewBusinessError(404, "任务不存在"))
 	}
+	// 检查任务是否存在次数
+	if cpTask.Frequency-cpTask.SuccessCount <= 0 {
+		utils.OkOrPanic(errors2.NewBusinessError(400, "任务没有执行次数"))
+	}
+
+	// 检查任务是否正在执行中
+	if cpTask.TaskStatus != cptask.TaskStatusNotStart && cpTask.TaskStatus != cptask.TaskStatusDone && cpTask.TaskStatus != cptask.TaskStatusFail {
+		utils.OkOrPanic(errors2.NewBusinessError(400, "任务正在执行中，请稍后再试"))
+	}
 
 	payload := &schedule.CpdailyTaskPayload{
 		CpTaskID: req.Id,
@@ -238,22 +272,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(cpTask.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/service/user_login/user_info_service.go

@@ -12,6 +12,7 @@ import (
 	"likedream-goframe/internal/ent/sysrole"
 	"likedream-goframe/internal/ent/sysuser"
 	"likedream-goframe/internal/framework/ent2"
+	"likedream-goframe/internal/service/balance"
 	"likedream-goframe/internal/utils"
 	"strings"
 	"time"
@@ -132,22 +133,25 @@ func (s *UserInfoServiceV1) CardRecharge(ctx context.Context, req *v1.CardRechar
 	tx, err := db.Global.Tx(ctx)
 	utils.OkOrPanic(err)
 	parent := user.QueryParent().WithAgent().FirstX(ctx)
+	// 返现20%
+	r, _ := decimal.NewFromString("0.2")
+	cashbackValue := decimal.NewFromUint64(card.Value).Mul(r)
 	if parent != nil {
-		// 返现20%
-		r, _ := decimal.NewFromString("0.2")
-		cashbackValue := decimal.NewFromUint64(card.Value).Mul(r)
 		// 这里查询父级是否是代理，代理的话要x代理费率
 		if parent.Edges.Agent != nil {
 			cashbackValue = cashbackValue.Mul(parent.Edges.Agent.Rate)
 		}
-		tx.SysUser.UpdateOneID(user.Edges.Parent.ID).SetBalance(parent.Balance.Add(cashbackValue)).SetInviteCashback(parent.InviteCashback.Add(cashbackValue)).ExecX(ctx)
+		tx.SysUser.UpdateOneID(user.Edges.Parent.ID).SetInviteCashback(parent.InviteCashback.Add(cashbackValue)).ExecX(ctx)
 		tx.SysCard.UpdateOneID(card.ID).SetStatus(syscard.StatusCashback).SetUsedID(user.ID).SetCashUserID(user.Edges.Parent.ID).SetCashValue(cashbackValue).ExecX(ctx)
 	} else {
 		//tx.SysCard.UpdateOneID(card.ID).SetStatus(cst.CardStatusUsed).SetUsedID(user.ID).ExecX(ctx)
 		tx.SysCard.UpdateOneID(card.ID).SetStatus(syscard.StatusUsed).SetUsedID(user.ID).ExecX(ctx)
 	}
-	tx.SysUser.UpdateOneID(user.ID).SetBalance(user.Balance.Add(decimal.NewFromUint64(card.Value))).ExecX(ctx)
 	err = tx.Commit()
+	balance.BalanceService.ChangeUserBalance(ctx, user.ID, decimal.NewFromUint64(card.Value))
+	if parent != nil {
+		balance.BalanceService.ChangeUserBalance(ctx, parent.ID, cashbackValue)
+	}
 	utils.OkOrPanic(err, func(err error) error {
 		return oops.
 			WithContext(ctx).

internal/utils/schedule.go

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

internal/utils/user.go

@@ -28,6 +28,14 @@ func CalculateBalance(balance decimal.Decimal, rate decimal.Decimal, amount deci
 	return balance.Sub(amount.Div(amountRate).Mul(rate)).Truncate(2)
 }
 
+func NeedBalance(rate decimal.Decimal, amount decimal.Decimal, amountRate decimal.Decimal) decimal.Decimal {
+	// 如果当前级别费率为0(超级管理员)
+	if rate.Equal(decimal.NewFromInt(0)) {
+		return decimal.Zero
+	}
+	return amount.Mul(rate).Div(amountRate)
+}
+
 // ConsumeBalance 消费余额
 func ConsumeBalance(balance decimal.Decimal, rate decimal.Decimal) decimal.Decimal {
 	return balance.Mul(rate)