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

Implement autoheal feature

Browse Source
This commit is contained in:
Ben Sarmiento
2023-10-23 20:01:55 +02:00
parent 6298830ea6
commit 21cbb16b88
6 changed files with 218 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
internal/dav/getfile.go
View File

@@ -64,24 +64,20 @@ func HandleGetRequest(w http.ResponseWriter, r *http.Request, t *torrent.Torrent
}
resp := realdebrid.RetryUntilOk(unrestrictFn)
if resp == nil {
// TODO: Readd the file
// when unrestricting fails, it means the file is not available anymore, but still in their database
// if it's the only file, tough luck
// if it's the only file, try to readd it
// delete the old one, add a new one
log.Println("Cannot unrestrict link", link, filenameV2)
t.MarkFileAsDeleted(torrent, file)
http.Error(w, "Cannot find file", http.StatusNotFound)
return
}
if resp.Filename != filenameV2 {
// TODO: Redo the logic to handle mismatch
// [SRS] Pokemon S22E01-35 1080p WEBRip AAC 2.0 x264 CC.rar
// Pokemon.S22E24.The.Secret.Princess.DUBBED.1080p.WEBRip.AAC.2.0.x264-SRS.mkv
// Action: schedule a "cleanup" job for the parent torrent
// If the file extension changed, that means it's a different file
actualExt := filepath.Ext(resp.Filename)
expectedExt := filepath.Ext(filenameV2)
if actualExt != expectedExt {
log.Println("File extension mismatch", resp.Filename, filenameV2)
} else {
log.Println("Filename mismatch", resp.Filename, filenameV2)
}
}
cache.Add(requestPath, resp.Download)
http.Redirect(w, r, resp.Download, http.StatusFound)
}

3
internal/dav/response.go
View File

