112 lines
2.8 KiB
Go
112 lines
2.8 KiB
Go
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))
|
|
}
|