Fixers
This commit is contained in:
Ben Sarmiento
@@ -22,99 +22,112 @@ func (t *TorrentManager) refreshTorrents() []string {
|
||||
return nil
|
||||
}
|
||||
t.log.Infof("Fetched %d torrents", len(instances))
|
||||
torChan := make(chan *Torrent, len(instances))
|
||||
var wg sync.WaitGroup
|
||||
|
||||
var wg sync.WaitGroup
|
||||
var mergeChan = make(chan *Torrent, len(instances))
|
||||
|
||||
updatedPaths := mapset.NewSet[string]()
|
||||
|
||||
allTorrents, _ := t.DirectoryMap.Get(INT_ALL)
|
||||
|
||||
freshHashes := mapset.NewSet[string]()
|
||||
freshIDs := mapset.NewSet[string]()
|
||||
freshAccessKeys := mapset.NewSet[string]()
|
||||
|
||||
counter := 0
|
||||
for i := range instances {
|
||||
if t.trashBin.Contains(instances[i].ID) {
|
||||
t.api.DeleteTorrent(instances[i].ID)
|
||||
t.log.Infof("Skipping trashed torrent %s", instances[i].Name)
|
||||
torChan <- nil
|
||||
}
|
||||
idx := i
|
||||
wg.Add(1)
|
||||
idx := i
|
||||
_ = t.workerPool.Submit(func() {
|
||||
defer wg.Done()
|
||||
torChan <- t.getMoreInfo(instances[idx])
|
||||
if t.trashBin.Contains(instances[idx].ID) {
|
||||
t.api.DeleteTorrent(instances[idx].ID)
|
||||
t.log.Debugf("Skipping trashed torrent %s (id=%s)", instances[idx].Name, instances[idx].ID)
|
||||
counter++
|
||||
mergeChan <- nil
|
||||
return
|
||||
}
|
||||
|
||||
if instances[idx].Progress != 100 {
|
||||
t.log.Debugf("Skipping incomplete torrent %s (id=%s)", instances[idx].Name, instances[idx].ID)
|
||||
counter++
|
||||
mergeChan <- nil
|
||||
return
|
||||
}
|
||||
|
||||
freshHashes.Add(instances[idx].Hash)
|
||||
freshIDs.Add(instances[idx].ID)
|
||||
|
||||
tInfo := t.getMoreInfo(instances[idx])
|
||||
torrent := t.convertToTorrent(tInfo)
|
||||
accessKey := t.GetKey(torrent)
|
||||
freshAccessKeys.Add(accessKey)
|
||||
|
||||
var forMerging *Torrent
|
||||
mainTorrent, exists := allTorrents.Get(accessKey)
|
||||
if !exists {
|
||||
allTorrents.Set(accessKey, torrent)
|
||||
|
||||
t.assignDirectory(torrent, func(directory string) {
|
||||
listing, _ := t.DirectoryMap.Get(directory)
|
||||
listing.Set(accessKey, torrent)
|
||||
|
||||
updatedPaths.Add(fmt.Sprintf("%s/%s", directory, accessKey))
|
||||
})
|
||||
} else if !mainTorrent.Components.Has(tInfo.ID) {
|
||||
forMerging = torrent
|
||||
}
|
||||
|
||||
counter++
|
||||
mergeChan <- forMerging
|
||||
})
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
close(torChan)
|
||||
t.log.Infof("Fetched info for %d torrents", len(instances))
|
||||
close(mergeChan)
|
||||
t.log.Infof("Finished fetching info for %d torrents, proceeding to merge", counter)
|
||||
|
||||
var updatedPaths []string
|
||||
noInfoCount := 0
|
||||
|
||||
allTorrents, _ := t.DirectoryMap.Get(INT_ALL)
|
||||
freshAccessKeys := mapset.NewSet[string]()
|
||||
|
||||
allHashes := mapset.NewSet[string]()
|
||||
t.getTorrentFiles().Each(func(path string) bool {
|
||||
path = filepath.Base(path)
|
||||
hash := strings.TrimSuffix(path, ".torrent_zurg")
|
||||
allHashes.Add(hash)
|
||||
return false
|
||||
})
|
||||
freshHashes := mapset.NewSet[string]()
|
||||
|
||||
allIDs := mapset.NewSet[string]()
|
||||
t.getInfoFiles().Each(func(path string) bool {
|
||||
path = filepath.Base(path)
|
||||
torrentID := strings.TrimSuffix(path, ".info_zurg")
|
||||
allIDs.Add(torrentID)
|
||||
return false
|
||||
})
|
||||
freshIDs := mapset.NewSet[string]()
|
||||
|
||||
for torrent := range torChan {
|
||||
for torrent := range mergeChan {
|
||||
if torrent == nil {
|
||||
noInfoCount++
|
||||
continue
|
||||
}
|
||||
|
||||
// there's only 1 component torrent at this point, let's get it
|
||||
var tInfo *realdebrid.TorrentInfo
|
||||
for _, tInfo = range torrent.Components {
|
||||
break
|
||||
}
|
||||
t.log.Debugf("Merging %s", t.GetKey(torrent))
|
||||
|
||||
accessKey := t.GetKey(torrent)
|
||||
freshAccessKeys.Add(accessKey)
|
||||
freshHashes.Add(torrent.Hash)
|
||||
freshIDs.Add(tInfo.ID)
|
||||
|
||||
// update allTorrents
|
||||
isNewID := false
|
||||
mainTorrent, exists := allTorrents.Get(accessKey)
|
||||
if !exists {
|
||||
allTorrents.Set(accessKey, torrent)
|
||||
mainTorrent = torrent
|
||||
isNewID = true
|
||||
} else if _, ok := mainTorrent.Components[tInfo.ID]; !ok {
|
||||
merged := t.mergeToMain(mainTorrent, torrent)
|
||||
allTorrents.Set(accessKey, merged)
|
||||
mainTorrent = merged
|
||||
isNewID = true
|
||||
existing, ok := allTorrents.Get(accessKey)
|
||||
if !ok {
|
||||
t.log.Warnf("Cannot merge %s", accessKey)
|
||||
continue
|
||||
}
|
||||
mainTorrent := t.mergeTorrents(existing, torrent)
|
||||
allTorrents.Set(accessKey, mainTorrent)
|
||||
t.writeTorrentToFile(mainTorrent)
|
||||
|
||||
if isNewID && tInfo.Progress == 100 {
|
||||
// assign to directory
|
||||
t.assignedDirectoryCb(mainTorrent, func(directory string) {
|
||||
listing, _ := t.DirectoryMap.Get(directory)
|
||||
listing.Set(accessKey, mainTorrent)
|
||||
|
||||
updatedPaths = append(updatedPaths, fmt.Sprintf("%s/%s", directory, accessKey))
|
||||
})
|
||||
// write torrent to file
|
||||
if !allHashes.Contains(mainTorrent.Hash) {
|
||||
t.writeTorrentToFile(mainTorrent)
|
||||
t.DirectoryMap.IterCb(func(directory string, torrents cmap.ConcurrentMap[string, *Torrent]) {
|
||||
if strings.HasPrefix(directory, "int__") {
|
||||
return
|
||||
}
|
||||
}
|
||||
torrents.Remove(accessKey)
|
||||
})
|
||||
|
||||
t.assignDirectory(mainTorrent, func(directory string) {
|
||||
listing, _ := t.DirectoryMap.Get(directory)
|
||||
listing.Set(accessKey, mainTorrent)
|
||||
|
||||
updatedPaths.Add(fmt.Sprintf("%s/%s", directory, accessKey))
|
||||
})
|
||||
|
||||
t.log.Debugf("Merging %s done!", t.GetKey(existing))
|
||||
}
|
||||
|
||||
t.log.Infof("Fetched info for %d torrents", len(instances))
|
||||
|
||||
noInfoCount := 0
|
||||
|
||||
// removed torrents
|
||||
allAccessKeys := mapset.NewSet[string](allTorrents.Keys()...)
|
||||
allAccessKeys.Difference(freshAccessKeys).Each(func(accessKey string) bool {
|
||||
oldPlusNewKeys := mapset.NewSet[string](allTorrents.Keys()...)
|
||||
oldPlusNewKeys.Difference(freshAccessKeys).Each(func(accessKey string) bool {
|
||||
t.Delete(accessKey, false)
|
||||
return false
|
||||
})
|
||||
@@ -122,18 +135,32 @@ func (t *TorrentManager) refreshTorrents() []string {
|
||||
t.log.Infof("Compiled into %d torrents, %d were missing info", allTorrents.Count(), noInfoCount)
|
||||
|
||||
// data directory cleanup
|
||||
allHashes.Difference(freshHashes).Each(func(hash string) bool {
|
||||
existingHashes := mapset.NewSet[string]()
|
||||
t.getTorrentFiles().Each(func(path string) bool {
|
||||
path = filepath.Base(path)
|
||||
hash := strings.TrimSuffix(path, ".torrent_zurg")
|
||||
existingHashes.Add(hash)
|
||||
return false
|
||||
})
|
||||
existingHashes.Difference(freshHashes).Each(func(hash string) bool {
|
||||
t.log.Infof("Deleting stale torrent file %s", hash)
|
||||
t.deleteTorrentFile(hash)
|
||||
return false
|
||||
})
|
||||
allIDs.Difference(freshIDs).Each(func(id string) bool {
|
||||
existingIDs := mapset.NewSet[string]()
|
||||
t.getInfoFiles().Each(func(path string) bool {
|
||||
path = filepath.Base(path)
|
||||
torrentID := strings.TrimSuffix(path, ".info_zurg")
|
||||
existingIDs.Add(torrentID)
|
||||
return false
|
||||
})
|
||||
existingIDs.Difference(freshIDs).Each(func(id string) bool {
|
||||
t.log.Infof("Deleting stale info file %s", id)
|
||||
t.deleteInfoFile(id)
|
||||
return false
|
||||
})
|
||||
|
||||
return updatedPaths
|
||||
return updatedPaths.ToSlice()
|
||||
}
|
||||
|
||||
// StartRefreshJob periodically refreshes the torrents
|
||||
@@ -165,16 +192,7 @@ func (t *TorrentManager) StartRefreshJob() {
|
||||
}()
|
||||
}
|
||||
|
||||
// getMoreInfo gets original name, size and files for a torrent
|
||||
func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
|
||||
diskTor := t.readTorrentFromFile(rdTorrent.Hash)
|
||||
if diskTor != nil {
|
||||
if diskInfo, ok := diskTor.Components[rdTorrent.ID]; ok && diskInfo.Progress == 100 {
|
||||
diskTor.Components = map[string]*realdebrid.TorrentInfo{rdTorrent.ID: diskInfo}
|
||||
return diskTor
|
||||
}
|
||||
}
|
||||
|
||||
func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *realdebrid.TorrentInfo {
|
||||
info := t.readInfoFromFile(rdTorrent.ID)
|
||||
if info == nil {
|
||||
var err error
|
||||
@@ -183,25 +201,40 @@ func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
|
||||
t.log.Warnf("Cannot get info for id=%s: %v", rdTorrent.ID, err)
|
||||
return nil
|
||||
}
|
||||
if info.Progress == 100 {
|
||||
t.writeInfoToFile(info)
|
||||
}
|
||||
t.writeInfoToFile(info)
|
||||
}
|
||||
return info
|
||||
}
|
||||
|
||||
func (t *TorrentManager) convertToTorrent(info *realdebrid.TorrentInfo) *Torrent {
|
||||
torrent := t.readTorrentFromFile(info.Hash)
|
||||
if torrent != nil && torrent.Components.Has(info.ID) {
|
||||
return torrent
|
||||
}
|
||||
|
||||
torrent := Torrent{
|
||||
torrent = &Torrent{
|
||||
Name: info.Name,
|
||||
OriginalName: info.OriginalName,
|
||||
Added: info.Added,
|
||||
Hash: info.Hash,
|
||||
State: NewTorrentState("broken_torrent"),
|
||||
Components: cmap.New[*realdebrid.TorrentInfo](),
|
||||
}
|
||||
|
||||
// SelectedFiles is a subset of Files with only the selected ones
|
||||
// it also has a Link field, which can be empty
|
||||
// if it is empty, it means the file is no longer available
|
||||
// Files+Links together are the same as SelectedFiles
|
||||
allFilenames := mapset.NewSet[string]()
|
||||
dupeFilenames := mapset.NewSet[string]()
|
||||
var selectedFiles []*File
|
||||
for _, file := range info.Files {
|
||||
filename := filepath.Base(file.Path)
|
||||
if allFilenames.Contains(filename) {
|
||||
dupeFilenames.Add(filename)
|
||||
} else {
|
||||
allFilenames.Add(filename)
|
||||
}
|
||||
if file.Selected == 0 {
|
||||
continue
|
||||
}
|
||||
@@ -216,6 +249,9 @@ func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
|
||||
// all links are still intact! good!
|
||||
for i, file := range selectedFiles {
|
||||
file.Link = info.Links[i]
|
||||
if strings.HasPrefix(file.Link, "https://real-debrid.com/d/") {
|
||||
file.Link = file.Link[0:39]
|
||||
}
|
||||
if err := file.State.Event(context.Background(), "repair_file"); err != nil {
|
||||
t.log.Errorf("Cannot repair file %s: %v", file.Path, err)
|
||||
}
|
||||
@@ -225,47 +261,34 @@ func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
|
||||
t.log.Errorf("Cannot repair torrent %s: %v", torrent.Hash, err)
|
||||
}
|
||||
} else {
|
||||
torrent.UnassignedLinks = mapset.NewSet[string](info.Links...)
|
||||
torrent.UnassignedLinks = mapset.NewSet[string]()
|
||||
for _, link := range info.Links {
|
||||
if strings.HasPrefix(link, "https://real-debrid.com/d/") {
|
||||
link = link[0:39]
|
||||
}
|
||||
torrent.UnassignedLinks.Add(link)
|
||||
}
|
||||
}
|
||||
torrent.SelectedFiles = cmap.New[*File]()
|
||||
for _, file := range selectedFiles {
|
||||
filename := t.GetPath(file)
|
||||
baseFilename := t.GetPath(file)
|
||||
// todo better handling of duplicate filenames
|
||||
if torrent.SelectedFiles.Has(filename) {
|
||||
oldName := filename
|
||||
ext := filepath.Ext(oldName)
|
||||
noExtension := strings.TrimSuffix(oldName, ext)
|
||||
newName := fmt.Sprintf("%s (%d)%s", noExtension, file.ID, ext)
|
||||
if dupeFilenames.Contains(baseFilename) {
|
||||
extension := filepath.Ext(baseFilename)
|
||||
filenameNoExt := strings.TrimSuffix(baseFilename, extension)
|
||||
newName := fmt.Sprintf("%s (%d)%s", filenameNoExt, file.ID, extension)
|
||||
torrent.SelectedFiles.Set(newName, file)
|
||||
} else {
|
||||
torrent.SelectedFiles.Set(filename, file)
|
||||
torrent.SelectedFiles.Set(baseFilename, file)
|
||||
}
|
||||
}
|
||||
torrent.Components = map[string]*realdebrid.TorrentInfo{rdTorrent.ID: info}
|
||||
|
||||
return &torrent
|
||||
torrent.Components.Set(info.ID, info)
|
||||
|
||||
return torrent
|
||||
}
|
||||
|
||||
// ResetSelectedFiles resets the selected files for a torrent
|
||||
func (t *TorrentManager) ResetSelectedFiles(torrent *Torrent) {
|
||||
// reset selected files
|
||||
newSelectedFiles := cmap.New[*File]()
|
||||
torrent.SelectedFiles.IterCb(func(_ string, file *File) {
|
||||
filename := t.GetPath(file)
|
||||
if newSelectedFiles.Has(filename) {
|
||||
oldName := filename
|
||||
ext := filepath.Ext(oldName)
|
||||
noExtension := strings.TrimSuffix(oldName, ext)
|
||||
newName := fmt.Sprintf("%s (%d)%s", noExtension, file.ID, ext)
|
||||
newSelectedFiles.Set(newName, file)
|
||||
} else {
|
||||
newSelectedFiles.Set(filename, file)
|
||||
}
|
||||
})
|
||||
torrent.SelectedFiles = newSelectedFiles
|
||||
}
|
||||
|
||||
func (t *TorrentManager) mergeToMain(existing, toMerge *Torrent) *Torrent {
|
||||
func (t *TorrentManager) mergeTorrents(existing, toMerge *Torrent) *Torrent {
|
||||
var newer, older *Torrent
|
||||
if existing.Added < toMerge.Added {
|
||||
newer = toMerge
|
||||
@@ -275,21 +298,25 @@ func (t *TorrentManager) mergeToMain(existing, toMerge *Torrent) *Torrent {
|
||||
older = toMerge
|
||||
}
|
||||
|
||||
mergedComponents := map[string]*realdebrid.TorrentInfo{}
|
||||
for k, v := range older.Components {
|
||||
mergedComponents[k] = v
|
||||
}
|
||||
for k, v := range newer.Components {
|
||||
mergedComponents[k] = v
|
||||
}
|
||||
// components
|
||||
numComponents := 0
|
||||
mergedComponents := cmap.New[*realdebrid.TorrentInfo]()
|
||||
older.Components.IterCb(func(torrentID string, info *realdebrid.TorrentInfo) {
|
||||
numComponents++
|
||||
mergedComponents.Set(torrentID, info)
|
||||
})
|
||||
newer.Components.IterCb(func(torrentID string, info *realdebrid.TorrentInfo) {
|
||||
numComponents++
|
||||
mergedComponents.Set(torrentID, info)
|
||||
})
|
||||
|
||||
// build the main torrent
|
||||
mainTorrent := Torrent{
|
||||
Name: newer.Name,
|
||||
OriginalName: newer.OriginalName,
|
||||
Rename: newer.Rename,
|
||||
Hash: newer.Hash,
|
||||
Added: newer.Added,
|
||||
// base of the merged torrent
|
||||
mergedTorrent := Torrent{
|
||||
Name: older.Name,
|
||||
OriginalName: older.OriginalName,
|
||||
Rename: older.Rename,
|
||||
Hash: older.Hash,
|
||||
Added: older.Added,
|
||||
Components: mergedComponents,
|
||||
|
||||
State: older.State,
|
||||
@@ -299,44 +326,58 @@ func (t *TorrentManager) mergeToMain(existing, toMerge *Torrent) *Torrent {
|
||||
reasons := mapset.NewSet[string]()
|
||||
reasons.Add(older.UnrepairableReason)
|
||||
reasons.Add(newer.UnrepairableReason)
|
||||
mainTorrent.UnrepairableReason = strings.Join(reasons.ToSlice(), ", ")
|
||||
mergedTorrent.UnrepairableReason = strings.Join(reasons.ToSlice(), ", ")
|
||||
|
||||
mainTorrent.SelectedFiles = cmap.New[*File]()
|
||||
older.SelectedFiles.IterCb(func(key string, olderFile *File) {
|
||||
mainTorrent.SelectedFiles.Set(key, olderFile)
|
||||
})
|
||||
newer.SelectedFiles.IterCb(func(key string, newerFile *File) {
|
||||
if f, ok := mainTorrent.SelectedFiles.Get(key); ok && f.State.Is("deleted_file") {
|
||||
return
|
||||
// selected files
|
||||
mergedTorrent.SelectedFiles = cmap.New[*File]()
|
||||
newer.SelectedFiles.IterCb(func(key string, file *File) {
|
||||
if f, ok := mergedTorrent.SelectedFiles.Get(key); !ok || !f.State.Is("ok_file") {
|
||||
mergedTorrent.SelectedFiles.Set(key, file)
|
||||
}
|
||||
})
|
||||
older.SelectedFiles.IterCb(func(key string, file *File) {
|
||||
if f, ok := mergedTorrent.SelectedFiles.Get(key); !ok || !f.State.Is("ok_file") {
|
||||
mergedTorrent.SelectedFiles.Set(key, file)
|
||||
}
|
||||
mainTorrent.SelectedFiles.Set(key, newerFile)
|
||||
})
|
||||
t.CheckDeletedStatus(&mainTorrent)
|
||||
|
||||
// unassigned links
|
||||
mainTorrent.UnassignedLinks = mapset.NewSet[string]()
|
||||
newer.UnassignedLinks.Union(older.UnassignedLinks).Each(func(link string) bool {
|
||||
mergedTorrent.UnassignedLinks = mapset.NewSet[string]()
|
||||
links := newer.UnassignedLinks.Union(older.UnassignedLinks)
|
||||
links.Each(func(link string) bool {
|
||||
found := false
|
||||
mainTorrent.SelectedFiles.IterCb(func(key string, file *File) {
|
||||
mergedTorrent.SelectedFiles.IterCb(func(key string, file *File) {
|
||||
if !found && file.Link == link {
|
||||
found = true
|
||||
return
|
||||
}
|
||||
})
|
||||
if !found {
|
||||
mainTorrent.UnassignedLinks.Add(link)
|
||||
mergedTorrent.UnassignedLinks.Add(link)
|
||||
}
|
||||
return false
|
||||
})
|
||||
|
||||
return &mainTorrent
|
||||
brokenCount := 0
|
||||
okCount := 0
|
||||
wtfCount := 0
|
||||
mergedTorrent.SelectedFiles.IterCb(func(key string, file *File) {
|
||||
if file.State.Is("broken_file") {
|
||||
brokenCount++
|
||||
} else if file.State.Is("ok_file") {
|
||||
okCount++
|
||||
} else {
|
||||
wtfCount++
|
||||
}
|
||||
})
|
||||
|
||||
// todo
|
||||
t.log.Debugf("Merging %s (%d comps) - selected files: %d ; unassigned: %d ; broken: %d ; ok %d ; wtf %d", t.GetKey(&mergedTorrent), numComponents, brokenCount+okCount+wtfCount, mergedTorrent.UnassignedLinks.Cardinality(), brokenCount, okCount, wtfCount)
|
||||
|
||||
return &mergedTorrent
|
||||
}
|
||||
|
||||
func (t *TorrentManager) assignedDirectoryCb(tor *Torrent, cb func(string)) {
|
||||
torrentIDs := []string{}
|
||||
for id := range tor.Components {
|
||||
torrentIDs = append(torrentIDs, id)
|
||||
}
|
||||
func (t *TorrentManager) assignDirectory(tor *Torrent, cb func(string)) {
|
||||
torrentIDs := tor.Components.Keys()
|
||||
// get filenames needed for directory conditions
|
||||
var filenames []string
|
||||
var fileSizes []int64
|
||||
|
||||
Reference in New Issue
Block a user