告别命令行！用Python+JSON-RPC打造你的Aria2远程下载管理器（附完整封装类）

用Python构建Aria2远程下载控制中心：从RPC配置到完整封装

每次在服务器上手动输入命令行启动下载任务，是不是已经让你感到厌倦？特别是当需要管理多个下载任务时，反复登录服务器操作既低效又容易出错。本文将带你彻底告别命令行，使用Python和JSON-RPC打造一个功能完善的远程下载管理系统。

1. 为什么选择Aria2+RPC方案？

Aria2作为一款轻量级多协议下载工具，其核心优势在于支持JSON-RPC远程控制接口。与传统的命令行调用相比，RPC方式带来了几个显著优势：

• 跨平台控制：通过HTTP协议，可以在任何设备上管理下载任务
• 自动化集成：轻松与其他系统或脚本集成，实现自动化工作流
• 状态可监控：实时获取下载进度、速度等详细信息
• 多任务管理：批量添加、暂停、删除任务变得轻而易举

以下是一个简单的功能对比表：

2. 配置Aria2的RPC服务

要让Aria2支持远程控制，首先需要正确配置RPC服务。以下是关键配置步骤：

2.1 创建配置文件

在Linux系统上，推荐将配置文件放在/etc/aria2/aria2.conf：

sudo mkdir -p /etc/aria2 sudo touch /etc/aria2/aria2.conf sudo chmod 644 /etc/aria2/aria2.conf

2.2 基础RPC配置

编辑配置文件，添加以下核心参数：

# 启用RPC接口 enable-rpc=true # RPC监听端口 rpc-listen-port=6800 # 允许所有来源的RPC请求 rpc-allow-origin-all=true # RPC密钥（建议设置） rpc-secret=YourSecretKey # 最大同时下载数 max-concurrent-downloads=5 # 断点续传 continue=true # 下载目录 dir=/path/to/your/downloads

提示：rpc-secret提供了基本的安全验证，生产环境务必设置

2.3 启动Aria2服务

使用systemd创建持久化服务：

# 创建服务文件 sudo tee /etc/systemd/system/aria2.service <<EOF [Unit] Description=Aria2 Download Manager After=network.target [Service] User=aria2 Group=aria2 Type=simple ExecStart=/usr/bin/aria2c --conf-path=/etc/aria2/aria2.conf [Install] WantedBy=multi-user.target EOF # 创建专用用户 sudo useradd -r -s /bin/false aria2 sudo chown -R aria2:aria2 /etc/aria2 # 启动服务 sudo systemctl enable --now aria2

3. Python与Aria2 RPC交互基础

理解了RPC协议后，我们可以用Python的requests库直接与Aria2交互。JSON-RPC请求的基本结构包括：

• jsonrpc: 协议版本（固定为"2.0"）
• method: 调用的Aria2方法名
• params: 方法参数（数组形式）
• id: 请求ID（用于匹配响应）

3.1 基本请求示例

import requests import json def add_download(url, save_path=None): headers = {'Content-Type': 'application/json'} payload = { 'jsonrpc': '2.0', 'id': '1', 'method': 'aria2.addUri', 'params': [ [url], {'dir': '/downloads'} if not save_path else {'dir': save_path} ] } if hasattr(add_download, 'secret'): payload['params'].insert(0, f'token:{add_download.secret}') response = requests.post( 'http://localhost:6800/jsonrpc', headers=headers, data=json.dumps(payload) ) return response.json() # 设置RPC密钥 add_download.secret = 'YourSecretKey' # 添加下载任务 result = add_download('https://example.com/large-file.zip') print(f'任务已添加，GID: {result["result"]}')

3.2 常用RPC方法

Aria2提供了丰富的RPC方法，以下是几个最常用的：

• 任务管理：

  • aria2.addUri- 添加HTTP/FTP下载
  • aria2.addTorrent- 添加种子文件
  • aria2.addMetalink- 添加Metalink文件

• 任务控制：

  • aria2.pause- 暂停任务
  • aria2.unpause- 恢复任务
  • aria2.remove- 删除任务

