package mongo import ( "context" "fmt" "fwh-toolbox-server/database" "fwh-toolbox-server/domain" "fwh-toolbox-server/dto" "time" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" "go.uber.org/zap" ) const ( CollectionName_SimCardManage = "SimCardManage" CollectionName_SimCardMonthlyData = "SimCardMonthlyData" ) // 查询SIM卡信息(批量) func GetSimCardBatchInfo(page, size uint, search string, status *string) (*dto.ResPage[domain.SimCardInfo], error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() collection := database.GetCollection(CollectionName_SimCardManage) // 查询条件 filter := bson.M{} // 搜索 _id (GICCID) if search != "" { filter["_id"] = bson.M{ "$regex": search, "$options": "i", // 不区分大小写,可去掉 } } // 过滤状态 if status != nil { filter["status"] = bson.M{ "$eq": *status, } } findOptions := options.Find(). SetSkip(int64((page - 1) * size)). SetLimit(int64(size)). SetSort(bson.D{ {Key: "allDataUsage", Value: -1}, // 降序 }) cursor, err := collection.Find(ctx, filter, findOptions) if err != nil { return nil, err } defer cursor.Close(ctx) var infoList []domain.SimCardInfo for cursor.Next(ctx) { var info domain.SimCardInfo if err := cursor.Decode(&info); err != nil { return nil, err } infoList = append(infoList, info) } // 查询总数 total, err := collection.CountDocuments(ctx, bson.M{}) if err != nil { return nil, err } res := dto.ResPage[domain.SimCardInfo]{ Data: infoList, PageSize: size, PageNum: page, Total: uint(total), } return &res, nil } // 查询SIM卡信息(单卡) func GetSimCardInfo(id string) (*domain.SimCardInfo, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() collection := database.GetCollection(CollectionName_SimCardManage) var info domain.SimCardInfo err := collection.FindOne(ctx, bson.M{"_id": id}).Decode(&info) if err != nil { return nil, err } return &info, nil } // 查询SIM卡月数据(单卡) func GetSimCardMonthlyData(month string, id string) (*domain.SimCardMonthlyData, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() collection := database.GetCollection(CollectionName_SimCardMonthlyData) var info domain.SimCardMonthlyData err := collection.FindOne(ctx, bson.M{"simId": id, "month": month}).Decode(&info) if err != nil { return nil, err } return &info, nil } // 插入SIM卡基本信息数据 func UpdateSimCardInfo(info domain.SimCardInfo) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() collection := database.GetCollection(CollectionName_SimCardManage) filter := bson.M{"_id": info.GICCID} // 按ID匹配 _, err := collection.ReplaceOne(ctx, filter, info, options.Replace().SetUpsert(true)) if err != nil { return err } return nil } // 批量插入SIM卡基本信息数据 func UpdateSimCardBatchInfo(infoList []domain.SimCardInfo) (int64, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() collection := database.GetCollection(CollectionName_SimCardManage) batchSize := 1000 var totalInserted int64 for start := 0; start < len(infoList); start += batchSize { end := start + batchSize if end > len(infoList) { end = len(infoList) } models := make([]mongo.WriteModel, 0, end-start) for _, doc := range infoList[start:end] { model := mongo.NewReplaceOneModel(). SetFilter(bson.M{ "_id": doc.GICCID, }). SetReplacement(doc). SetUpsert(true) models = append(models, model) } result, err := collection.BulkWrite( ctx, models, options.BulkWrite(). SetOrdered(false), ) if err != nil { zap.L().Error("UpdateSimCardBatchInfo bulk insert failed.", zap.String("err", err.Error()), zap.Int("start", start), zap.Int("end", end), zap.Int64("totalInserted", totalInserted)) return totalInserted, fmt.Errorf( "bulk insert failed (%d~%d): %w", start, end, err, ) } totalInserted += result.InsertedCount } return totalInserted, nil } // 插入SIM卡日流量数据 func UpdateSimCardDayData(data domain.SimCardDayData) error { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() collection := database.GetCollection(CollectionName_SimCardMonthlyData) dailyUsageField := fmt.Sprintf( "dailyUsage.%d", data.Day, ) update := bson.M{ "$set": bson.M{ "allDataUsage": data.AllDataUsage, "updateTime": data.UpdateTime, dailyUsageField: data.DailyUsage, }, "$setOnInsert": bson.M{ "simId": data.SimId, "month": data.Month, }, } filter := bson.M{"simId": data.SimId, "month": data.Month} // 按ID和month匹配 _, err := collection.UpdateOne(ctx, filter, update, options.Update().SetUpsert(true)) if err != nil { return err } return nil } // 批量插入SIM卡日流量数据 func UpdateSimCardBatchDayData(dataList []domain.SimCardDayData) (int64, error) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() collection := database.GetCollection(CollectionName_SimCardMonthlyData) batchSize := 1000 var totalInserted int64 for start := 0; start < len(dataList); start += batchSize { end := start + batchSize if end > len(dataList) { end = len(dataList) } models := make([]mongo.WriteModel, 0, end-start) for _, doc := range dataList[start:end] { dailyUsageField := fmt.Sprintf("dailyUsage.%d", doc.Day) update := bson.M{ "$set": bson.M{ "allDataUsage": doc.AllDataUsage, "updateTime": doc.UpdateTime, dailyUsageField: doc.DailyUsage, }, "$setOnInsert": bson.M{ "simId": doc.SimId, "month": doc.Month, }, } model := mongo.NewUpdateOneModel(). SetFilter(bson.M{ "simId": doc.SimId, "month": doc.Month, }). SetUpdate(update). SetUpsert(true) models = append(models, model) } result, err := collection.BulkWrite( ctx, models, options.BulkWrite(). SetOrdered(false), ) if err != nil { zap.L().Error("UpdateSimCardBatchMonthlyData bulk insert failed.", zap.String("err", err.Error()), zap.Int("start", start), zap.Int("end", end), zap.Int64("totalInserted", totalInserted)) return totalInserted, fmt.Errorf( "bulk insert failed (%d~%d): %w", start, end, err, ) } totalInserted += result.InsertedCount } return totalInserted, nil }