package srv_mediaresource import ( "context" "encoding/json" "fmt" "time" "videoweb/pub/common/log" "videoweb/pub/common/srvdef" "videoweb/pub/models/playletchaptermod" "videoweb/pub/models/playletmod" "videoweb/pub/pubdef" "videoweb/pub/redis" "videoweb/web/api/proto" goredis "github.com/go-redis/redis" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo/options" ) const ( // PlayletSerialSyncCronSpec 连载中短剧同步扫描频率:每1小时扫描一次,扫描阶段只投递 Redis 任务 PlayletSerialSyncCronSpec = "0 0 * * * ?" // playletSerialSyncQueueKeyFmt 解耦定时扫描和三方同步调用,避免 cron 扫描过程直接阻塞在外部接口上 playletSerialSyncQueueKeyFmt = "playlet:serial:sync:queue:%d" playletSerialSyncLockKeyFmt = "playlet:serial:sync:lock:%d:%d" playletSerialSyncJHLockKeyFmt = "playlet:serial:sync:jh:lock:%d:%d" playletSerialSyncLockTTL = 55 * time.Minute playletSerialSyncPopWait = 5 * time.Second playletSerialSyncRetryWait = time.Second playletSerialSyncBatchSize = int64(100) playletManualSyncMaxCount = 100 ) func playletSerialSyncQueueKey() string { return fmt.Sprintf(playletSerialSyncQueueKeyFmt, pubdef.AppId) } func playletSerialSyncLockKey(mediaId uint) string { return fmt.Sprintf(playletSerialSyncLockKeyFmt, pubdef.AppId, mediaId) } func playletSerialSyncJHLockKey(mediaId uint) string { return fmt.Sprintf(playletSerialSyncJHLockKeyFmt, pubdef.AppId, mediaId) } type playletSerialSyncJob struct { PlayletId int64 `json:"playletId"` MediaCenterId uint `json:"mediaCenterId"` MediaSource string `json:"mediaSource"` MediaSyncTime int64 `json:"mediaSyncTime"` CreateTime int64 `json:"createTime"` } // EnqueueSerialPlayletSyncJobs 扫描连载中短剧并投递 Redis 同步任务 // // 这里不直接请求三方媒资中心:扫描只负责发现任务,实际同步由后台消费者逐条处理 // 可以降低定时任务执行时间,也方便 Redis 队列在三方接口短暂异常时继续缓冲任务 func EnqueueSerialPlayletSyncJobs() { ctx := context.Background() filter := bson.M{ playletmod.FN.Playlet.MediaCenterId: bson.M{"$gt": 0}, playletmod.FN.Playlet.UpdateStatus: 1, // 原创短剧由后台维护剧集,自动同步只处理媒资导入剧 playletmod.FN.Playlet.IsOriginal: bson.M{"$ne": true}, } opts := options.Find(). SetSort(bson.D{{Key: playletmod.FN.Playlet.UpdatedAt, Value: 1}, {Key: playletmod.FN.Playlet.Id, Value: 1}}). SetLimit(playletSerialSyncBatchSize) list, err := playletmod.FindManyPlaylet(ctx, filter, opts) if err != nil { log.Error("扫描连载中短剧失败", log.Any("err", err)) return } enqueued := 0 for _, item := range list { if item == nil || item.Id == 0 || item.MediaCenterId == 0 || item.IsOriginal { continue } if err := pushPlayletSerialSyncJob(ctx, item); err != nil { log.Error("投递连载短剧同步任务失败", log.Any("playletId", item.Id), log.Any("mediaCenterId", item.MediaCenterId), log.E(err)) continue } enqueued++ } log.Info("连载中短剧同步任务扫描完成", log.Any("scanCount", len(list)), log.Any("enqueued", enqueued)) } // pushPlayletSerialSyncJob 将单部连载短剧投递到 Redis 队列 func pushPlayletSerialSyncJob(ctx context.Context, playlet *playletmod.Playlet) error { select { case <-ctx.Done(): return ctx.Err() default: } job := playletSerialSyncJob{ PlayletId: playlet.Id, MediaCenterId: playlet.MediaCenterId, MediaSource: playlet.MediaSource, MediaSyncTime: playlet.MediaSyncTime, CreateTime: time.Now().Unix(), } payload, err := json.Marshal(job) if err != nil { return err } return redis.RPush(playletSerialSyncQueueKey(), string(payload)) } // runPlayletSerialSyncWorker 常驻消费连载短剧同步队列 func runPlayletSerialSyncWorker(ctx context.Context) { for { select { case <-ctx.Done(): return default: } result, err := redis.Client().BLPop(playletSerialSyncPopWait, playletSerialSyncQueueKey()).Result() if err != nil { if err == goredis.Nil { continue } log.Error("消费连载短剧同步任务失败", log.E(err)) time.Sleep(playletSerialSyncRetryWait) continue } if len(result) < 2 { log.Error("连载短剧同步任务格式异常", log.Any("result", result)) continue } handlePlayletSerialSyncPayload(ctx, result[1]) } } // handlePlayletSerialSyncPayload 解析单个 Redis 任务并执行同步,panic 会被捕获避免 worker 退出 func handlePlayletSerialSyncPayload(ctx context.Context, payload string) { defer func() { if err := recover(); err != nil { log.Error("处理连载短剧同步任务异常", log.Any("err", err), log.Any("payload", payload)) } }() var job playletSerialSyncJob if err := json.Unmarshal([]byte(payload), &job); err != nil { log.Error("解析连载短剧同步任务失败", log.Any("payload", payload), log.E(err)) return } if job.PlayletId <= 0 || job.MediaCenterId == 0 { log.Error("连载短剧同步任务参数异常", log.Any("job", job)) return } if _, reason := syncSerialPlaylet(ctx, job, false, false); reason != "" { log.Error("连载短剧同步任务失败", log.Any("job", job), log.Any("reason", reason)) return } log.Info("连载短剧同步任务完成", log.Any("job", job)) } // SyncPlayletChapters 手动同步指定短剧的媒资剧集,支持单/批量,允许已完结短剧补集或修正状态。 func SyncPlayletChapters(ctx context.Context, req *proto.PlayletChapterSyncReq) (*proto.PlayletChapterSyncResp, *srvdef.SrvErr) { ids := normalizePlayletChapterSyncIds(req) if len(ids) == 0 { return nil, srvdef.New(srvdef.ErrParamError, "短剧ID不能为空", "") } if len(ids) > playletManualSyncMaxCount { return nil, srvdef.New(srvdef.ErrParamError, "单次最多同步100部短剧", "") } resp := &proto.PlayletChapterSyncResp{ Total: len(ids), List: make([]*proto.PlayletChapterSyncResult, 0, len(ids)), } for _, playletId := range ids { item := syncOnePlayletChapters(ctx, playletId) if item.Success { resp.SuccessCount++ } else { resp.FailedCount++ } resp.List = append(resp.List, item) } return resp, nil } // normalizePlayletChapterSyncIds 过滤重复或无效短剧ID。 func normalizePlayletChapterSyncIds(req *proto.PlayletChapterSyncReq) []int64 { if req == nil { return nil } result := make([]int64, 0, len(req.PlayletIds)) seen := make(map[int64]struct{}, len(req.PlayletIds)) for _, id := range req.PlayletIds { if id <= 0 { continue } if _, ok := seen[id]; ok { continue } seen[id] = struct{}{} result = append(result, id) } return result } // syncOnePlayletChapters 同步单部短剧的媒资剧集,单部失败不影响批量里其他短剧。 func syncOnePlayletChapters(ctx context.Context, playletId int64) *proto.PlayletChapterSyncResult { result := &proto.PlayletChapterSyncResult{ PlayletId: playletId, } playlet, err := playletmod.FindOnePlaylet(ctx, bson.M{ playletmod.FN.Playlet.Id: playletId, }, options.FindOne().SetProjection(bson.D{ {Key: playletmod.FN.Playlet.Id, Value: 1}, {Key: playletmod.FN.Playlet.MediaCenterId, Value: 1}, {Key: playletmod.FN.Playlet.MediaSource, Value: 1}, {Key: playletmod.FN.Playlet.MediaSyncTime, Value: 1}, })) if err != nil { result.Reason = playletChapterSyncErrReason(err) return result } if playlet.MediaCenterId == 0 { result.Reason = "非媒资导入短剧不能同步剧集" return result } newChapterCount, reason := syncSerialPlaylet(ctx, playletSerialSyncJob{ PlayletId: playlet.Id, MediaCenterId: playlet.MediaCenterId, MediaSource: playlet.MediaSource, MediaSyncTime: playlet.MediaSyncTime, CreateTime: time.Now().Unix(), }, true, true) if reason != "" { result.Reason = reason return result } result.Success = true result.NewChapterCount = newChapterCount return result } // playletChapterSyncErrReason 转换单部短剧同步失败原因,避免批量结果里 reason 为空。 func playletChapterSyncErrReason(err *srvdef.SrvErr) string { if err == nil { return "" } if err.Code == srvdef.ErrNoData { return "短剧不存在" } if reason := err.GetTip(); reason != "" { return reason } if reason := err.GetMsg(); reason != "" { return reason } return err.Error() } // syncSerialPlaylet 更新主表集数/状态,只追加新增章节,不覆盖已有章节。 // 手动同步使用 forceFullSync=true 全量拉取,便于恢复本地误删的旧章节;定时任务继续使用增量同步。 func syncSerialPlaylet(ctx context.Context, job playletSerialSyncJob, allowCompleted bool, forceFullSync bool) (int, string) { lockSource := job.MediaSource locked, err := lockPlayletSerialSyncBySource(lockSource, job.MediaCenterId) if err != nil { log.Error("获取连载短剧同步锁失败", log.Any("job", job), log.E(err)) return 0, "获取同步锁失败" } if !locked { return 0, "该媒资正在同步" } defer unlockPlayletSerialSyncBySource(lockSource, job.MediaCenterId) playlet, serr := playletmod.FindOnePlaylet(ctx, bson.M{ playletmod.FN.Playlet.Id: job.PlayletId, playletmod.FN.Playlet.MediaCenterId: job.MediaCenterId, }) if serr != nil { if serr.Code == srvdef.ErrNoData { return 0, "短剧不存在" } log.Error("查询连载短剧失败", log.Any("job", job), log.Any("err", serr)) return 0, "查询短剧失败" } if !allowCompleted && playlet.IsOriginal { return 0, "" } if !allowCompleted && playlet.UpdateStatus != 1 { return 0, "" } mediaSource := playlet.MediaSource if mediaSource == "" { mediaSource = job.MediaSource } ms := newPlayletMediaCenterBySource(mediaSource) syncTime := playlet.MediaSyncTime if forceFullSync { syncTime = 0 } result, err := ms.mediaSync(&playletMediaSyncReq{ Id: playlet.MediaCenterId, SyncTime: syncTime, }) if err != nil { log.Error("同步连载短剧媒资失败", log.Any("playletId", playlet.Id), log.Any("mediaCenterId", playlet.MediaCenterId), log.E(err)) return 0, "同步媒资失败" } if result.IsDelete { return 0, "媒资不存在或已删除" } if result.Media == nil { return 0, "媒资主资源为空" } newSyncTime := maxPlayletMediaContentUpdateTime(playlet.MediaSyncTime, result.MediaContentList) if reason := updateSerialPlayletInfo(ctx, playlet, result.Media, newSyncTime); reason != "" { return 0, reason } if len(result.MediaContentList) == 0 { return 0, "" } chapters, reason := buildNewSerialPlayletChapters(ctx, playlet.Id, mediaSource, result.MediaContentList) if reason != "" { return 0, reason } if len(chapters) == 0 { return 0, "" } // 上架且连载中时,新增剧集默认支付方式取剧级设置。 applyPlayletDefaultPayToChapters(playlet, chapters) if serr = playletchaptermod.InsertManyPlayletChapter(ctx, chapters); serr != nil { log.Error("连载短剧新增章节入库失败", log.Any("playletId", playlet.Id), log.Any("mediaCenterId", playlet.MediaCenterId), log.Any("count", len(chapters)), log.Any("err", serr)) return 0, "新增章节入库失败" } log.Info("连载短剧新增章节成功", log.Any("playletId", playlet.Id), log.Any("mediaCenterId", playlet.MediaCenterId), log.Any("count", len(chapters))) return len(chapters), "" } // updateSerialPlayletInfo 更新三方主资源里会随连载变化的字段 func updateSerialPlayletInfo(ctx context.Context, playlet *playletmod.Playlet, media *proto.PlayletResourceListItem, mediaSyncTime int64) string { updateStatus := normalizePlayletUpdateStatus(media.UpdateStatus) setDoc := bson.M{ playletmod.FN.Playlet.TotalEpisode: int64(media.TotalEpisode), playletmod.FN.Playlet.UpdateStatus: updateStatus, playletmod.FN.Playlet.MediaUpdatedAt: mediaResourceTime(media.UpdateTime), playletmod.FN.Playlet.UpdatedAt: time.Now(), } if mediaSyncTime > playlet.MediaSyncTime { setDoc[playletmod.FN.Playlet.MediaSyncTime] = mediaSyncTime } if business := playletMediaBusiness(media); business != "" && playlet.MediaBusiness != business { setDoc[playletmod.FN.Playlet.MediaBusiness] = business } if description := playletmod.BuildPlayletDescription(playlet.PlayName, playlet.Title, int64(media.TotalEpisode)); description != "" { setDoc[playletmod.FN.Playlet.Description] = description setDoc[playletmod.FN.Playlet.SeoInfo.Description] = description } if err := playletmod.UpdateOnePlaylet(ctx, bson.M{ playletmod.FN.Playlet.Id: playlet.Id, }, bson.M{"$set": setDoc}); err != nil { log.Error("更新连载短剧主表失败", log.Any("playletId", playlet.Id), log.Any("mediaCenterId", playlet.MediaCenterId), log.Any("err", err)) return "更新主表失败" } return "" } // buildNewSerialPlayletChapters 根据本地已存在章节做幂等过滤,只返回需要追加的新章节 func buildNewSerialPlayletChapters(ctx context.Context, playletId int64, mediaSource string, mediaContentList []playletMediaContent) (playletchaptermod.PlayletChapterList, string) { exists, err := loadExistingPlayletChapterKeys(ctx, playletId) if err != nil { return nil, "查询已有章节失败" } chapters := buildPlayletChaptersFromMedia(mediaContentList, mediaSource) result := make(playletchaptermod.PlayletChapterList, 0, len(chapters)) for _, chapter := range chapters { if chapter == nil || chapterDuplicateByKeys(chapter, exists) { continue } chapter.PlayletId = playletId markExistingPlayletChapterKeys(chapter, exists) result = append(result, chapter) } return result, "" } type existingPlayletChapterKeys struct { mediaContentIds map[uint]struct{} episodes map[int64]struct{} hashes map[string]struct{} videoUrls map[string]struct{} } // loadExistingPlayletChapterKeys 加载当前短剧已入库章节的去重键,避免连载同步重复插入章节 func loadExistingPlayletChapterKeys(ctx context.Context, playletId int64) (existingPlayletChapterKeys, error) { keys := existingPlayletChapterKeys{ mediaContentIds: make(map[uint]struct{}), episodes: make(map[int64]struct{}), hashes: make(map[string]struct{}), videoUrls: make(map[string]struct{}), } list, serr := playletchaptermod.FindManyPlayletChapter(ctx, bson.M{playletchaptermod.FN.PlayletChapter.PlayletId: playletId}) if serr != nil { return keys, fmt.Errorf("query playlet chapters failed: %+v", serr) } for _, chapter := range list { markExistingPlayletChapterKeys(chapter, keys) } return keys, nil } // chapterDuplicateByKeys 章节跨批次去重,优先用媒资章节ID,其次用集数、hash、播放地址兜底 func chapterDuplicateByKeys(chapter *playletchaptermod.PlayletChapter, keys existingPlayletChapterKeys) bool { if chapter.MediaContentId > 0 { if _, ok := keys.mediaContentIds[chapter.MediaContentId]; ok { return true } } if chapter.CurrentEpisode > 0 { if _, ok := keys.episodes[chapter.CurrentEpisode]; ok { return true } } if chapter.Hash != "" { if _, ok := keys.hashes[chapter.Hash]; ok { return true } } if chapter.VideoUrl != "" { if _, ok := keys.videoUrls[chapter.VideoUrl]; ok { return true } } return false } // markExistingPlayletChapterKeys 把已存在或本批次准备插入的章节登记到去重集合 func markExistingPlayletChapterKeys(chapter *playletchaptermod.PlayletChapter, keys existingPlayletChapterKeys) { if chapter == nil { return } if chapter.MediaContentId > 0 { keys.mediaContentIds[chapter.MediaContentId] = struct{}{} } if chapter.CurrentEpisode > 0 { keys.episodes[chapter.CurrentEpisode] = struct{}{} } if chapter.Hash != "" { keys.hashes[chapter.Hash] = struct{}{} } if chapter.VideoUrl != "" { keys.videoUrls[chapter.VideoUrl] = struct{}{} } } // maxPlayletMediaContentUpdateTime 记录本次同步返回章节中的最大更新时间,作为下次三方增量同步点 func maxPlayletMediaContentUpdateTime(syncTime int64, mediaContentList []playletMediaContent) int64 { result := syncTime for _, item := range mediaContentList { if item.UpdateTime > result { result = item.UpdateTime } } return result } // normalizePlayletUpdateStatus 三方 update_status=0 在本项目按 2-已完结 保存 func normalizePlayletUpdateStatus(updateStatus int) int { if updateStatus == 0 { return 2 } return updateStatus } // applyPlayletDefaultPayToChapters 上架且连载中时,新增章节默认沿用剧级支付设置。 func applyPlayletDefaultPayToChapters(playlet *playletmod.Playlet, chapters playletchaptermod.PlayletChapterList) { if playlet == nil || len(chapters) == 0 { return } if playlet.Status != playletmod.StatusUp || playlet.UpdateStatus != 1 { return } payType := playletchaptermod.PayType(playlet.PayType) singleCoins := playlet.SingleCoins if payType != playletchaptermod.PayTypeCoin { singleCoins = 0 } for _, chapter := range chapters { if chapter == nil { continue } chapter.PayType = payType chapter.SingleCoins = singleCoins } } // lockPlayletSerialSync 锁定单个媒资ID,避免同一部连载短剧被多个 worker 同时同步 func lockPlayletSerialSync(mediaId uint) (bool, error) { return redis.Setnx_NewOK(playletSerialSyncLockKey(mediaId), true, playletSerialSyncLockTTL) } // unlockPlayletSerialSync 释放单个媒资ID的同步锁 func unlockPlayletSerialSync(mediaId uint) { if err := redis.Del(playletSerialSyncLockKey(mediaId)); err != nil { log.Error("释放连载短剧同步锁失败", log.Any("mediaId", mediaId), log.E(err)) } } // lockPlayletSerialSyncBySource 按媒资来源锁定单个媒资ID func lockPlayletSerialSyncBySource(mediaSource string, mediaId uint) (bool, error) { if mediaSource == playletmod.MediaSourceHG { return lockPlayletSerialSync(mediaId) } return redis.Setnx_NewOK(playletSerialSyncJHLockKey(mediaId), true, playletSerialSyncLockTTL) } // unlockPlayletSerialSyncBySource 按媒资来源释放单个媒资ID的同步锁 func unlockPlayletSerialSyncBySource(mediaSource string, mediaId uint) { if mediaSource == playletmod.MediaSourceHG { unlockPlayletSerialSync(mediaId) return } if err := redis.Del(playletSerialSyncJHLockKey(mediaId)); err != nil { log.Error("释放嘉华连载短剧同步锁失败", log.Any("mediaId", mediaId), log.E(err)) } }