FranklinWH-Toolbox/service/task/queue/fwhQueueOld/fwhTaskSiteInfoUpdate.go

package fwhQueueOld

import (
	"context"
	"fmt"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/dto"
	"fwh-toolbox-server/logger"
	"fwh-toolbox-server/repositories"

	"go.uber.org/zap"
)

var (
	TaskTypeSiteInfoUpdate = "task_SiteInfoUpdate"
)

// type TaskSiteInfoUpdate = types.TaskSiteInfoUpdate

func handleTaskSiteInfoUpdate(ctx context.Context, payload domain.TaskSiteInfoUpdate) error {
	// 更新运行状态
	fmt.Println("更新任务运行状态")
	if repoErr := repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressRunning, 0); repoErr != nil {
		zap.L().Error("更新站点信息Update结果状态失败(Running)", zap.String("err", repoErr.Error()))
		return repoErr
	}

	fmt.Printf("开始执行任务 %v\n", payload)
	total, err := executeTaskSiteInfoUpdate(ctx, payload)
	// 记录任务执行结果
	if err != nil {
		fmt.Printf("任务执行失败 %s\n", err.Error())
		if repoErr := repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressError, 0); repoErr != nil {
			zap.L().Error("更新站点信息Update结果状态失败(Error)", zap.String("err", repoErr.Error()))
			return repoErr
		}
	} else {
		fmt.Println("任务执行成功")
		if repoErr := repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressComplete, total); repoErr != nil {
			zap.L().Error("更新站点信息Update结果状态失败(Complete)", zap.String("err", repoErr.Error()))
			return repoErr
		}
	}

	return err
}

var executeTaskSiteInfoUpdate = func(ctx context.Context, payload domain.TaskSiteInfoUpdate) (uint64, error) {
	var pageSize uint = 50
	var pageNum uint = 1
	var total uint64 = 0
	var realTotal uint64 = 0

	fmt.Println("查询站点信息")
	for {
		resp, err := repositories.QuerySiteInfoList(dto.SiteInfoPara{
			PageSize: pageSize,
			PageNum:  pageNum,
		})
		if err != nil {
			zap.L().Error("查询站点信息失败", zap.String("err", err.Error()), zap.Uint("pageSize", pageSize), zap.Uint("pageNum", pageNum), zap.Uint64("total", total))
			return 0, err
		}

		for _, info := range *resp.Result {
			if err := repositories.InsertSiteInfoData(*domain.FwhSiteInfoToSiteInfoDomain(&info)); err != nil {
				zap.L().Error("保存站点信息失败", zap.String("err", err.Error()), zap.Uint("pageSize", pageSize), zap.Uint("pageNum", pageNum), zap.Uint64("total", total))
				return 0, err
			}
			realTotal++
		}

		pageNum++

		total = resp.Total
		if (int64(pageNum)-1)*int64(pageSize) >= int64(total) {
			break
		}
	}

	return realTotal, nil
}

// 插入站点分析队列
func EnqueueSiteInfoUpdateTask(payload domain.TaskSiteInfoUpdate) error {
	p := TaskPayload[domain.TaskSiteInfoUpdate]{
		TaskPayloadBase: TaskPayloadBase{
			Type: TaskTypeSiteInfoUpdate,
		},
		Payload: payload,
	}
	if err := EnqueueUniqueTask[domain.TaskSiteInfoUpdate](p); err != nil {
		logger.Std.Printf("EnqueueSiteInfoUpdateTask error : %s\n", err.Error())
		return err
	}
	return nil
}