change default videos to sync

add length check for short videos
potentially fix blockchain.db issue

go.mod

@@ -33,6 +33,8 @@ require (
 	github.com/stretchr/testify v1.4.0
 	google.golang.org/appengine v1.6.5 // indirect
 	gopkg.in/ini.v1 v1.60.2 // indirect
+	gopkg.in/vansante/go-ffprobe.v2 v2.0.2
+	gotest.tools v2.2.0+incompatible
 )
 
 go 1.13

go.sum

@@ -593,6 +593,8 @@ gopkg.in/nullbio/null.v6 v6.0.0-20161116030900-40264a2e6b79 h1:FpCr9V8wuOei4BAen
 gopkg.in/nullbio/null.v6 v6.0.0-20161116030900-40264a2e6b79/go.mod h1:gWkaRU7CoXpezCBWfWjm3999QqS+1pYPXGbqQCTMzo8=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
+gopkg.in/vansante/go-ffprobe.v2 v2.0.2 h1:DdxSfFnlqeawPIVbIQEI6LR6OQHQNR7tNgWb2mWuC4w=
+gopkg.in/vansante/go-ffprobe.v2 v2.0.2/go.mod h1:qF0AlAjk7Nqzqf3y333Ly+KxN3cKF2JqA3JT5ZheUGE=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.4 h1:/eiJrUcujPVeJ3xlSWaiNi3uSVmDGBK1pDHUHAnao1I=

main.go

