sugarBecomer
352 lines
13 KiB
Go
352 lines
13 KiB
Go
package cpdaily
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/gob"
|
|
"encoding/json"
|
|
v1 "likedream-goframe/internal/api/v1"
|
|
"likedream-goframe/internal/application/cpdaily/login"
|
|
"likedream-goframe/internal/application/cpdaily/school"
|
|
"likedream-goframe/internal/application/cpdaily/task"
|
|
"likedream-goframe/internal/application/cpdaily/task/formtype"
|
|
cst2 "likedream-goframe/internal/common/cst"
|
|
db2 "likedream-goframe/internal/common/db"
|
|
errors2 "likedream-goframe/internal/common/errors"
|
|
"likedream-goframe/internal/ent"
|
|
"likedream-goframe/internal/ent/cptask"
|
|
"likedream-goframe/internal/ent/cpuser"
|
|
"likedream-goframe/internal/ent/sysuser"
|
|
"likedream-goframe/internal/framework/queue"
|
|
"likedream-goframe/internal/framework/schedule"
|
|
"likedream-goframe/internal/service/balance"
|
|
"likedream-goframe/internal/utils"
|
|
"time"
|
|
|
|
"entgo.io/ent/dialect/sql"
|
|
"github.com/bytedance/sonic"
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/hibiken/asynq"
|
|
"github.com/samber/oops"
|
|
"github.com/shopspring/decimal"
|
|
)
|
|
|
|
type CpTaskServiceV1 struct {
|
|
}
|
|
|
|
func (v CpTaskServiceV1) GetRecentTasks(ctx context.Context, req *v1.GetRecentTasksReq) *utils.Response[[]string] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
// 检测是否属于自己的cpuser
|
|
cpUser := db2.Global.CpUser.Query().Where(cpuser.ID(req.UserId), cpuser.HasSysuserWith(sysuser.ID(user.ID))).WithSysuser().WithSchool().OnlyX(ctx)
|
|
if cpUser.Edges.Sysuser == nil {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
|
|
}
|
|
schoolDetail := school.GetSchoolDetail(ctx, cpUser.SchoolID)
|
|
client := login.GetClient(ctx, cpUser.Cookie)
|
|
if !login.CookieEnabled(ctx, schoolDetail.WecloudUrl, client) {
|
|
loginExecutor := login.GetCpdailyLoginBySchoolInfo(ctx, &schoolDetail)
|
|
loginExecutor.Login(ctx, &schoolDetail, cpUser.Username, cpUser.Password)
|
|
cookie := loginExecutor.GetCookieString()
|
|
db2.Global.CpUser.UpdateOne(cpUser).SetCookie(cookie).ExecX(ctx)
|
|
client = login.GetClient(ctx, cookie)
|
|
}
|
|
taskService := task.GetCpdailyTaskService(ctx, req.Type, client, schoolDetail)
|
|
titleList := taskService.GetRecentTaskTitle(ctx)
|
|
return utils.SuccessWithData[[]string](titleList)
|
|
}
|
|
|
|
func (v CpTaskServiceV1) AddTask(ctx context.Context, req *v1.AddTaskReq) *utils.Response[any] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
// 检测是否属于自己的cpuser
|
|
cpUser := db2.Global.CpUser.Query().Where(cpuser.ID(req.UserId), cpuser.HasSysuserWith(sysuser.ID(user.ID))).WithSysuser().OnlyX(ctx)
|
|
if cpUser.Edges.Sysuser == nil {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
|
|
}
|
|
db2.Global.CpTask.Create().
|
|
SetCpUserID(req.UserId).
|
|
SetTaskType(cptask.TaskType(req.Type)).
|
|
SetTitle(req.Title).
|
|
SetStep(cptask.StepForm).
|
|
SetTaskStatus(cptask.TaskStatusUnknown).
|
|
SetNotificationType(cptask.NotificationTypeNone).
|
|
SetUserID(user.ID).
|
|
SaveX(ctx)
|
|
return utils.SuccessWithMsg("添加成功")
|
|
}
|
|
|
|
func (v CpTaskServiceV1) List(ctx context.Context, req *v1.CpTaskListReq) *utils.Response[[]*ent.CpTask] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
query := db2.Global.CpTask.Query().Where(cptask.HasSysUserWith(sysuser.ID(user.ID))).WithCpUser().Order(ent.Desc(cptask.FieldCreateTime))
|
|
if req.TaskType != "" {
|
|
query.Where(cptask.TaskTypeEQ(req.TaskType))
|
|
}
|
|
if req.Username != "" {
|
|
query.Where(cptask.HasCpUserWith(cpuser.UsernameContains(req.Username)))
|
|
}
|
|
if req.TaskStatus != "" {
|
|
query.Where(cptask.TaskStatusEQ(req.TaskStatus))
|
|
}
|
|
total := query.CountX(ctx)
|
|
var data []*ent.CpTask
|
|
if total > 0 {
|
|
query.Limit(req.Size).Offset((req.Page - 1) * req.Size)
|
|
data = query.AllX(ctx)
|
|
for _, item := range data {
|
|
item.Edges.CpUser.Password = ""
|
|
item.Edges.CpUser.Cookie = ""
|
|
}
|
|
}
|
|
return utils.SuccessWithTotal[[]*ent.CpTask](data, total)
|
|
}
|
|
|
|
func (v CpTaskServiceV1) TaskFormData(ctx context.Context, req *v1.GetCpTaskFormReq) *utils.Response[any] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
// 检测是否属于自己的cpTask
|
|
cpTask := db2.Global.CpTask.Query().Where(cptask.ID(req.Id),
|
|
cptask.HasSysUserWith(sysuser.ID(user.ID))).
|
|
WithCpUser().WithSysUser().OnlyX(ctx)
|
|
cpUser := cpTask.Edges.CpUser
|
|
if cpUser == nil {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
|
|
}
|
|
if cpTask.FormData != "" {
|
|
var formData []*formtype.CpTaskFormItem
|
|
err := sonic.Unmarshal([]byte(cpTask.FormData), &formData)
|
|
utils.OkOrPanic(err)
|
|
return utils.SuccessWithData[any](formData)
|
|
}
|
|
schoolDetail := school.GetSchoolDetail(ctx, cpTask.Edges.CpUser.SchoolID)
|
|
cookie := cpUser.Cookie
|
|
client := login.GetClient(ctx, cookie)
|
|
if !login.CookieEnabled(ctx, schoolDetail.WecloudUrl, client) {
|
|
loginExecutor := login.GetCpdailyLoginBySchoolInfo(ctx, &schoolDetail)
|
|
loginExecutor.Login(ctx, &schoolDetail, cpUser.Username, cpUser.Password)
|
|
cookie = loginExecutor.GetCookieString()
|
|
db2.Global.CpUser.UpdateOneID(cpUser.ID).SetCookie(cookie).ExecX(ctx)
|
|
client = login.GetClient(ctx, cookie)
|
|
}
|
|
taskService := task.GetCpdailyTaskService(ctx, cpTask.TaskType, client, schoolDetail)
|
|
formMap := taskService.QueryFormByTitle(ctx, cpTask.Title)
|
|
formDetail := taskService.ParseFormJson(ctx, formMap)
|
|
return utils.SuccessWithData[any](formDetail)
|
|
}
|
|
|
|
func (v CpTaskServiceV1) PutTaskFormData(ctx context.Context, req *v1.PutCpTaskFormReq) *utils.Response[any] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
// 检测是否属于自己的cpTask
|
|
cpTask := db2.Global.CpTask.Query().Where(cptask.ID(req.Id),
|
|
cptask.HasSysUserWith(sysuser.ID(user.ID))).
|
|
WithCpUser().WithSysUser().OnlyX(ctx)
|
|
cpUser := cpTask.Edges.CpUser
|
|
if cpUser == nil {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
|
|
}
|
|
update := db2.Global.CpTask.UpdateOne(cpTask)
|
|
marshal, _ := json.Marshal(req.Forms)
|
|
update.SetFormData(string(marshal))
|
|
if cpTask.Step == cptask.StepForm {
|
|
update.SetStep(cptask.StepExecute)
|
|
}
|
|
update.SaveX(ctx)
|
|
return utils.SuccessWithMsg("保存成功")
|
|
}
|
|
|
|
func (v CpTaskServiceV1) TaskExecute(ctx context.Context, req *v1.GetCpTaskExecuteReq) *utils.Response[any] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
// 检测是否属于自己的cpTask
|
|
cpTask := db2.Global.CpTask.Query().Where(cptask.ID(req.Id),
|
|
cptask.HasSysUserWith(sysuser.ID(user.ID))).OnlyX(ctx)
|
|
if cpTask == nil {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
|
|
}
|
|
return utils.SuccessWithData[any](cpTask)
|
|
}
|
|
|
|
func (v CpTaskServiceV1) PostExecute(ctx context.Context, req *v1.PostCpTaskExecuteReq) *utils.Response[any] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
// 检测是否属于自己的cpTask
|
|
cpTask := db2.Global.CpTask.Query().Where(cptask.ID(req.Id),
|
|
cptask.HasSysUserWith(sysuser.ID(user.ID))).
|
|
WithCpUser().WithSysUser().OnlyX(ctx)
|
|
if cpTask == nil {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
|
|
}
|
|
if cpTask.Step == cptask.StepForm {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "请先完成前置参数设置"))
|
|
}
|
|
consume := req.Frequency - cpTask.Frequency
|
|
if req.Frequency < 1 || consume < 0 {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "执行次数应不低于原有次数"))
|
|
}
|
|
|
|
//扣积分
|
|
basePoint, err := decimal.NewFromString(g.Cfg().MustGet(ctx, cst2.ConfigBasePointCpdaily).String())
|
|
utils.OkOrPanic(err)
|
|
if user.Edges.Agent != nil {
|
|
basePoint = basePoint.Mul(user.Edges.Agent.Rate)
|
|
}
|
|
needPoint := basePoint.Mul(decimal.NewFromInt(int64(consume)))
|
|
balance.BalanceService.ChangeUserBalance(ctx, user.ID, needPoint.Neg())
|
|
// 更新任务
|
|
e := db2.WithTx(ctx, func(ctx context.Context, tx *ent.Tx) {
|
|
update := db2.Global.CpTask.UpdateOne(cpTask).
|
|
SetStep(cptask.StepComplete).
|
|
SetFrequency(req.Frequency).
|
|
SetTaskStatus(cptask.TaskStatusNotStart).
|
|
SetNotificationType(req.NotificationType).
|
|
SetExecuteHour(req.ExecuteHour).
|
|
SetExecuteMinute(req.ExecuteMinute)
|
|
if cpTask.Step == cptask.StepExecute {
|
|
update.SetTaskStatus(cptask.TaskStatusNotStart)
|
|
}
|
|
cpTask := update.SaveX(ctx)
|
|
|
|
// 其他状态（暂停、失败等），使用正常的计算逻辑
|
|
nextExecuteAt := utils.GetCpdailyNextExecuteTime(cpTask)
|
|
|
|
db2.Global.CpTask.UpdateOne(cpTask).SetNextExecuteTime(nextExecuteAt).ExecX(ctx)
|
|
}, &sql.TxOptions{})
|
|
utils.OkOrPanic(e)
|
|
|
|
return utils.SuccessWithMsg("更新成功")
|
|
}
|
|
|
|
func (v CpTaskServiceV1) Delete(ctx context.Context, req *v1.DeleteTaskReq) *utils.Response[any] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
// 检测是否属于自己的cpTask
|
|
cpTask := db2.Global.CpTask.Query().Where(cptask.ID(req.Id),
|
|
cptask.HasSysUserWith(sysuser.ID(user.ID))).
|
|
WithCpUser().WithSysUser().OnlyX(ctx)
|
|
if cpTask == nil {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
|
|
}
|
|
// 异步终止线程，不管是否成功
|
|
//go schedule.StopTask(context.Background(), cst2.TaskQueueCpdaily, fmt.Sprintf(cst2.TaskCpdailyId, req.Id))
|
|
db2.Global.CpTask.DeleteOne(cpTask).ExecX(ctx)
|
|
return utils.SuccessWithMsg("删除成功")
|
|
}
|
|
|
|
func (v CpTaskServiceV1) ExecuteNow(ctx context.Context, req *v1.ExecuteNowReq) *utils.Response[any] {
|
|
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
|
|
// 查询任务并检查状态
|
|
cpTask := db2.Global.CpTask.Query().
|
|
Where(cptask.ID(req.Id), cptask.UserID(user.ID)).
|
|
Select(cptask.FieldTaskStatus, cptask.FieldFrequency, cptask.FieldSuccessCount).
|
|
FirstX(ctx)
|
|
|
|
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,
|
|
}
|
|
|
|
payloadWriter := bytes.NewBuffer(nil)
|
|
|
|
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(cst2.TaskCpdaily, payloadWriter.Bytes()),
|
|
asynq.Unique(2*time.Minute),
|
|
asynq.MaxRetry(0),
|
|
asynq.Timeout(10*time.Minute),
|
|
asynq.Deadline(time.Now().Add(10*time.Minute)),
|
|
asynq.Queue(cst2.TaskQueueCpdaily),
|
|
)
|
|
// 更新未开始的任务状态为队列中
|
|
db2.Global.CpTask.Update().Where(cptask.ID(cpTask.ID), cptask.TaskStatusEQ(cptask.TaskStatusNotStart)).SetTaskStatus(cptask.TaskStatusQueue).Exec(ctx)
|
|
})
|
|
if e != nil {
|
|
g.Log().Errorf(ctx, "立即执行任务入列失败: %+v", e)
|
|
utils.OkOrPanic(errors2.NewBusinessError(500, "同一个任务请勿连续执行"))
|
|
}
|
|
return utils.SuccessWithMsg("执行成功")
|
|
}
|
|
|
|
func (v CpTaskServiceV1) PauseTask(ctx context.Context, req *v1.PauseTaskReq) *utils.Response[any] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
cpTask := db2.Global.CpTask.Query().Where(cptask.ID(req.Id),
|
|
cptask.HasSysUserWith(sysuser.ID(user.ID))).
|
|
WithCpUser().WithSysUser().OnlyX(ctx)
|
|
if cpTask == nil {
|
|
utils.OkOrPanic(errors2.NewBusinessError(400, "非法的今日校园账号"))
|
|
}
|
|
if cpTask.TaskStatus == cptask.TaskStatusPause {
|
|
nextExecuteAt := utils.GetCpdailyNextExecuteTime(cpTask)
|
|
db2.Global.CpTask.UpdateOne(cpTask).SetTaskStatus(cptask.TaskStatusNotStart).SetNextExecuteTime(nextExecuteAt).ExecX(ctx)
|
|
return utils.SuccessWithMsg("恢复成功")
|
|
|
|
} else {
|
|
//go schedule.StopTask(context.Background(), cst2.TaskQueueCpdaily, fmt.Sprintf(cst2.TaskCpdailyId, req.Id))
|
|
db2.Global.CpTask.UpdateOne(cpTask).SetTaskStatus(cptask.TaskStatusPause).SetNextExecuteTime(-1).ExecX(ctx)
|
|
return utils.SuccessWithMsg("暂停成功")
|
|
}
|
|
}
|
|
|
|
func (v CpTaskServiceV1) GetTaskStatistics(ctx context.Context, req *v1.GetTaskStatisticsReq) *utils.Response[[]*v1.TaskStatisticsItem] {
|
|
user := utils.GetUserInfoFromCtx(ctx)
|
|
|
|
// 计算近七日的日期范围
|
|
now := time.Now()
|
|
statistics := make([]*v1.TaskStatisticsItem, 7)
|
|
|
|
for i := 6; i >= 0; i-- {
|
|
date := now.AddDate(0, 0, -i)
|
|
dateStr := date.Format("2006-01-02")
|
|
|
|
// 计算当天的开始和结束时间戳
|
|
startOfDay := time.Date(date.Year(), date.Month(), date.Day(), 0, 0, 0, 0, date.Location())
|
|
endOfDay := startOfDay.AddDate(0, 0, 1)
|
|
|
|
// 查询需要完成的任务数：状态不是暂停的任务，且执行次数大于成功次数
|
|
pendingCount := db2.Global.CpTask.Query().
|
|
Where(
|
|
cptask.HasSysUserWith(sysuser.ID(user.ID)),
|
|
cptask.TaskStatusNEQ(cptask.TaskStatusPause),
|
|
cptask.FrequencyGT(0),
|
|
).
|
|
Where(func(s *sql.Selector) {
|
|
s.Where(sql.ExprP("frequency > success_count"))
|
|
}).
|
|
CountX(ctx)
|
|
|
|
// 查询成功执行的任务数：当天成功的任务日志数
|
|
successCount := db2.Global.CpTaskLog.Query().
|
|
Where(func(s *sql.Selector) {
|
|
s.Where(sql.GTE(s.C("create_time"), startOfDay))
|
|
s.Where(sql.LT(s.C("create_time"), endOfDay))
|
|
s.Where(sql.EQ(s.C("is_success"), true))
|
|
s.Where(sql.EQ(s.C("system_user_id"), user.ID))
|
|
}).
|
|
CountX(ctx)
|
|
|
|
statistics[6-i] = &v1.TaskStatisticsItem{
|
|
Date: dateStr,
|
|
PendingTask: pendingCount,
|
|
SuccessTask: successCount,
|
|
}
|
|
}
|
|
|
|
return utils.SuccessWithData[[]*v1.TaskStatisticsItem](statistics)
|
|
}
|
|
|
|
func NewCpTaskServiceV1() *CpTaskServiceV1 {
|
|
return &CpTaskServiceV1{}
|
|
}
|