• 状态查询：

  • aria2.tellStatus- 获取单个任务状态
  • aria2.tellActive- 获取活动任务列表
  • aria2.getGlobalStat- 获取全局统计信息

4. 构建完整的Python封装类

为了更方便地使用Aria2 RPC，我们可以创建一个功能完善的Python类。这个类将封装常用操作，并提供错误处理和结果解析。

4.1 基础封装类

import json import requests from typing import Union, List, Dict, Optional class Aria2RPC: def __init__(self, host: str = 'localhost', port: int = 6800, secret: str = None, timeout: int = 30): """ 初始化Aria2 RPC客户端 :param host: Aria2服务地址 :param port: RPC端口，默认6800 :param secret: RPC密钥（如有） :param timeout: 请求超时时间（秒） """ self.base_url = f'http://{host}:{port}/jsonrpc' self.secret = secret self.timeout = timeout self._id = 0 self.headers = {'Content-Type': 'application/json'} def _request(self, method: str, params: list = None) -> dict: """ 发送RPC请求 :param method: RPC方法名 :param params: 方法参数 :return: 响应数据 """ self._id += 1 payload = { 'jsonrpc': '2.0', 'id': str(self._id), 'method': method } # 添加密钥前缀（如有） if self.secret: params = [f'token:{self.secret}'] + (params or []) if params is not None: payload['params'] = params try: response = requests.post( self.base_url, headers=self.headers, data=json.dumps(payload), timeout=self.timeout ) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: raise Aria2RPCError(f'RPC请求失败: {str(e)}') from e def add_uri(self, uris: Union[str, List[str]], options: dict = None) -> str: """ 添加URI下载任务 :param uris: 单个URI或URI列表 :param options: 下载选项 :return: 任务GID """ if isinstance(uris, str): uris = [uris] params = [uris] if options: params.append(options) result = self._request('aria2.addUri', params) return result.get('result') def tell_status(self, gid: str, keys: List[str] = None) -> dict: """ 获取任务状态 :param gid: 任务GID :param keys: 要返回的字段列表 :return: 状态字典 """ params = [gid] if keys: params.append(keys) return self._request('aria2.tellStatus', params).get('result') # 更多方法实现...

4.2 高级功能扩展

基础类之上，我们可以添加一些实用功能：

class Aria2Controller(Aria2RPC): def batch_add(self, uri_list: List[str], options: dict = None) -> List[str]: """ 批量添加下载任务 :param uri_list: URI列表 :param options: 下载选项 :return: 任务GID列表 """ return [self.add_uri(uri, options) for uri in uri_list] def wait_for_completion(self, gid: str, interval: int = 5, timeout: int = None) -> dict: """ 等待任务完成 :param gid: 任务GID :param interval: 检查间隔（秒） :param timeout: 超时时间（秒） :return: 最终状态 """ import time start_time = time.time() while True: status = self.tell_status(gid) if status['status'] in ['complete', 'error', 'removed']: return status if timeout and (time.time() - start_time) > timeout: raise TimeoutError('等待任务完成超时') time.sleep(interval) def get_download_speed(self) -> float: """ 获取全局下载速度（KB/s） """ stats = self.get_global_stat() return float(stats['downloadSpeed']) / 1024 def get_active_tasks(self) -> List[dict]: """ 获取所有活动任务的详细信息 """ tasks = self.tell_active() return [self._parse_task(task) for task in tasks] def _parse_task(self, task_data: dict) -> dict: """ 解析任务数据为友好格式 """ return { 'gid': task_data['gid'], 'status': task_data['status'], 'progress': float(task_data['completedLength']) / float(task_data['totalLength']) * 100, 'speed': f"{int(task_data['downloadSpeed'])/1024:.1f} KB/s", 'filename': task_data['files'][0]['path'].split('/')[-1] }

4.3 错误处理与日志

完善的封装还需要考虑错误处理和日志记录：

