285 lines
9.0 KiB
Go
285 lines
9.0 KiB
Go
package playletwatermark
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"videoweb/pub/common/log"
|
|
"videoweb/pub/common/srvdef"
|
|
"videoweb/pub/models/playletchaptermod"
|
|
"videoweb/pub/models/playletmod"
|
|
"videoweb/pub/models/syncfilemod"
|
|
"videoweb/pub/pubdef"
|
|
"videoweb/pub/storage/proxy"
|
|
|
|
"go.mongodb.org/mongo-driver/bson"
|
|
"go.mongodb.org/mongo-driver/mongo/options"
|
|
)
|
|
|
|
const (
|
|
syncPageSize int64 = 1000
|
|
syncOverlapWindow = 10 * time.Minute
|
|
reconcileBatchSize int64 = 100
|
|
)
|
|
|
|
var (
|
|
syncRunning bool
|
|
syncLock sync.Mutex
|
|
)
|
|
|
|
// SyncCompletedChapters 定时从 media 拉取已完成水印任务，回写剧集地址与状态
|
|
func SyncCompletedChapters() {
|
|
syncLock.Lock()
|
|
if syncRunning {
|
|
syncLock.Unlock()
|
|
return
|
|
}
|
|
syncRunning = true
|
|
syncLock.Unlock()
|
|
defer func() {
|
|
syncLock.Lock()
|
|
syncRunning = false
|
|
syncLock.Unlock()
|
|
}()
|
|
|
|
ctx := context.Background()
|
|
log.Info("playletwatermark:SyncCompletedChapters begin")
|
|
|
|
record, err := ensureSyncRecord(ctx)
|
|
if err != nil {
|
|
log.Error("playletwatermark:SyncCompletedChapters 获取同步水印失败", log.E(err))
|
|
return
|
|
}
|
|
|
|
recordTime, syncOK := syncIncrementalCompletedChapters(ctx, record.RecordTime)
|
|
if syncOK {
|
|
if updateErr := syncfilemod.UpdateOneSyncRecord(ctx, bson.M{
|
|
syncfilemod.FN.SyncRecord.Job: syncfilemod.PlayletWatermarkJob,
|
|
}, bson.M{"$set": bson.M{
|
|
syncfilemod.FN.SyncRecord.RecordTime: recordTime,
|
|
syncfilemod.FN.SyncRecord.SuccessAt: time.Now(),
|
|
syncfilemod.FN.SyncRecord.UpdatedAt: time.Now(),
|
|
}}); updateErr != nil {
|
|
log.Error("playletwatermark:SyncCompletedChapters 更新同步水印失败", log.E(updateErr))
|
|
}
|
|
}
|
|
|
|
reconcileProcessingChapters(ctx)
|
|
log.Info("playletwatermark:SyncCompletedChapters end",
|
|
log.Any("recordTime", recordTime),
|
|
log.Any("syncOK", syncOK))
|
|
}
|
|
|
|
// syncIncrementalCompletedChapters 按更新时间水印增量拉取。
|
|
// 每次回退一段时间重复扫描，避免临界时间及同一时间戳分页造成漏数据；章节回写是幂等的。
|
|
func syncIncrementalCompletedChapters(ctx context.Context, recordTime time.Time) (time.Time, bool) {
|
|
cursor := recordTime
|
|
if !cursor.IsZero() {
|
|
cursor = cursor.Add(-syncOverlapWindow)
|
|
}
|
|
maxRecordTime := recordTime
|
|
for {
|
|
req := proxy.MediaTaskSyncReq{
|
|
AppId: pubdef.AppId,
|
|
LastUpdateTime: cursor,
|
|
PageSize: syncPageSize,
|
|
}
|
|
result, syncErr := proxy.SyncMedia(&req, "")
|
|
if syncErr != nil || result == nil || result.Code != http.StatusOK {
|
|
log.Error("playletwatermark:SyncCompletedChapters SyncMedia 失败", log.Any("req", req), log.E(syncErr))
|
|
return recordTime, false
|
|
}
|
|
pageMaxTime := cursor
|
|
for _, media := range result.Data {
|
|
if media.UpdatedAt.After(pageMaxTime) {
|
|
pageMaxTime = media.UpdatedAt
|
|
}
|
|
if media.UpdatedAt.After(maxRecordTime) {
|
|
maxRecordTime = media.UpdatedAt
|
|
}
|
|
applyCompletedMedia(ctx, media, "incremental")
|
|
}
|
|
if int64(len(result.Data)) < syncPageSize {
|
|
break
|
|
}
|
|
if !pageMaxTime.After(cursor) {
|
|
log.Error("playletwatermark:SyncCompletedChapters 增量分页水印未推进",
|
|
log.Any("cursor", cursor),
|
|
log.Any("count", len(result.Data)))
|
|
return recordTime, false
|
|
}
|
|
cursor = pageMaxTime
|
|
}
|
|
return maxRecordTime, true
|
|
}
|
|
|
|
// reconcileProcessingChapters 反向对账所有处理中章节。
|
|
// 它不依赖增量水印，可补偿本地更新失败、时间戳临界和历史水印越过造成的漏回写。
|
|
func reconcileProcessingChapters(ctx context.Context) {
|
|
var lastID int64
|
|
for {
|
|
list, err := playletchaptermod.FindManyPlayletChapter(ctx, bson.M{
|
|
playletchaptermod.FN.PlayletChapter.Id: bson.M{"$gt": lastID},
|
|
playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusProcessing,
|
|
playletchaptermod.FN.PlayletChapter.MediaSourceId: bson.M{"$gt": ""},
|
|
}, options.Find().
|
|
SetSort(bson.D{{Key: playletchaptermod.FN.PlayletChapter.Id, Value: 1}}).
|
|
SetLimit(reconcileBatchSize))
|
|
if err != nil {
|
|
log.Error("playletwatermark:reconcileProcessingChapters 查询章节失败", log.E(err))
|
|
return
|
|
}
|
|
if len(list) == 0 {
|
|
return
|
|
}
|
|
|
|
sourceIDs := make([]string, 0, len(list))
|
|
for _, item := range list {
|
|
if item == nil {
|
|
continue
|
|
}
|
|
sourceIDs = append(sourceIDs, item.MediaSourceId)
|
|
if item.Id > lastID {
|
|
lastID = item.Id
|
|
}
|
|
}
|
|
if len(sourceIDs) > 0 {
|
|
result, queryErr := proxy.QueryMedias(&proxy.MediaQueryesReq{
|
|
AppId: pubdef.AppId,
|
|
MediaSourceIds: sourceIDs,
|
|
}, "")
|
|
if queryErr != nil || result == nil || result.Code != http.StatusOK {
|
|
log.Error("playletwatermark:reconcileProcessingChapters 查询 media 任务失败",
|
|
log.Any("sourceCount", len(sourceIDs)),
|
|
log.Any("result", result),
|
|
log.E(queryErr))
|
|
return
|
|
}
|
|
for _, media := range result.Data {
|
|
applyCompletedMedia(ctx, media, "reconcile")
|
|
}
|
|
}
|
|
if int64(len(list)) < reconcileBatchSize {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func applyCompletedMedia(ctx context.Context, media proxy.MediaResp, source string) bool {
|
|
sourceID := media.MediaSourceId.Hex()
|
|
if sourceID == "" || sourceID == "000000000000000000000000" {
|
|
log.Warn("playletwatermark:applyCompletedMedia mediaSourceId 无效",
|
|
log.Any("mediaTaskId", media.Id.Hex()),
|
|
log.Any("source", source))
|
|
return false
|
|
}
|
|
fieldPath := strings.TrimSpace(media.FieldNameFs)
|
|
if fieldPath == "" {
|
|
log.Warn("playletwatermark:applyCompletedMedia 视频路径为空",
|
|
log.Any("mediaSourceId", sourceID),
|
|
log.Any("mediaTaskId", media.Id.Hex()),
|
|
log.Any("source", source))
|
|
return false
|
|
}
|
|
if !strings.HasSuffix(strings.ToLower(fieldPath), ".m3u8") {
|
|
fieldPath += ".m3u8"
|
|
}
|
|
taskID := media.Id.Hex()
|
|
setDoc := bson.M{
|
|
playletchaptermod.FN.PlayletChapter.WatermarkVideoUrl: fieldPath,
|
|
playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusDone,
|
|
playletchaptermod.FN.PlayletChapter.MediaTaskId: taskID,
|
|
playletchaptermod.FN.PlayletChapter.UpdatedAt: time.Now(),
|
|
}
|
|
if sourceName := strings.TrimSpace(media.SourceNameFs); sourceName != "" &&
|
|
isOriginalPlayletBySourceID(ctx, sourceID) {
|
|
// 原创才回填原片地址，供换水印/重置复用；非原创保留媒资原始 VideoUrl。
|
|
setDoc[playletchaptermod.FN.PlayletChapter.VideoUrl] = sourceName
|
|
}
|
|
updateErr := playletchaptermod.UpdateOnePlayletChapter(ctx, bson.M{
|
|
playletchaptermod.FN.PlayletChapter.MediaSourceId: sourceID,
|
|
playletchaptermod.FN.PlayletChapter.WatermarkStatus: playletchaptermod.WatermarkStatusProcessing,
|
|
}, bson.M{"$set": setDoc})
|
|
if updateErr == nil {
|
|
log.Info("playletwatermark:applyCompletedMedia 回写成功",
|
|
log.Any("mediaSourceId", sourceID),
|
|
log.Any("mediaTaskId", taskID),
|
|
log.Any("watermarkVideoUrl", fieldPath),
|
|
log.Any("source", source))
|
|
return true
|
|
}
|
|
if updateErr.Code != srvdef.ErrNoData {
|
|
log.Error("playletwatermark:applyCompletedMedia 更新章节失败",
|
|
log.Any("mediaSourceId", sourceID),
|
|
log.Any("source", source),
|
|
log.E(updateErr))
|
|
return false
|
|
}
|
|
|
|
chapter, findErr := playletchaptermod.FindOnePlayletChapter(ctx, bson.M{
|
|
playletchaptermod.FN.PlayletChapter.MediaSourceId: sourceID,
|
|
})
|
|
if findErr != nil {
|
|
if findErr.Code != srvdef.ErrNoData {
|
|
log.Error("playletwatermark:applyCompletedMedia 查询未匹配章节失败",
|
|
log.Any("mediaSourceId", sourceID),
|
|
log.E(findErr))
|
|
} else {
|
|
log.Warn("playletwatermark:applyCompletedMedia 未找到对应章节",
|
|
log.Any("mediaSourceId", sourceID),
|
|
log.Any("mediaTaskId", taskID),
|
|
log.Any("source", source))
|
|
}
|
|
return false
|
|
}
|
|
if chapter.WatermarkStatus != playletchaptermod.WatermarkStatusDone {
|
|
log.Warn("playletwatermark:applyCompletedMedia 章节状态不允许回写",
|
|
log.Any("chapterId", chapter.Id),
|
|
log.Any("mediaSourceId", sourceID),
|
|
log.Any("watermarkStatus", chapter.WatermarkStatus),
|
|
log.Any("source", source))
|
|
}
|
|
return false
|
|
}
|
|
|
|
func isOriginalPlayletBySourceID(ctx context.Context, sourceID string) bool {
|
|
chapter, err := playletchaptermod.FindOnePlayletChapter(ctx, bson.M{
|
|
playletchaptermod.FN.PlayletChapter.MediaSourceId: sourceID,
|
|
}, options.FindOne().SetProjection(bson.M{
|
|
playletchaptermod.FN.PlayletChapter.PlayletId: 1,
|
|
}))
|
|
if err != nil || chapter == nil || chapter.PlayletId <= 0 {
|
|
return false
|
|
}
|
|
playlet, err := playletmod.FindOnePlaylet(ctx, bson.M{
|
|
playletmod.FN.Playlet.Id: chapter.PlayletId,
|
|
}, options.FindOne().SetProjection(bson.M{
|
|
playletmod.FN.Playlet.IsOriginal: 1,
|
|
}))
|
|
return err == nil && playlet != nil && playlet.IsOriginal
|
|
}
|
|
|
|
func ensureSyncRecord(ctx context.Context) (*syncfilemod.SyncRecord, *srvdef.SrvErr) {
|
|
filter := bson.M{syncfilemod.FN.SyncRecord.Job: syncfilemod.PlayletWatermarkJob}
|
|
record, err := syncfilemod.FindOneSyncRecord(ctx, filter)
|
|
if err == nil {
|
|
return record, nil
|
|
}
|
|
if err.Code != srvdef.ObjectDayStatDoesNotExist {
|
|
return nil, err
|
|
}
|
|
now := time.Now()
|
|
record = &syncfilemod.SyncRecord{
|
|
Job: syncfilemod.PlayletWatermarkJob,
|
|
RecordTime: time.Time{},
|
|
CreatedAt: now,
|
|
UpdatedAt: now,
|
|
}
|
|
if insertErr := syncfilemod.InsertOneSyncRecord(ctx, record); insertErr != nil {
|
|
return nil, insertErr
|
|
}
|
|
return record, nil
|
|
}
|