web_hgtv_server/web/skd/videosync/videosync.go

package videosync

import (
	"context"
	"errors"
	nethttp "net/http"
	"sync"
	"time"
	"videoweb/pub/common/log"
	"videoweb/pub/common/srvdef"
	"videoweb/pub/models/syncfilemod"
	"videoweb/pub/pubdef"
	"videoweb/pub/storage/proxy"
	"videoweb/web/service/srv_videoinfo"

	"go.mongodb.org/mongo-driver/bson"
)

/*
1.小说采集后上传媒资库
2.业务后台记录最新更新章节ID(数据ID自增)
3.查询对应小说ID是否存在,如果不存在需要先同步媒资库小说信息，再更新小说
注意：业务后台和媒资库 存储的小说ID和章节ID需一致， 资源以媒资库为准， 不单独更新业务后台小说
*/

var (
	isRunning = false
	lock      sync.Mutex
)

func getIsRunning() bool {
	lock.Lock()
	defer lock.Unlock()
	return isRunning
}
func setRunning(b bool) {
	lock.Lock()
	defer lock.Unlock()
	isRunning = b
}
func VideoSync() {
	isRunning = getIsRunning()
	if isRunning {
		return
	}
	setRunning(true)
	ctx := context.Background()
	log.Info("VideoSync begin")
	pageSize := int64(200)
	recordFilter := bson.M{
		syncfilemod.FN.SyncRecord.Job: syncfilemod.VideoJob,
	}
	recordInfo, serr := syncfilemod.FindOneSyncRecord(ctx, recordFilter)
	if serr != nil {
		log.Error("VideoSync", log.Any("recordFilter", recordFilter), log.E(serr))
		setRunning(false)
		return
	}
	defer func() {
		setRunning(false)
		// 更新同步记录
		set := bson.M{
			"$set": bson.M{
				syncfilemod.FN.SyncRecord.RecordTime: recordInfo.RecordTime,
				syncfilemod.FN.SyncRecord.SuccessAt:  time.Now(),
			},
		}
		_ = syncfilemod.UpdateOneSyncRecord(ctx, recordFilter, set)
	}()
	// 循环同步
	for {
		// 查询接口
		req := proxy.MediaTaskSyncReq{
			AppId:          pubdef.AppId,
			LastUpdateTime: recordInfo.RecordTime,
			PageSize:       pageSize,
		}
		result, err := proxy.SyncMedia(&req, "")
		if err != nil || result.Code != nethttp.StatusOK {
			log.Error("VideoSync SyncMedia", log.Any("req", req), log.E(err))
			return
		}

		// 处理
		for _, media := range result.Data {
			// 先推进水位，确保单条失败后下次不会卡在同一位置
			if media.UpdatedAt.After(recordInfo.RecordTime) {
				recordInfo.RecordTime = media.UpdatedAt
			}
			se := srv_videoinfo.SyncOneVideo(ctx, media)
			if se != nil {
				log.Error("VideoSync SyncOneVideo", log.Any("media", media), log.E(se))
				continue
			}
		}

		if len(result.Data) < int(pageSize) {
			log.Info("VideoSync success", log.Any("RecordTime", recordInfo.RecordTime))
			break
		}
	}
	log.Info("VideoSync success", log.Any("RecordTime", recordInfo.RecordTime))
}

// SyncUnencryptedFields 全量补齐历史资源的 UnencryptedImage / UnencryptedPreMp4 字段。
// 从头分页拉取媒资库，仅在源端有值时写回，本地不存在或字段已有值不覆盖。
// 独立于增量同步的 SyncRecord，用于一次性修复；可安全重复执行。
func SyncUnencryptedFields() {
	ctx := context.Background()
	log.Info("SyncUnencryptedFields begin")
	pageSize := int64(200)
	lastUpdateTime := time.Time{}
	for {
		req := proxy.MediaTaskSyncReq{
			AppId:          pubdef.AppId,
			LastUpdateTime: lastUpdateTime,
			PageSize:       pageSize,
		}
		result, err := proxy.SyncMedia(&req, "")
		if err != nil || result.Code != nethttp.StatusOK {
			log.Error("SyncUnencryptedFields SyncMedia", log.Any("req", req), log.E(err))
			return
		}
		for _, media := range result.Data {
			// 先推进时间戳，保证 continue 也不阻塞分页
			if media.UpdatedAt.After(lastUpdateTime) {
				lastUpdateTime = media.UpdatedAt
			}
			se := srv_videoinfo.SyncUnencryptedOne(ctx, media)
			if se != nil && !errors.Is(se.Code, srvdef.ErrNoData) {
				log.Warn("SyncUnencryptedFields SyncUnencryptedOne", log.Any("mediaId", media.Id.Hex()), log.E(se))
			}
		}
		if len(result.Data) < int(pageSize) {
			break
		}
	}
	log.Info("SyncUnencryptedFields success", log.Any("lastUpdateTime", lastUpdateTime))
}