FranklinWH-Toolbox/service/task/queue/simCardQueue/simCardTaskAcceleronix.go

package simCardQueue

import (
	"context"
	"encoding/json"
	"fwh-toolbox-server/domain"
	"fwh-toolbox-server/repositories"
	"strconv"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/hibiken/asynq"
	"go.uber.org/zap"
)

const (
	TaskTypeSimCardAcceleronixSyncOnePage = "SimCard:Acceleronix:SyncOnePage"

	AcceleronixSimCardInfoPageSize = 50 // Acceleronix平台每次查询页数
)

// 数据类型转换 AcceleronixInfo -> SimCardInfo
func (processor *SimCardProcessor) acceleronixInfoToSimCardInfo(infoList []domain.AcceleronixSimCardInfoItem, dataUsage map[string]*domain.AcceleronixSimCardDataUsageInfoResponse) []domain.SimCardInfo {
	var simCardInfoList []domain.SimCardInfo

	for _, info := range infoList {
		var simCardInfo domain.SimCardInfo

		simCardInfo.GICCID = info.Iccid[:19] // 只取前19位不取检验码
		simCardInfo.IMEI = info.Imei
		simCardInfo.IMSI = info.Imsi
		simCardInfo.MSISDN = info.Msisdn

		switch info.SimState {
		case domain.SCS_PreActivation: // 2 预激活
			simCardInfo.Status = domain.SimStatusNotActivated
			break
		case domain.SCS_Test: // 3 测试
			simCardInfo.Status = domain.SimStatusNotActivated
			break
		case domain.SCS_Silence: // 4 沉默
			simCardInfo.Status = domain.SimStatusNotActivated
			break
		case domain.SCS_Standby: // 5 待机
			simCardInfo.Status = domain.SimStatusNotActivated
			break
		case domain.SCS_Activated: // 6 激活
			simCardInfo.Status = domain.SimStatusActivated
			break
		case domain.SCS_Deactivated: // 7 停机
			simCardInfo.Status = domain.SimStatusDeactivated
			break
		case domain.SCS_Pause: // 8 暂停
			simCardInfo.Status = domain.SimStatusDeactivated
			break
		case domain.SCS_PreCancellation: // 9 预注销
			simCardInfo.Status = domain.SimStatusDeactivated
			break
		case domain.SCS_Cancellation: // 10 注销
			simCardInfo.Status = domain.SimStatusRetired
			break
		default:
			simCardInfo.Status = domain.SimStatusUnknown
			break
		}

		simCardInfo.ExpirationTime = info.ExpirationTime // 到期时间(后付费卡没有到期时间) 格式：yyyy-MM-dd
		simCardInfo.Platform = domain.SimPlatformAcceleronix // 数据来源平台

		usageInfo := dataUsage[info.Iccid]
		if usageInfo == nil {
			zap.L().Warn("AcceleronixSimCardDataUsageInfo not exist.", zap.String("Iccid", info.Iccid), zap.Any("dataUsageMap", dataUsage))
		} else {
			simCardInfo.Month = usageInfo.Month // 统计月份 格式：yyyy-MM
			totalDataUsage, err := strconv.ParseFloat(usageInfo.TotalDataUsage, 64)
			if err != nil {
				zap.L().Warn("TotalDataUsage ParseFloat failed.", zap.String("err", err.Error()), zap.String("TotalDataUsage", usageInfo.TotalDataUsage))
				simCardInfo.AllDataUsage = 0
			} else {
				simCardInfo.AllDataUsage = totalDataUsage * 1024 // 月数据用量(KB)
			}
		}

		simCardInfo.UpdateTime = time.Now().UTC() // UTC时间

		simCardInfoList = append(simCardInfoList, simCardInfo)
	}

	return simCardInfoList
}

// 处理Acceleronix信息收集任务
func (processor *SimCardProcessor) handleSimCardAcceleronixSyncOnePageTask(ctx context.Context, task *asynq.Task) error {
	payloadStr := task.Payload()

	var payload SimCardSyncOnePagePayload
	err := json.Unmarshal(payloadStr, &payload)
	if err != nil {
		zap.L().Error("handleSimCardAcceleronixSyncOnePageTask payload error.", zap.String("err", err.Error()), zap.String("payload", string(payloadStr)))
		return err
	}

	// 请求批量查询Acceleronix批量数据信息
	// 查询Sim卡基本信息列表
	respInfoList, err := repositories.QueryAcceleronixSimCardList(uint(payload.Page), uint(payload.Size))
	if err != nil {
		zap.L().Error("handleSimCardAcceleronixSyncOnePageTask QueryAcceleronixSimCardList failed.", zap.String("err", err.Error()), zap.Uint64("page", payload.Page), zap.Uint64("size", payload.Size))
		return err
	}

	idList := []string{}
	for _, info := range respInfoList.Data.List {
		idList = append(idList, info.Iccid)
	}

	// 查询
	mouth := strings.ReplaceAll(time.Now().UTC().Format(time.DateOnly)[:7], "-", "") // 统计月份 格式：yyyyMM
	respUsageList, err := repositories.QueryAcceleronixBatchSimCardDataInfo(idList, mouth)
	if err != nil {
		zap.L().Error("handleSimCardAcceleronixSyncOnePageTask QueryAcceleronixBatchSimCardDataInfo failed.", zap.String("err", err.Error()), zap.Strings("idList", idList), zap.String("mouth", mouth))
		return err
	}

	// 生成数据用量信息map
	dataUsageMap := make(map[string]*domain.AcceleronixSimCardDataUsageInfoResponse)
	for _, info := range respUsageList {
		dataUsageMap[info.Iccid] = &info
	}

	// 转换数据结构
	infoList := processor.acceleronixInfoToSimCardInfo(respInfoList.Data.List, dataUsageMap)
	uuidStr := uuid.NewString()
	processor.infoCacheMap[uuidStr] = infoList
	zap.L().Debug("Acceleronix get infoList.", zap.Int("len", len(infoList)))

	// 创建聚合组任务
	aggTask := asynq.NewTask(
		TaskTypeSimCardAggregation,
		[]byte(uuidStr),
		asynq.Queue("default"),
		asynq.Unique(5*time.Minute),
		asynq.Group(GroupNameSimCardAggregation),
	)
	_, err = TaskClient.Enqueue(aggTask)
	if err != nil {
		zap.L().Error("TaskTypeSimCardAggregation Enqueue failed.", zap.String("err", err.Error()), zap.Uint64("size", payload.Size), zap.Uint64("page", payload.Page))
		return err
	}

	// 判断是否查完所有信息
	if payload.Page*payload.Size > uint64(respInfoList.Data.Total) {
		return nil
	}

	// 查询太快，间隔1s
	time.Sleep(time.Second)

	// 创建下一个任务（查询下一页）
	payload.Page++
	taskPayloadStr, err := json.Marshal(payload)

	nextTask := asynq.NewTask(
		TaskTypeSimCardAcceleronixSyncOnePage,
		taskPayloadStr,
		asynq.MaxRetry(5),
		asynq.Queue("default"),
		asynq.Timeout(time.Minute),
		asynq.Unique(5*time.Minute),
		asynq.Retention(7*24*time.Hour), // 成功后保留7天
	)
	_, err = TaskClient.Enqueue(nextTask)
	if err != nil {
		zap.L().Error("TaskTypeSimCardSyncOnePage Enqueue failed.", zap.String("err", err.Error()), zap.Uint64("size", payload.Size), zap.Uint64("page", payload.Page))
		return err
	}

	return nil
}