package playletwatermark import ( "context" "errors" "net/url" "path" "strings" "sync" "time" "videoweb/pub/common/log" "videoweb/pub/models/cfg" "videoweb/pub/models/playletchaptermod" "videoweb/pub/models/playletmod" "videoweb/pub/pubdef" "videoweb/pub/storage/proxy" "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 taskAddScanLimit = 200 mediaTaskAlreadyExists = 6007 mediaTaskSourceNotReady = 6015 sourceNotReadyRetryWait = 5 * time.Minute mediaFileVersionV1 = "v1" mediaFileVersionV2 = "v2" ) // watermarkSubmitResult 是一次 task/add 或重新加水印的结果。 type watermarkSubmitResult int const ( watermarkSubmitFailed watermarkSubmitResult = iota watermarkSubmitNotReady watermarkSubmitOK ) 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 } cfg.CacheAppKey(sysCfg.AppKey) 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, AppKey: cfg.CachedAppKey(context.Background()), 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。 // 同一轮按重试时间、剧集 id 最多检查 taskAddScanLimit 条,成功提交满 taskAddRetryBatch 条即停。 // 原片未处理完的剧集延后,不再占住队头。 func submitWatermarkTasks(ctx context.Context, sysCfg *cfg.SysCfg) { now := time.Now() list, err := playletchaptermod.FindManyPlayletChapter(ctx, dueRegisteredWatermarkFilter(now), options.Find(). SetSort(bson.D{ {Key: playletchaptermod.FN.PlayletChapter.WatermarkRetryAt, Value: 1}, {Key: playletchaptermod.FN.PlayletChapter.Id, Value: 1}, }). SetLimit(taskAddScanLimit)) if err != nil { log.Error("playletwatermark:submitWatermarkTasks 查询失败", log.E(err)) return } submitted := 0 scanned := 0 for _, item := range list { if submitted >= taskAddRetryBatch { break } if item == nil || item.MediaSourceId == "" { continue } scanned++ var result watermarkSubmitResult if strings.TrimSpace(item.MediaTaskId) != "" { result = restartWatermarkTask(ctx, item, sysCfg) } else { result = addWatermarkTask(ctx, item, sysCfg) } switch result { case watermarkSubmitNotReady: postponeWatermarkRetry(ctx, item, now) case watermarkSubmitOK: if markWatermarkProcessing(ctx, item, now) { submitted++ } } } log.Info("playletwatermark:submitWatermarkTasks end", log.Any("scanned", scanned), log.Any("submitted", submitted), log.Any("candidates", len(list))) } // dueRegisteredWatermarkFilter 选出已登记、且到了重试时间的剧集。 // 没有重试时间的记录优先,原片未就绪被延后的记录排在后面。 func dueRegisteredWatermarkFilter(now time.Time) bson.M { return bson.M{ playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusRegistered, playletchaptermod.FN.PlayletChapter.MediaSourceId: bson.M{"$gt": ""}, "$or": []bson.M{ {playletchaptermod.FN.PlayletChapter.WatermarkRetryAt: bson.M{"$exists": false}}, {playletchaptermod.FN.PlayletChapter.WatermarkRetryAt: bson.M{"$lte": now}}, }, } } func classifyMediaTaskCode(code int) watermarkSubmitResult { switch code { case proxy.Success, mediaTaskAlreadyExists: return watermarkSubmitOK case mediaTaskSourceNotReady: return watermarkSubmitNotReady default: return watermarkSubmitFailed } } func markWatermarkProcessing(ctx context.Context, item *playletchaptermod.PlayletChapter, now time.Time) bool { if updateErr := 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: now, }, "$unset": bson.M{ playletchaptermod.FN.PlayletChapter.WatermarkRetryAt: "", }, }); updateErr != nil { log.Error("playletwatermark:submitWatermarkTasks 更新为处理中失败", log.Any("chapterId", item.Id), log.E(updateErr)) return false } return true } func postponeWatermarkRetry(ctx context.Context, item *playletchaptermod.PlayletChapter, now time.Time) { retryAt := now.Add(sourceNotReadyRetryWait) if updateErr := 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.WatermarkRetryAt: retryAt, }}); updateErr != nil { log.Error("playletwatermark:submitWatermarkTasks 延后重试失败", log.Any("chapterId", item.Id), log.E(updateErr)) return } log.Info("playletwatermark:submitWatermarkTasks 原片未处理完,延后重试", log.Any("chapterId", item.Id), log.Any("mediaSourceId", item.MediaSourceId), log.Any("retryAt", retryAt)) } func addWatermarkTask(ctx context.Context, item *playletchaptermod.PlayletChapter, sysCfg *cfg.SysCfg) watermarkSubmitResult { req := &proxy.MediaTaskAddReq{ MediaSourceId: item.MediaSourceId, AppId: pubdef.AppId, AppKey: strings.TrimSpace(sysCfg.AppKey), 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 watermarkSubmitFailed } // 200=新建成功;6007=任务已存在。6015=原片未处理完,延后并不占住本轮名额。 switch classifyMediaTaskCode(result.Code) { case watermarkSubmitOK: return watermarkSubmitOK case watermarkSubmitNotReady: return watermarkSubmitNotReady default: log.Info("playletwatermark:submitWatermarkTasks task/add 暂未成功", log.Any("chapterId", item.Id), log.Any("mediaSourceId", item.MediaSourceId), log.Any("result", result)) markFailedIfMediaSourceFailed(ctx, item) return watermarkSubmitFailed } } func restartWatermarkTask(ctx context.Context, item *playletchaptermod.PlayletChapter, sysCfg *cfg.SysCfg) watermarkSubmitResult { watermarkImage := sysCfg.WatermarkImage watermarkMedia := sysCfg.WatermarkMedia watermarkMediaTop := sysCfg.WatermarkMediaTop watermarkMediaTopRight := sysCfg.WatermarkMediaTopRight watermarkMediaBottomLeft := sysCfg.WatermarkMediaBottomLeft req := &proxy.MediaTaskUpdateWatermarkReq{ Id: item.MediaTaskId, AppId: pubdef.AppId, AppKey: strings.TrimSpace(sysCfg.AppKey), 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 watermarkSubmitFailed } switch classifyMediaTaskCode(result.Code) { case watermarkSubmitOK: return watermarkSubmitOK case watermarkSubmitNotReady: return watermarkSubmitNotReady default: log.Info("playletwatermark:submitWatermarkTasks update/watermark 暂未成功", log.Any("chapterId", item.Id), log.Any("mediaTaskId", item.MediaTaskId), log.Any("result", result)) markFailedIfMediaSourceFailed(ctx, item) return watermarkSubmitFailed } } 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) != "" }