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 }