class Aria2RPCError(Exception): """Aria2 RPC操作异常""" pass class Aria2Controller(Aria2RPC): def __init__(self, *args, **kwargs): self.logger = kwargs.pop('logger', None) super().__init__(*args, **kwargs) def _request(self, method: str, params: list = None) -> dict: try: response = super()._request(method, params) if 'error' in response: error = response['error'] raise Aria2RPCError( f"Aria2错误 {error['code']}: {error['message']}" ) return response except Aria2RPCError as e: if self.logger: self.logger.error(f"RPC操作失败: {str(e)}") raise

5. 实战：构建Web管理界面

有了强大的Python封装，我们可以轻松构建一个基于Flask的Web管理界面：

from flask import Flask, render_template, request, jsonify from aria2_controller import Aria2Controller app = Flask(__name__) aria2 = Aria2Controller(host='localhost', secret='YourSecretKey') @app.route('/') def index(): tasks = aria2.get_active_tasks() stats = aria2.get_global_stat() return render_template('index.html', tasks=tasks, stats=stats) @app.route('/api/add', methods=['POST']) def add_task(): data = request.json gid = aria2.add_uri(data['url'], data.get('options')) return jsonify({'success': True, 'gid': gid}) @app.route('/api/status/<gid>') def task_status(gid): status = aria2.tell_status(gid) return jsonify(status) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

对应的HTML模板（templates/index.html）可以使用现代前端框架如Vue.js来实现实时状态更新：

<div id="app"> <div class="speed-indicator"> 下载速度: {{ stats.downloadSpeed }} KB/s </div> <table class="task-table"> <thead> <tr> <th>文件名</th> <th>进度</th> <th>速度</th> <th>操作</th> </tr> </thead> <tbody> <tr v-for="task in tasks" :key="task.gid"> <td>{{ task.filename }}</td> <td> <progress :value="task.progress" max="100"></progress> {{ task.progress.toFixed(1) }}% </td> <td>{{ task.speed }}</td> <td> <button @click="pauseTask(task.gid)">暂停</button> <button @click="removeTask(task.gid)">删除</button> </td> </tr> </tbody> </table> <div class="add-panel"> <input v-model="newUrl" placeholder="输入下载链接"> <button @click="addTask">添加下载</button> </div> </div>

6. 性能优化与高级配置

要让Aria2发挥最佳性能，还需要考虑以下配置优化：

6.1 网络参数调优

# 每个服务器的最大连接数 max-connection-per-server=16 # 最小分片大小 min-split-size=1M # 单个文件最大线程数 split=16 # 禁用IPv6（如不需要） disable-ipv6=true # 磁盘缓存（MB） disk-cache=64M # 文件预分配方式（减少碎片） file-allocation=falloc

6.2 内存与磁盘优化

对于大内存服务器，可以增加缓存提升性能：

# 内存缓存大小（MB） # 需要Aria2 1.16+ disk-cache=256M # 启用内存映射文件 enable-mmap=true

6.3 定时任务管理

结合Python的schedule库，可以实现自动化的下载管理：

import schedule import time def cleanup_completed(): aria2 = Aria2Controller() stats = aria2.get_global_stat() if int(stats['numStopped']) > 20: aria2.purge_download_result() # 每天凌晨3点清理已完成任务 schedule.every().day.at("03:00").do(cleanup_completed) while True: schedule.run_pending() time.sleep(60)

7. 安全加固方案

将Aria2暴露在公网时，必须考虑安全性：

7.1 HTTPS加密

启用HTTPS可以防止通信被窃听：

# 启用RPC加密 rpc-secure=true # 证书路径 rpc-certificate=/path/to/cert.pem rpc-private-key=/path/to/key.pem

7.2 访问控制

# 限制访问IP rpc-listen-all=false rpc-listen-address=192.168.1.100 # 启用认证 rpc-user=username rpc-passwd=password

7.3 防火墙规则

# 只允许特定IP访问6800端口 sudo iptables -A INPUT -p tcp --dport 6800 -s 192.168.1.0/24 -j ACCEPT sudo iptables -A INPUT -p tcp --dport 6800 -j DROP