@@ -52,8 +52,7 @@ func createSingleTorrentResponse(basePath string, torrents []torrent.Torrent) (*
for _, torrent := range torrents {
for _, file := range torrent.SelectedFiles {
if file.Link == "" {
// TODO: Fix this file
log.Println("File has no link, skipping", file.Path)
log.Println("File has no link, skipping (repairing links take time)", file.Path)
continue
}

16
internal/http/get.go
View File

@@ -141,24 +141,20 @@ func HandleGetRequest(w http.ResponseWriter, r *http.Request, t *torrent.Torrent
}
resp := realdebrid.RetryUntilOk(unrestrictFn)
if resp == nil {
// TODO: Readd the file
// when unrestricting fails, it means the file is not available anymore, but still in their database
// if it's the only file, tough luck
// if it's the only file, try to readd it
// delete the old one, add a new one
log.Println("Cannot unrestrict link", link, filenameV2)
t.MarkFileAsDeleted(torrent, file)
http.Error(w, "Cannot find file", http.StatusNotFound)
return
}
if resp.Filename != filenameV2 {
// TODO: Redo the logic to handle mismatch
// [SRS] Pokemon S22E01-35 1080p WEBRip AAC 2.0 x264 CC.rar
// Pokemon.S22E24.The.Secret.Princess.DUBBED.1080p.WEBRip.AAC.2.0.x264-SRS.mkv
// Action: schedule a "cleanup" job for the parent torrent
// If the file extension changed, that means it's a different file
actualExt := filepath.Ext(resp.Filename)
expectedExt := filepath.Ext(filenameV2)
if actualExt != expectedExt {
log.Println("File extension mismatch", resp.Filename, filenameV2)
} else {
log.Println("Filename mismatch", resp.Filename, filenameV2)
}
}
cache.Add(requestPath, resp.Download)
http.Redirect(w, r, resp.Download, http.StatusFound)
}

254
internal/torrent/manager.go
View File

@@ -4,6 +4,7 @@ import (
"encoding/gob"
"fmt"
"log"
"math"
"os"
"strings"
"sync"
@@ -91,6 +92,8 @@ func (t *TorrentManager) MarkFileAsDeleted(torrent *Torrent, file *File) {
log.Println("Marking file as deleted", file.Path)
file.Link = ""
t.writeToFile(torrent.ID, torrent)
log.Println("Healing a single file in the torrent", torrent.Name)
t.heal(torrent.ID, []File{*file})
}
// GetInfo returns the info for a torrent
@@ -114,7 +117,7 @@ func (t *TorrentManager) getChecksum() string {
log.Println("Huh, no torrents returned")
return t.checksum
}
return fmt.Sprintf("%d-%s", totalCount, torrents[0].ID)
return fmt.Sprintf("%d-%s-%v", totalCount, torrents[0].ID, torrents[0].Progress == 100)
}
// refreshTorrents periodically refreshes the torrents
@@ -221,10 +224,12 @@ func (t *TorrentManager) getInfo(torrentID string) *Torrent {
if torrentFromFile != nil {
torrent := t.getByID(torrentID)
if torrent != nil {
if len(torrentFromFile.SelectedFiles) == len(torrent.Links) {
torrent.SelectedFiles = torrentFromFile.SelectedFiles
}
return torrent
}
}
}
log.Println("Getting info for", torrentID)
info, err := realdebrid.GetTorrentInfo(t.config.GetToken(), torrentID)
if err != nil {
@@ -242,62 +247,9 @@ func (t *TorrentManager) getInfo(torrentID string) *Torrent {
})
}
if len(selectedFiles) != len(info.Links) {
// TODO: This means some files have expired
// we need to 'fix' this torrent then, at least the missing selected files
log.Println("Some links has expired for", info.Name)
type Result struct {
Response *realdebrid.UnrestrictResponse
}
resultsChan := make(chan Result, len(info.Links))
var wg sync.WaitGroup
// Limit concurrency
sem := make(chan struct{}, t.config.GetNumOfWorkers())
for _, link := range info.Links {
wg.Add(1)
sem <- struct{}{} // Acquire semaphore
go func(lnk string) {
defer wg.Done()
defer func() { <-sem }() // Release semaphore
unrestrictFn := func() (*realdebrid.UnrestrictResponse, error) {
return realdebrid.UnrestrictCheck(t.config.GetToken(), lnk)
}
resp := realdebrid.RetryUntilOk(unrestrictFn)
if resp != nil {
resultsChan <- Result{Response: resp}
}
}(link)
}
go func() {
wg.Wait()
close(sem)
close(resultsChan)
}()
for result := range resultsChan {
found := false
for i := range selectedFiles {
if strings.HasSuffix(selectedFiles[i].Path, result.Response.Filename) {
selectedFiles[i].Link = result.Response.Link
found = true
}
}
if !found {
selectedFiles = append(selectedFiles, File{
File: realdebrid.File{
Path: result.Response.Filename,
Bytes: result.Response.Filesize,
Selected: 1,
},
Link: result.Response.Link,
})
}
}
selectedFiles = t.organizeChaos(info, selectedFiles)
t.heal(torrentID, selectedFiles)
} else {
for i, link := range info.Links {
selectedFiles[i].Link = link
@@ -365,3 +317,191 @@ func (t *TorrentManager) readFromFile(torrentID string) *Torrent {
return &torrent
}
func (t *TorrentManager) reinsertTorrent(oldTorrentID string, missingFiles string, deleteIfFailed bool) bool {
torrent := t.GetInfo(oldTorrentID)
if torrent == nil {
return false
}
if missingFiles == "" {
var selection string
for _, file := range torrent.SelectedFiles {
if file.Link == "" {
selection += fmt.Sprintf("%d,", file.ID)
}
}
if selection == "" {
return false
}
missingFiles = selection[:len(selection)-1]
}
// reinsert torrent
resp, err := realdebrid.AddMagnetHash(t.config.GetToken(), torrent.Hash)
if err != nil {
log.Printf("Cannot reinsert torrent: %v\n", err)
return false
}
newTorrentID := resp.ID
err = realdebrid.SelectTorrentFiles(t.config.GetToken(), newTorrentID, missingFiles)
if err != nil {
log.Printf("Cannot select files on reinserted torrent: %v\n", err)
}
if deleteIfFailed {
if err != nil {
realdebrid.DeleteTorrent(t.config.GetToken(), newTorrentID)
return false
}
time.Sleep(1 * time.Second)
// see if the torrent is ready
info, err := realdebrid.GetTorrentInfo(t.config.GetToken(), newTorrentID)
if err != nil {
log.Printf("Cannot get info on reinserted torrent: %v\n", err)
if deleteIfFailed {
realdebrid.DeleteTorrent(t.config.GetToken(), newTorrentID)
}
return false
}
time.Sleep(1 * time.Second)
if info.Progress != 100 {
log.Printf("Torrent is not cached anymore, %d%%\n", info.Progress)
realdebrid.DeleteTorrent(t.config.GetToken(), newTorrentID)
return false
}
if len(info.Links) != len(torrent.SelectedFiles) {
log.Printf("It doesn't fix the problem, got %d but we need %d\n", len(info.Links), len(torrent.SelectedFiles))
realdebrid.DeleteTorrent(t.config.GetToken(), newTorrentID)
return false
}
log.Println("Reinsertion successful, deleting old torrent")
realdebrid.DeleteTorrent(t.config.GetToken(), oldTorrentID)
}
return true
}
func (t *TorrentManager) organizeChaos(info *realdebrid.Torrent, selectedFiles []File) []File {
type Result struct {
Response *realdebrid.UnrestrictResponse
}
resultsChan := make(chan Result, len(info.Links))
var wg sync.WaitGroup
// Limit concurrency
sem := make(chan struct{}, t.config.GetNumOfWorkers())
for _, link := range info.Links {
wg.Add(1)
sem <- struct{}{} // Acquire semaphore
go func(lnk string) {
defer wg.Done()
defer func() { <-sem }() // Release semaphore
unrestrictFn := func() (*realdebrid.UnrestrictResponse, error) {
return realdebrid.UnrestrictCheck(t.config.GetToken(), lnk)
}
resp := realdebrid.RetryUntilOk(unrestrictFn)
if resp != nil {
resultsChan <- Result{Response: resp}
}
}(link)
}
go func() {
wg.Wait()
close(sem)
close(resultsChan)
}()
for result := range resultsChan {
found := false
for i := range selectedFiles {
if strings.HasSuffix(selectedFiles[i].Path, result.Response.Filename) {
selectedFiles[i].Link = result.Response.Link
found = true
}
}
if !found {
selectedFiles = append(selectedFiles, File{
File: realdebrid.File{
Path: result.Response.Filename,
Bytes: result.Response.Filesize,
Selected: 1,
},
Link: result.Response.Link,
})
}
}
return selectedFiles
}
func (t *TorrentManager) heal(torrentID string, selectedFiles []File) {
// max waiting time is 45 minutes
const maxRetries = 50
const baseDelay = 1 * time.Second
const maxDelay = 60 * time.Second
retryCount := 0
for {
count, err := realdebrid.GetActiveTorrentCount(t.config.GetToken())
if err != nil {
log.Printf("Cannot get active torrent count: %v\n", err)
if retryCount >= maxRetries {
log.Println("Max retries reached. Exiting.")
return
}
delay := time.Duration(math.Pow(2, float64(retryCount))) * baseDelay
if delay > maxDelay {
delay = maxDelay
}
time.Sleep(delay)
retryCount++
continue
}
if count.DownloadingCount < count.MaxNumberOfTorrents {
log.Printf("We can still add a new torrent, %d/%d\n", count.DownloadingCount, count.MaxNumberOfTorrents)
break
}
if retryCount >= maxRetries {
log.Println("Max retries reached. Exiting.")
return
}
delay := time.Duration(math.Pow(2, float64(retryCount))) * baseDelay
if delay > maxDelay {
delay = maxDelay
}
time.Sleep(delay)
retryCount++
}
// now we can get the missing files
half := len(selectedFiles) / 2
missingFiles1 := getMissingFiles(0, half, selectedFiles)
missingFiles2 := getMissingFiles(half, len(selectedFiles), selectedFiles)
// first solution: add the same selection, maybe it can be fixed by reinsertion?
success := t.reinsertTorrent(torrentID, "", true)
if !success {
// if not, last resort: add only the missing files and do it in 2 batches
t.reinsertTorrent(torrentID, missingFiles1, false)
t.reinsertTorrent(torrentID, missingFiles2, false)
}
}
func getMissingFiles(start, end int, files []File) string {
var missingFiles string
for i := start; i < end; i++ {
if files[i].File.Selected == 1 && files[i].ID != 0 && files[i].Link == "" {
missingFiles += fmt.Sprintf("%d,", files[i].ID)
}
}
if len(missingFiles) > 0 {
missingFiles = missingFiles[:len(missingFiles)-1]
}
return missingFiles
}

8
pkg/realdebrid/api.go
View File

@@ -104,6 +104,7 @@ func GetTorrents(accessToken string, customLimit int) ([]Torrent, int, error) {
params := url.Values{}
params.Set("page", fmt.Sprintf("%d", page))
params.Set("limit", fmt.Sprintf("%d", limit))
params.Set("filter", "active")
reqURL := baseURL + "?" + params.Encode()
@@ -264,12 +265,11 @@ func DeleteTorrent(accessToken string, id string) error {
}
}
// AddMagnet adds a magnet link to download.
func AddMagnet(accessToken, magnet, host string) (*MagnetResponse, error) {
// AddMagnetHash adds a magnet link to download.
func AddMagnetHash(accessToken, magnet string) (*MagnetResponse, error) {
// Prepare request data
data := url.Values{}
data.Set("magnet", magnet)
data.Set("host", host)
data.Set("magnet", fmt.Sprintf("magnet:?xt=urn:btih:%s", magnet))
// Construct request URL
reqURL := "https://api.real-debrid.com/rest/1.0/torrents/addMagnet"

1
pkg/realdebrid/types.go
View File

@@ -25,6 +25,7 @@ type Torrent struct {
}
type File struct {
ID int `json:"id"`
Path string `json:"path"`
Bytes int64 `json:"bytes"`
Selected int `json:"selected"`