web_huakui_server/web/skd/playletwatermark/submit.go

package playletwatermark

import (
	"context"
	"errors"
	"hkdj-web-server/pub/common/log"
	"hkdj-web-server/pub/models/cfg"
	"hkdj-web-server/pub/models/playletchaptermod"
	"hkdj-web-server/pub/models/playletmod"
	"hkdj-web-server/pub/pubdef"
	"hkdj-web-server/pub/storage/proxy"
	"net/url"
	"path"
	"strings"
	"sync"
	"time"

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

const (
	submitBatchSize    = 10
	taskAddRetryBatch  = 20
	mediaFileVersionV1 = "v1"
	mediaFileVersionV2 = "v2"
)

var (
	submitRunning bool
	submitLock    sync.Mutex
)

// SubmitPendingChapters 定时提交未加水印剧集到 media（每10分钟最多10条）
func SubmitPendingChapters() {
	submitLock.Lock()
	if submitRunning {
		submitLock.Unlock()
		return
	}
	submitRunning = true
	submitLock.Unlock()
	defer func() {
		submitLock.Lock()
		submitRunning = false
		submitLock.Unlock()
	}()

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

	sysCfg, err := cfg.FindOneSysCfg(ctx, bson.M{cfg.FN.SysCfg.Id: 1})
	if err != nil || sysCfg == nil {
		log.Error("playletwatermark:SubmitPendingChapters 读取系统配置失败", log.E(err))
		return
	}
	if !hasVideoWatermark(sysCfg) {
		log.Warn("playletwatermark:SubmitPendingChapters 水印素材未配置，跳过")
		return
	}

	registerPendingSources(ctx)
	submitWatermarkTasks(ctx, sysCfg)
	log.Info("playletwatermark:SubmitPendingChapters end")
}

// registerPendingSources 将未处理剧集的相对 m3u8 登记到 media
func registerPendingSources(ctx context.Context) {
	list, err := playletchaptermod.FindManyPlayletChapter(ctx, bson.M{
		playletchaptermod.FN.PlayletChapter.WatermarkStatus: bson.M{
			"$in": []playletchaptermod.WatermarkStatus{
				playletchaptermod.WatermarkStatusPending,
				playletchaptermod.WatermarkStatusRegistered,
			},
		},
		playletchaptermod.FN.PlayletChapter.VideoUrl: bson.M{"$gt": ""},
		"$or": []bson.M{
			{playletchaptermod.FN.PlayletChapter.MediaSourceId: bson.M{"$exists": false}},
			{playletchaptermod.FN.PlayletChapter.MediaSourceId: ""},
		},
	}, options.Find().
		SetSort(bson.D{{Key: playletchaptermod.FN.PlayletChapter.Id, Value: 1}}).
		SetLimit(submitBatchSize))
	if err != nil {
		log.Error("playletwatermark:registerPendingSources 查询失败", log.E(err))
		return
	}
	for _, item := range list {
		if item == nil {
			continue
		}
		sourceID, syncErr := RegisterChapterSource(item)
		if syncErr != nil {
			log.Error("playletwatermark:registerPendingSources SyncMediaFile 失败",
				log.Any("chapterId", item.Id), log.Any("videoUrl", item.VideoUrl), log.E(syncErr))
			continue
		}
		now := time.Now()
		if updateErr := playletchaptermod.UpdateOnePlayletChapter(ctx,
			bson.M{playletchaptermod.FN.PlayletChapter.Id: item.Id},
			bson.M{"$set": bson.M{
				playletchaptermod.FN.PlayletChapter.MediaSourceId:   sourceID,
				playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusRegistered,
				playletchaptermod.FN.PlayletChapter.UpdatedAt:       now,
			}}); updateErr != nil {
			log.Error("playletwatermark:registerPendingSources 更新章节失败",
				log.Any("chapterId", item.Id), log.E(updateErr))
		}
	}
}

// RegisterChapterSource 将剧集原片登记到 media 切片服，返回 mediaSourceId。
func RegisterChapterSource(item *playletchaptermod.PlayletChapter) (string, error) {
	if item == nil {
		return "", errors.New("chapter is nil")
	}
	fieldPath, version := normalizeMediaFilePath(item.VideoUrl)
	if fieldPath == "" {
		return "", errors.New("invalid video path")
	}
	sourceID := primitive.NewObjectID()
	req := &proxy.MediaFileSyncReq{
		AppId: pubdef.AppId,
		VideoList: []proxy.MediaFileSyncItem{
			{
				ID:               sourceID.Hex(),
				Title:            mediaWatermarkTitle(playletForChapter(item), item),
				TaskID:           sourceID.Hex(),
				FieldNameFs:      fieldPath,
				Width:            item.Width,
				Height:           item.Height,
				VideoTime:        item.MediaTime,
				Size:             item.MediaSize,
				MediaFileVersion: version,
				IsPlaylet:        true,
			},
		},
	}
	result, syncErr := proxy.SyncMediaFile(req, "")
	if syncErr != nil {
		return "", syncErr
	}
	if result == nil || result.Code != proxy.Success {
		return "", errors.New("SyncMediaFile failed")
	}
	return sourceID.Hex(), nil
}

// submitWatermarkTasks 对已登记原片尝试 task/add，成功后进入处理中
func submitWatermarkTasks(ctx context.Context, sysCfg *cfg.SysCfg) {
	list, err := playletchaptermod.FindManyPlayletChapter(ctx, bson.M{
		playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusRegistered,
		playletchaptermod.FN.PlayletChapter.MediaSourceId:   bson.M{"$gt": ""},
	}, options.Find().
		SetSort(bson.D{{Key: playletchaptermod.FN.PlayletChapter.Id, Value: 1}}).
		SetLimit(taskAddRetryBatch))
	if err != nil {
		log.Error("playletwatermark:submitWatermarkTasks 查询失败", log.E(err))
		return
	}
	for _, item := range list {
		if item == nil || item.MediaSourceId == "" {
			continue
		}
		if strings.TrimSpace(item.MediaTaskId) != "" {
			if !restartWatermarkTask(ctx, item, sysCfg) {
				continue
			}
		} else if !addWatermarkTask(ctx, item, sysCfg) {
			continue
		}
		_ = playletchaptermod.UpdateOnePlayletChapter(ctx,
			bson.M{
				playletchaptermod.FN.PlayletChapter.Id:              item.Id,
				playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusRegistered,
			},
			bson.M{"$set": bson.M{
				playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusProcessing,
				playletchaptermod.FN.PlayletChapter.UpdatedAt:       time.Now(),
			}})
	}
}

func addWatermarkTask(ctx context.Context, item *playletchaptermod.PlayletChapter, sysCfg *cfg.SysCfg) bool {
	req := &proxy.MediaTaskAddReq{
		MediaSourceId:            item.MediaSourceId,
		AppId:                    pubdef.AppId,
		Title:                    mediaWatermarkTitle(playletForChapter(item), item),
		WatermarkImage:           sysCfg.WatermarkImage,
		WatermarkMedia:           sysCfg.WatermarkMedia,
		WatermarkMediaTop:        sysCfg.WatermarkMediaTop,
		WatermarkMediaTopRight:   sysCfg.WatermarkMediaTopRight,
		WatermarkMediaBottomLeft: sysCfg.WatermarkMediaBottomLeft,
	}
	result, addErr := proxy.AddMediaTask(req, "")
	if addErr != nil || result == nil {
		log.Info("playletwatermark:submitWatermarkTasks task/add 请求失败",
			log.Any("chapterId", item.Id),
			log.Any("mediaSourceId", item.MediaSourceId),
			log.E(addErr))
		return false
	}
	// 200=新建成功；6007=任务已存在。原片未处理完时保持已登记，下轮再试。
	if result.Code != proxy.Success && result.Code != 6007 {
		log.Info("playletwatermark:submitWatermarkTasks task/add 暂未成功",
			log.Any("chapterId", item.Id),
			log.Any("mediaSourceId", item.MediaSourceId),
			log.Any("result", result))
		markFailedIfMediaSourceFailed(ctx, item)
		return false
	}
	return true
}

func restartWatermarkTask(ctx context.Context, item *playletchaptermod.PlayletChapter, sysCfg *cfg.SysCfg) bool {
	watermarkImage := sysCfg.WatermarkImage
	watermarkMedia := sysCfg.WatermarkMedia
	watermarkMediaTop := sysCfg.WatermarkMediaTop
	watermarkMediaTopRight := sysCfg.WatermarkMediaTopRight
	watermarkMediaBottomLeft := sysCfg.WatermarkMediaBottomLeft
	req := &proxy.MediaTaskUpdateWatermarkReq{
		Id:                       item.MediaTaskId,
		AppId:                    pubdef.AppId,
		WatermarkImage:           &watermarkImage,
		WatermarkMedia:           &watermarkMedia,
		WatermarkMediaTop:        &watermarkMediaTop,
		WatermarkMediaTopRight:   &watermarkMediaTopRight,
		WatermarkMediaBottomLeft: &watermarkMediaBottomLeft,
	}
	result, updateErr := proxy.UpdateMediaWatermark(req, "")
	if updateErr != nil || result == nil {
		log.Info("playletwatermark:submitWatermarkTasks update/watermark 请求失败",
			log.Any("chapterId", item.Id),
			log.Any("mediaTaskId", item.MediaTaskId),
			log.E(updateErr))
		return false
	}
	if result.Code != proxy.Success {
		log.Info("playletwatermark:submitWatermarkTasks update/watermark 暂未成功",
			log.Any("chapterId", item.Id),
			log.Any("mediaTaskId", item.MediaTaskId),
			log.Any("result", result))
		markFailedIfMediaSourceFailed(ctx, item)
		return false
	}
	return true
}

func markFailedIfMediaSourceFailed(ctx context.Context, item *playletchaptermod.PlayletChapter) {
	if item == nil || item.MediaSourceId == "" {
		return
	}
	source, err := proxy.QueryMediaSource(item.MediaSourceId, pubdef.AppId, "")
	if err != nil {
		log.Info("playletwatermark:markFailedIfMediaSourceFailed 查询原片失败",
			log.Any("chapterId", item.Id),
			log.Any("mediaSourceId", item.MediaSourceId),
			log.E(err))
		return
	}
	if source == nil || source.Status != proxy.MediaSourceCompletedFail {
		return
	}
	now := time.Now()
	if updateErr := playletchaptermod.UpdateOnePlayletChapter(ctx,
		bson.M{
			playletchaptermod.FN.PlayletChapter.Id: item.Id,
			playletchaptermod.FN.PlayletChapter.WatermarkStatus: bson.M{
				"$in": []playletchaptermod.WatermarkStatus{
					playletchaptermod.WatermarkStatusRegistered,
					playletchaptermod.WatermarkStatusProcessing,
				},
			},
		},
		bson.M{"$set": bson.M{
			playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusFailed,
			playletchaptermod.FN.PlayletChapter.UpdatedAt:       now,
		}}); updateErr != nil {
		log.Error("playletwatermark:markFailedIfMediaSourceFailed 更新章节失败",
			log.Any("chapterId", item.Id),
			log.E(updateErr))
		return
	}
	log.Info("playletwatermark:markFailedIfMediaSourceFailed 原片已失败，章节标为处理失败",
		log.Any("chapterId", item.Id),
		log.Any("mediaSourceId", item.MediaSourceId),
		log.Any("failReason", source.FailReason))
}

// normalizeMediaFilePath 将相对/绝对 m3u8 地址规范化为 media 可识别的相对路径（不含 .m3u8）
func normalizeMediaFilePath(raw string) (fieldPath string, version string) {
	raw = strings.TrimSpace(raw)
	if raw == "" {
		return "", ""
	}
	version = mediaFileVersionV2
	if strings.HasPrefix(raw, "http://") || strings.HasPrefix(raw, "https://") {
		u, err := url.Parse(raw)
		if err != nil {
			return "", ""
		}
		raw = u.Path
	}
	raw = strings.TrimPrefix(raw, "/")
	if strings.HasPrefix(raw, "v1/") {
		version = mediaFileVersionV1
	}
	if strings.HasSuffix(strings.ToLower(raw), ".m3u8") {
		raw = strings.TrimSuffix(raw, ".m3u8")
		raw = strings.TrimSuffix(raw, ".M3U8")
	}
	raw = path.Clean(raw)
	if raw == "." || raw == "" {
		return "", ""
	}
	return raw, version
}

// mediaWatermarkTitle 传给 media 的标题：短剧剧名（没有则用标题）+ "-" + 剧集标题
func mediaWatermarkTitle(playlet *playletmod.Playlet, chapter *playletchaptermod.PlayletChapter) string {
	playletName := ""
	if playlet != nil {
		playletName = strings.TrimSpace(playlet.PlayName)
		if playletName == "" {
			playletName = strings.TrimSpace(playlet.Title)
		}
	}
	episodeTitle := ""
	if chapter != nil {
		episodeTitle = strings.TrimSpace(chapter.Title)
	}
	if playletName != "" && episodeTitle != "" {
		return playletName + "-" + episodeTitle
	}
	if playletName != "" {
		return playletName
	}
	return episodeTitle
}

func playletForChapter(item *playletchaptermod.PlayletChapter) *playletmod.Playlet {
	if item == nil || item.PlayletId <= 0 {
		return nil
	}
	playlet, err := playletmod.FindOnePlaylet(context.Background(), bson.M{
		playletmod.FN.Playlet.Id: item.PlayletId,
	}, options.FindOne().SetProjection(bson.M{
		playletmod.FN.Playlet.PlayName: 1,
		playletmod.FN.Playlet.Title:    1,
	}))
	if err != nil {
		return nil
	}
	return playlet
}

func hasVideoWatermark(sysCfg *cfg.SysCfg) bool {
	if sysCfg == nil {
		return false
	}
	return strings.TrimSpace(sysCfg.WatermarkMedia) != "" ||
		strings.TrimSpace(sysCfg.WatermarkMediaTop) != "" ||
		strings.TrimSpace(sysCfg.WatermarkMediaTopRight) != "" ||
		strings.TrimSpace(sysCfg.WatermarkMediaBottomLeft) != ""
}