<commit_before>package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"github.com/pschanely/scoredb"
	"log"
	"os"
	"path"
	"runtime"
	"strings"
	"time"
)

func MakeStandardDb(dataDir string, numShards int) (*scoredb.BaseDb, error) {
	var shards []scoredb.StreamingDb

	if scoredb.Exists(dataDir) && scoredb.Exists(path.Join(dataDir, "shard.0")) {
		i := 0
		shards = make([]scoredb.StreamingDb, 0, numShards)
		for {
			shardDir := path.Join(dataDir, fmt.Sprintf("shard.%d", i))
			if scoredb.Exists(shardDir) {
				shards = append(shards, scoredb.BaseStreamingDb{scoredb.NewFsScoreDb(shardDir)})
			} else {
				break
			}
			i += 1
		}
	} else {
		shards = make([]scoredb.StreamingDb, numShards)
		for i := range shards {
			shardDir := path.Join(dataDir, fmt.Sprintf("shard.%d", i))
			shards[i] = scoredb.BaseStreamingDb{scoredb.NewFsScoreDb(shardDir)}
		}
	}
	idDb, err := scoredb.NewBoltIdDb(path.Join(dataDir, "iddb"))
	if err != nil {
		return nil, err
	}
	return &scoredb.BaseDb{
		StreamingDb: scoredb.ShardedDb{
			Shards: shards,
		},
		IdDb: idDb,
	}, nil
}

func watchDir(db *scoredb.MigratableDb, baseDir string, namePrefix string) {
	log.Printf("Watching for databases at %s%s*\n", baseDir, namePrefix)
	var lastName = ""
	for {
		dir, err := os.Open(baseDir)
		var fileInfos []os.FileInfo
		if err == nil {
			fileInfos, err = dir.Readdir(0)
		}
		if err != nil {
			log.Printf("Unable to read %v: %v\n", dir, err)
			time.Sleep(55 * time.Second)
		} else {
			var newDbName = ""
			for _, fileInfo := range fileInfos {
				name := fileInfo.Name()
				if strings.HasPrefix(name, namePrefix) {
					if name > newDbName {
						newDbName = name
					}
				}
			}
			if newDbName > lastName {
				fmt.Printf("Detected database at %s%s\n", baseDir, newDbName)
				fullDbName := path.Join(baseDir, newDbName)
				newDb, err := MakeStandardDb(fullDbName, 1)
				if err != nil {
					log.Printf("Unable to load database at %s%s (%v); ignoring\n", dir, fullDbName, err)
				} else {
					fmt.Printf("The database at %s%s is live\n", baseDir, fullDbName)
					db.Current = newDb
					lastName = newDbName
				}
			}
		}
		time.Sleep(5 * time.Second)
	}
}

func SetupDirLoading(databaseDir string) *scoredb.MigratableDb {
	migratable := scoredb.MigratableDb{Current: nil}
	baseDir, namePrefix := path.Split(databaseDir)<commit_msg>close watch directory fd when done<commit_after>package main

import (
	"bufio"
	"encoding/json"
	"flag"
	"fmt"
	"github.com/pschanely/scoredb"
	"log"
	"os"
	"path"
	"runtime"
	"strings"
	"time"
)

func MakeStandardDb(dataDir string, numShards int) (*scoredb.BaseDb, error) {
	var shards []scoredb.StreamingDb

	if scoredb.Exists(dataDir) && scoredb.Exists(path.Join(dataDir, "shard.0")) {
		i := 0
		shards = make([]scoredb.StreamingDb, 0, numShards)
		for {
			shardDir := path.Join(dataDir, fmt.Sprintf("shard.%d", i))
			if scoredb.Exists(shardDir) {
				shards = append(shards, scoredb.BaseStreamingDb{scoredb.NewFsScoreDb(shardDir)})
			} else {
				break
			}
			i += 1
		}
	} else {
		shards = make([]scoredb.StreamingDb, numShards)
		for i := range shards {
			shardDir := path.Join(dataDir, fmt.Sprintf("shard.%d", i))
			shards[i] = scoredb.BaseStreamingDb{scoredb.NewFsScoreDb(shardDir)}
		}
	}
	idDb, err := scoredb.NewBoltIdDb(path.Join(dataDir, "iddb"))
	if err != nil {
		return nil, err
	}
	return &scoredb.BaseDb{
		StreamingDb: scoredb.ShardedDb{
			Shards: shards,
		},
		IdDb: idDb,
	}, nil
}

func watchDir(db *scoredb.MigratableDb, baseDir string, namePrefix string) {
	log.Printf("Watching for databases at %s%s*\n", baseDir, namePrefix)
	var lastName = ""
	for {
		dir, err := os.Open(baseDir)
		var fileInfos []os.FileInfo
		if err == nil {
			fileInfos, err = dir.Readdir(0)
			dir.Close()
		}
		if err != nil {
			log.Printf("Unable to read %v: %v\n", dir, err)
			time.Sleep(55 * time.Second)
		} else {
			var newDbName = ""
			for _, fileInfo := range fileInfos {
				name := fileInfo.Name()
				if strings.HasPrefix(name, namePrefix) {
					if name > newDbName {
						newDbName = name
					}
				}
			}
			if newDbName > lastName {
				fmt.Printf("Detected database at %s%s\n", baseDir, newDbName)
				fullDbName := path.Join(baseDir, newDbName)
				newDb, err := MakeStandardDb(fullDbName, 1)
				if err != nil {
					log.Printf("Unable to load database at %s%s (%v); ignoring\n", dir, fullDbName, err)
				} else {
					fmt.Printf("The database at %s%s is live\n", baseDir, fullDbName)
					db.Current = newDb
					lastName = newDbName
				}
			}
		}
		time.Sleep(10 * time.Second)
	}
}

func SetupDirLoading(databaseDir string) *scoredb.MigratableDb {
	migratable := scoredb.MigratableDb{Current: nil}
	baseDir, namePrefix := path.Split(databaseDir)
{'fraction_non_alphanumeric': 0.1447285020071836, 'fraction_numerical': 0.004014367208958377, 'mean_word_length': 3.5042816365366316, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7610736', 'n_tokens_mistral': 2091, 'n_tokens_neox': 1848, 'n_words': 393}