package statvidmod import ( "fmt" "time" "91porn-server/common" "91porn-server/common/db" "91porn-server/common/log" "91porn-server/models" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" ) var mdb *db.MongoDB const table = models.VideoStat func coll(t *db.MongoTool) *db.MongoTool { if t == nil { return mdb.Coll(table) } return t.Coll(table) } // initIndex 初始化索引 func initIndex() { many := []mongo.IndexModel{ { Keys: bson.D{{Key: "vid", Value: 1}}, }, { Keys: bson.D{{Key: "sumDate", Value: -1}, {Key: "vid", Value: 1}}, Options: options.Index().SetUnique(true), }, { Keys: bson.D{{Key: "playCount", Value: -1}}, Options: options.Index().SetSparse(true), }, { Keys: bson.D{{Key: "createdAt", Value: -1}}, }, { Keys: bson.D{{Key: "updatedAt", Value: -1}}, }, { Keys: bson.D{{Key: "recordAt", Value: -1}}, }, { Keys: bson.D{{Key: "sumDate", Value: -1}, {Key: "playCount", Value: -1}}, }, } if _, err := coll(nil).CreateIndex(many); err != nil { panic(fmt.Sprintf("%s model set index err ==>[%+v]", table, err)) } } func ChangeStatTrans(trans *db.MongoTool, sumDate time.Time, recordTime time.Time, statDocMap VideoStatDocMap) error { writes := make([]mongo.WriteModel, len(statDocMap)) i := 0 for vid, statDoc := range statDocMap { filter := M{ "sumDate": sumDate, "vid": vid, } setM, err := common.ToBsonM(statDoc.VideoStatSetDoc) if err != nil { return err } setM["recordAt"] = recordTime setM["updatedAt"] = time.Now() incM, err := common.ToBsonM(statDoc.VideoStatIncDoc) if err != nil { return err } update := M{ "$setOnInsert": M{ "sumDate": sumDate, "vid": vid, "createdAt": time.Now(), }, } if len(incM) != 0 { update["$inc"] = incM } if len(setM) != 0 { update["$set"] = setM } writes[i] = mongo.NewUpdateOneModel(). SetFilter(filter). SetUpdate(update). SetUpsert(true) i++ log.Debug(fmt.Sprintf("sumDate:%s table:%s [ vid:%+v filter:%+v update:%+v ]\n", sumDate, table, vid, filter, update)) } if len(writes) == 0 { return nil } //bulkWrite 不是原子操作 不具备事务性 opt := (&options.BulkWriteOptions{}).SetOrdered(false) //设为无序,触发并行写,提升写效率 _, err := coll(trans).Bulk(writes, opt) return err } // GetDailyPlayCountGroup // @return vid List, vid->dailyPlayCount Map, error func GetDailyPlayCountGroup(skip, limit int64) ([]ObjectID, map[ObjectID]int64, error) { playCountMin := int64(0) playCountGTMatch := PlayCountGTMatch{&playCountMin} sort := D{{Key: "sumDate", Value: -1}, {Key: "playCount", Value: -1}} list, err := List(sort, skip, limit, playCountGTMatch.New()) if err != nil { return nil, nil, err } listLen := len(list) vidList := make([]ObjectID, listLen) countMap := make(map[ObjectID]int64, listLen) for i, v := range list { vidList[i] = v.Vid countMap[v.Vid] = v.PlayCount } return vidList, countMap, nil } func SumByRecordAt(match RecordAtGTEAndLTMatch) ([]SumGroup, error) { return Sum(match.New()) } func DeleteBeforeCreatedAt(t *db.MongoTool, tm time.Time) error { _, err := coll(t).DeleteMany(bson.M{"createdAt": bson.M{"$lt": tm}}) return err }