package job import ( "context" "strings" "sync/atomic" "videoweb/pub/common/laosiji" "videoweb/pub/common/log" "videoweb/pub/models/videoinformationmod" "videoweb/web/service/srv_videoinfo" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo/options" ) const transcodeBatchSize = 5 const transcodeMaxFailCount = 3 var transcodeRunning int32 var transcodeCheckRunning int32 const transcodeCheckBatchSize = 100 // TranscodeVideoByWatchTimes 按观看次数挑选尚未转码成功的视频,批量提交 H265 转码。 func TranscodeVideoByWatchTimes() { if !atomic.CompareAndSwapInt32(&transcodeRunning, 0, 1) { log.Info("TranscodeVideoByWatchTimes skip: previous run still running") return } defer atomic.StoreInt32(&transcodeRunning, 0) ctx := context.Background() if laosiji.LaosijiClient == nil || laosiji.LaosijiDomains == nil { log.Warn("TranscodeVideoByWatchTimes skip: laosiji not ready") return } if _, err := laosiji.LaosijiDomains.Get(ctx); err != nil { log.Warn("TranscodeVideoByWatchTimes skip: domains not ready", log.E(err)) return } filter := bson.M{ videoinformationmod.FN.VideoInformation.HevcStatus: bson.M{ "$in": bson.A{videoinformationmod.HevcStatusNone, videoinformationmod.HevcStatusFailed}, }, videoinformationmod.FN.VideoInformation.HevcFailCount: bson.M{"$lt": transcodeMaxFailCount}, videoinformationmod.FN.VideoInformation.Video.Url: bson.M{"$ne": ""}, } opts := options.Find(). SetSort(bson.D{{Key: videoinformationmod.FN.VideoInformation.Real.WatchTimes, Value: -1}}). SetLimit(transcodeBatchSize) videoList, serr := videoinformationmod.FindManyVideoInformations(ctx, filter, opts) if serr != nil { log.Error("TranscodeVideoByWatchTimes find video failed", log.Any("serr", serr)) return } if len(videoList) == 0 { return } log.Info("TranscodeVideoByWatchTimes start", log.Any("count", len(videoList))) for _, v := range videoList { src := strings.TrimSpace(v.Video.Url) if src == "" { continue } if err := srv_videoinfo.SubmitTranscodeVideo(ctx, v.Id, src, v.Video.FileId); err != nil { log.Warn("TranscodeVideoByWatchTimes submit failed", log.Any("videoId", v.Id), log.Any("watchTimes", v.Real.WatchTimes), log.E(err)) } } log.Info("TranscodeVideoByWatchTimes done", log.Any("count", len(videoList))) } // CheckTranscodingResults 查询转码中任务并回写结果。 func CheckTranscodingResults() { if !atomic.CompareAndSwapInt32(&transcodeCheckRunning, 0, 1) { log.Info("CheckTranscodingResults skip: previous run still running") return } defer atomic.StoreInt32(&transcodeCheckRunning, 0) ctx := context.Background() if laosiji.LaosijiClient == nil || laosiji.LaosijiDomains == nil { return } filter := bson.M{videoinformationmod.FN.VideoInformation.HevcStatus: videoinformationmod.HevcStatusPending} opts := options.Find().SetLimit(transcodeCheckBatchSize) videoList, serr := videoinformationmod.FindManyVideoInformations(ctx, filter, opts) if serr != nil { log.Error("CheckTranscodingResults find video failed", log.Any("serr", serr)) return } if len(videoList) == 0 { return } for _, v := range videoList { src := strings.TrimSpace(v.Video.Url) if src == "" { continue } if err := srv_videoinfo.RefreshTranscodeVideo(ctx, v.Id, src, v.Video.FileId); err != nil { log.Warn("CheckTranscodingResults refresh failed", log.Any("videoId", v.Id), log.E(err)) } } }