package videosync import ( "context" "errors" "hkdj-web-server/pub/common/log" "hkdj-web-server/pub/common/srvdef" "hkdj-web-server/pub/models/syncfilemod" "hkdj-web-server/pub/pubdef" "hkdj-web-server/pub/storage/proxy" "hkdj-web-server/web/service/srv_videoinfo" nethttp "net/http" "sync" "time" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" ) /* 1.小说采集后上传媒资库 2.业务后台记录最新更新章节ID(数据ID自增) 3.查询对应小说ID是否存在,如果不存在需要先同步媒资库小说信息,再更新小说 注意:业务后台和媒资库 存储的小说ID和章节ID需一致, 资源以媒资库为准, 不单独更新业务后台小说 */ var ( isRunning = false lock sync.Mutex ) func getIsRunning() bool { lock.Lock() defer lock.Unlock() return isRunning } func setRunning(b bool) { lock.Lock() defer lock.Unlock() isRunning = b } func VideoSync() { isRunning = getIsRunning() if isRunning { return } setRunning(true) ctx := context.Background() log.Info("VideoSync begin") pageSize := int64(200) recordFilter := bson.M{ syncfilemod.FN.SyncRecord.Job: syncfilemod.VideoJob, } recordInfo, serr := syncfilemod.FindOneSyncRecord(ctx, recordFilter) if serr != nil { if serr.Code == srvdef.ObjectDayStatDoesNotExist { log.Info("VideoSync init record", log.Any("job", syncfilemod.VideoJob)) syncfilemod.InsertOneSyncRecord(ctx, &syncfilemod.SyncRecord{ ID: primitive.NewObjectID(), Job: syncfilemod.VideoJob, RecordId: 1, RecordTime: time.Time{}, SuccessAt: time.Time{}, CreatedAt: time.Time{}, UpdatedAt: time.Time{}, }) } else { log.Error("VideoSync FindOneSyncRecord failed", log.Any("recordFilter", recordFilter), log.E(serr)) } setRunning(false) return } defer func() { setRunning(false) // 更新同步记录 set := bson.M{ "$set": bson.M{ syncfilemod.FN.SyncRecord.RecordTime: recordInfo.RecordTime, syncfilemod.FN.SyncRecord.SuccessAt: time.Now(), }, } _ = syncfilemod.UpdateOneSyncRecord(ctx, recordFilter, set) }() // 循环同步 for { // 查询接口 req := proxy.MediaTaskSyncReq{ AppId: pubdef.AppId, LastUpdateTime: recordInfo.RecordTime, PageSize: pageSize, } result, err := proxy.SyncMedia(&req, "") if err != nil || result.Code != nethttp.StatusOK { log.Error("VideoSync SyncMedia", log.Any("req", req), log.E(err)) return } // 处理 for _, media := range result.Data { // 先推进水位,确保单条失败后下次不会卡在同一位置 if media.UpdatedAt.After(recordInfo.RecordTime) { recordInfo.RecordTime = media.UpdatedAt } if srv_videoinfo.IsPlayletMedia(ctx, media) { log.Info("VideoSync skip playlet media", log.Any("mediaId", media.Id.Hex()), log.Any("mediaSourceId", media.MediaSourceId.Hex())) continue } se := srv_videoinfo.SyncOneVideo(ctx, media) if se != nil { log.Error("VideoSync SyncOneVideo", log.Any("media", media), log.E(se)) continue } } if len(result.Data) < int(pageSize) { log.Info("VideoSync success", log.Any("RecordTime", recordInfo.RecordTime)) break } } log.Info("VideoSync success", log.Any("RecordTime", recordInfo.RecordTime)) } // SyncUnencryptedFields 全量补齐历史资源的 UnencryptedImage / UnencryptedPreMp4 字段。 // 从头分页拉取媒资库,仅在源端有值时写回,本地不存在或字段已有值不覆盖。 // 独立于增量同步的 SyncRecord,用于一次性修复;可安全重复执行。 func SyncUnencryptedFields() { ctx := context.Background() log.Info("SyncUnencryptedFields begin") pageSize := int64(200) lastUpdateTime := time.Time{} for { req := proxy.MediaTaskSyncReq{ AppId: pubdef.AppId, LastUpdateTime: lastUpdateTime, PageSize: pageSize, } result, err := proxy.SyncMedia(&req, "") if err != nil || result.Code != nethttp.StatusOK { log.Error("SyncUnencryptedFields SyncMedia", log.Any("req", req), log.E(err)) return } for _, media := range result.Data { // 先推进时间戳,保证 continue 也不阻塞分页 if media.UpdatedAt.After(lastUpdateTime) { lastUpdateTime = media.UpdatedAt } se := srv_videoinfo.SyncUnencryptedOne(ctx, media) if se != nil && !errors.Is(se.Code, srvdef.ErrNoData) { log.Warn("SyncUnencryptedFields SyncUnencryptedOne", log.Any("mediaId", media.Id.Hex()), log.E(se)) } } if len(result.Data) < int(pageSize) { break } } log.Info("SyncUnencryptedFields success", log.Any("lastUpdateTime", lastUpdateTime)) }