@@ -58,7 +58,7 @@ func main() {
 	cmd.Flags().Int64Var(&cliFlags.SyncFrom, "after", time.Unix(0, 0).Unix(), "Specify from when to pull jobs [Unix time](Default: 0)")
 	cmd.Flags().Int64Var(&cliFlags.SyncUntil, "before", time.Now().AddDate(1, 0, 0).Unix(), "Specify until when to pull jobs [Unix time](Default: current Unix time)")
 	cmd.Flags().IntVar(&cliFlags.ConcurrentJobs, "concurrent-jobs", 1, "how many jobs to process concurrently")
-	cmd.Flags().IntVar(&cliFlags.VideosLimit, "videos-limit", 1000, "how many videos to process per channel")
+	cmd.Flags().IntVar(&cliFlags.VideosLimit, "videos-limit", 0, "how many videos to process per channel (leave 0 for automatic detection)")
 	cmd.Flags().IntVar(&cliFlags.MaxVideoSize, "max-size", 2048, "Maximum video size to process (in MB)")
 	cmd.Flags().IntVar(&maxVideoLength, "max-length", 2, "Maximum video length to process (in hours)")
 

manager/setup.go

@@ -99,8 +99,8 @@ func (s *Sync) walletSetup() error {
 
 	log.Debugf("We already allocated credits for %d published videos and %d failed videos", publishedCount, failedCount)
 
-	if videosOnYoutube > s.Manager.CliFlags.VideosLimit {
+	if videosOnYoutube > s.Manager.CliFlags.VideosToSync(s.DbChannelData.TotalSubscribers) {
-		videosOnYoutube = s.Manager.CliFlags.VideosLimit
+		videosOnYoutube = s.Manager.CliFlags.VideosToSync(s.DbChannelData.TotalSubscribers)
 	}
 	unallocatedVideos := videosOnYoutube - (publishedCount + failedCount)
 	channelFee := channelClaimAmount

manager/ytsync.go

@@ -894,7 +894,7 @@ func (s *Sync) enqueueYoutubeVideos() error {
 		return err
 	}
 
-	videos, err := ytapi.GetVideosToSync(s.Manager.ApiConfig, s.DbChannelData.ChannelId, s.syncedVideos, s.Manager.CliFlags.QuickSync, s.Manager.CliFlags.VideosLimit, ytapi.VideoParams{
+	videos, err := ytapi.GetVideosToSync(s.Manager.ApiConfig, s.DbChannelData.ChannelId, s.syncedVideos, s.Manager.CliFlags.QuickSync, s.Manager.CliFlags.VideosToSync(s.DbChannelData.TotalSubscribers), ytapi.VideoParams{
 		VideoDir: s.videoDirectory,
 		S3Config: *s.Manager.AwsConfigs.GetS3AWSConfig(),
 		Stopper:  s.grp,
@@ -962,7 +962,7 @@ func (s *Sync) processVideo(v ytapi.Video) (err error) {
 		return nil
 	}
 
-	if !videoRequiresUpgrade && v.PlaylistPosition() >= s.Manager.CliFlags.VideosLimit {
+	if !videoRequiresUpgrade && v.PlaylistPosition() >= s.Manager.CliFlags.VideosToSync(s.DbChannelData.TotalSubscribers) {
 		log.Println(v.ID() + " is old: skipping")
 		return nil
 	}

shared/shared.go

@@ -65,6 +65,30 @@ type SyncFlags struct {
 	MaxVideoLength          time.Duration
 }
 
+// VideosToSync dynamically figures out how many videos should be synced for a given subs count if nothing was otherwise specified
+func (f *SyncFlags) VideosToSync(totalSubscribers uint) int {
+	if f.VideosLimit > 0 {
+		return f.VideosLimit
+	}
+	defaultVideosToSync := map[int]int{
+		10000: 1000,
+		5000:  500,
+		1000:  400,
+		800:   250,
+		600:   200,
+		200:   80,
+		100:   50,
+		1:     10,
+	}
+	videosToSync := 0
+	for s, r := range defaultVideosToSync {
+		if int(totalSubscribers) >= s && r > videosToSync {
+			videosToSync = r
+		}
+	}
+	return videosToSync
+}
+
 func (f *SyncFlags) IsSingleChannelSync() bool {
 	return f.ChannelID != ""
 }

shared/shared_test.go

@@ -0,0 +1,20 @@
+package shared
+
+import (
+	"testing"
+
+	"gotest.tools/assert"
+)
+
+func TestSyncFlags_VideosToSync(t *testing.T) {
+	f := SyncFlags{}
+	assert.Equal(t, f.VideosToSync(0), 0)
+	assert.Equal(t, f.VideosToSync(1), 10)
+	assert.Equal(t, f.VideosToSync(5), 10)
+	assert.Equal(t, f.VideosToSync(10), 10)
+	assert.Equal(t, f.VideosToSync(101), 50)
+	assert.Equal(t, f.VideosToSync(500), 80)
+	assert.Equal(t, f.VideosToSync(21000), 1000)
+	f.VideosLimit = 1337
+	assert.Equal(t, f.VideosToSync(21), 1337)
+}

sources/youtubeVideo.go

@@ -1,6 +1,7 @@
 package sources
 
 import (
+	"context"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -15,6 +16,7 @@ import (
 	"github.com/abadojack/whatlanggo"
 	"github.com/lbryio/ytsync/v5/downloader/ytdl"
 	"github.com/lbryio/ytsync/v5/shared"
+	"gopkg.in/vansante/go-ffprobe.v2"
 
 	"github.com/lbryio/ytsync/v5/ip_manager"
 	"github.com/lbryio/ytsync/v5/namer"
@@ -492,7 +494,17 @@ func (v *YoutubeVideo) downloadAndPublish(daemon *jsonrpc.Client, params SyncPar
 	}
 
 	log.Debugln("Downloaded " + v.id)
+	ctx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancelFn()
 
+	data, err := ffprobe.ProbeURL(ctx, v.getFullPath())
+	if err != nil {
+		log.Errorf("failure in probing downloaded video: %s", err.Error())
+	} else {
+		if data.Format.Duration() < minDuration {
+			return nil, errors.Err("video is too short to process")
+		}
+	}
 	err = v.triggerThumbnailSave()
 	if err != nil {
 		return nil, errors.Prefix("thumbnail error", err)

util/util.go

@@ -248,25 +248,15 @@ func CleanupLbrynet() error {
 	}
 	lbryumDir = lbryumDir + ledger
 
-	db, err := os.Stat(lbryumDir + "/blockchain.db")
+	files, err = filepath.Glob(lbryumDir + "/blockchain.db*")
 	if err != nil {
-		if os.IsNotExist(err) {
-			return nil
-		}
 		return errors.Err(err)
 	}
-	dbSizeLimit := int64(2 * 1024 * 1024 * 1024)
+	for _, f := range files {
-	if db.Size() > dbSizeLimit {
+		err = os.Remove(f)
-		files, err := filepath.Glob(lbryumDir + "/blockchain.db*")
 		if err != nil {
 			return errors.Err(err)
 		}
-		for _, f := range files {
-			err = os.Remove(f)
-			if err != nil {
-				return errors.Err(err)
-			}
-		}
 	}
 	return nil
 }