package torrent import ( "context" "fmt" "path/filepath" "strings" "sync" "time" "github.com/debridmediamanager/zurg/internal/config" "github.com/debridmediamanager/zurg/pkg/realdebrid" "github.com/debridmediamanager/zurg/pkg/utils" mapset "github.com/deckarep/golang-set/v2" cmap "github.com/orcaman/concurrent-map/v2" ) func inProgressStatus(status string) bool { return status == "downloading" || status == "uploading" || status == "queued" || status == "compressing" } func (t *TorrentManager) refreshTorrents() []string { t.inProgressHashes = mapset.NewSet[string]() instances, _, err := t.api.GetTorrents(false) if err != nil { t.log.Warnf("Cannot get torrents: %v", err) return nil } var wg sync.WaitGroup var mergeChan = make(chan *Torrent, len(instances)) updatedPaths := mapset.NewSet[string]() allTorrents, _ := t.DirectoryMap.Get(INT_ALL) freshIDs := mapset.NewSet[string]() freshAccessKeys := mapset.NewSet[string]() for i := range instances { freshIDs.Add(instances[i].ID) wg.Add(1) idx := i _ = t.workerPool.Submit(func() { defer wg.Done() if t.binImmediately(instances[idx].ID) || t.binOnceDoneErrorCheck(instances[idx].ID, instances[idx].Status) || instances[idx].Progress != 100 { if inProgressStatus(instances[idx].Status) { t.inProgressHashes.Add(instances[idx].Hash) } mergeChan <- nil return } 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)) }) t.writeTorrentToFile(torrent) } else if !mainTorrent.DownloadedIDs.Contains(tInfo.ID) { forMerging = torrent } mergeChan <- forMerging }) } t.workerPool.Submit(func() { t.cleanupBins(freshIDs) }) wg.Wait() close(mergeChan) t.log.Infof("Compiling %d torrents", len(instances)) for torrent := range mergeChan { if torrent == nil { continue } accessKey := t.GetKey(torrent) 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) 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)) }) } // removed torrents oldPlusNewKeys := mapset.NewSet[string](allTorrents.Keys()...) oldPlusNewKeys.Difference(freshAccessKeys).Each(func(accessKey string) bool { t.Delete(accessKey, false) return false }) t.log.Infof("Compiled into %d unique torrents", allTorrents.Count()) t.workerPool.Submit(func() { // delete info files that are no longer present t.getInfoFiles().Each(func(path string) bool { path = filepath.Base(path) torrentID := strings.TrimSuffix(path, ".zurginfo") // if binOnceDone returns true, it means the info file is deleted // if false, then we check if it's one of the torrents we just fetched // if not, then we delete the info file if !t.binOnceDone(torrentID) && !freshIDs.Contains(torrentID) { t.deleteInfoFile(torrentID) } return false }) }) t.workerPool.Submit(func() { // update DownloadedIDs field of torrents allTorrents.IterCb(func(accessKey string, torrent *Torrent) { deletedIDs := torrent.DownloadedIDs.Difference(freshIDs) if deletedIDs.Cardinality() > 0 { deletedIDs.Each(func(id string) bool { torrent.DownloadedIDs.Remove(id) return false }) t.writeTorrentToFile(torrent) } }) }) t.workerPool.Submit(func() { allTorrents.IterCb(func(accessKey string, torrent *Torrent) { if torrent.UnassignedLinks.Cardinality() > 0 { t.assignLinks(torrent) } }) }) return updatedPaths.ToSlice() } // StartRefreshJob periodically refreshes the torrents func (t *TorrentManager) StartRefreshJob() { go func() { t.log.Debug("Starting periodic refresh job") refreshTicker := time.NewTicker(time.Duration(t.Config.GetRefreshEverySecs()) * time.Second) defer refreshTicker.Stop() for { select { case <-refreshTicker.C: checksum := t.getCurrentState() if t.latestState.Eq(checksum) { continue } t.setNewLatestState(checksum) updatedPaths := t.refreshTorrents() t.log.Info("Finished refreshing torrents") t.TriggerHookOnLibraryUpdate(updatedPaths) case <-t.RefreshKillSwitch: t.log.Info("Stopping periodic refresh job") return } } }() } func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *realdebrid.TorrentInfo { info := t.readInfoFromFile(rdTorrent.ID) if info == nil { var err error info, err = t.api.GetTorrentInfo(rdTorrent.ID) if err != nil { t.log.Warnf("Cannot get info for id=%s: %v", rdTorrent.ID, err) return nil } t.writeInfoToFile(info) } return info } func (t *TorrentManager) convertToTorrent(info *realdebrid.TorrentInfo) *Torrent { torrent := t.readTorrentFromFile("data/" + info.Hash + ".zurgtorrent") if torrent != nil && torrent.DownloadedIDs.Contains(info.ID) { return torrent } torrent = &Torrent{ Name: info.Name, OriginalName: info.OriginalName, Added: info.Added, Hash: info.Hash, State: NewTorrentState("broken_torrent"), DownloadedIDs: mapset.NewSet[string](), } // 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 } selectedFiles = append(selectedFiles, &File{ File: file, Ended: info.Ended, Link: "", // no link yet, consider it broken State: NewFileState("broken_file"), }) } if len(selectedFiles) == len(info.Links) { // 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] } file.State.Event(context.Background(), "repair_file") } torrent.UnassignedLinks = mapset.NewSet[string]() torrent.State.Event(context.Background(), "mark_as_repaired") } else { 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 { baseFilename := t.GetPath(file) // todo better handling of duplicate filenames 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(baseFilename, file) } } torrent.DownloadedIDs.Add(info.ID) return torrent } func (t *TorrentManager) mergeTorrents(existing, toMerge *Torrent) *Torrent { var newer, older *Torrent if existing.Added < toMerge.Added { newer = toMerge older = existing } else { newer = existing older = toMerge } // base of the merged torrent mergedTorrent := &Torrent{ Name: older.Name, OriginalName: older.OriginalName, Rename: older.Rename, Hash: older.Hash, Added: older.Added, DownloadedIDs: older.DownloadedIDs.Union(newer.DownloadedIDs), State: older.State, } // unrepairable reason reasons := mapset.NewSet[string]() reasons.Add(older.UnrepairableReason) reasons.Add(newer.UnrepairableReason) mergedTorrent.UnrepairableReason = strings.Join(reasons.ToSlice(), ", ") // selected files mergedTorrent.SelectedFiles = cmap.New[*File]() newer.SelectedFiles.IterCb(func(key string, file *File) { mergedTorrent.SelectedFiles.SetIfAbsent(key, file) }) older.SelectedFiles.IterCb(func(key string, file *File) { mediaInfo := file.MediaInfo f, ok := mergedTorrent.SelectedFiles.Get(key) if !ok || !f.State.Is("ok_file") { mergedTorrent.SelectedFiles.Set(key, file) } if ok && f.MediaInfo == nil && mediaInfo != nil { f.MediaInfo = mediaInfo } }) // unassigned links mergedTorrent.UnassignedLinks = mapset.NewSet[string]() links := newer.UnassignedLinks.Union(older.UnassignedLinks) links.Each(func(link string) bool { found := false mergedTorrent.SelectedFiles.IterCb(func(key string, file *File) { if !found && file.Link == link { found = true } }) if !found { mergedTorrent.UnassignedLinks.Add(link) } return false }) brokenCount := 0 okCount := 0 mergedTorrent.SelectedFiles.IterCb(func(key string, file *File) { if file.State.Is("broken_file") { brokenCount++ } else if file.State.Is("ok_file") { okCount++ } }) if brokenCount == 0 && okCount > 0 { mergedTorrent.State.Event(context.Background(), "mark_as_repaired") } return mergedTorrent } func (t *TorrentManager) assignDirectory(tor *Torrent, cb func(string)) { torrentIDs := tor.DownloadedIDs.ToSlice() // get filenames needed for directory conditions var filenames []string var fileSizes []int64 unplayable := true tor.SelectedFiles.IterCb(func(key string, file *File) { filenames = append(filenames, filepath.Base(file.Path)) fileSizes = append(fileSizes, file.Bytes) if utils.IsPlayable(file.Path) || t.IsPlayable(file.Path) { unplayable = false } }) if unplayable { t.markAsUnplayable(tor, "no playable files") return } // Map torrents to directories switch t.Config.GetVersion() { case "v1": configV1 := t.Config.(*config.ZurgConfigV1) for _, directories := range configV1.GetGroupMap() { for _, directory := range directories { if t.Config.MeetsConditions(directory, t.GetKey(tor), tor.ComputeTotalSize(), torrentIDs, filenames, fileSizes) { cb(directory) break } } } } } func (t *TorrentManager) IsPlayable(filePath string) bool { filePath = strings.ToLower(filePath) playableExts := t.Config.GetPlayableExtensions() for _, ext := range playableExts { if strings.HasSuffix(filePath, fmt.Sprintf(".%s", ext)) { return true } } return false }