package mediaService import ( "91jd-server/common/log" "91jd-server/common/mediaCenter" "91jd-server/models/v/mediacontentmod" "91jd-server/models/v/mediamod" "fmt" "strings" "time" "go.mongodb.org/mongo-driver/bson/primitive" ) func findExistingMediaContent( mediaID primitive.ObjectID, mediaType string, hashID string, episodeNumber int, findByHash func(primitive.ObjectID, string) (mediacontentmod.MediaContent, error), findByEpisode func(primitive.ObjectID, int) (mediacontentmod.MediaContent, error), ) (mediacontentmod.MediaContent, bool, error) { content, err := findByHash(mediaID, hashID) if err != mediacontentmod.MediaContentNotFound || mediaType != mediamod.MediaTypeDrama { return content, false, err } content, err = findByEpisode(mediaID, episodeNumber) return content, err == nil, err } // SyncMediaData 同步acg动漫 func (m *mediaSync) SyncMediaData(media *mediamod.Media) (err error) { if media == nil { return } // 加锁,防止多个同步出错 // 判断当前是否有正在同步的任务,有的话就不执行 key := fmt.Sprintf("MediaSync:%v", media.MediaCenterId) has, err := m.rd.SetNX(key, 1, time.Minute*10) if err != nil { log.Error("SyncAll setNx fail", log.E(err)) return } if !has { log.Info("【ACG同步任务】正在进行", log.Any("task", key)) return } // 释放锁 defer m.rd.Del(key) log.Info("【ACG同步任务】开始", log.Any("mediaId", media.ID), log.Any("mediaCenterId", media.MediaCenterId)) // 同步数据 data, err := m.Ms.MediaSync(&mediaCenter.MediaSyncReq{ Id: media.MediaCenterId, SyncTime: media.SyncTime, }) if err != nil { log.Error("SyncMediaData call mediaSync Fail", log.Any("id", media.MediaCenterId), log.E(err)) return err } if data.IsDelete { // 数据已经删除,不需要同步 log.Info("【ACG同步任务】 数据在媒资库已经删除,不需要同步", log.Any("mediaId", media.ID), log.Any("mediaCenterId", media.MediaCenterId)) return } err = m.OnSyncMediaData(media.ID, data.Media) if err != nil { log.Error("SyncMediaData call OnSyncMediaData Fail", log.Any("id", media.MediaCenterId), log.E(err)) return err } newSyncTime, isAdd, err := m.OnSyncMediaContentData(media, data.MediaContentList) if err != nil { return } err = OnMediaContentChange(media.ID, media.MediaType, isAdd) if err != nil { log.Error("【ACG同步任务】 OnMediaContentChange Fail", log.Any("mediaId", media.ID), log.E(err)) return err } // 修改下一次需要同步的时间点位 m.OnSyncTimeChange(media.ID, newSyncTime) log.Info("【ACG同步任务】完成", log.Any("mediaId", media.ID), log.Any("mediaCenterId", media.MediaCenterId)) return nil } func (m *mediaSync) OnSyncMediaContentData(media *mediamod.Media, mediaContentDataList []mediaCenter.MediaContent) (int64, bool, error) { isAdd := false syncTime := media.SyncTime updatedIDs := make([]primitive.ObjectID, 0) if len(mediaContentDataList) == 0 { // 已经没有需要同步的数据了,直接返回 return syncTime, false, nil } log.Info("【ACG同步任务】获取到需要同步的子集", log.Any("mediaId", media.ID), log.Any("mediaCenterId", media.MediaCenterId), log.Any("子集数量", len(mediaContentDataList))) for _, mediaContentData := range mediaContentDataList { h265URL := normalizeH265URL(mediaContentData.H265Url) // 更改下一次同步的时间点 if mediaContentData.UpdateTime > syncTime { syncTime = mediaContentData.UpdateTime } log.Info("【ACG同步任务】同步子集", log.Any("mediaId", media.ID), log.Any("mediaCenterId", media.MediaCenterId), log.Any("子集ID", mediaContentData.ID)) // 判断是否已经存在该记录 mediaContent, hashChanged, err := findExistingMediaContent( media.ID, media.MediaType, mediaContentData.HashId, mediaContentData.EpisodeNumber, mediacontentmod.GetInfoByHashId, mediacontentmod.GetInfoByMediaIDAndEpisodeNumber, ) if hashChanged { log.Warn("【ACG同步任务】媒资hashId已变更,按剧集编号复用原子集", log.Any("mediaId", media.ID), log.Any("episodeNumber", mediaContentData.EpisodeNumber), log.Any("oldHashId", mediaContent.HashId), log.Any("newHashId", mediaContentData.HashId)) } if err != nil && err != mediacontentmod.MediaContentNotFound { log.Error("【ACG同步任务】同步子集失败", log.Any("mediaId", media.ID), log.Any("mediaCenterId", media.MediaCenterId), log.Any("子集ID", mediaContentData.ID), log.Any("err", err)) return 0, false, err } // TODO 后续同步,只做追加,不对之前的子集数据进行更新,避免数据出错 if err == mediacontentmod.MediaContentNotFound { isAdd = true // 不存在则入库 mediaContent = mediacontentmod.MediaContent{ MediaID: media.ID, MediaType: media.MediaType, EpisodeNumber: mediaContentData.EpisodeNumber, ListenPermission: 0, // 收听权限 0:会员 1:金币购买 2:免费 Name: mediaContentData.Title, // 章节名 Text: mediaContentData.Text, // 内容(小说专用) Author: media.Anchor, // 作者 Height: int(mediaContentData.Height), // 高 Weight: int(mediaContentData.Width), // 宽 MediaSize: int64(mediaContentData.MediaSize), // 资源大小 PlayTime: mediaContentData.PlayTime, // 影片长度 Status: 1, // 状态 0:转码中 1:转码成功 2:转码失败, 视频资源使用 UpdatedAct: "系统自动同步", // 操作用户账号 CreatedAt: time.Now(), // 文档创建时间 UpdateTime: time.Now(), // 文档更新时间 IsActive: true, // 同步过来的默认上架 HashId: mediaContentData.HashId, // 在媒资库中的唯一标识 VideoUrl: mediaContentData.M3u8Url, H265Url: h265URL, AudioUrl: mediaContentData.AudioUrl, } if media.MediaType == mediamod.MediaTypeDrama { mediaContent.Price = int64(mediaContentData.SaleCoin) mediaContent.ListenPermission = 1 if mediaContentData.SaleCoin == 0 { mediaContent.ListenPermission = 2 } } if strings.TrimSpace(mediaContentData.Urls) != "" { mediaContent.UrlSet = strings.Split(mediaContentData.Urls, ",") } else { mediaContent.UrlSet = []string{} } _, err = mediacontentmod.Insert(nil, mediaContent) if err != nil { log.Error("【ACG同步任务】同步子集失败", log.Any("mediaId", media.ID), log.Any("mediaCenterId", media.MediaCenterId), log.Any("子集ID", mediaContentData.ID), log.Any("err", err)) return 0, false, err } } else { updateData := make(map[string]interface{}) if mediaContent.HashId != mediaContentData.HashId { updateData["hashId"] = mediaContentData.HashId } if shouldUpdateH265URL(mediaContent.H265Url, h265URL) { updateData["h265Url"] = h265URL } if len(updateData) > 0 { // 保留运营维护的价格、权限等数据,只更新媒资标识和 H.265 地址。 updateData["updateTime"] = time.Now() if _, err = mediacontentmod.UpdateByID(nil, mediaContent.ID, updateData); err != nil { log.Error("【ACG同步任务】更新子集媒资信息失败", log.Any("mediaId", media.ID), log.Any("mediaContentId", mediaContent.ID), log.Any("子集ID", mediaContentData.ID), log.Any("err", err)) return 0, false, err } updatedIDs = append(updatedIDs, mediaContent.ID) } } } return syncTime, isAdd, nil } func normalizeH265URL(raw string) string { return strings.TrimSpace(raw) } func shouldUpdateH265URL(current, incoming string) bool { incoming = normalizeH265URL(incoming) return incoming != "" && normalizeH265URL(current) != incoming } func (m *mediaSync) OnSyncMediaData(mediaId primitive.ObjectID, mediaData *mediaCenter.Media) (err error) { // 只同步连载状态,不覆盖运营后台维护的总集数等配置。 updateData := make(map[string]interface{}) //updateData["title"] = data.Media.Title //updateData["horizontalCover"] = data.Media.HorizontalCover //updateData["verticalCover"] = data.Media.VerticalCover //updateData["summary"] = data.Media.Summary updateData["updateStatus"] = mediaData.UpdateStatus if mediaData.UpdateStatus == 0 { // 直接是已完结 updateData["updateStatus"] = 2 } //updateData["anchor"] = data.Media.Anchor updateData["updateTime"] = time.Now() _, err = mediamod.UpdateByID(nil, mediaId, updateData) if err != nil { return } return }