web_huangguo_server/web/skd/playletwatermark/sync.go

package playletwatermark

import (
	"context"
	"net/http"
	"strings"
	"sync"
	"time"
	"videoweb/pub/common/log"
	"videoweb/pub/common/srvdef"
	"videoweb/pub/models/playletchaptermod"
	"videoweb/pub/models/playletmod"
	"videoweb/pub/models/syncfilemod"
	"videoweb/pub/pubdef"
	"videoweb/pub/storage/proxy"

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

const (
	syncPageSize       int64 = 1000
	syncOverlapWindow        = 10 * time.Minute
	reconcileBatchSize int64 = 100
)

var (
	syncRunning bool
	syncLock    sync.Mutex
)

// SyncCompletedChapters 定时从 media 拉取已完成水印任务，回写剧集地址与状态
func SyncCompletedChapters() {
	syncLock.Lock()
	if syncRunning {
		syncLock.Unlock()
		return
	}
	syncRunning = true
	syncLock.Unlock()
	defer func() {
		syncLock.Lock()
		syncRunning = false
		syncLock.Unlock()
	}()

	ctx := context.Background()
	log.Info("playletwatermark:SyncCompletedChapters begin")

	record, err := ensureSyncRecord(ctx)
	if err != nil {
		log.Error("playletwatermark:SyncCompletedChapters 获取同步水印失败", log.E(err))
		return
	}

	recordTime, syncOK := syncIncrementalCompletedChapters(ctx, record.RecordTime)
	if syncOK {
		if updateErr := syncfilemod.UpdateOneSyncRecord(ctx, bson.M{
			syncfilemod.FN.SyncRecord.Job: syncfilemod.PlayletWatermarkJob,
		}, bson.M{"$set": bson.M{
			syncfilemod.FN.SyncRecord.RecordTime: recordTime,
			syncfilemod.FN.SyncRecord.SuccessAt:  time.Now(),
			syncfilemod.FN.SyncRecord.UpdatedAt:  time.Now(),
		}}); updateErr != nil {
			log.Error("playletwatermark:SyncCompletedChapters 更新同步水印失败", log.E(updateErr))
		}
	}

	reconcileProcessingChapters(ctx)
	log.Info("playletwatermark:SyncCompletedChapters end",
		log.Any("recordTime", recordTime),
		log.Any("syncOK", syncOK))
}

// syncIncrementalCompletedChapters 按更新时间水印增量拉取。
// 每次回退一段时间重复扫描，避免临界时间及同一时间戳分页造成漏数据；章节回写是幂等的。
func syncIncrementalCompletedChapters(ctx context.Context, recordTime time.Time) (time.Time, bool) {
	cursor := recordTime
	if !cursor.IsZero() {
		cursor = cursor.Add(-syncOverlapWindow)
	}
	maxRecordTime := recordTime
	for {
		req := proxy.MediaTaskSyncReq{
			AppId:          pubdef.AppId,
			LastUpdateTime: cursor,
			PageSize:       syncPageSize,
		}
		result, syncErr := proxy.SyncMedia(&req, "")
		if syncErr != nil || result == nil || result.Code != http.StatusOK {
			log.Error("playletwatermark:SyncCompletedChapters SyncMedia 失败", log.Any("req", req), log.E(syncErr))
			return recordTime, false
		}
		pageMaxTime := cursor
		for _, media := range result.Data {
			if media.UpdatedAt.After(pageMaxTime) {
				pageMaxTime = media.UpdatedAt
			}
			if media.UpdatedAt.After(maxRecordTime) {
				maxRecordTime = media.UpdatedAt
			}
			applyCompletedMedia(ctx, media, "incremental")
		}
		if int64(len(result.Data)) < syncPageSize {
			break
		}
		if !pageMaxTime.After(cursor) {
			log.Error("playletwatermark:SyncCompletedChapters 增量分页水印未推进",
				log.Any("cursor", cursor),
				log.Any("count", len(result.Data)))
			return recordTime, false
		}
		cursor = pageMaxTime
	}
	return maxRecordTime, true
}

// reconcileProcessingChapters 反向对账所有处理中章节。
// 它不依赖增量水印，可补偿本地更新失败、时间戳临界和历史水印越过造成的漏回写。
func reconcileProcessingChapters(ctx context.Context) {
	var lastID int64
	for {
		list, err := playletchaptermod.FindManyPlayletChapter(ctx, bson.M{
			playletchaptermod.FN.PlayletChapter.Id:              bson.M{"$gt": lastID},
			playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusProcessing,
			playletchaptermod.FN.PlayletChapter.MediaSourceId:   bson.M{"$gt": ""},
		}, options.Find().
			SetSort(bson.D{{Key: playletchaptermod.FN.PlayletChapter.Id, Value: 1}}).
			SetLimit(reconcileBatchSize))
		if err != nil {
			log.Error("playletwatermark:reconcileProcessingChapters 查询章节失败", log.E(err))
			return
		}
		if len(list) == 0 {
			return
		}

		sourceIDs := make([]string, 0, len(list))
		for _, item := range list {
			if item == nil {
				continue
			}
			sourceIDs = append(sourceIDs, item.MediaSourceId)
			if item.Id > lastID {
				lastID = item.Id
			}
		}
		if len(sourceIDs) > 0 {
			result, queryErr := proxy.QueryMedias(&proxy.MediaQueryesReq{
				AppId:          pubdef.AppId,
				MediaSourceIds: sourceIDs,
			}, "")
			if queryErr != nil || result == nil || result.Code != http.StatusOK {
				log.Error("playletwatermark:reconcileProcessingChapters 查询 media 任务失败",
					log.Any("sourceCount", len(sourceIDs)),
					log.Any("result", result),
					log.E(queryErr))
				return
			}
			for _, media := range result.Data {
				applyCompletedMedia(ctx, media, "reconcile")
			}
		}
		if int64(len(list)) < reconcileBatchSize {
			return
		}
	}
}

func applyCompletedMedia(ctx context.Context, media proxy.MediaResp, source string) bool {
	sourceID := media.MediaSourceId.Hex()
	if sourceID == "" || sourceID == "000000000000000000000000" {
		log.Warn("playletwatermark:applyCompletedMedia mediaSourceId 无效",
			log.Any("mediaTaskId", media.Id.Hex()),
			log.Any("source", source))
		return false
	}
	fieldPath := strings.TrimSpace(media.FieldNameFs)
	if fieldPath == "" {
		log.Warn("playletwatermark:applyCompletedMedia 视频路径为空",
			log.Any("mediaSourceId", sourceID),
			log.Any("mediaTaskId", media.Id.Hex()),
			log.Any("source", source))
		return false
	}
	if !strings.HasSuffix(strings.ToLower(fieldPath), ".m3u8") {
		fieldPath += ".m3u8"
	}
	taskID := media.Id.Hex()
	setDoc := bson.M{
		playletchaptermod.FN.PlayletChapter.WatermarkVideoUrl: fieldPath,
		playletchaptermod.FN.PlayletChapter.WatermarkStatus:   playletchaptermod.WatermarkStatusDone,
		playletchaptermod.FN.PlayletChapter.MediaTaskId:       taskID,
		playletchaptermod.FN.PlayletChapter.UpdatedAt:         time.Now(),
	}
	if sourceName := strings.TrimSpace(media.SourceNameFs); sourceName != "" &&
		isOriginalPlayletBySourceID(ctx, sourceID) {
		// 原创才回填原片地址，供换水印/重置复用；非原创保留媒资原始 VideoUrl。
		setDoc[playletchaptermod.FN.PlayletChapter.VideoUrl] = sourceName
	}
	updateErr := playletchaptermod.UpdateOnePlayletChapter(ctx, bson.M{
		playletchaptermod.FN.PlayletChapter.MediaSourceId:   sourceID,
		playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusProcessing,
	}, bson.M{"$set": setDoc})
	if updateErr == nil {
		log.Info("playletwatermark:applyCompletedMedia 回写成功",
			log.Any("mediaSourceId", sourceID),
			log.Any("mediaTaskId", taskID),
			log.Any("watermarkVideoUrl", fieldPath),
			log.Any("source", source))
		return true
	}
	if updateErr.Code != srvdef.ErrNoData {
		log.Error("playletwatermark:applyCompletedMedia 更新章节失败",
			log.Any("mediaSourceId", sourceID),
			log.Any("source", source),
			log.E(updateErr))
		return false
	}

	chapter, findErr := playletchaptermod.FindOnePlayletChapter(ctx, bson.M{
		playletchaptermod.FN.PlayletChapter.MediaSourceId: sourceID,
	})
	if findErr != nil {
		if findErr.Code != srvdef.ErrNoData {
			log.Error("playletwatermark:applyCompletedMedia 查询未匹配章节失败",
				log.Any("mediaSourceId", sourceID),
				log.E(findErr))
		} else {
			log.Warn("playletwatermark:applyCompletedMedia 未找到对应章节",
				log.Any("mediaSourceId", sourceID),
				log.Any("mediaTaskId", taskID),
				log.Any("source", source))
		}
		return false
	}
	if chapter.WatermarkStatus != playletchaptermod.WatermarkStatusDone {
		log.Warn("playletwatermark:applyCompletedMedia 章节状态不允许回写",
			log.Any("chapterId", chapter.Id),
			log.Any("mediaSourceId", sourceID),
			log.Any("watermarkStatus", chapter.WatermarkStatus),
			log.Any("source", source))
	}
	return false
}

func isOriginalPlayletBySourceID(ctx context.Context, sourceID string) bool {
	chapter, err := playletchaptermod.FindOnePlayletChapter(ctx, bson.M{
		playletchaptermod.FN.PlayletChapter.MediaSourceId: sourceID,
	}, options.FindOne().SetProjection(bson.M{
		playletchaptermod.FN.PlayletChapter.PlayletId: 1,
	}))
	if err != nil || chapter == nil || chapter.PlayletId <= 0 {
		return false
	}
	playlet, err := playletmod.FindOnePlaylet(ctx, bson.M{
		playletmod.FN.Playlet.Id: chapter.PlayletId,
	}, options.FindOne().SetProjection(bson.M{
		playletmod.FN.Playlet.IsOriginal: 1,
	}))
	return err == nil && playlet != nil && playlet.IsOriginal
}

func ensureSyncRecord(ctx context.Context) (*syncfilemod.SyncRecord, *srvdef.SrvErr) {
	filter := bson.M{syncfilemod.FN.SyncRecord.Job: syncfilemod.PlayletWatermarkJob}
	record, err := syncfilemod.FindOneSyncRecord(ctx, filter)
	if err == nil {
		return record, nil
	}
	if err.Code != srvdef.ObjectDayStatDoesNotExist {
		return nil, err
	}
	now := time.Now()
	record = &syncfilemod.SyncRecord{
		Job:        syncfilemod.PlayletWatermarkJob,
		RecordTime: time.Time{},
		CreatedAt:  now,
		UpdatedAt:  now,
	}
	if insertErr := syncfilemod.InsertOneSyncRecord(ctx, record); insertErr != nil {
		return nil, insertErr
	}
	return record, nil
}