mirror of
https://github.com/evilhero/mylar
synced 2024-12-23 08:12:41 +00:00
633 lines
19 KiB
Python
633 lines
19 KiB
Python
import requests
|
|
import json
|
|
|
|
|
|
class LoginRequired(Exception):
|
|
def __str__(self):
|
|
return 'Please login first.'
|
|
|
|
|
|
class Client(object):
|
|
"""class to interact with qBittorrent WEB API"""
|
|
def __init__(self, url):
|
|
if not url.endswith('/'):
|
|
url += '/'
|
|
self.url = url
|
|
|
|
session = requests.Session()
|
|
check_prefs = session.get(url+'query/preferences')
|
|
|
|
if check_prefs.status_code == 200:
|
|
self._is_authenticated = True
|
|
self.session = session
|
|
|
|
elif check_prefs.status_code == 404:
|
|
self._is_authenticated = False
|
|
raise RuntimeError("""
|
|
This wrapper only supports qBittorrent applications
|
|
with version higher than 3.1.x.
|
|
Please use the latest qBittorrent release.
|
|
""")
|
|
|
|
else:
|
|
self._is_authenticated = False
|
|
|
|
def _get(self, endpoint, **kwargs):
|
|
"""
|
|
Method to perform GET request on the API.
|
|
|
|
:param endpoint: Endpoint of the API.
|
|
:param kwargs: Other keyword arguments for requests.
|
|
|
|
:return: Response of the GET request.
|
|
"""
|
|
return self._request(endpoint, 'get', **kwargs)
|
|
|
|
def _post(self, endpoint, data, **kwargs):
|
|
"""
|
|
Method to perform POST request on the API.
|
|
|
|
:param endpoint: Endpoint of the API.
|
|
:param data: POST DATA for the request.
|
|
:param kwargs: Other keyword arguments for requests.
|
|
|
|
:return: Response of the POST request.
|
|
"""
|
|
return self._request(endpoint, 'post', data, **kwargs)
|
|
|
|
def _request(self, endpoint, method, data=None, **kwargs):
|
|
"""
|
|
Method to hanle both GET and POST requests.
|
|
|
|
:param endpoint: Endpoint of the API.
|
|
:param method: Method of HTTP request.
|
|
:param data: POST DATA for the request.
|
|
:param kwargs: Other keyword arguments.
|
|
|
|
:return: Response for the request.
|
|
"""
|
|
final_url = self.url + endpoint
|
|
|
|
if not self._is_authenticated:
|
|
raise LoginRequired
|
|
|
|
rq = self.session
|
|
if method == 'get':
|
|
request = rq.get(final_url, **kwargs)
|
|
else:
|
|
request = rq.post(final_url, data, **kwargs)
|
|
|
|
request.raise_for_status()
|
|
request.encoding = 'utf_8'
|
|
|
|
if len(request.text) == 0:
|
|
data = json.loads('{}')
|
|
else:
|
|
try:
|
|
data = json.loads(request.text)
|
|
except ValueError:
|
|
data = request.text
|
|
|
|
return data
|
|
|
|
def login(self, username='admin', password='admin'):
|
|
"""
|
|
Method to authenticate the qBittorrent Client.
|
|
|
|
Declares a class attribute named ``session`` which
|
|
stores the authenticated session if the login is correct.
|
|
Else, shows the login error.
|
|
|
|
:param username: Username.
|
|
:param password: Password.
|
|
|
|
:return: Response to login request to the API.
|
|
"""
|
|
self.session = requests.Session()
|
|
login = self.session.post(self.url+'login',
|
|
data={'username': username,
|
|
'password': password})
|
|
if login.text == 'Ok.':
|
|
self._is_authenticated = True
|
|
else:
|
|
return login.text
|
|
|
|
def logout(self):
|
|
"""
|
|
Logout the current session.
|
|
"""
|
|
response = self._get('logout')
|
|
self._is_authenticated = False
|
|
return response
|
|
|
|
@property
|
|
def qbittorrent_version(self):
|
|
"""
|
|
Get qBittorrent version.
|
|
"""
|
|
return self._get('version/qbittorrent')
|
|
|
|
@property
|
|
def api_version(self):
|
|
"""
|
|
Get WEB API version.
|
|
"""
|
|
return self._get('version/api')
|
|
|
|
@property
|
|
def api_min_version(self):
|
|
"""
|
|
Get minimum WEB API version.
|
|
"""
|
|
return self._get('version/api_min')
|
|
|
|
def shutdown(self):
|
|
"""
|
|
Shutdown qBittorrent.
|
|
"""
|
|
return self._get('command/shutdown')
|
|
|
|
def torrents(self, **filters):
|
|
"""
|
|
Returns a list of torrents matching the supplied filters.
|
|
|
|
:param filter: Current status of the torrents.
|
|
:param category: Fetch all torrents with the supplied label.
|
|
:param sort: Sort torrents by.
|
|
:param reverse: Enable reverse sorting.
|
|
:param limit: Limit the number of torrents returned.
|
|
:param offset: Set offset (if less than 0, offset from end).
|
|
|
|
:return: list() of torrent with matching filter.
|
|
"""
|
|
params = {}
|
|
for name, value in filters.items():
|
|
# make sure that old 'status' argument still works
|
|
name = 'filter' if name == 'status' else name
|
|
params[name] = value
|
|
|
|
return self._get('query/torrents', params=params)
|
|
|
|
def get_torrent(self, infohash):
|
|
"""
|
|
Get details of the torrent.
|
|
|
|
:param infohash: INFO HASH of the torrent.
|
|
"""
|
|
return self._get('query/propertiesGeneral/' + infohash.lower())
|
|
|
|
def get_torrent_trackers(self, infohash):
|
|
"""
|
|
Get trackers for the torrent.
|
|
|
|
:param infohash: INFO HASH of the torrent.
|
|
"""
|
|
return self._get('query/propertiesTrackers/' + infohash.lower())
|
|
|
|
def get_torrent_webseeds(self, infohash):
|
|
"""
|
|
Get webseeds for the torrent.
|
|
|
|
:param infohash: INFO HASH of the torrent.
|
|
"""
|
|
return self._get('query/propertiesWebSeeds/' + infohash.lower())
|
|
|
|
def get_torrent_files(self, infohash):
|
|
"""
|
|
Get list of files for the torrent.
|
|
|
|
:param infohash: INFO HASH of the torrent.
|
|
"""
|
|
return self._get('query/propertiesFiles/' + infohash.lower())
|
|
|
|
@property
|
|
def global_transfer_info(self):
|
|
"""
|
|
Get JSON data of the global transfer info of qBittorrent.
|
|
"""
|
|
return self._get('query/transferInfo')
|
|
|
|
@property
|
|
def preferences(self):
|
|
"""
|
|
Get the current qBittorrent preferences.
|
|
Can also be used to assign individual preferences.
|
|
For setting multiple preferences at once,
|
|
see ``set_preferences`` method.
|
|
|
|
Note: Even if this is a ``property``,
|
|
to fetch the current preferences dict, you are required
|
|
to call it like a bound method.
|
|
|
|
Wrong::
|
|
|
|
qb.preferences
|
|
|
|
Right::
|
|
|
|
qb.preferences()
|
|
|
|
"""
|
|
prefs = self._get('query/preferences')
|
|
|
|
class Proxy(Client):
|
|
"""
|
|
Proxy class to to allow assignment of individual preferences.
|
|
this class overrides some methods to ease things.
|
|
|
|
Because of this, settings can be assigned like::
|
|
|
|
In [5]: prefs = qb.preferences()
|
|
|
|
In [6]: prefs['autorun_enabled']
|
|
Out[6]: True
|
|
|
|
In [7]: prefs['autorun_enabled'] = False
|
|
|
|
In [8]: prefs['autorun_enabled']
|
|
Out[8]: False
|
|
|
|
"""
|
|
|
|
def __init__(self, url, prefs, auth, session):
|
|
super(Proxy, self).__init__(url)
|
|
self.prefs = prefs
|
|
self._is_authenticated = auth
|
|
self.session = session
|
|
|
|
def __getitem__(self, key):
|
|
return self.prefs[key]
|
|
|
|
def __setitem__(self, key, value):
|
|
kwargs = {key: value}
|
|
return self.set_preferences(**kwargs)
|
|
|
|
def __call__(self):
|
|
return self.prefs
|
|
|
|
return Proxy(self.url, prefs, self._is_authenticated, self.session)
|
|
|
|
def sync(self, rid=0):
|
|
"""
|
|
Sync the torrents by supplied LAST RESPONSE ID.
|
|
Read more @ http://git.io/vEgXr
|
|
|
|
:param rid: Response ID of last request.
|
|
"""
|
|
return self._get('sync/maindata', params={'rid': rid})
|
|
|
|
def download_from_link(self, link, **kwargs):
|
|
"""
|
|
Download torrent using a link.
|
|
|
|
:param link: URL Link or list of.
|
|
:param savepath: Path to download the torrent.
|
|
:param category: Label or Category of the torrent(s).
|
|
|
|
:return: Empty JSON data.
|
|
"""
|
|
# old:new format
|
|
old_arg_map = {'save_path': 'savepath'} # , 'label': 'category'}
|
|
|
|
# convert old option names to new option names
|
|
options = kwargs.copy()
|
|
for old_arg, new_arg in old_arg_map.items():
|
|
if options.get(old_arg) and not options.get(new_arg):
|
|
options[new_arg] = options[old_arg]
|
|
|
|
options['urls'] = link
|
|
|
|
# workaround to send multipart/formdata request
|
|
# http://stackoverflow.com/a/23131823/4726598
|
|
dummy_file = {'_dummy': (None, '_dummy')}
|
|
|
|
return self._post('command/download', data=options, files=dummy_file)
|
|
|
|
def download_from_file(self, file_buffer, **kwargs):
|
|
"""
|
|
Download torrent using a file.
|
|
|
|
:param file_buffer: Single file() buffer or list of.
|
|
:param save_path: Path to download the torrent.
|
|
:param label: Label of the torrent(s).
|
|
|
|
:return: Empty JSON data.
|
|
"""
|
|
if isinstance(file_buffer, list):
|
|
torrent_files = {}
|
|
for i, f in enumerate(file_buffer):
|
|
torrent_files.update({'torrents%s' % i: f})
|
|
else:
|
|
torrent_files = {'torrents': file_buffer}
|
|
|
|
data = kwargs.copy()
|
|
|
|
if data.get('save_path'):
|
|
data.update({'savepath': data['save_path']})
|
|
return self._post('command/upload', data=data, files=torrent_files)
|
|
|
|
def add_trackers(self, infohash, trackers):
|
|
"""
|
|
Add trackers to a torrent.
|
|
|
|
:param infohash: INFO HASH of torrent.
|
|
:param trackers: Trackers.
|
|
"""
|
|
data = {'hash': infohash.lower(),
|
|
'urls': trackers}
|
|
return self._post('command/addTrackers', data=data)
|
|
|
|
@staticmethod
|
|
def _process_infohash_list(infohash_list):
|
|
"""
|
|
Method to convert the infohash_list to qBittorrent API friendly values.
|
|
|
|
:param infohash_list: List of infohash.
|
|
"""
|
|
if isinstance(infohash_list, list):
|
|
data = {'hashes': '|'.join([h.lower() for h in infohash_list])}
|
|
else:
|
|
data = {'hashes': infohash_list.lower()}
|
|
return data
|
|
|
|
def pause(self, infohash):
|
|
"""
|
|
Pause a torrent.
|
|
|
|
:param infohash: INFO HASH of torrent.
|
|
"""
|
|
return self._post('command/pause', data={'hash': infohash.lower()})
|
|
|
|
def pause_all(self):
|
|
"""
|
|
Pause all torrents.
|
|
"""
|
|
return self._get('command/pauseAll')
|
|
|
|
def pause_multiple(self, infohash_list):
|
|
"""
|
|
Pause multiple torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/pauseAll', data=data)
|
|
|
|
def set_label(self, infohash_list, label):
|
|
"""
|
|
Set the label on multiple torrents.
|
|
IMPORTANT: OLD API method, kept as it is to avoid breaking stuffs.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
data['label'] = label
|
|
return self._post('command/setLabel', data=data)
|
|
|
|
def set_category(self, infohash_list, category):
|
|
"""
|
|
Set the category on multiple torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
data['category'] = category
|
|
return self._post('command/setCategory', data=data)
|
|
|
|
def resume(self, infohash):
|
|
"""
|
|
Resume a paused torrent.
|
|
|
|
:param infohash: INFO HASH of torrent.
|
|
"""
|
|
return self._post('command/resume', data={'hash': infohash.lower()})
|
|
|
|
def resume_all(self):
|
|
"""
|
|
Resume all torrents.
|
|
"""
|
|
return self._get('command/resumeAll')
|
|
|
|
def resume_multiple(self, infohash_list):
|
|
"""
|
|
Resume multiple paused torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/resumeAll', data=data)
|
|
|
|
def delete(self, infohash_list):
|
|
"""
|
|
Delete torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/delete', data=data)
|
|
|
|
def delete_permanently(self, infohash_list):
|
|
"""
|
|
Permanently delete torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/deletePerm', data=data)
|
|
|
|
def recheck(self, infohash_list):
|
|
"""
|
|
Recheck torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/recheck', data=data)
|
|
|
|
def increase_priority(self, infohash_list):
|
|
"""
|
|
Increase priority of torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/increasePrio', data=data)
|
|
|
|
def decrease_priority(self, infohash_list):
|
|
"""
|
|
Decrease priority of torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/decreasePrio', data=data)
|
|
|
|
def set_max_priority(self, infohash_list):
|
|
"""
|
|
Set torrents to maximum priority level.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/topPrio', data=data)
|
|
|
|
def set_min_priority(self, infohash_list):
|
|
"""
|
|
Set torrents to minimum priority level.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/bottomPrio', data=data)
|
|
|
|
def set_file_priority(self, infohash, file_id, priority):
|
|
"""
|
|
Set file of a torrent to a supplied priority level.
|
|
|
|
:param infohash: INFO HASH of torrent.
|
|
:param file_id: ID of the file to set priority.
|
|
:param priority: Priority level of the file.
|
|
"""
|
|
if priority not in [0, 1, 2, 7]:
|
|
raise ValueError("Invalid priority, refer WEB-UI docs for info.")
|
|
elif not isinstance(file_id, int):
|
|
raise TypeError("File ID must be an int")
|
|
|
|
data = {'hash': infohash.lower(),
|
|
'id': file_id,
|
|
'priority': priority}
|
|
|
|
return self._post('command/setFilePrio', data=data)
|
|
|
|
# Get-set global download and upload speed limits.
|
|
|
|
def get_global_download_limit(self):
|
|
"""
|
|
Get global download speed limit.
|
|
"""
|
|
return self._get('command/getGlobalDlLimit')
|
|
|
|
def set_global_download_limit(self, limit):
|
|
"""
|
|
Set global download speed limit.
|
|
|
|
:param limit: Speed limit in bytes.
|
|
"""
|
|
return self._post('command/setGlobalDlLimit', data={'limit': limit})
|
|
|
|
global_download_limit = property(get_global_download_limit,
|
|
set_global_download_limit)
|
|
|
|
def get_global_upload_limit(self):
|
|
"""
|
|
Get global upload speed limit.
|
|
"""
|
|
return self._get('command/getGlobalUpLimit')
|
|
|
|
def set_global_upload_limit(self, limit):
|
|
"""
|
|
Set global upload speed limit.
|
|
|
|
:param limit: Speed limit in bytes.
|
|
"""
|
|
return self._post('command/setGlobalUpLimit', data={'limit': limit})
|
|
|
|
global_upload_limit = property(get_global_upload_limit,
|
|
set_global_upload_limit)
|
|
|
|
# Get-set download and upload speed limits of the torrents.
|
|
def get_torrent_download_limit(self, infohash_list):
|
|
"""
|
|
Get download speed limit of the supplied torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/getTorrentsDlLimit', data=data)
|
|
|
|
def set_torrent_download_limit(self, infohash_list, limit):
|
|
"""
|
|
Set download speed limit of the supplied torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
:param limit: Speed limit in bytes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
data.update({'limit': limit})
|
|
return self._post('command/setTorrentsDlLimit', data=data)
|
|
|
|
def get_torrent_upload_limit(self, infohash_list):
|
|
"""
|
|
Get upoload speed limit of the supplied torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/getTorrentsUpLimit', data=data)
|
|
|
|
def set_torrent_upload_limit(self, infohash_list, limit):
|
|
"""
|
|
Set upload speed limit of the supplied torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
:param limit: Speed limit in bytes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
data.update({'limit': limit})
|
|
return self._post('command/setTorrentsUpLimit', data=data)
|
|
|
|
# setting preferences
|
|
def set_preferences(self, **kwargs):
|
|
"""
|
|
Set preferences of qBittorrent.
|
|
Read all possible preferences @ http://git.io/vEgDQ
|
|
|
|
:param kwargs: set preferences in kwargs form.
|
|
"""
|
|
json_data = "json={}".format(json.dumps(kwargs))
|
|
headers = {'content-type': 'application/x-www-form-urlencoded'}
|
|
return self._post('command/setPreferences', data=json_data,
|
|
headers=headers)
|
|
|
|
def get_alternative_speed_status(self):
|
|
"""
|
|
Get Alternative speed limits. (1/0)
|
|
"""
|
|
return self._get('command/alternativeSpeedLimitsEnabled')
|
|
|
|
alternative_speed_status = property(get_alternative_speed_status)
|
|
|
|
def toggle_alternative_speed(self):
|
|
"""
|
|
Toggle alternative speed limits.
|
|
"""
|
|
return self._get('command/toggleAlternativeSpeedLimits')
|
|
|
|
def toggle_sequential_download(self, infohash_list):
|
|
"""
|
|
Toggle sequential download in supplied torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/toggleSequentialDownload', data=data)
|
|
|
|
def toggle_first_last_piece_priority(self, infohash_list):
|
|
"""
|
|
Toggle first/last piece priority of supplied torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
return self._post('command/toggleFirstLastPiecePrio', data=data)
|
|
|
|
def force_start(self, infohash_list, value=True):
|
|
"""
|
|
Force start selected torrents.
|
|
|
|
:param infohash_list: Single or list() of infohashes.
|
|
:param value: Force start value (bool)
|
|
"""
|
|
data = self._process_infohash_list(infohash_list)
|
|
data.update({'value': json.dumps(value)})
|
|
return self._post('command/setForceStart', data=data)
|