feat: indexers and download clients you add and configure (M502 #5518)
release / govulncheck (push) Successful in 34s
release / web (push) Successful in 1m13s
release / go (push) Successful in 2m12s
release / integration (push) Successful in 4m6s
release / Build signed APK (releases and dev) (push) Successful in 8m41s
release / android (push) Successful in 9m47s
release / Attach APK to the Release (tag releases only) (push) Skipped
release / Build + push container image (push) Successful in 59s
release / Verify release artifacts (tag releases only) (push) Skipped

Admin → Integrations gains "Indexers and download clients": two lists of
connections the operator adds, Fabled Librarian's model. Nothing is seeded
or copied from Lidarr.

- Indexers: Prowlarr (its native /api/v1/search, which asks every indexer
  at once; Prowlarr has no aggregate Torznab feed) and any Torznab feed.
- Download clients: qBittorrent (WebUI v2, with re-login on an expired
  session) and FabledExchange, one entry with a switch per job:
  downloads (/api/downloads), search (/api/search) and release details
  (/api/torrents/{infohash}). All three start on; its key is optional.
- Each entry: enable, name, address, a write-only key or password, Test.
  A download client's Test lists its torrents and checks the downloads
  folder is visible and writable from Minstrel; an Exchange's Test
  reports each job switched on.
- Download clients carry a path mapping (Minstrel's /downloads vs the
  client's own name for it); search sources carry searches per hour.

Migration 0079 adds acquisition_services with its CHECKs. Ported from
Fabled Librarian: internal/indexer (Torznab, 429 Retry-After, magnet
redirects), internal/prowlarr, internal/exchange, internal/downloader,
and internal/acquire's one builder per kind, which Test already uses
so a passing Test vouches for the path a grab will take.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-09 20:34:51 -04:00
co-authored by Claude Opus 5.5
parent 5306a261c1
commit a2e1890d4c
35 changed files with 4966 additions and 2 deletions
+156
View File
@@ -0,0 +1,156 @@
package downloader
import (
"context"
"errors"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
)
func TestPathMap(t *testing.T) {
m := PathMap{Local: "/downloads", Remote: "/data/torrents/"}
for in, want := range map[string]string{
"/downloads": "/data/torrents",
"/downloads/abc/x.mp3": "/data/torrents/abc/x.mp3",
"/downloads2/abc": "/downloads2/abc",
"/music/a": "/music/a",
"": "",
} {
if got := m.ToRemote(in); got != want {
t.Errorf("ToRemote(%q) = %q, want %q", in, got, want)
}
}
if got := m.ToLocal("/data/torrents/abc"); got != "/downloads/abc" {
t.Errorf("ToLocal = %q", got)
}
if (PathMap{Local: "/downloads", Remote: "/downloads/"}).Active() {
t.Error("the same folder under one name is not a mapping")
}
}
// fakeQbit is a qBittorrent WebUI that wants a login, and that expires the
// session once to make the client log in again.
func fakeQbit(t *testing.T, expireOnce bool) (*httptest.Server, *atomic.Int32) {
t.Helper()
var logins atomic.Int32
var expired atomic.Bool
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Referer") == "" {
w.WriteHeader(http.StatusForbidden)
return
}
switch r.URL.Path {
case "/api/v2/auth/login":
_ = r.ParseForm()
if r.Form.Get("password") != "pw" {
_, _ = w.Write([]byte("Fails."))
return
}
logins.Add(1)
http.SetCookie(w, &http.Cookie{Name: "SID", Value: "s", Path: "/"})
_, _ = w.Write([]byte("Ok."))
case "/api/v2/torrents/info":
if _, err := r.Cookie("SID"); err != nil {
w.WriteHeader(http.StatusForbidden)
return
}
if expireOnce && !expired.Swap(true) {
w.WriteHeader(http.StatusForbidden)
return
}
if r.URL.Query().Get("category") != "minstrel" {
t.Errorf("category = %q", r.URL.Query().Get("category"))
}
_, _ = w.Write([]byte(`[{"hash": "ABC", "name": "x", "save_path": "/data/torrents/abc", "progress": 1}]`))
case "/api/v2/torrents/add":
body, _ := io.ReadAll(r.Body)
if !strings.Contains(string(body), "/data/torrents/new") {
t.Errorf("add body has no translated save path: %s", body)
}
_, _ = w.Write([]byte("Ok."))
default:
w.WriteHeader(http.StatusNotFound)
}
}))
t.Cleanup(srv.Close)
return srv, &logins
}
func TestQBittorrent_LogsInAndLogsInAgainWhenTheSessionExpires(t *testing.T) {
srv, logins := fakeQbit(t, true)
d := Mapped(&Client{Config: Config{BaseURL: srv.URL, Username: "u", Password: "pw", Category: "minstrel"}},
PathMap{Local: "/downloads", Remote: "/data/torrents"})
ts, err := d.Torrents(context.Background())
if err != nil {
t.Fatalf("torrents: %v", err)
}
if logins.Load() != 2 {
t.Errorf("logins = %d, want 2 (the second after the 403)", logins.Load())
}
if len(ts) != 1 || ts[0].Hash != "abc" || ts[0].SavePath != "/downloads/abc" || !ts[0].Done() {
t.Errorf("torrents = %+v, want one, hash lowercased, path in Minstrel's terms", ts)
}
if err := d.AddTorrent(context.Background(), Add{Magnet: "magnet:?xt=urn:btih:abc", SavePath: "/downloads/new"}); err != nil {
t.Errorf("add: %v", err)
}
}
func TestQBittorrent_AWrongPasswordIsAuth(t *testing.T) {
srv, _ := fakeQbit(t, false)
c := &Client{Config: Config{BaseURL: srv.URL, Password: "nope"}}
if _, err := c.Torrents(context.Background()); !errors.Is(err, ErrAuth) {
t.Fatalf("err = %v, want ErrAuth", err)
}
}
func TestExchange_MapsDownloadsAndRefusesWhileRestoring(t *testing.T) {
var restoring, sentKey atomic.Bool
restoring.Store(true)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Header.Get("Authorization") != "" {
sentKey.Store(true)
}
pending := "0"
if restoring.Load() {
pending = "3"
}
_, _ = io.WriteString(w, `{"enabled": true, "running": false, "held_reason": "disk full", "restore_pending": `+pending+`, "downloads": [
{"infohash": "ABC", "name": "Album", "save_path": "/dl", "state": "seeding", "progress": 1, "length": 10, "bytes": 10, "category": "minstrel"},
{"infohash": "DEF", "name": "Other", "save_path": "/dl", "state": "fast", "category": "someone-else"},
{"infohash": "123", "name": "Broken", "save_path": "/dl", "failed": true, "fail_reason": "tracker gone", "category": "minstrel"}
]}`)
}))
defer srv.Close()
c := &Exchange{Config: Config{BaseURL: srv.URL, Category: "minstrel"}}
if _, err := c.Torrents(context.Background()); !errors.Is(err, ErrRestoring) {
t.Fatalf("while restoring: err = %v, want ErrRestoring", err)
}
if sentKey.Load() {
t.Error("an Authorization header was sent with no key set")
}
restoring.Store(false)
ts, err := c.Torrents(context.Background())
if err != nil {
t.Fatalf("torrents: %v", err)
}
if len(ts) != 2 {
t.Fatalf("torrents = %+v, want the two in Minstrel's category", ts)
}
if ts[0].Hash != "abc" || ts[0].State != "uploading" || ts[0].ContentPath != "/dl/Album" || !strings.Contains(ts[0].EngineDown, "disk full") {
t.Errorf("first = %+v", ts[0])
}
if !ts[1].Failed() || ts[1].FailReason != "tracker gone" {
t.Errorf("second = %+v, want failed with Exchange's reason", ts[1])
}
}
func TestFor_RefusesAnUnknownProtocol(t *testing.T) {
if _, err := For("transmission", Config{}); err == nil {
t.Fatal("want an error for a client Minstrel cannot drive")
}
}
+198
View File
@@ -0,0 +1,198 @@
// Package downloader drives the download clients Minstrel hands a grab to
// (M502 #5518): qBittorrent's WebUI v2, and FabledExchange's native
// /api/downloads. Ported from Fabled Librarian's internal/downloader.
//
// Both answer through one Driver, and Torrent is qBittorrent's shape: the
// Exchange driver maps its own downloads onto it, so the grab, the importer
// and the stall watchdog read one kind of record whichever client holds the
// torrent.
package downloader
import (
"context"
"errors"
"fmt"
"time"
)
// Protocols a download client row may name.
const (
KindQBittorrent = "qbittorrent"
KindExchange = "exchange"
)
// ErrUnconfigured is a client with no address.
var ErrUnconfigured = errors.New("download client is not configured")
// ErrUnreachable is a call that could not be made, or a refusal that is not
// about credentials.
var ErrUnreachable = errors.New("download client unreachable")
// ErrAuth is credentials the client refused.
var ErrAuth = errors.New("download client refused the credentials")
// ErrRestoring is an Exchange still bringing its downloads back after a
// restart: its list is incomplete, so nothing may be concluded from it.
var ErrRestoring = errors.New("download client is still restoring its downloads after a restart")
// ErrDuplicate is a torrent the client already holds.
var ErrDuplicate = errors.New("download client already holds this torrent")
// ErrUnsupported is a call this client has no equivalent for.
var ErrUnsupported = errors.New("download client cannot do that")
// UserAgent names Minstrel in the client's logs.
const UserAgent = "Minstrel (self-hosted music server; download client)"
// DefaultTimeout is what an unset Timeout means: a download client answers
// from its own state.
const DefaultTimeout = 15 * time.Second
// Config is one download client.
type Config struct {
BaseURL string
// Username and Password are qBittorrent's. Both may be empty for a WebUI
// that bypasses auth for local addresses.
Username string
Password string
// APIKey is Exchange's, sent as a bearer only when set.
APIKey string
// Category files everything Minstrel adds, so its torrents can be told
// from the operator's own.
Category string
Timeout time.Duration
}
// Driver is what Minstrel needs from a download client.
type Driver interface {
AddTorrent(ctx context.Context, a Add) error
// Torrents lists this client's torrents in Config.Category, or only the
// given hashes.
Torrents(ctx context.Context, hashes ...string) ([]Torrent, error)
// Connected says whether the client can reach any peer at all.
Connected(ctx context.Context) (bool, error)
Stop(ctx context.Context, hashes ...string) error
Start(ctx context.Context, hashes ...string) error
Reannounce(ctx context.Context, hashes ...string) error
Recheck(ctx context.Context, hashes ...string) error
Delete(ctx context.Context, deleteFiles bool, hashes ...string) error
}
// FileSkipper sets which files of a torrent are downloaded.
type FileSkipper interface {
SkipFiles(ctx context.Context, hash string, indices []int) error
KeepFiles(ctx context.Context, hash string, indices []int) error
}
// File is one file inside a torrent.
type File struct {
Index int
Path string
Size int64
}
// FileLister lists a torrent's files.
type FileLister interface {
Files(ctx context.Context, hash string) ([]File, error)
}
// FileState is one file's progress.
type FileState struct {
Path string
Size int64
Done int64
Skipped bool
}
// Unfinished says whether the file is wanted and not yet whole.
func (f FileState) Unfinished() bool { return !f.Skipped && f.Done < f.Size }
// FileProgress reports per-file progress.
type FileProgress interface {
FileStates(ctx context.Context, hash string) ([]FileState, error)
}
var (
_ Driver = (*Client)(nil)
_ Driver = (*Exchange)(nil)
_ FileSkipper = (*Client)(nil)
_ FileSkipper = (*Exchange)(nil)
_ FileSkipper = (*mapped)(nil)
_ FileLister = (*Client)(nil)
_ FileLister = (*Exchange)(nil)
_ FileLister = (*mapped)(nil)
_ FileProgress = (*Client)(nil)
_ FileProgress = (*Exchange)(nil)
_ FileProgress = (*mapped)(nil)
)
// For builds the driver a protocol names.
func For(protocol string, cfg Config) (Driver, error) {
switch protocol {
case KindQBittorrent:
return &Client{Config: cfg}, nil
case KindExchange:
return &Exchange{Config: cfg}, nil
}
return nil, fmt.Errorf("download client %q is not one Minstrel can drive", protocol)
}
// Torrent is one torrent in qBittorrent's JSON shape.
type Torrent struct {
Hash string `json:"hash"`
Name string `json:"name"`
Size int64 `json:"size"`
Progress float64 `json:"progress"`
State string `json:"state"`
Category string `json:"category"`
Tags string `json:"tags"`
SavePath string `json:"save_path"`
ContentPath string `json:"content_path"`
AddedOn int64 `json:"added_on"`
NumSeeds int `json:"num_seeds"`
NumLeechs int `json:"num_leechs"`
AmountLeft int64 `json:"amount_left"`
Ratio float64 `json:"ratio"`
DLSpeed int64 `json:"dlspeed"`
ETA int64 `json:"eta"`
// EngineDown is why the client's whole engine is not moving anything, in
// words; empty when it is running. Exchange says so; qBittorrent does not.
EngineDown string `json:"-"`
// FailReason is the client's own reason for a failed torrent.
FailReason string `json:"-"`
}
// Done says the torrent is whole.
func (t Torrent) Done() bool { return t.Progress >= 1 }
// Failed says the client gave up on it.
func (t Torrent) Failed() bool { return t.State == "error" || t.State == "missingFiles" }
// Fetching says it is still fetching metadata.
func (t Torrent) Fetching() bool { return t.State == "metaDL" || t.State == "forcedMetaDL" }
// Alone says nobody is sharing an unfinished torrent.
func (t Torrent) Alone() bool { return !t.Done() && t.NumSeeds == 0 && t.NumLeechs == 0 }
// Added is when the client took it.
func (t Torrent) Added() time.Time {
if t.AddedOn <= 0 {
return time.Time{}
}
return time.Unix(t.AddedOn, 0)
}
// Add is one torrent to hand a client: a magnet or the .torrent bytes.
type Add struct {
Magnet string
Torrent []byte
// SavePath is where it lands, as Minstrel sees it; Mapped translates it.
// Required: a download with no save path lands outside the folder
// Minstrel imports from.
SavePath string
Tags []string
Paused bool
// StopAtMetadata asks qBittorrent to stop once it has the file list, so
// unwanted files can be skipped before any data moves.
StopAtMetadata bool
}
+340
View File
@@ -0,0 +1,340 @@
package downloader
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/url"
"path"
"strings"
"time"
)
// Exchange drives FabledExchange's native /api/downloads: not its qBittorrent
// facade, so its own states, failure reasons and engine status come through.
type Exchange struct {
Config Config
// HTTP is injectable for tests. Nil takes a plain client.
HTTP *http.Client
}
type exchangeItem struct {
InfoHash string `json:"infohash"`
Name string `json:"name"`
Bytes int64 `json:"bytes"`
Length int64 `json:"length"`
Progress float64 `json:"progress"`
State string `json:"state"`
Peers int `json:"peers"`
Category string `json:"category"`
Tags []string `json:"tags"`
SavePath string `json:"save_path"`
AddedAt time.Time `json:"added_at"`
Failed bool `json:"failed"`
FailReason string `json:"fail_reason"`
Ratio float64 `json:"ratio"`
}
type exchangeSnapshot struct {
Enabled bool `json:"enabled"`
Running bool `json:"running"`
HeldReason string `json:"held_reason"`
InboundConns int64 `json:"inbound_conns"`
RestorePending int `json:"restore_pending"`
Downloads []exchangeItem `json:"downloads"`
}
func (c *Exchange) AddTorrent(ctx context.Context, a Add) error {
if a.Magnet == "" && len(a.Torrent) == 0 {
return fmt.Errorf("add: neither a magnet nor torrent bytes")
}
if a.SavePath == "" {
return fmt.Errorf("add: no save path, which would land the download outside the folder Minstrel imports from")
}
if len(a.Torrent) > 0 {
q := url.Values{"save_path": {a.SavePath}}
if c.Config.Category != "" {
q.Set("category", c.Config.Category)
}
for _, t := range a.Tags {
q.Add("tag", t)
}
if a.Paused {
q.Set("paused", "true")
}
_, err := c.call(ctx, http.MethodPost, "/api/downloads/torrent?"+q.Encode(), "application/x-bittorrent", a.Torrent)
return err
}
body, err := json.Marshal(map[string]any{
"magnet": a.Magnet, "category": c.Config.Category, "save_path": a.SavePath,
"tags": a.Tags, "paused": a.Paused,
})
if err != nil {
return fmt.Errorf("add: %w", err)
}
_, err = c.call(ctx, http.MethodPost, "/api/downloads", "application/json", body)
return err
}
// Torrents maps Exchange's downloads onto qBittorrent's shape. While Exchange
// is still restoring after a restart its list is incomplete, so it answers
// ErrRestoring rather than a list that looks like downloads vanished.
func (c *Exchange) Torrents(ctx context.Context, hashes ...string) ([]Torrent, error) {
snap, err := c.snapshot(ctx)
if err != nil {
return nil, err
}
if snap.RestorePending > 0 {
return nil, fmt.Errorf("%w (%d still to come back)", ErrRestoring, snap.RestorePending)
}
want := map[string]bool{}
for _, h := range hashes {
want[strings.ToLower(strings.TrimSpace(h))] = true
}
down := ""
switch {
case snap.Running:
case !snap.Enabled:
down = "the download engine is switched off in FabledExchange"
case snap.HeldReason != "":
down = "the download engine is held (" + snap.HeldReason + ") and resumes by itself"
default:
down = "the download engine is enabled but not running; FabledExchange is retrying it"
}
out := make([]Torrent, 0, len(snap.Downloads))
for _, it := range snap.Downloads {
if c.Config.Category != "" && it.Category != c.Config.Category {
continue
}
t := asTorrent(it)
t.EngineDown = down
if len(want) > 0 && !want[t.Hash] {
continue
}
out = append(out, t)
}
return out, nil
}
func asTorrent(it exchangeItem) Torrent {
content := it.SavePath
if it.Name != "" {
content = path.Join(it.SavePath, it.Name)
}
left := max(it.Length-it.Bytes, 0)
state := exchangeState(it.State, it.Progress)
if it.Failed {
state = "error"
}
var added int64
if !it.AddedAt.IsZero() {
added = it.AddedAt.Unix()
}
return Torrent{
Hash: strings.ToLower(it.InfoHash), Name: it.Name, Size: it.Length,
Progress: it.Progress, State: state, Category: it.Category,
Tags: strings.Join(it.Tags, ","), SavePath: it.SavePath, ContentPath: content,
AddedOn: added, NumLeechs: it.Peers, AmountLeft: left,
FailReason: it.FailReason, Ratio: it.Ratio,
}
}
// exchangeState names Exchange's state in qBittorrent's vocabulary.
func exchangeState(tier string, progress float64) string {
switch tier {
case "fast", "slow":
return "downloading"
case "metadata":
return "metaDL"
case "stalled":
return "stalledDL"
case "queued":
return "queuedDL"
case "seeding":
return "uploading"
case "relocating":
return "moving"
case "paused":
if progress >= 1 {
return "pausedUP"
}
return "pausedDL"
}
return "unknown"
}
// Connected is a running engine that has had an inbound connection: the
// nearest Exchange comes to qBittorrent's "connected".
func (c *Exchange) Connected(ctx context.Context) (bool, error) {
snap, err := c.snapshot(ctx)
if err != nil {
return false, err
}
return snap.Running && snap.InboundConns > 0, nil
}
func (c *Exchange) Stop(ctx context.Context, hashes ...string) error {
return c.each(ctx, hashes, func(h string) error {
_, err := c.call(ctx, http.MethodPost, "/api/downloads/"+url.PathEscape(h)+"/pause", "", nil)
return err
})
}
func (c *Exchange) Start(ctx context.Context, hashes ...string) error {
return c.each(ctx, hashes, func(h string) error {
_, err := c.call(ctx, http.MethodPost, "/api/downloads/"+url.PathEscape(h)+"/resume", "", nil)
return err
})
}
func (c *Exchange) SkipFiles(ctx context.Context, hash string, indices []int) error {
return c.filePriority(ctx, hash, indices, "skip")
}
func (c *Exchange) KeepFiles(ctx context.Context, hash string, indices []int) error {
return c.filePriority(ctx, hash, indices, "normal")
}
func (c *Exchange) filePriority(ctx context.Context, hash string, indices []int, priority string) error {
if len(indices) == 0 {
return nil
}
body, err := json.Marshal(map[string]any{"indices": indices, "priority": priority})
if err != nil {
return err
}
_, err = c.call(ctx, http.MethodPatch, "/api/downloads/"+url.PathEscape(hash)+"/files", "application/json", body)
return err
}
func (c *Exchange) Files(ctx context.Context, hash string) ([]File, error) {
raw, err := c.files(ctx, hash)
if err != nil {
return nil, err
}
out := make([]File, len(raw))
for i, f := range raw {
out[i] = File{Index: f.Index, Path: f.Path, Size: f.Length}
}
return out, nil
}
func (c *Exchange) FileStates(ctx context.Context, hash string) ([]FileState, error) {
raw, err := c.files(ctx, hash)
if err != nil {
return nil, err
}
out := make([]FileState, len(raw))
for i, f := range raw {
if f.Bytes == nil {
return nil, fmt.Errorf("%w: FabledExchange does not say how much of each file it has", ErrUnsupported)
}
out[i] = FileState{Path: f.Path, Size: f.Length, Done: *f.Bytes, Skipped: f.Priority == "skip"}
}
return out, nil
}
type exchangeFile struct {
Index int `json:"index"`
Path string `json:"path"`
Length int64 `json:"length"`
Bytes *int64 `json:"bytes"`
Priority string `json:"priority"`
}
func (c *Exchange) files(ctx context.Context, hash string) ([]exchangeFile, error) {
body, err := c.call(ctx, http.MethodGet, "/api/downloads/"+url.PathEscape(hash), "", nil)
if err != nil {
return nil, err
}
var detail struct {
Files []exchangeFile `json:"files"`
}
if err := json.Unmarshal(body, &detail); err != nil {
return nil, fmt.Errorf("%w: parse download detail: %v", ErrUnreachable, err)
}
return detail.Files, nil
}
func (c *Exchange) Reannounce(context.Context, ...string) error {
return fmt.Errorf("%w: FabledExchange schedules its own announces", ErrUnsupported)
}
func (c *Exchange) Recheck(context.Context, ...string) error {
return fmt.Errorf("%w: FabledExchange has no recheck", ErrUnsupported)
}
func (c *Exchange) Delete(ctx context.Context, deleteFiles bool, hashes ...string) error {
return c.each(ctx, hashes, func(h string) error {
p := "/api/downloads/" + url.PathEscape(h)
if deleteFiles {
p += "?delete_files=true"
}
_, err := c.call(ctx, http.MethodDelete, p, "", nil)
return err
})
}
func (c *Exchange) each(ctx context.Context, hashes []string, do func(string) error) error {
for _, h := range hashes {
if ctx.Err() != nil {
return ctx.Err()
}
if err := do(strings.ToLower(strings.TrimSpace(h))); err != nil {
return err
}
}
return nil
}
func (c *Exchange) snapshot(ctx context.Context) (exchangeSnapshot, error) {
var snap exchangeSnapshot
body, err := c.call(ctx, http.MethodGet, "/api/downloads", "", nil)
if err != nil {
return snap, err
}
if err := json.Unmarshal(body, &snap); err != nil {
return snap, fmt.Errorf("%w: parse download list: %v", ErrUnreachable, err)
}
return snap, nil
}
func (c *Exchange) call(ctx context.Context, method, p, contentType string, body []byte) ([]byte, error) {
if c.Config.BaseURL == "" {
return nil, ErrUnconfigured
}
httpc := c.HTTP
if httpc == nil {
httpc = &http.Client{}
}
status, out, err := send(ctx, httpc, c.Config.Timeout, method,
strings.TrimRight(c.Config.BaseURL, "/")+p, contentType, body, func(h http.Header) {
h.Set("Accept", "application/json")
if c.Config.APIKey != "" {
h.Set("Authorization", "Bearer "+c.Config.APIKey)
}
})
if err != nil {
return nil, err
}
if status >= 200 && status < 300 {
return out, nil
}
var e struct {
Error string `json:"error"`
Reason string `json:"reason"`
}
_ = json.Unmarshal(out, &e)
said := strings.TrimSpace(e.Error)
if said == "" {
said = fmt.Sprintf("HTTP %d", status)
}
switch {
case status == http.StatusUnauthorized || status == http.StatusForbidden:
return nil, fmt.Errorf("%w: FabledExchange refused the API key (%s)", ErrAuth, said)
case status == http.StatusConflict && e.Reason == "duplicate":
return nil, fmt.Errorf("%w: %s", ErrDuplicate, said)
}
return nil, fmt.Errorf("%w: %s", ErrUnreachable, said)
}
+107
View File
@@ -0,0 +1,107 @@
package downloader
import (
"context"
"path"
"strings"
)
// PathMap is one downloads folder under two names: where Minstrel has it
// mounted (Local) and where the client sees it (Remote). The two usually run
// in different containers. Every path crossing between them swaps one prefix
// for the other.
type PathMap struct {
Local string
Remote string
}
// Active says the two names differ, so there is something to translate.
func (m PathMap) Active() bool {
return m.Remote != "" && m.Local != "" && path.Clean(m.Remote) != path.Clean(m.Local)
}
// ToRemote is a Minstrel path as the client names it.
func (m PathMap) ToRemote(p string) string { return swap(p, m.Local, m.Remote) }
// ToLocal is a client's path as Minstrel names it.
func (m PathMap) ToLocal(p string) string { return swap(p, m.Remote, m.Local) }
// swap replaces the prefix from with to on segment boundaries
// (/downloads2 is not inside /downloads). A path outside from is unchanged.
func swap(p, from, to string) string {
if p == "" || from == "" || to == "" {
return p
}
from, to = path.Clean(from), path.Clean(to)
clean := path.Clean(p)
if clean == from {
return to
}
if from == "/" {
return path.Join(to, clean)
}
if rest, ok := strings.CutPrefix(clean, from+"/"); ok {
return path.Join(to, rest)
}
return p
}
// Mapped wraps d so every path it is given or answers with is translated.
// File indices and in-torrent paths need no translation.
func Mapped(d Driver, m PathMap) Driver {
if !m.Active() {
return d
}
return &mapped{Driver: d, m: m}
}
type mapped struct {
Driver
m PathMap
}
func (c *mapped) AddTorrent(ctx context.Context, a Add) error {
a.SavePath = c.m.ToRemote(a.SavePath)
return c.Driver.AddTorrent(ctx, a)
}
func (c *mapped) Torrents(ctx context.Context, hashes ...string) ([]Torrent, error) {
ts, err := c.Driver.Torrents(ctx, hashes...)
for i := range ts {
ts[i].SavePath = c.m.ToLocal(ts[i].SavePath)
ts[i].ContentPath = c.m.ToLocal(ts[i].ContentPath)
}
return ts, err
}
func (c *mapped) SkipFiles(ctx context.Context, hash string, indices []int) error {
fs, ok := c.Driver.(FileSkipper)
if !ok {
return ErrUnsupported
}
return fs.SkipFiles(ctx, hash, indices)
}
func (c *mapped) KeepFiles(ctx context.Context, hash string, indices []int) error {
fs, ok := c.Driver.(FileSkipper)
if !ok {
return ErrUnsupported
}
return fs.KeepFiles(ctx, hash, indices)
}
func (c *mapped) Files(ctx context.Context, hash string) ([]File, error) {
fl, ok := c.Driver.(FileLister)
if !ok {
return nil, ErrUnsupported
}
return fl.Files(ctx, hash)
}
func (c *mapped) FileStates(ctx context.Context, hash string) ([]FileState, error) {
fp, ok := c.Driver.(FileProgress)
if !ok {
return nil, ErrUnsupported
}
return fp.FileStates(ctx, hash)
}
+367
View File
@@ -0,0 +1,367 @@
package downloader
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"mime/multipart"
"net/http"
"net/http/cookiejar"
"net/url"
"strconv"
"strings"
"sync"
"time"
)
// Client drives one qBittorrent through its WebUI API v2.
type Client struct {
Config Config
mu sync.Mutex
// http carries the SID cookie once logged in; nil until then.
http *http.Client
}
// AddTorrent hands qBittorrent a magnet or .torrent bytes, filed under the
// configured category with the save path already set.
func (c *Client) AddTorrent(ctx context.Context, a Add) error {
if a.Magnet == "" && len(a.Torrent) == 0 {
return fmt.Errorf("add: neither a magnet nor torrent bytes")
}
if a.SavePath == "" {
return fmt.Errorf("add: no save path, which would land the download outside the folder Minstrel imports from")
}
body := &bytes.Buffer{}
form := multipart.NewWriter(body)
fields := map[string]string{
"savepath": a.SavePath,
"category": c.Config.Category,
// Both, because qBittorrent 5 renamed paused to stopped and an older
// one ignores the field it does not know.
"paused": strconv.FormatBool(a.Paused),
"stopped": strconv.FormatBool(a.Paused),
}
if len(a.Tags) > 0 {
fields["tags"] = strings.Join(a.Tags, ",")
}
if a.StopAtMetadata && !a.Paused {
fields["stopCondition"] = "MetadataReceived"
}
if a.Magnet != "" {
fields["urls"] = a.Magnet
}
for k, v := range fields {
if v == "" {
continue
}
if err := form.WriteField(k, v); err != nil {
return fmt.Errorf("add: build form: %w", err)
}
}
if len(a.Torrent) > 0 {
part, err := form.CreateFormFile("torrents", "release.torrent")
if err != nil {
return fmt.Errorf("add: build form: %w", err)
}
if _, err := part.Write(a.Torrent); err != nil {
return fmt.Errorf("add: build form: %w", err)
}
}
if err := form.Close(); err != nil {
return fmt.Errorf("add: build form: %w", err)
}
resp, err := c.post(ctx, "/api/v2/torrents/add", form.FormDataContentType(), body.Bytes())
if err != nil {
return err
}
// qBittorrent refuses with HTTP 200 and the body "Fails.".
if strings.HasPrefix(strings.TrimSpace(string(resp)), "Fails") {
return fmt.Errorf("%w: the client refused the torrent", ErrUnreachable)
}
return nil
}
// Torrents lists the configured category's torrents, or the given hashes.
func (c *Client) Torrents(ctx context.Context, hashes ...string) ([]Torrent, error) {
params := url.Values{}
if c.Config.Category != "" {
params.Set("category", c.Config.Category)
}
if len(hashes) > 0 {
params.Set("hashes", strings.Join(hashes, "|"))
}
body, err := c.get(ctx, "/api/v2/torrents/info", params)
if err != nil {
return nil, err
}
var out []Torrent
if err := json.Unmarshal(body, &out); err != nil {
return nil, fmt.Errorf("%w: parse torrent list: %v", ErrUnreachable, err)
}
for i := range out {
out[i].Hash = strings.ToLower(out[i].Hash)
}
return out, nil
}
// Connected reads qBittorrent's own connection status.
func (c *Client) Connected(ctx context.Context) (bool, error) {
body, err := c.get(ctx, "/api/v2/transfer/info", nil)
if err != nil {
return false, err
}
var info struct {
ConnectionStatus string `json:"connection_status"`
}
if err := json.Unmarshal(body, &info); err != nil {
return false, fmt.Errorf("%w: parse transfer info: %v", ErrUnreachable, err)
}
return info.ConnectionStatus == "connected", nil
}
// Stop and Start try qBittorrent 5's names first and fall back to 4's.
func (c *Client) Stop(ctx context.Context, hashes ...string) error {
return c.command(ctx, hashes, "/api/v2/torrents/stop", "/api/v2/torrents/pause")
}
func (c *Client) Start(ctx context.Context, hashes ...string) error {
return c.command(ctx, hashes, "/api/v2/torrents/start", "/api/v2/torrents/resume")
}
func (c *Client) Reannounce(ctx context.Context, hashes ...string) error {
return c.command(ctx, hashes, "/api/v2/torrents/reannounce", "/api/v2/torrents/reannounce")
}
func (c *Client) Recheck(ctx context.Context, hashes ...string) error {
return c.command(ctx, hashes, "/api/v2/torrents/recheck", "/api/v2/torrents/recheck")
}
func (c *Client) Delete(ctx context.Context, deleteFiles bool, hashes ...string) error {
if len(hashes) == 0 {
return nil
}
params := url.Values{
"hashes": {strings.Join(hashes, "|")},
"deleteFiles": {strconv.FormatBool(deleteFiles)},
}
_, err := c.post(ctx, "/api/v2/torrents/delete", "application/x-www-form-urlencoded", []byte(params.Encode()))
return err
}
func (c *Client) SkipFiles(ctx context.Context, hash string, indices []int) error {
return c.filePriority(ctx, hash, indices, "0")
}
func (c *Client) KeepFiles(ctx context.Context, hash string, indices []int) error {
return c.filePriority(ctx, hash, indices, "1")
}
func (c *Client) filePriority(ctx context.Context, hash string, indices []int, priority string) error {
if len(indices) == 0 {
return nil
}
ids := make([]string, len(indices))
for i, n := range indices {
ids[i] = strconv.Itoa(n)
}
params := url.Values{"hash": {hash}, "id": {strings.Join(ids, "|")}, "priority": {priority}}
_, err := c.post(ctx, "/api/v2/torrents/filePrio", "application/x-www-form-urlencoded", []byte(params.Encode()))
return err
}
func (c *Client) Files(ctx context.Context, hash string) ([]File, error) {
raw, err := c.files(ctx, hash)
if err != nil {
return nil, err
}
out := make([]File, len(raw))
for i, f := range raw {
out[i] = File{Index: f.Index, Path: f.Name, Size: f.Size}
}
return out, nil
}
func (c *Client) FileStates(ctx context.Context, hash string) ([]FileState, error) {
raw, err := c.files(ctx, hash)
if err != nil {
return nil, err
}
out := make([]FileState, len(raw))
for i, f := range raw {
done := f.Size
if f.Progress < 1 {
// Progress is a float; never round an unfinished file up to whole.
done = min(int64(f.Progress*float64(f.Size)), f.Size-1)
}
out[i] = FileState{Path: f.Name, Size: f.Size, Done: max(done, 0), Skipped: f.Priority == 0}
}
return out, nil
}
type qbitFile struct {
Index int `json:"index"`
Name string `json:"name"`
Size int64 `json:"size"`
Progress float64 `json:"progress"`
Priority int `json:"priority"`
}
func (c *Client) files(ctx context.Context, hash string) ([]qbitFile, error) {
body, err := c.get(ctx, "/api/v2/torrents/files", url.Values{"hash": {hash}})
if err != nil {
return nil, err
}
var raw []qbitFile
if err := json.Unmarshal(body, &raw); err != nil {
return nil, fmt.Errorf("%w: parse file list: %v", ErrUnreachable, err)
}
return raw, nil
}
func (c *Client) command(ctx context.Context, hashes []string, primary, fallback string) error {
if len(hashes) == 0 {
return nil
}
params := []byte(url.Values{"hashes": {strings.Join(hashes, "|")}}.Encode())
_, err := c.post(ctx, primary, "application/x-www-form-urlencoded", params)
var status statusError
if errors.As(err, &status) && (status.code == http.StatusNotFound || status.code == http.StatusMethodNotAllowed) {
_, err = c.post(ctx, fallback, "application/x-www-form-urlencoded", params)
}
return err
}
type statusError struct {
code int
what string
}
func (e statusError) Error() string { return fmt.Sprintf("%s: HTTP %d", e.what, e.code) }
func (e statusError) Unwrap() error { return ErrUnreachable }
func (c *Client) get(ctx context.Context, path string, params url.Values) ([]byte, error) {
endpoint := path
if len(params) > 0 {
endpoint += "?" + params.Encode()
}
return c.do(ctx, http.MethodGet, endpoint, "", nil)
}
func (c *Client) post(ctx context.Context, path, contentType string, body []byte) ([]byte, error) {
return c.do(ctx, http.MethodPost, path, contentType, body)
}
// do logs in when there is no session, and once more on a 403, which is how
// qBittorrent says a session expired.
func (c *Client) do(ctx context.Context, method, path, contentType string, body []byte) ([]byte, error) {
if c.Config.BaseURL == "" {
return nil, ErrUnconfigured
}
if err := c.ensureSession(ctx, false); err != nil {
return nil, err
}
out, err := c.call(ctx, method, path, contentType, body)
var status statusError
if errors.As(err, &status) && status.code == http.StatusForbidden {
if err := c.ensureSession(ctx, true); err != nil {
return nil, err
}
return c.call(ctx, method, path, contentType, body)
}
return out, err
}
func (c *Client) ensureSession(ctx context.Context, force bool) error {
c.mu.Lock()
defer c.mu.Unlock()
if c.http != nil && !force {
return nil
}
jar, err := cookiejar.New(nil)
if err != nil {
return fmt.Errorf("cookie jar: %w", err)
}
httpc := &http.Client{Jar: jar}
params := url.Values{"username": {c.Config.Username}, "password": {c.Config.Password}}
out, err := c.callWith(ctx, httpc, http.MethodPost, "/api/v2/auth/login",
"application/x-www-form-urlencoded", []byte(params.Encode()))
if err != nil {
var status statusError
if errors.As(err, &status) && status.code == http.StatusForbidden {
return fmt.Errorf("%w: too many failed attempts, or this address is banned", ErrAuth)
}
return err
}
if strings.HasPrefix(strings.TrimSpace(string(out)), "Fails") {
return fmt.Errorf("%w: wrong username or password", ErrAuth)
}
c.http = httpc
return nil
}
func (c *Client) call(ctx context.Context, method, path, contentType string, body []byte) ([]byte, error) {
c.mu.Lock()
httpc := c.http
c.mu.Unlock()
if httpc == nil {
httpc = &http.Client{}
}
return c.callWith(ctx, httpc, method, path, contentType, body)
}
func (c *Client) callWith(ctx context.Context, httpc *http.Client, method, path, contentType string, body []byte) ([]byte, error) {
base := strings.TrimRight(c.Config.BaseURL, "/")
status, out, err := send(ctx, httpc, c.Config.Timeout, method, base+path, contentType, body, func(h http.Header) {
// qBittorrent's CSRF check refuses a request without a matching
// Referer.
h.Set("Referer", base)
})
if err != nil {
return nil, err
}
if status < 200 || status >= 300 {
return nil, statusError{code: status, what: path}
}
return out, nil
}
// send makes one request to a download client under its timeout, lets the
// caller add headers, and reads at most 16 MB. It returns the status for the
// caller to read, because each client's refusals are its own.
func send(ctx context.Context, httpc *http.Client, timeout time.Duration, method, target, contentType string, body []byte, set func(http.Header)) (int, []byte, error) {
if timeout <= 0 {
timeout = DefaultTimeout
}
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
var reader io.Reader
if body != nil {
reader = bytes.NewReader(body)
}
req, err := http.NewRequestWithContext(ctx, method, target, reader)
if err != nil {
return 0, nil, fmt.Errorf("%w: build request: %v", ErrUnreachable, err)
}
req.Header.Set("User-Agent", UserAgent)
if contentType != "" {
req.Header.Set("Content-Type", contentType)
}
if set != nil {
set(req.Header)
}
resp, err := httpc.Do(req)
if err != nil {
return 0, nil, fmt.Errorf("%w: %v", ErrUnreachable, err)
}
defer func() { _ = resp.Body.Close() }()
out, err := io.ReadAll(io.LimitReader(resp.Body, 16<<20))
if err != nil {
return 0, nil, fmt.Errorf("%w: read body: %v", ErrUnreachable, err)
}
return resp.StatusCode, out, nil
}