johno/decypharr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix mounts; backward compatibility

Browse Source
This commit is contained in:
Mukhtar Akere
2025-02-13 05:07:14 +01:00
parent 6f4f72d781
commit bfd2596367
21 changed files with 71 additions and 466 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/debrid/alldebrid/alldebrid.go
-11
View File
@@ -29,10 +29,6 @@ type AllDebrid struct {
CheckCached bool
}
func (ad *AllDebrid) GetMountPath() string {
return ad.MountPath
}
func (ad *AllDebrid) GetName() string {
return ad.Name
}
@@ -77,7 +73,6 @@ func (ad *AllDebrid) SubmitMagnet(torrent *torrent.Torrent) (*torrent.Torrent, e
}
magnet := magnets[0]
torrentId := strconv.Itoa(magnet.ID)
ad.logger.Info().Msgf("Torrent: %s added with id: %s", torrent.Name, torrentId)
torrent.Id = torrentId
return torrent, nil
@@ -170,12 +165,6 @@ func (ad *AllDebrid) GetTorrent(id string) (*torrent.Torrent, error) {
t.Seeders = data.Seeders
index := -1
files := flattenFiles(data.Files, "", &index)
parentFolder := data.Filename
if data.NbLinks == 1 {
// All debrid doesn't return the parent folder for single file torrents
parentFolder = ""
}
t.OriginalFilename = parentFolder
t.Files = files
}
return t, nil
pkg/debrid/cache/cache.go
Vendored
-360
View File
@@ -1,360 +0,0 @@
package cache
import (
"bufio"
"encoding/json"
"fmt"
"github.com/rs/zerolog"
"github.com/sirrobot01/debrid-blackhole/internal/logger"
"os"
"path/filepath"
"runtime"
"sync"
"time"
"github.com/sirrobot01/debrid-blackhole/internal/config"
"github.com/sirrobot01/debrid-blackhole/pkg/debrid/engine"
"github.com/sirrobot01/debrid-blackhole/pkg/debrid/torrent"
)
type DownloadLinkCache struct {
Link string `json:"download_link"`
}
type CachedTorrent struct {
*torrent.Torrent
LastRead time.Time `json:"last_read"`
IsComplete bool `json:"is_complete"`
DownloadLinks map[string]DownloadLinkCache `json:"download_links"`
}
var (
_logInstance zerolog.Logger
once sync.Once
)
func getLogger() zerolog.Logger {
once.Do(func() {
_logInstance = logger.NewLogger("cache", "info", os.Stdout)
})
return _logInstance
}
type Cache struct {
dir string
client engine.Service
torrents *sync.Map // key: torrent.Id, value: *CachedTorrent
torrentsNames *sync.Map // key: torrent.Name, value: torrent.Id
LastUpdated time.Time `json:"last_updated"`
}
type Manager struct {
caches map[string]*Cache
}
func NewManager(debridService *engine.Engine) *Manager {
cfg := config.GetConfig()
cm := &Manager{
caches: make(map[string]*Cache),
}
for _, debrid := range debridService.GetDebrids() {
c := New(debrid, cfg.Path)
cm.caches[debrid.GetName()] = c
}
return cm
}
func (m *Manager) GetCaches() map[string]*Cache {
return m.caches
}
func (m *Manager) GetCache(debridName string) *Cache {
return m.caches[debridName]
}
func New(debridService engine.Service, basePath string) *Cache {
return &Cache{
dir: filepath.Join(basePath, "cache", debridService.GetName(), "torrents"),
torrents: &sync.Map{},
torrentsNames: &sync.Map{},
client: debridService,
}
}
func (c *Cache) Start() error {
_logger := getLogger()
_logger.Info().Msg("Starting cache for: " + c.client.GetName())
if err := c.Load(); err != nil {
return fmt.Errorf("failed to load cache: %v", err)
}
if err := c.Sync(); err != nil {
return fmt.Errorf("failed to sync cache: %v", err)
}
return nil
}
func (c *Cache) Load() error {
_logger := getLogger()
if err := os.MkdirAll(c.dir, 0755); err != nil {
return fmt.Errorf("failed to create cache directory: %w", err)
}
files, err := os.ReadDir(c.dir)
if err != nil {
return fmt.Errorf("failed to read cache directory: %w", err)
}
for _, file := range files {
if file.IsDir() || filepath.Ext(file.Name()) != ".json" {
continue
}
filePath := filepath.Join(c.dir, file.Name())
data, err := os.ReadFile(filePath)
if err != nil {
_logger.Debug().Err(err).Msgf("Failed to read file: %s", filePath)
continue
}
var ct CachedTorrent
if err := json.Unmarshal(data, &ct); err != nil {
_logger.Debug().Err(err).Msgf("Failed to unmarshal file: %s", filePath)
continue
}
if len(ct.Files) > 0 {
c.torrents.Store(ct.Torrent.Id, &ct)
c.torrentsNames.Store(ct.Torrent.Name, ct.Torrent.Id)
}
}
return nil
}
func (c *Cache) GetTorrent(id string) *CachedTorrent {
if value, ok := c.torrents.Load(id); ok {
return value.(*CachedTorrent)
}
return nil
}
func (c *Cache) GetTorrentByName(name string) *CachedTorrent {
if id, ok := c.torrentsNames.Load(name); ok {
return c.GetTorrent(id.(string))
}
return nil
}
func (c *Cache) SaveTorrent(ct *CachedTorrent) error {
data, err := json.MarshalIndent(ct, "", " ")
if err != nil {
return fmt.Errorf("failed to marshal torrent: %w", err)
}
fileName := ct.Torrent.Id + ".json"
filePath := filepath.Join(c.dir, fileName)
tmpFile := filePath + ".tmp"
f, err := os.Create(tmpFile)
if err != nil {
return fmt.Errorf("failed to create temp file: %w", err)
}
defer f.Close()
w := bufio.NewWriter(f)
if _, err := w.Write(data); err != nil {
return fmt.Errorf("failed to write data: %w", err)
}
if err := w.Flush(); err != nil {
return fmt.Errorf("failed to flush data: %w", err)
}
return os.Rename(tmpFile, filePath)
}
func (c *Cache) SaveAll() error {
const batchSize = 100
var wg sync.WaitGroup
_logger := getLogger()
tasks := make(chan *CachedTorrent, batchSize)
for i := 0; i < runtime.NumCPU(); i++ {
wg.Add(1)
go func() {
defer wg.Done()
for ct := range tasks {
if err := c.SaveTorrent(ct); err != nil {
_logger.Error().Err(err).Msg("failed to save torrent")
}
}
}()
}
c.torrents.Range(func(_, value interface{}) bool {
tasks <- value.(*CachedTorrent)
return true
})
close(tasks)
wg.Wait()
c.LastUpdated = time.Now()
return nil
}
func (c *Cache) Sync() error {
_logger := getLogger()
torrents, err := c.client.GetTorrents()
if err != nil {
return fmt.Errorf("failed to sync torrents: %v", err)
}
workers := runtime.NumCPU() * 200
workChan := make(chan *torrent.Torrent, len(torrents))
errChan := make(chan error, len(torrents))
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for t := range workChan {
if err := c.processTorrent(t); err != nil {
errChan <- err
}
}
}()
}
for _, t := range torrents {
workChan <- t
}
close(workChan)
wg.Wait()
close(errChan)
for err := range errChan {
_logger.Error().Err(err).Msg("sync error")
}
_logger.Info().Msgf("Synced %d torrents", len(torrents))
return nil
}
func (c *Cache) processTorrent(t *torrent.Torrent) error {
if existing, ok := c.torrents.Load(t.Id); ok {
ct := existing.(*CachedTorrent)
if ct.IsComplete {
return nil
}
}
c.AddTorrent(t)
return nil
}
func (c *Cache) AddTorrent(t *torrent.Torrent) {
_logger := getLogger()
if len(t.Files) == 0 {
tNew, err := c.client.GetTorrent(t.Id)
_logger.Debug().Msgf("Getting torrent files for %s", t.Id)
if err != nil {
_logger.Debug().Msgf("Failed to get torrent files for %s: %v", t.Id, err)
return
}
t = tNew
}
if len(t.Files) == 0 {
_logger.Debug().Msgf("No files found for %s", t.Id)
return
}
ct := &CachedTorrent{
Torrent: t,
LastRead: time.Now(),
IsComplete: len(t.Files) > 0,
DownloadLinks: make(map[string]DownloadLinkCache),
}
c.torrents.Store(t.Id, ct)
c.torrentsNames.Store(t.Name, t.Id)
go func() {
if err := c.SaveTorrent(ct); err != nil {
_logger.Debug().Err(err).Msgf("Failed to save torrent %s", t.Id)
}
}()
}
func (c *Cache) RefreshTorrent(torrentId string) *CachedTorrent {
_logger := getLogger()
t, err := c.client.GetTorrent(torrentId)
if err != nil {
_logger.Debug().Msgf("Failed to get torrent files for %s: %v", torrentId, err)
return nil
}
if len(t.Files) == 0 {
return nil
}
ct := &CachedTorrent{
Torrent: t,
LastRead: time.Now(),
IsComplete: len(t.Files) > 0,
DownloadLinks: make(map[string]DownloadLinkCache),
}
c.torrents.Store(t.Id, ct)
c.torrentsNames.Store(t.Name, t.Id)
go func() {
if err := c.SaveTorrent(ct); err != nil {
_logger.Debug().Err(err).Msgf("Failed to save torrent %s", t.Id)
}
}()
return ct
}
func (c *Cache) GetFileDownloadLink(t *CachedTorrent, file *torrent.File) (string, error) {
_logger := getLogger()
if linkCache, ok := t.DownloadLinks[file.Id]; ok {
return linkCache.Link, nil
}
if file.Link == "" {
t = c.RefreshTorrent(t.Id)
if t == nil {
return "", fmt.Errorf("torrent not found")
}
file = t.Torrent.GetFile(file.Id)
}
_logger.Debug().Msgf("Getting download link for %s", t.Name)
link := c.client.GetDownloadLink(t.Torrent, file)
if link == nil {
return "", fmt.Errorf("download link not found")
}
t.DownloadLinks[file.Id] = DownloadLinkCache{
Link: link.DownloadLink,
}
go func() {
if err := c.SaveTorrent(t); err != nil {
_logger.Debug().Err(err).Msgf("Failed to save torrent %s", t.Id)
}
}()
return link.DownloadLink, nil
}
func (c *Cache) GetTorrents() *sync.Map {
return c.torrents
}
pkg/debrid/debrid.go
+1 -1
View File
@@ -81,7 +81,7 @@ func ProcessTorrent(d *engine.Engine, magnet *utils.Magnet, a *arr.Arr, isSymlin
errs = append(errs, err)
continue
}
logger.Info().Msgf("Torrent: %s submitted to %s", dbt.Name, db.GetName())
logger.Info().Msgf("Torrent: %s(id=%s) submitted to %s", dbt.Name, dbt.Id, db.GetName())
d.LastUsed = index
return db.CheckStatus(dbt, isSymlink)
}
pkg/debrid/debrid_link/debrid_link.go
-6
View File
@@ -11,7 +11,6 @@ import (
"github.com/sirrobot01/debrid-blackhole/internal/request"
"github.com/sirrobot01/debrid-blackhole/pkg/debrid/torrent"
"log"
"net/http"
"os"
"strings"
@@ -29,10 +28,6 @@ type DebridLink struct {
CheckCached bool
}
func (dl *DebridLink) GetMountPath() string {
return dl.MountPath
}
func (dl *DebridLink) GetName() string {
return dl.Name
}
@@ -176,7 +171,6 @@ func (dl *DebridLink) SubmitMagnet(t *torrent.Torrent) (*torrent.Torrent, error)
}
data := *res.Value
status := "downloading"
log.Printf("Torrent: %s added with id: %s", t.Name, data.ID)
name := common.RemoveInvalidChars(data.Name)
t.Id = data.ID
t.Name = name
pkg/debrid/engine/service.go
-1
View File
@@ -12,7 +12,6 @@ type Service interface {
GetDownloadLink(tr *torrent.Torrent, file *torrent.File) *torrent.DownloadLinks
DeleteTorrent(tr *torrent.Torrent)
IsAvailable(infohashes []string) map[string]bool
GetMountPath() string
GetCheckCached() bool
GetTorrent(id string) (*torrent.Torrent, error)
GetTorrents() ([]*torrent.Torrent, error)
pkg/debrid/realdebrid/realdebrid.go
+4 -6
View File
@@ -31,10 +31,6 @@ type RealDebrid struct {
CheckCached bool
}
func (r *RealDebrid) GetMountPath() string {
return r.MountPath
}
func (r *RealDebrid) GetName() string {
return r.Name
}
@@ -156,9 +152,9 @@ func (r *RealDebrid) SubmitMagnet(t *torrent.Torrent) (*torrent.Torrent, error)
return nil, err
}
err = json.Unmarshal(resp, &data)
r.logger.Info().Msgf("Torrent: %s added with id: %s", t.Name, data.Id)
t.Id = data.Id
t.Debrid = r.Name
t.MountPath = r.MountPath
return t, nil
}
@@ -218,6 +214,8 @@ func (r *RealDebrid) CheckStatus(t *torrent.Torrent, isSymlink bool) (*torrent.T
t.Seeders = data.Seeders
t.Links = data.Links
t.Status = status
t.Debrid = r.Name
t.MountPath = r.MountPath
downloadingStatus := []string{"downloading", "magnet_conversion", "queued", "compressing", "uploading"}
if status == "waiting_files_selection" {
files := GetTorrentFiles(data, true) // Validate files to be selected
pkg/debrid/torbox/torbox.go
+2 -6
View File
@@ -11,7 +11,6 @@ import (
"github.com/sirrobot01/debrid-blackhole/internal/request"
"github.com/sirrobot01/debrid-blackhole/pkg/debrid/torrent"
"log"
"mime/multipart"
"net/http"
gourl "net/url"
@@ -35,10 +34,6 @@ type Torbox struct {
CheckCached bool
}
func (tb *Torbox) GetMountPath() string {
return tb.MountPath
}
func (tb *Torbox) GetName() string {
return tb.Name
}
@@ -130,8 +125,9 @@ func (tb *Torbox) SubmitMagnet(torrent *torrent.Torrent) (*torrent.Torrent, erro
}
dt := *data.Data
torrentId := strconv.Itoa(dt.Id)
log.Printf("Torrent: %s added with id: %s", torrent.Name, torrentId)
torrent.Id = torrentId
torrent.MountPath = tb.MountPath
torrent.Debrid = tb.Name
return torrent, nil
}