deluge.py

# This code is a wrapper for the API exposed by the deluge-webapi plugin.
# https://github.com/idlesign/deluge-webapi

import subprocess
import requests as http
import json
from dataclasses import dataclass

# replace as necessary with values appropriate for your use case
endpoint = 'https://deluge.example.com/json'
user_id = 1
password = 'hunter2'
blackhole_path = '/path/to/blackhole/'

headers = {'accept': 'application/json', 'content-type': 'application/json'}

_json = lambda **kwargs: json.dumps(kwargs)

class DelugeException(Exception):
    def __init__(self, code, msg):
        super(Exception, self).__init__(f'Error {code}: {msg}')

def _get_cookie():
    r = http.post(endpoint,
                  headers=headers,
                  data=_json(id=user_id,
                             method='auth.login',
                             params=[password]))
    r.raise_for_status()
    return r.cookies['_session_id']

def _invoke_method(method, params):
    # TODO: hold onto cookies
    cookies = {'_session_id': _get_cookie()}
    r = http.post(endpoint,
                  headers=headers,
                  cookies=cookies,
                  data=_json(id=user_id,
                             method=method,
                             params=params))
    r.raise_for_status()
    j = r.json()
    err = j['error']
    if err:
        raise DelugeException(err['code'], err['message'])
    else:
        return j

def add_torrent(url):
    subprocess.run(('wget', '-q', '-t', '10', '-P', blackhole_path, '--', url), check=True)

def add_magnet(link, options=None):
    _invoke_method('webapi.add_torrent', [link, options])

# convenience function for scripts that, for instance, just add "torrent links" from argv or stdin
# TODO: support for description pages on various trackers? nyaa would be easy, at least
def add(link):
    if link.startswith('magnet:'):
        add_magnet(link)
    else:
        add_torrent(link)

def get_api_version():
    r = _invoke_method('webapi.get_api_version', [])
    return r['result']

def remove_torrent(torrent_id, remove_data=False):
    _invoke_method('webapi.remove_torrent', [torrent_id, remove_data])

@dataclass(frozen=True)
class Torrent:
    name: str
    infohash: str
    save_path: str
    comment: str

def get_torrents(ids=None, params=None):
    r = _invoke_method('webapi.get_torrents', [ids, params])
    return [Torrent(x['name'], x['hash'], x['save_path'], x['comment']) for x in r['result']['torrents']]