package playletbackflowstatmod import ( "context" "fmt" "time" "videoweb/pub/common/log" "videoweb/pub/common/srvdef" "videoweb/pub/mongos" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" ) var coll *mongos.Collection // Init 初始化原创短剧 AI 内容回流日统计表和查询索引。 func Init(ctx context.Context) { coll = mongos.Shop.Coll(TN.PlayletBackflowDailyStat) initIndex(ctx) } func initIndex(ctx context.Context) { indexes := []mongo.IndexModel{ { Keys: bson.D{ {Key: FN.PlayletBackflowDailyStat.DataDate, Value: 1}, {Key: FN.PlayletBackflowDailyStat.ReportStatus, Value: 1}, {Key: FN.PlayletBackflowDailyStat.Id, Value: 1}, }, }, { Keys: bson.D{ {Key: FN.PlayletBackflowDailyStat.DataDate, Value: 1}, {Key: FN.PlayletBackflowDailyStat.PlayletId, Value: 1}, }, Options: options.Index().SetUnique(true).SetName("uniq_dataDate_playletId"), }, } if _, err := coll.CreateIndex(ctx, indexes); err != nil { panic(fmt.Sprintf("(playletbackflowstatmod:initIndex) Create Index failed! err:%+v:", err)) } } // UpsertManyPlayletBackflowDailyStats 批量保存每天原创短剧回流数据快照。 func UpsertManyPlayletBackflowDailyStats(ctx context.Context, docs PlayletBackflowDailyStatList) *srvdef.SrvErr { if len(docs) == 0 { return nil } now := time.Now() writes := make([]mongo.WriteModel, 0, len(docs)) for _, doc := range docs { if doc == nil || doc.DataDate == "" || doc.PlayletId <= 0 { continue } doc.UpdatedAt = now doc.LastStatAt = now update := bson.M{ "$set": bson.M{ FN.PlayletBackflowDailyStat.DataDate: doc.DataDate, FN.PlayletBackflowDailyStat.PlayletId: doc.PlayletId, FN.PlayletBackflowDailyStat.ChapterId: doc.ChapterId, FN.PlayletBackflowDailyStat.CurrentEpisode: doc.CurrentEpisode, FN.PlayletBackflowDailyStat.MediaCenterId: doc.MediaCenterId, FN.PlayletBackflowDailyStat.MediaContentId: doc.MediaContentId, FN.PlayletBackflowDailyStat.Title: doc.Title, FN.PlayletBackflowDailyStat.PublishedData: doc.PublishedData, FN.PlayletBackflowDailyStat.ScriptName: doc.ScriptName, FN.PlayletBackflowDailyStat.DramaName: doc.DramaName, FN.PlayletBackflowDailyStat.DramaNo: doc.DramaNo, FN.PlayletBackflowDailyStat.VideoURL: doc.VideoURL, FN.PlayletBackflowDailyStat.PlatformURL: doc.PlatformURL, FN.PlayletBackflowDailyStat.Views: doc.Views, FN.PlayletBackflowDailyStat.Clicks: doc.Clicks, FN.PlayletBackflowDailyStat.Likes: doc.Likes, FN.PlayletBackflowDailyStat.LikeRate: doc.LikeRate, FN.PlayletBackflowDailyStat.CompletionRate: doc.CompletionRate, FN.PlayletBackflowDailyStat.PaidRate: doc.PaidRate, FN.PlayletBackflowDailyStat.Comments: doc.Comments, FN.PlayletBackflowDailyStat.Shares: doc.Shares, FN.PlayletBackflowDailyStat.CompletionRateScope: doc.CompletionRateScope, FN.PlayletBackflowDailyStat.ReportStatus: doc.ReportStatus, FN.PlayletBackflowDailyStat.ReportError: "", FN.PlayletBackflowDailyStat.LastStatAt: doc.LastStatAt, FN.PlayletBackflowDailyStat.UpdatedAt: doc.UpdatedAt, }, "$setOnInsert": bson.M{ FN.PlayletBackflowDailyStat.Id: primitive.NewObjectID(), FN.PlayletBackflowDailyStat.ReportRetryCount: int64(0), FN.PlayletBackflowDailyStat.CreatedAt: now, }, } model := mongo.NewUpdateOneModel(). SetFilter(bson.M{ FN.PlayletBackflowDailyStat.DataDate: doc.DataDate, FN.PlayletBackflowDailyStat.PlayletId: doc.PlayletId, }). SetUpdate(update). SetUpsert(true) writes = append(writes, model) } if len(writes) == 0 { return nil } if _, err := coll.BulkWrite(ctx, writes, options.BulkWrite().SetOrdered(false)); err != nil { log.Warn("playletbackflowstatmod:UpsertManyPlayletBackflowDailyStats failed", log.E(err), log.Any("count", len(writes))) return srvdef.New(srvdef.ErrDbUpdateError, "", "") } return nil } // FindManyPlayletBackflowDailyStats 查询原创短剧回流日统计。 func FindManyPlayletBackflowDailyStats(ctx context.Context, filter bson.M, opts ...*options.FindOptions) (PlayletBackflowDailyStatList, *srvdef.SrvErr) { data := make(PlayletBackflowDailyStatList, 0) if err := coll.Find(ctx, &data, filter, opts...); err != nil { log.Warn("playletbackflowstatmod:FindManyPlayletBackflowDailyStats failed", log.E(err), log.Any("filter", filter)) return nil, srvdef.New(srvdef.ErrDbQueryError, "", "") } return data, nil } // CountPlayletBackflowDailyStats 统计原创短剧回流日统计条数。 func CountPlayletBackflowDailyStats(ctx context.Context, filter bson.M, opts ...*options.CountOptions) (int64, *srvdef.SrvErr) { count, err := coll.Count(ctx, filter, opts...) if err != nil { log.Warn("playletbackflowstatmod:CountPlayletBackflowDailyStats failed", log.E(err), log.Any("filter", filter)) return 0, srvdef.New(srvdef.ErrDbQueryError, "", "") } return count, nil } // MarkPlayletBackflowDailyStatsReported 把一批统计记录标记为上报成功。 func MarkPlayletBackflowDailyStatsReported(ctx context.Context, ids []primitive.ObjectID, reportAt time.Time) *srvdef.SrvErr { if len(ids) == 0 { return nil } _, err := coll.UpdateMany(ctx, bson.M{FN.PlayletBackflowDailyStat.Id: bson.M{"$in": ids}}, bson.M{"$set": bson.M{ FN.PlayletBackflowDailyStat.ReportStatus: ReportStatusSuccess, FN.PlayletBackflowDailyStat.ReportAt: reportAt, FN.PlayletBackflowDailyStat.ReportError: "", FN.PlayletBackflowDailyStat.UpdatedAt: time.Now(), }}) if err != nil { log.Warn("playletbackflowstatmod:MarkPlayletBackflowDailyStatsReported failed", log.E(err), log.Any("count", len(ids))) return srvdef.New(srvdef.ErrDbUpdateError, "", "") } return nil } // MarkPlayletBackflowDailyStatsFailed 把一批统计记录标记为上报失败。 func MarkPlayletBackflowDailyStatsFailed(ctx context.Context, ids []primitive.ObjectID, reportErr string) *srvdef.SrvErr { if len(ids) == 0 { return nil } _, err := coll.UpdateMany(ctx, bson.M{FN.PlayletBackflowDailyStat.Id: bson.M{"$in": ids}}, bson.M{ "$set": bson.M{ FN.PlayletBackflowDailyStat.ReportStatus: ReportStatusFailed, FN.PlayletBackflowDailyStat.ReportError: reportErr, FN.PlayletBackflowDailyStat.UpdatedAt: time.Now(), }, "$inc": bson.M{ FN.PlayletBackflowDailyStat.ReportRetryCount: int64(1), }, }) if err != nil { log.Warn("playletbackflowstatmod:MarkPlayletBackflowDailyStatsFailed failed", log.E(err), log.Any("count", len(ids))) return srvdef.New(srvdef.ErrDbUpdateError, "", "") } return nil }