package videosync import ( "context" nethttp "net/http" "sync" "time" "videoweb/pub/common/log" "videoweb/pub/common/srvdef" "videoweb/pub/models/syncfilemod" "videoweb/pub/pubdef" "videoweb/pub/storage/proxy" "videoweb/web/service/srv_videoinfo" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" ) 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 srv_videoinfo.IsPlayletMedia(ctx, media) { log.Info("VideoSync skip playlet media", log.Any("mediaId", media.Id.Hex()), log.Any("mediaSourceId", media.MediaSourceId.Hex())) } se := srv_videoinfo.SyncOneVideo(ctx, media) if se != nil { log.Error("VideoSync SyncOneVideo", log.Any("media", media), log.E(se)) return } if media.UpdatedAt.After(recordInfo.RecordTime) { recordInfo.RecordTime = media.UpdatedAt } } 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)) }