Files
web_huangguo_server/web/service/srv_mediaresource/serial_sync.go
T
rootandClaude Opus 5.5 7974a42301 Initial commit
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-23 11:07:42 +08:00

526 lines
18 KiB
Go

package srv_mediaresource
import (
"context"
"encoding/json"
"fmt"
"strings"
"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, 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 playName := strings.TrimSpace(playlet.PlayName); playName != "" {
description := playletmod.BuildPlayNameDescription(playlet.Desc, playName, updateStatus, int64(media.TotalEpisode))
if 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, mediaContentList []playletMediaContent) (playletchaptermod.PlayletChapterList, string) {
exists, err := loadExistingPlayletChapterKeys(ctx, playletId)
if err != nil {
return nil, "查询已有章节失败"
}
chapters := buildPlayletChaptersFromMedia(mediaContentList)
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))
}
}