tcjlj/MeBox
1
0
Fork 0
mirror of https://github.com/truewhile/MeBox.git synced 2026-10-12 16:26:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(downloads): support Transmission and aria2 end to end

Browse Source 
Fixes #65
This commit is contained in:
ShukeBta
2026-08-10 21:04:04 +08:00
parent c0d743d89d
commit 42ef19b7e5
43 changed files with 2350 additions and 480 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/service/downloads.go
+93 -32
View File
@@ -1,7 +1,7 @@
// Package service — download manager.
//
// DownloadService persists user-initiated downloads, dispatches them to
// the configured client (currently qBittorrent) and pushes live progress
// the configured qBittorrent, Transmission, or aria2 client and pushes live progress
// to the WS hub so the React UI can render a live table.
//
// Settings consumed (system Setting table):
@@ -18,6 +18,7 @@ package service
import (
"context"
"errors"
"fmt"
"strings"
"sync"
"time"
@@ -34,6 +35,7 @@ type DownloadService struct {
repo *repository.Container
hub *Hub
qb *QBitClient
manager *DownloadManager
organizer *OrganizerService
organizePipeline *OrganizePipelineService
scanner *ScannerService
@@ -45,7 +47,7 @@ type DownloadService struct {
stopCh chan struct{}
pollOnce sync.Once
organizeOnce sync.Once
prevStates map[string]bool // hash -> wasCompleted
prevStates map[string]bool // client/task identity -> wasCompleted
pollInitialized bool
liveTorrents []QBitTorrent
liveTorrentsAt time.Time
@@ -70,8 +72,12 @@ func (d *DownloadService) SetNotifyChannels(notify *NotifyChannelService) {
d.notify = notify
}
func (d *DownloadService) SetDownloadManager(manager *DownloadManager) {
d.manager = manager
}
// ErrDownloadAlreadyExists tells callers that the requested resource is already
// tracked locally or present in qBittorrent. Subscriptions treat this as a
// tracked locally or present in a downloader. Subscriptions treat this as a
// successful dedup hit, not as a retryable enqueue failure.
var ErrDownloadAlreadyExists = errors.New("download already exists")
@@ -80,6 +86,8 @@ var ErrDownloadAlreadyExists = errors.New("download already exists")
// downloader again.
var ErrMediaAlreadyInLibrary = errors.New("media already exists in library")
var ErrDownloadOperationUnsupported = errors.New("download client operation unsupported")
func IsDownloadDedupError(err error) bool {
return errors.Is(err, ErrDownloadAlreadyExists) || errors.Is(err, ErrMediaAlreadyInLibrary)
}
@@ -108,7 +116,9 @@ func NewDownloadService(log *zap.Logger, repo *repository.Container, hub *Hub, o
// Start kicks off the background poller (idempotent).
func (d *DownloadService) Start(ctx context.Context) {
d.pollOnce.Do(func() {
_ = d.ReloadConfig(ctx)
if err := d.ReloadConfig(ctx); err != nil && d.log != nil {
d.log.Warn("initial download client reload failed", zap.Error(err))
}
d.startAutoOrganizeWorker(ctx)
go d.poll(ctx)
})
@@ -124,8 +134,8 @@ func (d *DownloadService) TorrentExistsByName(ctx context.Context, name string)
if query == "" {
return false
}
live, err := d.qb.List(ctx, "")
if err != nil {
live, err := d.listLiveTorrents(ctx, "")
if err != nil && len(live) == 0 {
return false
}
for _, torrent := range live {
@@ -143,43 +153,52 @@ func (d *DownloadService) TorrentExistsByName(ctx context.Context, name string)
return false
}
// List returns every persisted download task augmented with live data
// from qBittorrent when available.
// List returns every persisted download task augmented with live downloader data.
func (d *DownloadService) List(ctx context.Context) ([]model.DownloadTask, []QBitTorrent, error) {
rows, err := d.repo.Download.List(ctx)
if err != nil {
return nil, nil, err
}
live, err := d.qb.List(ctx, "")
live, err := d.listLiveTorrents(ctx, "")
if err != nil {
// Network failure shouldn't break the page — return rows with no
// live data and let the UI render the persisted snapshot.
d.log.Debug("qbittorrent list failed", zap.Error(err))
return rows, nil, nil
// A failed client must not hide healthy clients or persisted rows.
d.log.Debug("download client list failed", zap.Error(err))
if len(live) == 0 {
return rows, nil, nil
}
}
return rows, live, nil
}
// Delete removes a torrent (and optionally its files) from qBittorrent.
func (d *DownloadService) Delete(ctx context.Context, hash string, withFiles bool) error {
// Delete removes a task from its native downloader. clientID is optional for
// legacy callers, but disambiguates equal native IDs across multiple clients.
func (d *DownloadService) Delete(ctx context.Context, hash string, withFiles bool, clientID ...string) error {
hash = strings.TrimSpace(hash)
if hash == "" {
return errors.New("hash is required")
}
var torrentName string
if live, err := d.qb.List(ctx, ""); err == nil {
for _, torrent := range live {
if strings.EqualFold(torrent.Hash, hash) || len(live) == 1 {
torrentName = torrent.Name
break
}
}
requestedClientID := ""
if len(clientID) > 0 {
requestedClientID = clientID[0]
}
if err := d.qb.Delete(ctx, hash, withFiles); err != nil {
resolvedClientID, torrentName, err := d.resolveOperationClientID(ctx, hash, requestedClientID)
if err != nil {
return err
}
d.markDownloadTaskDeleted(ctx, hash, torrentName)
stateKey := strings.ToLower(hash)
target, err := d.downloadTargetByID(ctx, resolvedClientID)
if err != nil {
return err
}
if target.legacyQB {
err = d.qb.Delete(ctx, hash, withFiles)
} else {
err = target.adapter.Remove(ctx, hash, withFiles)
}
if err != nil {
return err
}
d.markDownloadTaskDeleted(ctx, hash, torrentName, resolvedClientID)
stateKey := completedTorrentQueueKey(QBitTorrent{ClientID: resolvedClientID, Hash: hash})
d.mu.Lock()
delete(d.prevStates, stateKey)
delete(d.organizeQueued, stateKey)
@@ -187,7 +206,7 @@ func (d *DownloadService) Delete(ctx context.Context, hash string, withFiles boo
return nil
}
func (d *DownloadService) markDownloadTaskDeleted(ctx context.Context, hash, torrentName string) {
func (d *DownloadService) markDownloadTaskDeleted(ctx context.Context, hash, torrentName string, clientID ...string) {
if d == nil || d.repo == nil || d.repo.DB == nil {
return
}
@@ -195,7 +214,7 @@ func (d *DownloadService) markDownloadTaskDeleted(ctx context.Context, hash, tor
if err != nil {
return
}
if matched, ok := findDownloadTaskByHash(rows, hash); ok {
if matched, ok := findDownloadTaskByHash(rows, hash, clientID...); ok {
_ = d.repo.DB.WithContext(ctx).Model(&model.DownloadTask{}).
Where("id = ?", matched.ID).
Updates(map[string]any{
@@ -208,7 +227,7 @@ func (d *DownloadService) markDownloadTaskDeleted(ctx context.Context, hash, tor
return
}
taskByKey := tasksByTorrentIdentity(rows)
matched, ok := findMatchingTaskByTorrentIdentity(torrentName, taskByKey)
matched, ok := findMatchingTaskForTorrent(QBitTorrent{Name: torrentName, Hash: hash, ClientID: firstString(clientID)}, taskByKey)
if !ok {
return
}
@@ -220,12 +239,24 @@ func (d *DownloadService) markDownloadTaskDeleted(ctx context.Context, hash, tor
}).Error
}
func findDownloadTaskByHash(rows []model.DownloadTask, hash string) (model.DownloadTask, bool) {
func findDownloadTaskByHash(rows []model.DownloadTask, hash string, clientID ...string) (model.DownloadTask, bool) {
hash = strings.ToLower(strings.TrimSpace(hash))
if hash == "" {
return model.DownloadTask{}, false
}
wantClientID := strings.TrimSpace(firstString(clientID))
for _, row := range rows {
if wantClientID != "" && strings.TrimSpace(row.DownloadClientID) != "" && row.DownloadClientID != wantClientID {
continue
}
if strings.EqualFold(strings.TrimSpace(row.ExternalID), hash) {
return row, true
}
}
for _, row := range rows {
if wantClientID != "" && strings.TrimSpace(row.DownloadClientID) != "" && row.DownloadClientID != wantClientID {
continue
}
if strings.Contains(strings.ToLower(row.URL), hash) {
return row, true
}
@@ -233,15 +264,45 @@ func findDownloadTaskByHash(rows []model.DownloadTask, hash string) (model.Downl
return model.DownloadTask{}, false
}
func firstString(values []string) string {
if len(values) == 0 {
return ""
}
return values[0]
}
// RelocateTorrent moves a torrent's data to a new save directory while keeping
// it seeding (qBittorrent performs the physical move and resumes seeding).
// 用于「移动 PT 种子文件且转移后继续做种上传」的整盘迁移场景。
func (d *DownloadService) RelocateTorrent(ctx context.Context, hash, location string) error {
func (d *DownloadService) RelocateTorrent(ctx context.Context, hash, location string, clientID ...string) error {
if strings.TrimSpace(hash) == "" {
return errors.New("hash is required")
}
if strings.TrimSpace(location) == "" {
return errors.New("location is required")
}
return d.qb.SetLocation(ctx, hash, strings.TrimSpace(location))
requestedClientID := firstString(clientID)
resolvedClientID := strings.TrimSpace(requestedClientID)
if resolvedClientID == "" {
var err error
resolvedClientID, _, err = d.resolveOperationClientID(ctx, hash, "")
if err != nil {
return err
}
}
target, err := d.downloadTargetByID(ctx, resolvedClientID)
if err != nil {
return err
}
if target.typ != "qbittorrent" {
return fmt.Errorf("%w: %s does not support torrent relocation; only qBittorrent is supported", ErrDownloadOperationUnsupported, target.typ)
}
if target.legacyQB {
return d.qb.SetLocation(ctx, hash, strings.TrimSpace(location))
}
relocator, ok := target.adapter.(TorrentRelocateAdapter)
if !ok {
return fmt.Errorf("%w: configured qBittorrent adapter cannot relocate torrents", ErrDownloadOperationUnsupported)
}
return relocator.Relocate(ctx, hash, strings.TrimSpace(location))
}