hpz937/zurg
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix rewrites

Browse Source
This commit is contained in:
Ben Sarmiento
2024-05-22 02:26:20 +02:00
parent a9cc689702
commit 9990bf90ca
10 changed files with 149 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
internal/dav/delete.go
View File

@@ -37,8 +37,7 @@ func HandleDeleteFile(directory, torrentName, fileName string, torMgr *torrent.T
if dirCfg.OnlyShowTheBiggestFile {
torMgr.Delete(torrentName, true)
} else {
err := file.State.Event(context.Background(), "delete_file")
if err != nil {
if err := file.State.Event(context.Background(), "delete_file"); err != nil {
return fmt.Errorf("cannot delete file %s: %v", fileName, err)
}
if torMgr.CheckDeletedStatus(torrent) {

2
internal/torrent/fixer.go
View File

@@ -66,7 +66,7 @@ func (t *TorrentManager) processFixers(instances []realdebrid.Torrent) {
}
for _, torrent := range toRedownload {
t.redownloadTorrent(torrent, "")
t.redownloadTorrent(torrent, []string{})
}
t.writeFixersToFile()

15
internal/torrent/latestState.go
View File

@@ -1,16 +1,29 @@
package torrent
import "github.com/debridmediamanager/zurg/pkg/logutil"
type LibraryState struct {
TotalCount int
ActiveCount int
FirstTorrentId string
log *logutil.Logger
}
func (ls *LibraryState) Eq(a LibraryState) bool {
if ls.TotalCount == 0 || ls.FirstTorrentId == "" {
ls.log.Debugf("Checksum is empty")
return false
} else if a.TotalCount != ls.TotalCount {
ls.log.Debugf("Checksum total count mismatch: %d != %d", a.TotalCount, ls.TotalCount)
return false
} else if a.ActiveCount != ls.ActiveCount {
ls.log.Debugf("Checksum active count mismatch: %d != %d", a.ActiveCount, ls.ActiveCount)
return false
} else if a.FirstTorrentId != ls.FirstTorrentId {
ls.log.Debugf("Checksum first torrent id mismatch: %s != %s", a.FirstTorrentId, ls.FirstTorrentId)
return false
}
return a.TotalCount == ls.TotalCount && a.ActiveCount == ls.ActiveCount && a.FirstTorrentId == ls.FirstTorrentId
return true
}
func (t *TorrentManager) setNewLatestState(checksum LibraryState) {

12
internal/torrent/manager.go
View File

@@ -68,7 +68,7 @@ func NewTorrentManager(cfg config.ConfigInterface, api *realdebrid.RealDebrid, w
RepairKillSwitch: make(chan struct{}, 1),
RemountTrigger: make(chan struct{}, 1),
latestState: &LibraryState{},
latestState: &LibraryState{log: log},
}
t.fixers = t.readFixersFromFile()
@@ -200,10 +200,7 @@ func (t *TorrentManager) readTorrentFromFile(hash string) *Torrent {
func (t *TorrentManager) deleteTorrentFile(hash string) {
filePath := "data/" + hash + ".torrent_zurg"
err := os.Remove(filePath)
if err != nil {
t.log.Warnf("Cannot delete file %s: %v", filePath, err)
}
_ = os.Remove(filePath)
}
/// end torrent functions
@@ -265,10 +262,7 @@ func (t *TorrentManager) readInfoFromFile(torrentID string) *realdebrid.TorrentI
func (t *TorrentManager) deleteInfoFile(torrentID string) {
filePath := "data/" + torrentID + ".info_zurg"
err := os.Remove(filePath)
if err != nil {
t.log.Warnf("Cannot delete info file %s: %v", filePath, err)
}
_ = os.Remove(filePath)
}
/// end info functions

32
internal/torrent/refresh.go
View File

@@ -220,15 +220,13 @@ func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
// all links are still intact! good!
for i, file := range selectedFiles {
file.Link = info.Links[i]
err := file.State.Event(context.Background(), "repair_file")
if err != nil {
t.log.Warnf("Cannot repair file %s: %v", file.Path, err)
if err := file.State.Event(context.Background(), "repair_file"); err != nil {
t.log.Errorf("Cannot repair file %s: %v", file.Path, err)
}
}
torrent.UnassignedLinks = mapset.NewSet[string]()
err := torrent.State.Event(context.Background(), "repair_torrent")
if err != nil {
t.log.Warnf("Cannot repair torrent %s: %v", torrent.Hash, err)
if err := torrent.State.Event(context.Background(), "mark_as_repaired"); err != nil {
t.log.Errorf("Cannot repair torrent %s: %v", torrent.Hash, err)
}
} else {
torrent.UnassignedLinks = mapset.NewSet[string](info.Links...)
@@ -298,7 +296,6 @@ func (t *TorrentManager) mergeToMain(existing, toMerge *Torrent) *Torrent {
Added: newer.Added,
Components: mergedComponents,
UnassignedLinks: newer.UnassignedLinks.Union(older.UnassignedLinks),
UnrepairableReason: newer.UnrepairableReason,
State: older.State,
@@ -323,14 +320,29 @@ func (t *TorrentManager) mergeToMain(existing, toMerge *Torrent) *Torrent {
mainTorrent.SelectedFiles.Set(key, olderFile)
} else if olderFile.State.Is("deleted_file") {
newerFile, _ := mainTorrent.SelectedFiles.Get(key)
err := newerFile.State.Event(context.Background(), "delete_file")
if err != nil {
t.log.Warnf("Cannot delete file %s: %v", key, err)
if err := newerFile.State.Event(context.Background(), "delete_file"); err != nil {
t.log.Errorf("Cannot delete file %s: %v", key, err)
}
}
})
t.CheckDeletedStatus(&mainTorrent)
// unassigned links
mainTorrent.UnassignedLinks = mapset.NewSet[string]()
newer.UnassignedLinks.Union(older.UnassignedLinks).Each(func(link string) bool {
found := false
mainTorrent.SelectedFiles.IterCb(func(key string, file *File) {
if !found && file.Link == link {
found = true
return
}
})
if !found {
mainTorrent.UnassignedLinks.Add(link)
}
return false
})
return &mainTorrent
}

170
internal/torrent/repair.go
View File

@@ -95,23 +95,25 @@ func (t *TorrentManager) repairAll(torrent *Torrent) {
toRepair := mapset.NewSet[*Torrent]()
haystack.IterCb(func(_ string, torrent *Torrent) {
if torrent.AnyInProgress() || torrent.AllInProgress() || torrent.UnrepairableReason != "" {
if torrent.UnrepairableReason != "" {
return
}
// check 1: for broken files
brokenFileIDs := mapset.NewSet[int]()
brokenFileCount := 0
torrent.SelectedFiles.IterCb(func(_ string, file *File) {
if file.State.Is("broken_file") {
brokenFileIDs.Add(file.ID)
brokenFileCount++
}
})
if brokenFileIDs.Cardinality() > 0 {
if brokenFileCount > 0 {
t.log.Debugf("Torrent %s has %d broken files, adding to repair list", t.GetKey(torrent), brokenFileCount)
toRepair.Add(torrent)
return
}
// check 2: for unassigned links (this means the torrent has started to deteriorate)
if torrent.UnassignedLinks.Cardinality() > 0 {
t.log.Debugf("Torrent %s has unassigned links, adding to repair list", t.GetKey(torrent))
unassignedCount := torrent.UnassignedLinks.Cardinality()
if unassignedCount > 0 {
t.log.Debugf("Torrent %s has %d unassigned links, adding to repair list", t.GetKey(torrent), unassignedCount)
toRepair.Add(torrent)
return
}
@@ -125,21 +127,10 @@ func (t *TorrentManager) repairAll(torrent *Torrent) {
})
wg.Wait()
t.log.Infof("Finished repairing %d broken torrents", toRepair.Cardinality())
t.log.Infof("Finished repair sequence for %d broken torrents", toRepair.Cardinality())
}
func (t *TorrentManager) Repair(torrent *Torrent, wg *sync.WaitGroup) {
if torrent.UnrepairableReason != "" {
t.log.Warnf("Torrent %s is unfixable (%s), skipping repair", t.GetKey(torrent), torrent.UnrepairableReason)
wg.Done()
return
}
if torrent.AnyInProgress() || torrent.AllInProgress() {
t.log.Infof("Torrent %s is in progress, skipping repair until download is done", t.GetKey(torrent))
wg.Done()
return
}
// blocks for approx 45 minutes if active torrents are full
if !t.canCapacityHandle() {
t.log.Error("Blocked for too long due to limit of active torrents, cannot continue with the repair")
@@ -150,6 +141,10 @@ func (t *TorrentManager) Repair(torrent *Torrent, wg *sync.WaitGroup) {
// assign to a worker
_ = t.workerPool.Submit(func() {
defer wg.Done()
if err := torrent.State.Event(context.Background(), "repair_torrent"); err != nil {
t.log.Errorf("Failed to mark torrent %s as under repair: %v", t.GetKey(torrent), err)
return
}
t.repair(torrent)
})
}
@@ -163,27 +158,36 @@ func (t *TorrentManager) repair(torrent *Torrent) {
// handle torrents with incomplete links for selected files
// torrent can be rar'ed by RD, so we need to check for that
if !t.assignUnassignedLinks(torrent) {
if torrent.UnassignedLinks.Cardinality() > 0 && !t.assignUnassignedLinks(torrent) {
return
}
// get all broken files
brokenFiles, allBroken := getBrokenFiles(torrent)
brokenFileIDs := getFileIDs(brokenFiles)
t.log.Debugf("Torrent %s has %d broken files (ids=%s; total is %d), repairing by redownloading", t.GetKey(torrent), len(brokenFiles), brokenFileIDs, torrent.SelectedFiles.Count())
// first step: redownload the whole torrent
info, err := t.redownloadTorrent(torrent, "") // reinsert the whole torrent, passing ""
if info != nil && info.Progress != 100 {
t.log.Infof("Torrent %s (files=%s) is still in progress after redownloading but it should be repaired once done", t.GetKey(torrent), brokenFileIDs)
t.log.Debugf("Torrent %s has %d broken files (out of %d), repairing by redownloading whole torrent", t.GetKey(torrent), len(brokenFiles), torrent.SelectedFiles.Count())
info, err := t.redownloadTorrent(torrent, []string{}) // reinsert the whole torrent, passing empty selection
if info != nil && info.Progress == 100 && !t.isStillBroken(info, brokenFiles) {
// successful repair
if err := torrent.State.Event(context.Background(), "mark_as_repaired"); err != nil {
t.log.Errorf("Cannot repair torrent %s: %v", torrent.Hash, err)
return
}
t.log.Infof("Successfully repaired torrent %s by redownloading all files", t.GetKey(torrent))
return
} else if info != nil && info.Progress == 100 && !t.isStillBroken(info, brokenFiles) {
t.log.Infof("Successfully repaired torrent %s (files=%s) by redownloading", t.GetKey(torrent), brokenFileIDs)
} else if info != nil && info.Progress != 100 {
t.log.Infof("Torrent %s is still in progress after redownloading but it should be repaired once done", t.GetKey(torrent))
return
}
t.log.Warnf("Cannot repair torrent %s by redownloading all files (error=%s)", t.GetKey(torrent), err.Error())
if err != nil {
t.log.Warnf("Cannot repair torrent %s by redownloading all files (error=%s)", t.GetKey(torrent), err.Error())
} else {
t.log.Warnf("Cannot repair torrent %s by redownloading all files", t.GetKey(torrent))
}
if torrent.UnrepairableReason != "" {
t.log.Debugf("Torrent %s has been marked as unfixable during redownload (%s), ending repair process early", t.GetKey(torrent), torrent.UnrepairableReason)
@@ -193,41 +197,29 @@ func (t *TorrentManager) repair(torrent *Torrent) {
// second step: download just the broken files
if len(brokenFiles) == 1 && allBroken {
// if all files are broken, we can't do anything
t.log.Warnf("Torrent %s has only 1 cached file and it's broken, marking as unfixable", t.GetKey(torrent))
// if all files are broken, we can't do anything!
t.log.Warnf("Torrent %s has only 1 cached file and it's broken; marking as unfixable (to fix, select other files)", t.GetKey(torrent))
t.markAsUnfixable(torrent, "the lone cached file is broken")
return
}
} else if len(brokenFiles) > 1 {
t.log.Infof("Repairing by downloading 2 batches of the %d broken files of torrent %s", len(brokenFiles), t.GetKey(torrent))
t.log.Infof("Repairing by downloading %d batches of the %d broken files of torrent %s", int(math.Ceil(float64(len(brokenFiles))/130)), len(brokenFiles), t.GetKey(torrent))
oldTorrentIDs := []string{}
for id := range torrent.Components {
oldTorrentIDs = append(torrentIDs, id)
}
oldTorrentIDs := []string{}
for id := range torrent.Components {
oldTorrentIDs = append(oldTorrentIDs, id)
}
newlyDownloadedIds := make([]string, 0)
group := make([]*File, 0)
for _, file := range brokenFiles {
group = append(group, file)
if len(group) >= 200 {
brokenFileIDs := getFileIDs(group)
redownloadedInfo, err := t.redownloadTorrent(torrent, brokenFileIDs)
if err != nil {
t.log.Warnf("Cannot repair torrent %s by downloading broken files (error=%s) giving up", t.GetKey(torrent), err.Error())
for _, newId := range newlyDownloadedIds {
t.registerFixer(newId, "download_failed")
}
return
}
newlyDownloadedIds = append(newlyDownloadedIds, redownloadedInfo.ID)
group = make([]*File, 0)
}
}
if len(group) > 0 {
newlyDownloadedIds := make([]string, 0)
group := make([]*File, 0)
batchNum := 1
for _, file := range brokenFiles {
group = append(group, file)
if len(group) >= 130 {
t.log.Debugf("Downloading batch %d of broken files of torrent %s", batchNum, t.GetKey(torrent))
batchNum++
brokenFileIDs := getFileIDs(group)
_, err := t.redownloadTorrent(torrent, brokenFileIDs)
redownloadedInfo, err := t.redownloadTorrent(torrent, brokenFileIDs)
if err != nil {
t.log.Warnf("Cannot repair torrent %s by downloading broken files (error=%s) giving up", t.GetKey(torrent), err.Error())
for _, newId := range newlyDownloadedIds {
@@ -235,13 +227,28 @@ func (t *TorrentManager) repair(torrent *Torrent) {
}
return
}
}
for _, oldId := range oldTorrentIDs {
t.registerFixer(oldId, "replaced")
newlyDownloadedIds = append(newlyDownloadedIds, redownloadedInfo.ID)
group = make([]*File, 0)
}
}
t.log.Debugf("Downloading last batch of broken files of torrent %s", t.GetKey(torrent))
if len(group) > 0 {
brokenFileIDs := getFileIDs(group)
_, err := t.redownloadTorrent(torrent, brokenFileIDs)
if err != nil {
t.log.Warnf("Cannot repair torrent %s by downloading broken files (error=%s) giving up", t.GetKey(torrent), err.Error())
for _, newId := range newlyDownloadedIds {
t.registerFixer(newId, "download_failed")
}
return
}
}
for _, oldId := range oldTorrentIDs {
t.registerFixer(oldId, "replaced")
}
}
func (t *TorrentManager) assignUnassignedLinks(torrent *Torrent) bool {
@@ -267,10 +274,9 @@ func (t *TorrentManager) assignUnassignedLinks(torrent *Torrent) bool {
assigned := false
torrent.SelectedFiles.IterCb(func(_ string, file *File) {
// base it on size because why not?
if !assigned && file.State.Is("broken_file") && ((unrestrict.Filesize > 1_000_000 && file.Bytes == unrestrict.Filesize) || strings.HasSuffix(strings.ToLower(file.Path), strings.ToLower(unrestrict.Filename))) {
if !assigned && file.State.Is("broken_file") && file.Bytes == unrestrict.Filesize && strings.HasSuffix(strings.ToLower(file.Path), strings.ToLower(unrestrict.Filename)) {
file.Link = link
err := file.State.Event(context.Background(), "repair_file")
if err != nil {
if err := file.State.Event(context.Background(), "repair_file"); err != nil {
t.log.Errorf("Failed to mark file %s as repaired: %v", file.Path, err)
return
}
@@ -331,6 +337,9 @@ func (t *TorrentManager) assignUnassignedLinks(torrent *Torrent) bool {
torrent.UnassignedLinks = mapset.NewSet[string]()
t.markAsUnfixable(torrent, "rar'ed by RD")
t.markAsUnplayable(torrent, "rar'ed by RD")
if err := torrent.State.Event(context.Background(), "mark_as_repaired"); err != nil {
t.log.Errorf("Cannot repair torrent %s: %v", torrent.Hash, err)
}
}
return false // end repair
}
@@ -338,11 +347,13 @@ func (t *TorrentManager) assignUnassignedLinks(torrent *Torrent) bool {
return true
}
func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (*realdebrid.TorrentInfo, error) {
func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection []string) (*realdebrid.TorrentInfo, error) {
// broken files means broken links
// if brokenFiles is not provided, we will redownload all files
oldTorrentIDs := make([]string, 0)
if selection == "" {
finalSelection := strings.Join(selection, ",")
selectionCount := len(selection)
if selectionCount == 0 {
// only delete the old torrent if we are redownloading all files
for id := range torrent.Components {
oldTorrentIDs = append(oldTorrentIDs, id)
@@ -350,11 +361,12 @@ func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (
tmpSelection := ""
torrent.SelectedFiles.IterCb(func(_ string, file *File) {
tmpSelection += fmt.Sprintf("%d,", file.ID) // select all files
selectionCount++
})
if tmpSelection == "" {
return nil, nil // nothing to repair
} else {
selection = tmpSelection[:len(tmpSelection)-1]
finalSelection = tmpSelection[:len(tmpSelection)-1]
}
}
@@ -380,18 +392,18 @@ func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (
newTorrentID := resp.ID
// sleep for 1 second to let RD process the magnet
time.Sleep(10 * time.Second)
time.Sleep(1 * time.Second)
var info *realdebrid.TorrentInfo
for {
// select files
err = t.api.SelectTorrentFiles(newTorrentID, selection)
err = t.api.SelectTorrentFiles(newTorrentID, finalSelection)
if err != nil {
t.registerFixer(newTorrentID, "download_failed")
return nil, fmt.Errorf("cannot start redownloading: %v", err)
return nil, fmt.Errorf("cannot start redownloading torrent %s (id=%s): %v", t.GetKey(torrent), newTorrentID, err)
}
// sleep for 5 second to let RD process the magnet
time.Sleep(10 * time.Second)
// sleep for 2 second to let RD process the magnet
time.Sleep(2 * time.Second)
// see if the torrent is ready
info, err = t.api.GetTorrentInfo(newTorrentID)
@@ -400,9 +412,14 @@ func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (
return nil, fmt.Errorf("cannot get info on redownloaded torrent %s (id=%s) : %v", t.GetKey(torrent), newTorrentID, err)
}
if info.Status != "magnet_conversion" && info.Status != "waiting_files_selection" {
break
if info.Status == "magnet_conversion" {
time.Sleep(60 * time.Second)
continue
} else if info.Status == "waiting_files_selection" {
time.Sleep(10 * time.Second)
return nil, fmt.Errorf("torrent %s (id=%s) is stuck on waiting_files_selection", t.GetKey(torrent), newTorrentID)
}
break
}
// documented status: magnet_error, magnet_conversion, waiting_files_selection, queued, downloading, downloaded, error, virus, compressing, uploading, dead
@@ -422,7 +439,6 @@ func (t *TorrentManager) redownloadTorrent(torrent *Torrent, selection string) (
}
// check if incorrect number of links
selectionCount := len(strings.Split(selection, ","))
if info.Progress == 100 && len(info.Links) != selectionCount {
t.registerFixer(newTorrentID, "download_failed")
return nil, fmt.Errorf("torrent %s only got %d links but we need %d", t.GetKey(torrent), len(info.Links), selectionCount)
@@ -484,8 +500,7 @@ func (t *TorrentManager) markAsUnplayable(torrent *Torrent, reason string) {
return
}
t.log.Warnf("Marking torrent %s as unplayable - %s", t.GetKey(torrent), reason)
err := torrent.State.Event(context.Background(), "mark_as_unplayable_torrent")
if err != nil {
if err := torrent.State.Event(context.Background(), "mark_as_unplayable_torrent"); err != nil {
t.log.Errorf("Failed to mark torrent %s as unplayable: %v", t.GetKey(torrent), err)
return
}
@@ -537,8 +552,7 @@ func (t *TorrentManager) isStillBroken(info *realdebrid.TorrentInfo, brokenFiles
// all links are still intact! good!
for i, file := range selectedFiles {
file.Link = info.Links[i]
err := file.State.Event(context.Background(), "repair_file")
if err != nil {
if err := file.State.Event(context.Background(), "repair_file"); err != nil {
t.log.Errorf("Failed to mark file %s as repaired: %v", file.Path, err)
return true
}

3
internal/torrent/states.go
View File

@@ -12,7 +12,8 @@ func NewTorrentState(initial string) *fsm.FSM {
initial,
fsm.Events{
{Name: "break_torrent", Src: []string{"ok_torrent", "unplayable_torrent"}, Dst: "broken_torrent"},
{Name: "repair_torrent", Src: []string{"broken_torrent"}, Dst: "ok_torrent"},
{Name: "repair_torrent", Src: []string{"broken_torrent"}, Dst: "under_repair_torrent"},
{Name: "mark_as_repaired", Src: []string{"broken_torrent", "under_repair_torrent"}, Dst: "ok_torrent"},
{Name: "mark_as_unplayable_torrent", Src: []string{"ok_torrent"}, Dst: "unplayable_torrent"},
},
fsm.Callbacks{},

15
internal/torrent/util.go
View File

@@ -2,15 +2,20 @@ package torrent
import (
"fmt"
"strings"
"sort"
)
func getFileIDs(files []*File) string {
var fileIDs []string
func getFileIDs(files []*File) []string {
var fileIDs []int
for _, file := range files {
if file.ID != 0 {
fileIDs = append(fileIDs, fmt.Sprintf("%d", file.ID))
fileIDs = append(fileIDs, file.ID)
}
}
return strings.Join(fileIDs, ",")
sort.Ints(fileIDs)
var ret []string
for _, id := range fileIDs {
ret = append(ret, fmt.Sprintf("%d", id))
}
return ret
}

9
internal/universal/downloader.go
View File

@@ -63,8 +63,7 @@ func (dl *Downloader) DownloadFile(
unrestrict := torMgr.UnrestrictFileUntilOk(file)
if unrestrict == nil {
err := file.State.Event(context.Background(), "break_file")
if err != nil {
if err := file.State.Event(context.Background(), "break_file"); err != nil {
log.Errorf("File %s is stale: %v", fileName, err)
http.Error(resp, "File is stale, please try again", http.StatusLocked)
return
@@ -154,8 +153,7 @@ func (dl *Downloader) streamFileToResponse(
downloadResp, err := dl.client.Do(dlReq)
if err != nil {
if file != nil && unrestrict.Streamable == 1 {
err := file.State.Event(context.Background(), "break_file")
if err != nil {
if err := file.State.Event(context.Background(), "break_file"); err != nil {
log.Errorf("File %s is stale: %v", file.Path, err)
http.Error(resp, "File is stale, please try again", http.StatusLocked)
return
@@ -177,8 +175,7 @@ func (dl *Downloader) streamFileToResponse(
// Check if the download was not successful
if downloadResp.StatusCode/100 != 2 {
if file != nil && unrestrict.Streamable == 1 {
err := file.State.Event(context.Background(), "break_file")
if err != nil {
if err := file.State.Event(context.Background(), "break_file"); err != nil {
log.Errorf("File %s is stale: %v", file.Path, err)
http.Error(resp, "File is stale, please try again", http.StatusLocked)
return

2
pkg/realdebrid/api.go
View File

@@ -266,7 +266,7 @@ func (rd *RealDebrid) SelectTorrentFiles(id string, files string) error {
}
defer resp.Body.Close()
rd.log.Debugf("Selected %d files and started the download for torrent id=%s", len(strings.Split(files, ",")), id)
rd.log.Debugf("Selected %d files and started the download for torrent id=%s (status code: %d)", len(strings.Split(files, ",")), id, resp.StatusCode)
return nil
}