package srv_mediaresource import ( "context" "sync" "videoweb/pub/common/log" "videoweb/pub/common/srvdef" "videoweb/pub/models/playletmod" "videoweb/web/api/proto" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo/options" ) const ( // defaultOriginalPlayletBackfillPageSize 与媒资库列表默认分页一致。 defaultOriginalPlayletBackfillPageSize = int64(20) // defaultOriginalPlayletBackfillMaxPages 限制启动回填最多扫描 2000 条媒资。 defaultOriginalPlayletBackfillMaxPages = int64(100) // maxOriginalPlayletBackfillMaxPages 防止误配置导致启动任务长时间扫全量。 maxOriginalPlayletBackfillMaxPages = int64(1000) ) // originalPlayletBackfillOnce 保证每个进程只触发一次启动回填。 var originalPlayletBackfillOnce sync.Once // originalPlayletBackfillOptions 控制 CGDJ 原创短剧回填范围。 type originalPlayletBackfillOptions struct { DryRun bool // 只预览不写库 PageSize int64 // 媒资库分页大小 MaxPages int64 // 最多扫描页数 } // originalPlayletBackfillItem 记录单部已导入短剧的回填结果。 type originalPlayletBackfillItem struct { MediaCenterId int64 // 媒资库主资源 ID PlayletId int64 // 本地短剧 ID Title string // 本地短剧标题 MediaTitle string // 媒资库原始标题 Source string // 本次识别出的媒资来源 Before string // 回填前媒资来源 Result string // updated / preview / already_marked / conflict / failed } // originalPlayletBackfillResp 汇总一次 CGDJ 回填统计。 type originalPlayletBackfillResp struct { DryRun bool PageSize int64 MaxPages int64 Pages int64 SourceTotal int64 // 媒资库返回总数 SourceScanned int64 // 已扫描媒资条数 SourceMatched int64 // 匹配 CGDJ 前缀的媒资条数 Imported int64 // 已导入本地短剧库的条数 Matched int64 // 需要写入媒资来源的条数 Updated int64 // 实际更新成功条数 Skipped int64 // 已有相同来源或并发冲突跳过条数 HasMore bool // 达到 MaxPages 后仍可能存在后续数据 List []*originalPlayletBackfillItem } // originalPlayletBackfillStore 抽象外部依赖,方便单测覆盖媒资库和 Mongo。 type originalPlayletBackfillStore struct { mediaList func(*playletMediaListReq) (playletMediaListResp, error) findMany func(context.Context, bson.M, *options.FindOptions) (playletmod.PlayletList, *srvdef.SrvErr) updateOne func(context.Context, bson.M, bson.M) *srvdef.SrvErr audit func(context.Context, *originalPlayletBackfillItem) } // StartCGDJOriginalBackfillOnce 在 web 启动后异步执行一次 CGDJ 原创短剧标记回填。 func StartCGDJOriginalBackfillOnce(ctx context.Context) { originalPlayletBackfillOnce.Do(func() { go runCGDJOriginalBackfillOnce(ctx) }) } // runCGDJOriginalBackfillOnce 执行启动回填,成功后顺手触发一次昨日数据回流。 func runCGDJOriginalBackfillOnce(ctx context.Context) { resp, serr := backfillCGDJOriginalPlaylets(ctx, originalPlayletBackfillOptions{DryRun: false}) if serr != nil { log.Error("CGDJ 原创短剧标记启动回填失败", log.Any("err", serr), log.Any("resp", resp)) return } log.Info("CGDJ 原创短剧标记启动回填完成", log.Any("pages", resp.Pages), log.Any("sourceTotal", resp.SourceTotal), log.Any("sourceScanned", resp.SourceScanned), log.Any("sourceMatched", resp.SourceMatched), log.Any("imported", resp.Imported), log.Any("matched", resp.Matched), log.Any("updated", resp.Updated), log.Any("skipped", resp.Skipped), log.Any("hasMore", resp.HasMore)) ReportAIContentBackflowYesterday() } // backfillCGDJOriginalPlaylets 从媒资库 CGDJ 短剧列表回填本地短剧 mediaBusiness。 // 判断依据只使用源媒资标题前缀和 mediaCenterId,不依赖本地短剧标题,避免运营改名后漏判。 func backfillCGDJOriginalPlaylets(ctx context.Context, opts originalPlayletBackfillOptions) (*originalPlayletBackfillResp, *srvdef.SrvErr) { ms := newPlayletMediaCenterFromConfig() return backfillCGDJOriginalPlayletsWithStore(ctx, opts, originalPlayletBackfillStore{ mediaList: ms.mediaList, findMany: func(ctx context.Context, filter bson.M, opts *options.FindOptions) (playletmod.PlayletList, *srvdef.SrvErr) { return playletmod.FindManyPlaylet(ctx, filter, opts) }, updateOne: playletmod.UpdateOnePlaylet, audit: func(ctx context.Context, item *originalPlayletBackfillItem) { log.Info("回填短剧媒资业务来源", log.Any("playletId", item.PlayletId), log.Any("mediaCenterId", item.MediaCenterId), log.Any("title", item.Title), log.Any("mediaTitle", item.MediaTitle), log.Any("mediaBusiness", item.Source)) }, }) } // backfillCGDJOriginalPlayletsWithStore 按媒资库 CGDJ 搜索结果回填本地短剧媒资来源。 func backfillCGDJOriginalPlayletsWithStore(ctx context.Context, opts originalPlayletBackfillOptions, store originalPlayletBackfillStore) (*originalPlayletBackfillResp, *srvdef.SrvErr) { pageSize, maxPages, serr := normalizeOriginalPlayletBackfillOptions(opts) if serr != nil { return nil, serr } resp := &originalPlayletBackfillResp{ DryRun: opts.DryRun, PageSize: pageSize, MaxPages: maxPages, List: make([]*originalPlayletBackfillItem, 0), } for _, title := range playletMediaBusinessSearchTitles() { searchTitle := title for page := int64(1); page <= maxPages; page++ { data, err := store.mediaList(&playletMediaListReq{ MediaType: playletResourceMediaTypeShort, Title: &searchTitle, SyncType: defaultResourceSyncType, Page: uint(page), Size: uint(pageSize), }) if err != nil { log.Error("查询 CGDJ 短剧媒资失败", log.Any("title", searchTitle), log.Any("page", page), log.E(err)) return resp, srvdef.New(srvdef.Failure, "查询 CGDJ 短剧媒资失败", "查询 CGDJ 短剧媒资失败") } resp.Pages++ if page == 1 { resp.SourceTotal += int64(data.Pagination.Count) } if len(data.Data) == 0 { break } if serr = backfillOriginalPlayletPage(ctx, data.Data, resp, store); serr != nil { return resp, serr } if data.Pagination.IsLastPage || int64(len(data.Data)) < pageSize { break } if page == maxPages { resp.HasMore = true } } } return resp, nil } // normalizeOriginalPlayletBackfillOptions 修正分页参数并限制启动扫描范围。 func normalizeOriginalPlayletBackfillOptions(opts originalPlayletBackfillOptions) (int64, int64, *srvdef.SrvErr) { pageSize := opts.PageSize if pageSize <= 0 { pageSize = defaultOriginalPlayletBackfillPageSize } maxPages := opts.MaxPages if maxPages <= 0 { maxPages = defaultOriginalPlayletBackfillMaxPages } if pageSize > int64(maxResourcePageSize) { return 0, 0, srvdef.New(srvdef.ErrParamError, "pageSize最多100", "pageSize最多100") } if maxPages > maxOriginalPlayletBackfillMaxPages { return 0, 0, srvdef.New(srvdef.ErrParamError, "maxPages最多1000", "maxPages最多1000") } return pageSize, maxPages, nil } // backfillOriginalPlayletPage 只更新已导入本地短剧库的 CGDJ 媒资。 func backfillOriginalPlayletPage(ctx context.Context, mediaList []*proto.PlayletResourceListItem, resp *originalPlayletBackfillResp, store originalPlayletBackfillStore) *srvdef.SrvErr { mediaTitleByID := make(map[uint]string, len(mediaList)) mediaBusinessByID := make(map[uint]string, len(mediaList)) mediaIDs := make([]uint, 0, len(mediaList)) for _, media := range mediaList { resp.SourceScanned++ business := playletMediaBusiness(media) if business == "" { continue } if _, exists := mediaTitleByID[media.ID]; exists { continue } resp.SourceMatched++ mediaTitleByID[media.ID] = media.Title mediaBusinessByID[media.ID] = business mediaIDs = append(mediaIDs, media.ID) } if len(mediaIDs) == 0 { return nil } playlets, serr := store.findMany(ctx, bson.M{ playletmod.FN.Playlet.MediaCenterId: bson.M{"$in": mediaIDs}, playletmod.FN.Playlet.Status: playletmod.StatusUp, }, options.Find().SetProjection(bson.M{ playletmod.FN.Playlet.Id: 1, playletmod.FN.Playlet.MediaCenterId: 1, playletmod.FN.Playlet.Title: 1, playletmod.FN.Playlet.MediaBusiness: 1, })) if serr != nil { log.Error("查询已导入 CGDJ 短剧失败", log.Any("mediaIds", mediaIDs), log.Any("err", serr)) return serr } for _, playlet := range playlets { if playlet == nil || playlet.MediaCenterId == 0 { continue } business := mediaBusinessByID[playlet.MediaCenterId] if business == "" { continue } resp.Imported++ item := &originalPlayletBackfillItem{ MediaCenterId: int64(playlet.MediaCenterId), PlayletId: playlet.Id, Title: playlet.Title, MediaTitle: mediaTitleByID[playlet.MediaCenterId], Source: business, Before: playlet.MediaBusiness, } if playlet.MediaBusiness == business { resp.Skipped++ item.Result = "already_marked" resp.List = append(resp.List, item) continue } resp.Matched++ if resp.DryRun { item.Result = "preview" resp.List = append(resp.List, item) continue } serr = store.updateOne(ctx, bson.M{ playletmod.FN.Playlet.Id: playlet.Id, playletmod.FN.Playlet.MediaCenterId: playlet.MediaCenterId, playletmod.FN.Playlet.Status: playletmod.StatusUp, playletmod.FN.Playlet.MediaBusiness: bson.M{"$ne": business}, }, bson.M{"$set": bson.M{ playletmod.FN.Playlet.MediaBusiness: business, }}) if serr != nil { if serr.Code == srvdef.ErrNoData { resp.Skipped++ item.Result = "conflict" resp.List = append(resp.List, item) continue } item.Result = "failed" resp.List = append(resp.List, item) return serr } resp.Updated++ item.Result = "updated" resp.List = append(resp.List, item) if store.audit != nil { store.audit(ctx, item) } } return nil }