Aria2 Python Client

aria2c.py

import json
import requests

class Aria2c:
    '''
    Example : 
    
      client = Aria2c('localhost', '6800')
 
      # print server version
      print(client.getVer())
      
      # add a task to server 
      client.addUri('http://example.com/file.iso')
      
      # provide addtional options 
      option = {"out": "new_file_name.iso"}
      client.addUri('http://example.com/file.iso', option)
    '''
    IDPREFIX = "pyaria2c"
    ADD_URI = 'aria2.addUri'
    GET_VER = 'aria2.getVersion'

    def __init__(self, host, port, token=None):
        self.host = host
        self.port = port
        self.token = token
        self.serverUrl = "http://{host}:{port}/jsonrpc".format(**locals())
    
    def _genPayload(self, method, uris=None, options=None, cid=None):
        cid = IDPREFIX + cid if cid else Aria2c.IDPREFIX
        p = {
            'jsonrpc': '2.0', 
            'id': cid,
            'method': method,
            'params': []
            }

        if self.token:
            p['params'].append("token:"+self.token)
        if uris:
            p['params'].append(uris)
        if options:
            p['params'].append(options)
        return p

    @staticmethod
    def _defaultErrorHandler(code, message):
        print("ERROR: {}, {}".format(code, message))
        return None
    
    def _post(self, action, params, onSuc, onFail=None):
        if onFail is None:
            onFail = Aria2c._defaultErrorHandler
        payloads = self._genPayload(action, *params)
        resp = requests.post(self.serverUrl, data=json.dumps(payloads))
        result = resp.json()
        if "error" in result:
            return onFail(result["error"]["code"], result["error"]["message"])
        else:
            return onSuc(resp)

    def addUri(self, uri, options=None):
        def success(response):
            return response.text
        return self._post(Aria2c.ADD_URI, [[uri,], options], success)

    def getVer(self):
        def success(response):
            return response.json()['result']['version']
        return self._post(Aria2c.GET_VER, [], success)

client = Aria2c('localhost', '6800')
服务器没设置token， token为空时，添加任务会报错，建议修改一下。

    if self.token is not None:
            p['params'] = ["token:"+self.token]
    else:
            p['params'] = ["token:" + '']

或实例修改一下

client = Aria2c('localhost', '6800','')

---

03/13/2019 lpimem: 谢谢. Fixed.

>>> import aria2c
>>> c = aria2c.Aria2c('localhost', '6800')
>>> uri = "https://gist.githubusercontent.com/lpimem/3b505fa85739671a8dde504f607a25d3/raw/202e3e8e99813edd4dc86a3128c1573174ec784f/aria2c.py"
>>> c.addUri(uri)
'{"id":"pyaria2c","jsonrpc":"2.0","result":"74d4b94dd8ae5f32"}'
>>> c = aria2c.Aria2c('localhost', '6800', '')
>>> c.addUri(uri)
'{"id":"pyaria2c","jsonrpc":"2.0","result":"8daf641c46157089"}'
>>> c = aria2c.Aria2c('localhost', '6800', 'TestPass12345.B')
>>> c.addUri(uri)
'{"id":"pyaria2c","jsonrpc":"2.0","result":"d5bfaabf039d76a0"}'

萌新问一下。。。。发送出去的 id 是什么？

这个能否支持ws协议？