🐛 邮件调整

config.toml

@@ -29,10 +29,10 @@ url = "redis://:jhkdjhkjdhsIUTYURTU_pEGmFP@database.out:56379"
 smtpHost = "smtp-mail.outlook.com"
 smtpPort = 587
 # 发件人邮箱和密码（需要使用授权码，不是登录密码）
-smtpUsername = "demo@likedream.xyz"
+smtpUsername = "reply@likedream.xyz"
 smtpPassword = "baca122..."
 # 发件人邮箱和显示名称
-fromEmail = "demo@likedream.xyz"
+fromEmail = "reply@likedream.xyz"
 fromName = "若梦系统"
 
 
@@ -55,4 +55,4 @@ cpdailyTask = "0.5"
 cpdailyLeave = "4"
 cpdailyPhone = "0.25"
 [task]
-concurrency = 20
+concurrency = 80

internal/application/cpdaily/cryptoutil/aes_decrypt_test.go

@@ -7,7 +7,7 @@ import (
 )
 
 func TestAesDecrypt(t *testing.T) {
-	str := "457xXTd3ZBTY0017ueFiMR0gFff7W2NuXH39lkT/kvTTCKqhXA1vjJwgtmzRgZOC+qEpA3O9GmAroR+A1lORFVqe+q7o3alKOKj4rxPUUSwcyqGUF2LMNbxFXrmWUEPzeedbdUtlI+ufR8SLXNHbZSs0ECPy864bwUAuSDhFcD235c/3AbiTNacXIZZ+paK0T6GWxS291XbffgNCpGFa+5jCaP7/t9erCUjId9Zm4hmoaHkhqGRKAgZrLwuICa4lTFNOQqLA1N/Dljd/ZPVbZGVZaJQ0oV3zEaQQDCWuSTcWF4JVVeaCw9s/8mqkarg+RIz3/OA8dCvc8Wo6auMGzXT6IlisCX+S5TuPlhKKo1ttAW1KrWNg2nkZMxyD+kTKtNijvZRsdQnA8znVp0q2p0byyvzevaB6qSYUM9NBcaxyxkes5Bd1lKCTUHPaG93Jz4SQ+B8MQ1rLm13eWY8uME/WBTOraG7Rgmba32zBccC6BWwV3ncNGhu7hLL0dIhdgIpuxnrFmXDw4HX/IM+cnCxAFo7+EfzKb4l3yDMyr2d+0tDlVl4UbUR2uRTGz5r/7vVI1WtLgM0fiRTk56JU5BwRrutS7KonsLzugbEp/E8="
+	str := "gTIjjNBTwN0Vdv2o5ZlTZ6gHbFcIA0MV0caRqvv/50tptfoSO+VASLxpQZWy Xa86lY21QtmXhwlvTsfd3XR819O0IUi2FynPOM6s0pDsSndM2JhhqoRNXBTK 4WqfMgt0RCQJ38KP8XtM5O+Ktkj1YHHM/OVEGqOxRcUh2M++JJCheJbSs+aQ 2T3va13/GWJH7PEd8JVEJgBiO3wXLOb9+CvvWuU/Frc9nNNxQL7LrgZTdQAv C9Yt+5/a2n41reOeOs7Bu5C4FFlHQSs71Rom/HtIhL67x1ecIre1rbgK9xxz 1AhTEHY6nhdZmktPGJD5CfxvLVWIeLv09CjKqdSaNodkP4e9YTScNEZ2s8TK y8Xhv8L0r6gllC9B3nla4GRe+lqMvhDPDZx7PFu7kxFspf9yoM4eOpgWfCr9 xbDwxO9cXphiktfWsQiX28W4g2jCFFEazh01WBtpjM9LsLfbBgZA+z3jgfYa AXZuOvFQMJuUnLyOIGPtsaSTwXFDlJwNRgEw+Ctl22JSJbbu1yOoM9S6Ge27 rUVMhbAS5snVbNbLhgAymQLQQVLRwfsMLbtUfLTZuCC8zO8d16eOYw56+DqR QWNG1EXQz3CLeFDs2hFqcNWrAt773qANQ3zLPA65"
 	str = strings.ReplaceAll(str, " ", "")
 	decodeString, _ := base64.StdEncoding.DecodeString(str)
 	decrypt := AesDecryptForCat(decodeString)

internal/application/cpdaily/leave/report.go

@@ -10,6 +10,7 @@ import (
 	"likedream-goframe/internal/common/errors"
 	"likedream-goframe/internal/utils"
 
+	"github.com/bytedance/sonic"
 	"github.com/gogf/gf/v2/container/gvar"
 	"github.com/google/uuid"
 	"github.com/samber/oops"
@@ -104,6 +105,11 @@ func (c *CpdailyReportApp) LeaveDetail(leaveId string) gjson.Result {
 	return datas
 }
 
+type Attach struct {
+	Type string `json:"type"`
+	Url  string `json:"url"`
+}
+
 func (c *CpdailyReportApp) FillReportForm(detail gjson.Result) map[string]*gvar.Var {
 	resultForm := make(map[string]*gvar.Var)
 	resultForm["id"] = gvar.New(detail.Get("detail.id").String())
@@ -112,9 +118,17 @@ func (c *CpdailyReportApp) FillReportForm(detail gjson.Result) map[string]*gvar.
 	resultForm["reportLocation"] = gvar.New(detail.Get("detail.applyLocation.address").String())
 	resultForm["approverWid"] = gvar.New("")
 	resultForm["reportApplyExplain"] = gvar.New(nil)
-	//resultForm["reportAttach"] = gvar.New(detail.Get("detail.applyAttach").Array())
+	//resultForm["reportAttach"] = gvar.New(.Array())
 	resultForm["reportAttach"] = gvar.New([]string{})
 	resultForm["uaIsCpadaily"] = gvar.New(true)
+	attachStr := detail.Get("detail.applyAttach").String()
+	if attachStr != "" {
+		var attach []Attach
+		_ = sonic.Unmarshal([]byte(attachStr), &attach)
+		resultForm["reportAttach"] = gvar.New(attach)
+	} else {
+		resultForm["reportAttach"] = gvar.New([]string{})
+	}
 	return resultForm
 }
 

internal/framework/schedule/cpdaily_handler.go

@@ -40,19 +40,15 @@ func (c *CpdailyTaskHandler) ProcessTask(ctx context.Context, task *asynq.Task)
 	e := gob.NewDecoder(bytes.NewReader(task.Payload())).Decode(&taskPayload)
 	utils.OkOrPanic(e, oops.WithContext(ctx).With("payload", task.Payload(), "type", task.Type()).In("CpdailyTask.DecodePayload").Wrap)
 
-	// 查询指定的任务和关联今日校园用户
-	targetTask, e := db.Global.CpTask.Query().Where(cptask.ID(taskPayload.CpTaskID)).
-		WithCpUser().
-		WithSysUser().
-		First(ctx)
-	if ent.IsNotFound(e) {
-		utils.OkOrPanic(oops.WithContext(ctx).With("task_id", taskPayload.CpTaskID).In("CpdailyTask.ProcessTask").Wrap(e))
-	}
-	utils.OkOrPanic(e, oops.WithContext(ctx).With("task_id", taskPayload.CpTaskID).In("CpdailyTask.ProcessTask").Wrap)
-
-	db.Global.CpTask.UpdateOneID(taskPayload.CpTaskID).SetTaskStatus(cptask.TaskStatusDoing).ExecX(ctx)
-
+	// 先注册错误处理，确保任何阶段的错误都能被捕获并重置状态
+	var targetTask *ent.CpTask
 	defer utils.HandleError(func(err error) {
+		// 如果 targetTask 还没查询到，只能重置状态为 not_start，让任务重新入队
+		if targetTask == nil {
+			db.Global.CpTask.UpdateOneID(taskPayload.CpTaskID).SetTaskStatus(cptask.TaskStatusNotStart).Exec(ctx)
+			panic(err)
+		}
+
 		// 执行失败应该更新下次执行时间
 		executeTime := utils.GetCpdailyNextExecuteTime(targetTask)
 		e := db.Global.CpTask.UpdateOneID(taskPayload.CpTaskID).SetNextExecuteTime(executeTime).SetTaskStatus(cptask.TaskStatusFail).Exec(ctx)
@@ -90,6 +86,15 @@ func (c *CpdailyTaskHandler) ProcessTask(ctx context.Context, task *asynq.Task)
 		panic(err)
 	})
 
+	// 查询指定的任务和关联今日校园用户
+	targetTask, e = db.Global.CpTask.Query().Where(cptask.ID(taskPayload.CpTaskID)).
+		WithCpUser().
+		WithSysUser().
+		First(ctx)
+	utils.OkOrPanic(e, oops.WithContext(ctx).With("task_id", taskPayload.CpTaskID).In("CpdailyTask.ProcessTask").Wrap)
+
+	db.Global.CpTask.UpdateOneID(taskPayload.CpTaskID).SetTaskStatus(cptask.TaskStatusDoing).ExecX(ctx)
+
 	// 查询学校详情
 	schoolDetail := school.GetSchoolDetail(ctx, targetTask.Edges.CpUser.SchoolID)
 	cookie := targetTask.Edges.CpUser.Cookie

internal/framework/schedule/email_handler.go

@@ -7,6 +7,7 @@ import (
 	"likedream-goframe/internal/framework/logger"
 	"likedream-goframe/internal/framework/model/email"
 	"likedream-goframe/internal/utils"
+	"time"
 
 	"github.com/hibiken/asynq"
 	"github.com/samber/oops"
@@ -44,7 +45,10 @@ func (h *EmailTaskHandler) ProcessTask(ctx context.Context, task *asynq.Task) (e
 
 	message.SetBodyString(mail.TypeTextHTML, payload.Body)
 
-	client, e := mail.NewClient(payload.SMTPHost, mail.WithPort(payload.SMTPPort), mail.WithSMTPAuth(mail.SMTPAuthAutoDiscover),
+	client, e := mail.NewClient(payload.SMTPHost, mail.WithPort(payload.SMTPPort),
+		mail.WithSMTPAuth(mail.SMTPAuthLogin), // Outlook 使用 LOGIN 认证
+		mail.WithTLSPortPolicy(mail.TLSMandatory),
+		mail.WithTimeout(30*time.Second), // 连接超时 30 秒
 		mail.WithUsername(payload.SMTPUsername), mail.WithPassword(payload.SMTPPassword))
 	utils.OkOrPanic(e, oops.WithContext(ctx).Wrap)
 

internal/framework/schedule/init.go

@@ -2,21 +2,85 @@ package schedule
 
 import (
 	"context"
+	cst2 "likedream-goframe/internal/common/cst"
+	"likedream-goframe/internal/common/db"
+	"likedream-goframe/internal/ent/cptask"
 	"likedream-goframe/internal/framework/queue"
 	"likedream-goframe/internal/utils"
 
 	"github.com/gogf/gf/v2/frame/g"
 	"github.com/gogf/gf/v2/os/gctx"
+	"github.com/hibiken/asynq"
 )
 
 func Init(ctx context.Context) {
 	queue.InitClient(ctx)
 	initServer(ctx)
 	initInspector(ctx)
+	cleanupStuckTasks(ctx) // 清理卡住的任务
 	initScanner()
 	initPeriodicTasks(ctx)
 }
 
+// cleanupStuckTasks 清理服务重启时卡在 queue 或 doing 状态的任务
+// 检查 asynq 队列中是否有 archived 的任务，如果有则删除并重置数据库状态
+func cleanupStuckTasks(ctx context.Context) {
+	// 获取所有状态为 queue 或 doing 的任务
+	stuckTasks, err := db.Global.CpTask.Query().
+		Where(
+			cptask.Or(
+				cptask.TaskStatusEQ(cptask.TaskStatusQueue),
+				cptask.TaskStatusEQ(cptask.TaskStatusDoing),
+			),
+		).
+		All(ctx)
+
+	if err != nil {
+		g.Log().Warningf(ctx, "查询卡住的任务失败: %v", err)
+		return
+	}
+
+	if len(stuckTasks) == 0 {
+		return
+	}
+
+	// 检查 archived 队列中的任务数量
+	archivedTasks, err := globalInspector.ListArchivedTasks(cst2.TaskQueueCpdaily, asynq.PageSize(10000))
+	if err != nil {
+		g.Log().Warningf(ctx, "获取 archived 任务列表失败: %v", err)
+	}
+
+	// 删除所有 archived 的任务，让它们可以重新入队
+	for _, task := range archivedTasks {
+		if err := globalInspector.DeleteTask(cst2.TaskQueueCpdaily, task.ID); err != nil {
+			g.Log().Debugf(ctx, "删除 archived 任务失败: %v", err)
+		}
+	}
+
+	if len(archivedTasks) > 0 {
+		g.Log().Infof(ctx, "已清理 %d 个 archived 状态的 asynq 任务", len(archivedTasks))
+	}
+
+	// 重置数据库中卡住的任务状态
+	affected, err := db.Global.CpTask.Update().
+		Where(
+			cptask.Or(
+				cptask.TaskStatusEQ(cptask.TaskStatusQueue),
+			),
+		).
+		SetTaskStatus(cptask.TaskStatusNotStart).
+		Save(ctx)
+
+	if err != nil {
+		g.Log().Warningf(ctx, "重置卡住的任务失败: %v", err)
+		return
+	}
+
+	if affected > 0 {
+		g.Log().Infof(ctx, "已重置 %d 个卡住的任务状态为 not_start", affected)
+	}
+}
+
 func Stop() {
 	globalScanner.StopScan()
 	g.Log().Info(gctx.GetInitCtx(), "task scanner stopped")
@@ -30,7 +94,10 @@ func Stop() {
 	g.Log().Info(gctx.GetInitCtx(), "asynq client closed")
 
 	globalServer.Shutdown()
-	g.Log().Info(gctx.GetInitCtx(), "asynq server stopped")
+	g.Log().Info(gctx.GetInitCtx(), "asynq main server stopped")
+
+	auxiliaryServer.Shutdown()
+	g.Log().Info(gctx.GetInitCtx(), "asynq auxiliary server stopped")
 
 	e = globalInspector.Close()
 	utils.OkOrPanic(e)

internal/framework/schedule/scanner.go

@@ -8,6 +8,7 @@ import (
 	cst2 "likedream-goframe/internal/common/cst"
 	"likedream-goframe/internal/common/db"
 	"likedream-goframe/internal/ent"
+	"likedream-goframe/internal/ent/courseentity"
 	"likedream-goframe/internal/ent/cptask"
 	"likedream-goframe/internal/framework/queue"
 	"likedream-goframe/internal/utils"
@@ -65,6 +66,7 @@ func (ts *TaskScanner) StartScan() {
 		}
 
 		ts.scanCpdailyTask()
+		ts.scanXuexitongTask()
 	}
 }
 
@@ -121,12 +123,16 @@ func (ts *TaskScanner) scanCpdailyTask() {
 		year, month, day := currentTime.Date()
 		execTime := time.Date(year, month, day, *task.ExecuteHour, *task.ExecuteMinute, 0, 0, cst2.TimeZoneShangHai)
 
+		// 添加随机延迟（0-30秒），分散同一时间点的任务执行，避免瞬间占满所有 worker
+		randomDelay := time.Duration(task.ID%30) * time.Second
+		execTime = execTime.Add(randomDelay)
+
 		e = utils.Catch(func() {
 			queue.AddTask(ts.ctx, asynq.NewTask(cst2.TaskCpdaily, payloadWriter.Bytes()),
-				asynq.Unique(5*time.Minute),
+				asynq.Unique(2*time.Minute),
 				asynq.MaxRetry(4),
 				asynq.ProcessAt(execTime),
-				asynq.Timeout(10*time.Minute),
+				asynq.Timeout(5*time.Minute),
 				asynq.Queue(cst2.TaskQueueCpdaily),
 			)
 			// 更新未开始的任务状态为队列中
@@ -134,7 +140,8 @@ func (ts *TaskScanner) scanCpdailyTask() {
 		})
 		if e != nil {
 			if errors.Is(e, asynq.ErrDuplicateTask) {
-				// 任务已在队列中，不更新下次执行时间，等任务执行完成后再处理
+				// 任务已在队列中，确保数据库状态也是 queue，然后跳过
+				db.Global.CpTask.Update().Where(cptask.ID(task.ID)).SetTaskStatus(cptask.TaskStatusQueue).Exec(ts.ctx)
 				g.Log().Debugf(ts.ctx, "任务ID %d 已在队列中，跳过本次扫描", task.ID)
 				failedIDS = append(failedIDS, task.ID)
 				continue
@@ -186,3 +193,76 @@ func (ts *TaskScanner) scanCpdailyTask() {
 		utils.LogError(ts.ctx, "重置过期任务下次执行时间失败", oops.With("task_id", task.ID, "next_execute_at", nextExecuteAt).Wrap(e))
 	}
 }
+
+// scanXuexitongTask 扫描学习通刷课任务
+func (ts *TaskScanner) scanXuexitongTask() {
+	// 查询所有状态为not_start或waiting的课程
+	allCourses := db.Global.CourseEntity.Query().
+		Where(
+			courseentity.Or(
+				courseentity.StatusEQ(courseentity.StatusNotStart),
+				courseentity.StatusEQ(courseentity.StatusWaiting),
+			),
+		).
+		AllX(ts.ctx)
+
+	if len(allCourses) == 0 {
+		return
+	}
+
+	// 按暴力模式区分任务
+	for _, course := range allCourses {
+		// 检查是否暂停
+		if course.Status == courseentity.StatusPause {
+			continue
+		}
+
+		payload := &XuexitongTaskPayload{
+			CourseID: course.ID,
+		}
+
+		payloadWriter := bytes.NewBuffer(nil)
+
+		e := gob.NewEncoder(payloadWriter).Encode(payload)
+		utils.OkOrPanic(e, oops.WithContext(ts.ctx).With("payload", payload).In("XuexitongTaskScanner.scanXuexitongTask").Wrap)
+
+		// 根据violence字段选择队列和任务类型
+		var queueName string
+		var taskType string
+
+		if course.Violence == courseentity.ViolenceYes {
+			queueName = cst2.TaskQueueXuexitongVio
+			taskType = cst2.TaskXuexitongVio
+		} else {
+			queueName = cst2.TaskQueueXuexitong
+			taskType = cst2.TaskXuexitong
+		}
+
+		e = utils.Catch(func() {
+			queue.AddTask(ts.ctx, asynq.NewTask(taskType, payloadWriter.Bytes()),
+				//asynq.Unique(30*time.Minute), // 30分钟内去重
+				asynq.MaxRetry(3),
+				asynq.Timeout(2*time.Hour), // 刷课可能需要较长时间
+				asynq.Queue(queueName),
+			)
+			// 更新状态为等待中
+			db.Global.CourseEntity.Update().
+				Where(courseentity.ID(course.ID), courseentity.StatusEQ(courseentity.StatusNotStart)).
+				SetStatus(courseentity.StatusWaiting).
+				Exec(ts.ctx)
+		})
+		if e != nil {
+			if errors.Is(e, asynq.ErrDuplicateTask) {
+				// 任务已在队列中，跳过
+				g.Log().Debugf(ts.ctx, "学习通课程ID %d 已在队列中，跳过本次扫描", course.ID)
+				continue
+			} else if errors.Is(e, context.Canceled) {
+				return
+			} else {
+				utils.LogError(ts.ctx, "扫描学习通任务时，入列失败", oops.With("course_id", course.ID).Wrap(e))
+			}
+		} else {
+			g.Log().Debugf(ts.ctx, "学习通课程 %s (ID: %d) 已加入队列: %s", course.CourseName, course.ID, queueName)
+		}
+	}
+}

internal/framework/schedule/server.go

@@ -12,13 +12,19 @@ import (
 	"github.com/hibiken/asynq"
 )
 
-var globalServer *asynq.Server
+var (
+	// 主任务服务器：处理 cpdaily、xuexitong 等业务任务
+	globalServer *asynq.Server
+	// 辅助任务服务器：处理 email、reset 等轻量级任务
+	auxiliaryServer *asynq.Server
+)
 
 func initServer(ctx context.Context) {
 	cfg := g.Cfg()
 
 	taskConcurrency := cfg.MustGetWithEnv(ctx, cst2.ConfigTaskConcurrency).Int()
 
+	// 主服务器：专门处理业务任务，使用大部分 worker
 	globalServer = asynq.NewServerFromRedisClient(redis.GetClient(), asynq.Config{
 		Concurrency:       taskConcurrency,
 		BaseContext:       gctx.GetInitCtx,
@@ -28,24 +34,45 @@ func initServer(ctx context.Context) {
 		},
 		DelayedTaskCheckInterval: 2 * time.Second,
 		ErrorHandler:             taskErrorHandler{},
+		// 只处理业务队列，使用严格优先级模式
+		StrictPriority: true,
 		Queues: map[string]int{
-			cst2.TaskQueueEmail:        2,
-			cst2.TaskQueueCpdaily:      10,
-			cst2.TaskQueueXuexitong:    5,
+			cst2.TaskQueueCpdaily:      10, // 最高优先级
 			cst2.TaskQueueXuexitongVio: 8,
-			cst2.TaskQueueDefault:      1,
+			cst2.TaskQueueXuexitong:    5,
 		},
 	})
 
-	mux := asynq.NewServeMux()
+	// 辅助服务器：专门处理邮件等轻量级任务，独立的 worker 池
+	auxiliaryServer = asynq.NewServerFromRedisClient(redis.GetClient(), asynq.Config{
+		Concurrency:       5, // 邮件任务不需要太多并发
+		BaseContext:       gctx.GetInitCtx,
+		TaskCheckInterval: 10 * time.Second,
+		RetryDelayFunc: func(n int, _ error, _ *asynq.Task) time.Duration {
+			return time.Duration(n) * 2 * time.Second
+		},
+		DelayedTaskCheckInterval: 2 * time.Second,
+		Queues: map[string]int{
+			cst2.TaskQueueEmail:   2,
+			cst2.TaskQueueDefault: 1,
+		},
+	})
 
-	mux.Handle(cst2.TaskCpdaily, &CpdailyTaskHandler{})
-	mux.Handle(cst2.TaskEmail, &EmailTaskHandler{})
-	mux.Handle(cst2.TaskResetDaily, &ResetTaskHandler{})
-	mux.Handle(cst2.TaskXuexitong, &XuexitongTaskHandler{})
-	mux.Handle(cst2.TaskXuexitongVio, &XuexitongTaskHandler{})
+	// 主服务器的任务处理器
+	mainMux := asynq.NewServeMux()
+	mainMux.Handle(cst2.TaskCpdaily, &CpdailyTaskHandler{})
+	mainMux.Handle(cst2.TaskXuexitong, &XuexitongTaskHandler{})
+	mainMux.Handle(cst2.TaskXuexitongVio, &XuexitongTaskHandler{})
 
-	e := globalServer.Start(mux)
+	// 辅助服务器的任务处理器
+	auxMux := asynq.NewServeMux()
+	auxMux.Handle(cst2.TaskEmail, &EmailTaskHandler{})
+	auxMux.Handle(cst2.TaskResetDaily, &ResetTaskHandler{})
+
+	e := globalServer.Start(mainMux)
+	utils.OkOrPanic(e)
+
+	e = auxiliaryServer.Start(auxMux)
 	utils.OkOrPanic(e)
 }
 

internal/service/cpdaily/cptask.go

@@ -265,7 +265,7 @@ func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq)
 		queue.AddTask(ctx, asynq.NewTask(cst2.TaskCpdaily, payloadWriter.Bytes()),
 			asynq.Unique(2*time.Minute),
 			asynq.MaxRetry(0),
-			asynq.Timeout(10*time.Minute),
+			asynq.Timeout(5*time.Minute),
 			asynq.Deadline(time.Now().Add(10*time.Minute)),
 			asynq.Queue(cst2.TaskQueueCpdaily),
 		)

internal/service/email/email_service.go

@@ -144,7 +144,7 @@ func (s *EmailServiceV1) sendEmail(ctx context.Context, to, subject, body string
 
 	task := queue.AddTask(ctx, asynq.NewTask(cst2.TaskEmail, payload.Bytes()), asynq.Queue(cst2.TaskQueueEmail),
 		asynq.MaxRetry(3),
-		asynq.Timeout(10*time.Minute),
+		asynq.Timeout(5*time.Minute),
 	)
 	logger.Logger.Info("添加任务: ", zap.String("task_id", task.ID), zap.String("邮箱", to))
 	return nil