From b3c8b38f7a99bed8765712b6f7843704ab46b73b Mon Sep 17 00:00:00 2001 From: ShukeBta <272197458+ShukeBta@users.noreply.github.com> Date: Sat, 27 Jun 2026 01:45:18 +0800 Subject: [PATCH] split aria2 adapter helpers --- internal/service/aria2_adp.go | 269 -------------------------------- internal/service/aria2_parse.go | 118 ++++++++++++++ internal/service/aria2_rpc.go | 166 ++++++++++++++++++++ 3 files changed, 284 insertions(+), 269 deletions(-) create mode 100644 internal/service/aria2_parse.go create mode 100644 internal/service/aria2_rpc.go diff --git a/internal/service/aria2_adp.go b/internal/service/aria2_adp.go index 71b6fb7..ade3ca8 100644 --- a/internal/service/aria2_adp.go +++ b/internal/service/aria2_adp.go @@ -5,39 +5,15 @@ package service import ( - "bytes" "context" "encoding/json" "fmt" - "io" "net/http" "strings" "sync" "time" ) -// aria2Request 是 Aria2 JSON-RPC 请求结构。 -type aria2Request struct { - JSONRPC string `json:"jsonrpc"` - Method string `json:"method"` - ID string `json:"id"` - Params []interface{} `json:"params"` -} - -// aria2Response 是 Aria2 JSON-RPC 响应结构。 -type aria2Response struct { - JSONRPC string `json:"jsonrpc"` - ID string `json:"id"` - Result json.RawMessage `json:"result"` - Error *aria2Error `json:"error"` -} - -// aria2Error 是 Aria2 JSON-RPC 错误结构。 -type aria2Error struct { - Code int `json:"code"` - Message string `json:"message"` -} - // Aria2Adapter 是 Aria2 的 DownloadAdapter 实现。 type Aria2Adapter struct { mu sync.Mutex @@ -53,133 +29,6 @@ func NewAria2Adapter() *Aria2Adapter { } } -// Initialize 配置并初始化 Aria2 RPC 连接。 -func (a *Aria2Adapter) Initialize(ctx context.Context, cfg DownloadClientConfig) error { - a.mu.Lock() - defer a.mu.Unlock() - endpoint, err := normalizeDownloadClientEndpoint("aria2", cfg.Host) - if err != nil { - return err - } - cfg.Host = endpoint - a.cfg = cfg - a.idSeq = 0 - return a.getVersionLocked(ctx) -} - -// Ping 测试连接。 -func (a *Aria2Adapter) Ping(ctx context.Context) error { - a.mu.Lock() - defer a.mu.Unlock() - return a.getVersionLocked(ctx) -} - -// getVersionLocked 内部版本检查（调用者必须持有锁）。 -func (a *Aria2Adapter) getVersionLocked(ctx context.Context) error { - rpcURL, err := downloadClientRPCURL("aria2", a.cfg.Host) - if err != nil { - return err - } - - req := &aria2Request{ - JSONRPC: "2.0", - Method: "aria2.getVersion", - ID: a.nextID(), - Params: []interface{}{"token:" + a.cfg.Password}, - } - - body, err := json.Marshal(req) - if err != nil { - return err - } - - httpReq, err := newDownloadClientHTTPRequest(ctx, http.MethodPost, rpcURL, bytes.NewReader(body)) - if err != nil { - return err - } - httpReq.Header.Set("Content-Type", "application/json") - if a.cfg.Username != "" { - httpReq.SetBasicAuth(a.cfg.Username, a.cfg.Password) - } - - resp, err := a.client.Do(httpReq) - if err != nil { - return err - } - defer resp.Body.Close() - if resp.StatusCode >= 400 { - return fmt.Errorf("aria2 rpc: %d", resp.StatusCode) - } - return nil -} - -// rpcLocked 发送 JSON-RPC 请求（调用者必须持有锁）。 -func (a *Aria2Adapter) rpcLocked(ctx context.Context, method string, params []interface{}) (json.RawMessage, error) { - rpcURL, err := downloadClientRPCURL("aria2", a.cfg.Host) - if err != nil { - return nil, err - } - - if params == nil { - params = []interface{}{} - } - - // 如果 secret 不在 params 中，添加到第一位 - if len(params) > 0 { - if secret, ok := params[0].(string); ok && strings.HasPrefix(secret, "token:") { - // 已经有 secret - } else { - newParams := make([]interface{}, 0, len(params)+1) - newParams = append(newParams, "token:"+a.cfg.Password) - newParams = append(newParams, params...) - params = newParams - } - } else { - params = []interface{}{"token:" + a.cfg.Password} - } - - req := &aria2Request{ - JSONRPC: "2.0", - Method: method, - ID: a.nextID(), - Params: params, - } - - body, err := json.Marshal(req) - if err != nil { - return nil, err - } - - httpReq, err := newDownloadClientHTTPRequest(ctx, http.MethodPost, rpcURL, bytes.NewReader(body)) - if err != nil { - return nil, err - } - httpReq.Header.Set("Content-Type", "application/json") - if a.cfg.Username != "" { - httpReq.SetBasicAuth(a.cfg.Username, a.cfg.Password) - } - - resp, err := a.client.Do(httpReq) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - respBody, err := io.ReadAll(resp.Body) - if err != nil { - return nil, err - } - - var rpcResp aria2Response - if err := json.Unmarshal(respBody, &rpcResp); err != nil { - return nil, err - } - if rpcResp.Error != nil { - return nil, fmt.Errorf("aria2 rpc error [%d]: %s", rpcResp.Error.Code, rpcResp.Error.Message) - } - return rpcResp.Result, nil -} - // AddTorrent 通过 URL 添加种子或磁力链接。 func (a *Aria2Adapter) AddTorrent(ctx context.Context, torrentURL, savePath string) (string, error) { a.mu.Lock() @@ -309,121 +158,3 @@ func (a *Aria2Adapter) GetInfo(ctx context.Context, hash string) (*TorrentInfo, } return info, nil } - -// parseAria2Items 解析 Aria2 返回的任务列表。 -func (a *Aria2Adapter) parseAria2Items(raw json.RawMessage) []TorrentInfo { - var items []map[string]interface{} - if err := json.Unmarshal(raw, &items); err != nil { - return nil - } - - result := make([]TorrentInfo, 0, len(items)) - for _, item := range items { - info := a.parseSingleItem(item) - if info != nil { - result = append(result, *info) - } - } - return result -} - -// parseSingleItem 解析单个 Aria2 任务项。 -func (a *Aria2Adapter) parseSingleItem(item map[string]interface{}) *TorrentInfo { - gid := strVal(item["gid"]) - totalLength := toInt64(item["totalLength"]) - completedLength := toInt64(item["completedLength"]) - dlSpeed := toInt64(item["downloadSpeed"]) - upSpeed := toInt64(item["uploadSpeed"]) - status := strVal(item["status"]) - dir := strVal(item["dir"]) - numSeeders := int(toInt64(item["numSeeders"])) - connections := int(toInt64(item["connections"])) - - var name string - var hash string - - // 尝试从 bittorrent info 获取名称和 hash - if bt, ok := item["bittorrent"].(map[string]interface{}); ok { - if info, ok := bt["info"].(map[string]interface{}); ok { - name = strVal(info["name"]) - } - hash = strVal(bt["infoHash"]) - } - - // 如果没有 bittorrent 信息，使用 GID 作为 hash - if hash == "" { - hash = gid - } - if name == "" { - // 尝试从 files 获取文件名 - if files, ok := item["files"].([]interface{}); ok && len(files) > 0 { - if f, ok := files[0].(map[string]interface{}); ok { - paths, ok := f["path"].([]interface{}) - if ok && len(paths) > 0 { - name = strVal(paths[len(paths)-1]) - } - if name == "" { - name = strVal(f["uris"]) - } - } - } - } - if name == "" { - name = gid - } - - var progress float64 - if totalLength > 0 { - progress = float64(completedLength) / float64(totalLength) * 100 - } - - // Aria2 状态映射 - state := aria2StatusStr(status) - - return &TorrentInfo{ - Hash: hash, - Name: name, - Size: totalLength, - Progress: progress, - DLSpeed: dlSpeed, - UPSpeed: upSpeed, - State: state, - SavePath: dir, - NumSeeds: numSeeders, - NumLeechs: max(connections-numSeeders, 0), - AddedOn: time.Now(), - } -} - -// aria2StatusStr 将 Aria2 状态转为可读字符串。 -func aria2StatusStr(status string) string { - switch status { - case "active": - return "downloading" - case "waiting": - return "queued" - case "paused": - return "paused" - case "error": - return "error" - case "complete": - return "seeding" - case "removed": - return "removed" - default: - return status - } -} - -// nextID 生成递增的请求 ID。 -func (a *Aria2Adapter) nextID() string { - a.idSeq++ - return fmt.Sprintf("msg-%d", a.idSeq) -} - -func max(a, b int) int { - if a > b { - return a - } - return b -} diff --git a/internal/service/aria2_parse.go b/internal/service/aria2_parse.go new file mode 100644 index 0000000..b2ba067 --- /dev/null +++ b/internal/service/aria2_parse.go @@ -0,0 +1,118 @@ +package service + +import ( + "encoding/json" + "time" +) + +// parseAria2Items 解析 Aria2 返回的任务列表。 +func (a *Aria2Adapter) parseAria2Items(raw json.RawMessage) []TorrentInfo { + var items []map[string]interface{} + if err := json.Unmarshal(raw, &items); err != nil { + return nil + } + + result := make([]TorrentInfo, 0, len(items)) + for _, item := range items { + info := a.parseSingleItem(item) + if info != nil { + result = append(result, *info) + } + } + return result +} + +// parseSingleItem 解析单个 Aria2 任务项。 +func (a *Aria2Adapter) parseSingleItem(item map[string]interface{}) *TorrentInfo { + gid := strVal(item["gid"]) + totalLength := toInt64(item["totalLength"]) + completedLength := toInt64(item["completedLength"]) + dlSpeed := toInt64(item["downloadSpeed"]) + upSpeed := toInt64(item["uploadSpeed"]) + status := strVal(item["status"]) + dir := strVal(item["dir"]) + numSeeders := int(toInt64(item["numSeeders"])) + connections := int(toInt64(item["connections"])) + + var name string + var hash string + + // 尝试从 bittorrent info 获取名称和 hash + if bt, ok := item["bittorrent"].(map[string]interface{}); ok { + if info, ok := bt["info"].(map[string]interface{}); ok { + name = strVal(info["name"]) + } + hash = strVal(bt["infoHash"]) + } + + // 如果没有 bittorrent 信息，使用 GID 作为 hash + if hash == "" { + hash = gid + } + if name == "" { + // 尝试从 files 获取文件名 + if files, ok := item["files"].([]interface{}); ok && len(files) > 0 { + if f, ok := files[0].(map[string]interface{}); ok { + paths, ok := f["path"].([]interface{}) + if ok && len(paths) > 0 { + name = strVal(paths[len(paths)-1]) + } + if name == "" { + name = strVal(f["uris"]) + } + } + } + } + if name == "" { + name = gid + } + + var progress float64 + if totalLength > 0 { + progress = float64(completedLength) / float64(totalLength) * 100 + } + + // Aria2 状态映射 + state := aria2StatusStr(status) + + return &TorrentInfo{ + Hash: hash, + Name: name, + Size: totalLength, + Progress: progress, + DLSpeed: dlSpeed, + UPSpeed: upSpeed, + State: state, + SavePath: dir, + NumSeeds: numSeeders, + NumLeechs: aria2MaxInt(connections-numSeeders, 0), + AddedOn: time.Now(), + } +} + +// aria2StatusStr 将 Aria2 状态转为可读字符串。 +func aria2StatusStr(status string) string { + switch status { + case "active": + return "downloading" + case "waiting": + return "queued" + case "paused": + return "paused" + case "error": + return "error" + case "complete": + return "seeding" + case "removed": + return "removed" + default: + return status + } +} + +func aria2MaxInt(a, b int) int { + if a > b { + return a + } + return b +} diff --git a/internal/service/aria2_rpc.go b/internal/service/aria2_rpc.go new file mode 100644 index 0000000..0444e73 --- /dev/null +++ b/internal/service/aria2_rpc.go @@ -0,0 +1,166 @@ +package service + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" +) + +// aria2Request 是 Aria2 JSON-RPC 请求结构。 +type aria2Request struct { + JSONRPC string `json:"jsonrpc"` + Method string `json:"method"` + ID string `json:"id"` + Params []interface{} `json:"params"` +} + +// aria2Response 是 Aria2 JSON-RPC 响应结构。 +type aria2Response struct { + JSONRPC string `json:"jsonrpc"` + ID string `json:"id"` + Result json.RawMessage `json:"result"` + Error *aria2Error `json:"error"` +} + +// aria2Error 是 Aria2 JSON-RPC 错误结构。 +type aria2Error struct { + Code int `json:"code"` + Message string `json:"message"` +} + +// Initialize 配置并初始化 Aria2 RPC 连接。 +func (a *Aria2Adapter) Initialize(ctx context.Context, cfg DownloadClientConfig) error { + a.mu.Lock() + defer a.mu.Unlock() + endpoint, err := normalizeDownloadClientEndpoint("aria2", cfg.Host) + if err != nil { + return err + } + cfg.Host = endpoint + a.cfg = cfg + a.idSeq = 0 + return a.getVersionLocked(ctx) +} + +// Ping 测试连接。 +func (a *Aria2Adapter) Ping(ctx context.Context) error { + a.mu.Lock() + defer a.mu.Unlock() + return a.getVersionLocked(ctx) +} + +// getVersionLocked 内部版本检查（调用者必须持有锁）。 +func (a *Aria2Adapter) getVersionLocked(ctx context.Context) error { + rpcURL, err := downloadClientRPCURL("aria2", a.cfg.Host) + if err != nil { + return err + } + + req := &aria2Request{ + JSONRPC: "2.0", + Method: "aria2.getVersion", + ID: a.nextID(), + Params: []interface{}{"token:" + a.cfg.Password}, + } + + body, err := json.Marshal(req) + if err != nil { + return err + } + + httpReq, err := newDownloadClientHTTPRequest(ctx, http.MethodPost, rpcURL, bytes.NewReader(body)) + if err != nil { + return err + } + httpReq.Header.Set("Content-Type", "application/json") + if a.cfg.Username != "" { + httpReq.SetBasicAuth(a.cfg.Username, a.cfg.Password) + } + + resp, err := a.client.Do(httpReq) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + return fmt.Errorf("aria2 rpc: %d", resp.StatusCode) + } + return nil +} + +// rpcLocked 发送 JSON-RPC 请求（调用者必须持有锁）。 +func (a *Aria2Adapter) rpcLocked(ctx context.Context, method string, params []interface{}) (json.RawMessage, error) { + rpcURL, err := downloadClientRPCURL("aria2", a.cfg.Host) + if err != nil { + return nil, err + } + + if params == nil { + params = []interface{}{} + } + + // 如果 secret 不在 params 中，添加到第一位 + if len(params) > 0 { + if secret, ok := params[0].(string); ok && strings.HasPrefix(secret, "token:") { + // 已经有 secret + } else { + newParams := make([]interface{}, 0, len(params)+1) + newParams = append(newParams, "token:"+a.cfg.Password) + newParams = append(newParams, params...) + params = newParams + } + } else { + params = []interface{}{"token:" + a.cfg.Password} + } + + req := &aria2Request{ + JSONRPC: "2.0", + Method: method, + ID: a.nextID(), + Params: params, + } + + body, err := json.Marshal(req) + if err != nil { + return nil, err + } + + httpReq, err := newDownloadClientHTTPRequest(ctx, http.MethodPost, rpcURL, bytes.NewReader(body)) + if err != nil { + return nil, err + } + httpReq.Header.Set("Content-Type", "application/json") + if a.cfg.Username != "" { + httpReq.SetBasicAuth(a.cfg.Username, a.cfg.Password) + } + + resp, err := a.client.Do(httpReq) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(resp.Body) + if err != nil { + return nil, err + } + + var rpcResp aria2Response + if err := json.Unmarshal(respBody, &rpcResp); err != nil { + return nil, err + } + if rpcResp.Error != nil { + return nil, fmt.Errorf("aria2 rpc error [%d]: %s", rpcResp.Error.Code, rpcResp.Error.Message) + } + return rpcResp.Result, nil +} + +// nextID 生成递增的请求 ID。 +func (a *Aria2Adapter) nextID() string { + a.idSeq++ + return fmt.Sprintf("msg-%d", a.idSeq) +}