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 } // 请求批量查询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.infoCacheMap[uuidStr] = 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 { return nil } // 创建下一个任务(查询下一页) 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 }