271 lines
9.8 KiB
Go
271 lines
9.8 KiB
Go
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
|
|
}
|