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

package simCardQueue

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

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

const (
	TaskTypeSimCardSmartHubSyncOnePage = "SimCard:SmartHub:SyncOnePage"

	SmartHubSimCardInfoPageSize = 50 // SmartHub平台每次查询页数
)

// 数据类型转换 SmartHubInfo -> SimCardInfo
func (processor *SimCardProcessor) smartHubInfoToSimCardInfo(infoList []domain.SmartHubSimCardInfoItem) []domain.SimCardInfo {
	var simCardInfoList []domain.SimCardInfo
	for _, info := range infoList {
		var simCardInfo domain.SimCardInfo

		simCardInfo.GICCID = info.GICCID[:19] // 只取前19位不取检验码

		if len(info.DeviceInfo) == 0 {
			zap.L().Error("SmartHub SimCardInfo has no DeviceInfo", zap.String("GICCID", info.GICCID))
		} else {
			deviceInfo := info.DeviceInfo[0]
			simCardInfo.IMEI = deviceInfo.IMEI     // 国际移动用户识别码
			simCardInfo.IMSI = deviceInfo.IMSI     // 国际移动用户识别码
			simCardInfo.MSISDN = deviceInfo.MSISDN // 移动用户号码
		}

		switch info.Status {
		case domain.SmartHubSimStatusTestReady: // 可测试
			simCardInfo.Status = domain.SimStatusNotActivated
			break
		case domain.SmartHubSimStatusActivateReady: // 可激活
			simCardInfo.Status = domain.SimStatusNotActivated
			break
		case domain.SmartHubSimStatusActivated: // 已激活
			simCardInfo.Status = domain.SimStatusActivated
			break
		case domain.SmartHubSimStatusDeactivated: // 已停用
			simCardInfo.Status = domain.SimStatusDeactivated
			break
		case domain.SmartHubSimStatusInventory: // 库存
			simCardInfo.Status = domain.SimStatusNotActivated
			break
		case domain.SmartHubSimStatusRetired: // 失效
			simCardInfo.Status = domain.SimStatusRetired
			break
		default:
			simCardInfo.Status = domain.SimStatusUnknown
			break
		}

		simCardInfo.Platform = domain.SimPlatformSmartHub              // 数据来源平台
		simCardInfo.ExpirationTime = ""                                // SmartHub没有有效期
		simCardInfo.Month = time.Now().UTC().Format(time.DateOnly)[:7] // 统计月份 格式：yyyy-MM
		simCardInfo.AllDataUsage = info.CurrentCycleUsages.DataUsage   // 月数据用量(KB)
		simCardInfo.UpdateTime = time.Now().UTC()                      // UTC时间

		simCardInfoList = append(simCardInfoList, simCardInfo)
	}

	return simCardInfoList
}

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

	var payload SimCardSyncOnePagePayload
	err := json.Unmarshal(payloadStr, &payload)
	if err != nil {
		zap.L().Error("handleSimCardSmartHubSyncOnePageTask payload error.", zap.String("err", err.Error()), zap.String("payload", string(payloadStr)))
		return err
	}
	if payload.Date == "" {
		payload.Date = time.Now().UTC().Format(time.DateOnly)
	}

	// 请求批量查询SmartHub批量数据信息
	// 获取指定页信息
	resp, err := repositories.QuerySmartHubBatchSimCardInfo(uint(payload.Page), uint(payload.Size))
	if err != nil {
		zap.L().Error("handleSimCardSmartHubSyncOnePageTask QuerySmartHubBatchSimCardInfo failed.", zap.String("err", err.Error()), zap.Uint64("page", payload.Page), zap.Uint64("size", payload.Size))
		return err
	}

	// 转换数据结构
	infoList := processor.smartHubInfoToSimCardInfo(resp.Data)
	uuidStr := uuid.NewString()
	processor.putBatches(uuidStr, []simCardPageBatch{{
		Date: payload.Date, Pages: []domain.SimCardRunPage{{Platform: domain.SimPlatformSmartHub, Page: payload.Page}}, Info: 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 > resp.Total {
		if err := repositories.MarkSimCardLastPage(payload.Date, domain.SimPlatformSmartHub, payload.Page); err != nil {
			return err
		}
		return processor.enqueueSimCardReportIfComplete(payload.Date)
	}

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

	nextTask := asynq.NewTask(
		TaskTypeSimCardSmartHubSyncOnePage,
		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
}