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