web_huakui_server/web/skd/job/transcode_video.go

package job

import (
	"context"
	"hkdj-web-server/pub/common/laosiji"
	"hkdj-web-server/pub/common/log"
	"hkdj-web-server/pub/models/videoinformationmod"
	"hkdj-web-server/web/service/srv_videoinfo"
	"strings"
	"sync/atomic"

	"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))
		}
	}
}