KevinMidboe/delugeClient
1
0
Fork 0
mirror of https://github.com/KevinMidboe/delugeClient.git synced 2025-10-28 19:40:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Restructured project files.

Browse Source 
Moved all packages files to delugeClient folder.
Split contents of deluge_cli to __main__, utils, deluge & torrent.
Config default changed from config.ini --> delugeClient/default_config.ini.
Setup.py updated with new entry.
This commit is contained in:
Kevin
2021-07-04 16:25:44 +02:00
parent 9d34802957
commit 518b5b0e78
11 changed files with 486 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
config.ini
View File

@@ -1,10 +0,0 @@
[Deluge]
HOST = YOUR_DELUGE_HOST
PORT = YOUR_DELUGE_PORT
USER = YOUR_DELUGE_USER
PASSWORD = YOUR_DELUGE_PASSWORD
[ssh]
HOST = YOUR_DELUGE_SERVER_IP
USER = YOUR_SSH_USER
PKEY = YOUR_SSH_PRIVATE_KEY_DIRECTORY

20
delugeClient/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
from sys import path
from os.path import dirname
path.append(dirname(__file__))
__version__=0.1
import logging
logger = logging.getLogger('deluge_cli')
logger.setLevel(logging.DEBUG)
fh = logging.FileHandler(os.path.join(BASE_DIR, 'deluge_cli.log'))
fh.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
formatter = logging.Formatter('%(asctime)s %(levelname)8s %(name)s | %(message)s')
fh.setFormatter(formatter)
logger.addHandler(fh)
logger.addHandler(ch)

148
delugeClient/__main__.py Normal file
View File

@@ -0,0 +1,148 @@
#!/usr/bin/env python3.6
"""Custom delugeRPC client
Usage:
deluge_cli add MAGNET [DIR] [--json | --debug | --warning | --error]
deluge_cli search NAME [--json]
deluge_cli get TORRENT [--json | --debug | --warning | --error]
deluge_cli ls [--downloading | --seeding | --paused | --json]
deluge_cli toggle TORRENT
deluge_cli progress [--json]
deluge_cli rm NAME [--destroy] [--debug | --warning | --error]
deluge_cli (-h | --help)
deluge_cli --version
Arguments:
MAGNET Magnet link to add
DIR Directory to save to
TORRENT A selected torrent
Options:
-h --help Show this screen
--version Show version
--print Print response from commands
--json Print response as JSON
--debug Print all debug log
--warning Print only logged warnings
--error Print error messages (Error/Warning)
"""
import os
import sys
import signal
import logging
from docopt import docopt
from pprint import pprint
from deluge import Deluge
from utils import ColorizeFilter, BASE_DIR
from __init__ import __version__
logger = logging.getLogger('deluge_cli')
logger.setLevel(logging.DEBUG)
fh = logging.FileHandler(os.path.join(BASE_DIR, 'deluge_cli.log'))
fh.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
formatter = logging.Formatter('%(asctime)s %(levelname)8s %(name)s | %(message)s')
fh.setFormatter(formatter)
logger.addHandler(fh)
logger.addHandler(ch)
logger.addFilter(ColorizeFilter())
def signal_handler(signal, frame):
"""
Handle exit by Keyboardinterrupt
"""
logger.info('\nGood bye!')
sys.exit(0)
def main():
"""
Main function, parse the input
"""
signal.signal(signal.SIGINT, signal_handler)
arguments = docopt(__doc__, version=__version__)
# Set logging level for streamHandler
if arguments['--debug']:
ch.setLevel(logging.DEBUG)
elif arguments['--warning']:
ch.setLevel(logging.WARNING)
elif arguments['--error']:
ch.setLevel(logging.ERROR)
logger.info('Deluge client')
logger.debug(arguments)
# Get config settings
deluge = Deluge()
_id = arguments['TORRENT']
query = arguments['NAME']
magnet = arguments['MAGNET']
name = arguments['NAME']
_filter = [ a[2:] for a in ['--downloading', '--seeding', '--paused'] if arguments[a] ]
response = None
if arguments['add']:
logger.info('Add cmd selected with link {}'.format(magnet))
response = deluge.add(magnet)
if response is not None:
logger.info('Successfully added torrent.\nResponse from deluge: {}'.format(response))
else:
logger.warning('Add response returned empty: {}'.format(response))
elif arguments['search']:
logger.info('Search cmd selected for query: {}'.format(query))
response = deluge.search(query)
if response is not None or response != '[]':
logger.info('Search found {} torrents'.format(len(response)))
else:
logger.info('Empty response for search query.')
elif arguments['progress']:
logger.info('Progress cmd selected.')
response = deluge.progress()
elif arguments['get']:
logger.info('Get cmd selected for id: {}'.format(_id))
response = deluge.get(_id)
elif arguments['ls']:
logger.info('List cmd selected')
response = deluge.get_all(_filter=_filter)
elif arguments['toggle']:
logger.info('Toggling id: {}'.format(_id))
deluge.togglePaused(_id)
elif arguments['rm']:
destroy = arguments['--destroy']
logger.info('Remove by name: {}.'.format(name))
if destroy:
logger.info('Destroy set, removing files')
deluge.remove(name, destroy)
try:
if arguments['--json']:
if len(response) > 1:
print('[{}]'.format(','.join([t.toJSON() for t in response])))
else:
print(response[0].toJSON())
except KeyError as error:
logger.error('Unexpected error while trying to print')
raise error
return response
if __name__ == '__main__':
main()

11
delugeClient/default_config.ini Normal file
View File

@@ -0,0 +1,11 @@
[deluge]
host=
port=58846
user=
password=
[ssh]
host=
user=
password=
pkey=

167
delugeClient/deluge.py Normal file
View File

@@ -0,0 +1,167 @@
#!/usr/bin/env python3.6
import os
import re
import sys
import logging
import requests
import logging.config
from deluge_client import DelugeRPCClient
from sshtunnel import SSHTunnelForwarder
from delugeUtils import getConfig, BASE_DIR
from torrent import Torrent
logger = logging.getLogger('deluge_cli')
def split_words(string):
logger.debug('Splitting input: {} (type: {}) with split_words'.format(string, type(string)))
return re.findall(r"[\w\d']+", string.lower())
class Deluge(object):
"""docstring for ClassName"""
def __init__(self):
config = getConfig()
self.host = config['deluge']['host']
self.port = int(config['deluge']['port'])
self.user = config['deluge']['user']
self.password = config['deluge']['password']
self.ssh_host = config['ssh']['host']
self.ssh_user = config['ssh']['user']
self.ssh_pkey = config['ssh']['pkey']
self.ssh_password = config['ssh']['password']
self._connect()
def freeSpace(self):
return self.client.call('core.get_free_space')
def parseResponse(self, response):
torrents = []
for key in response:
torrent = response[key]
torrents.append(Torrent.fromDeluge(torrent))
return torrents
def _connect(self):
logger.info('Checking if script on same server as deluge RPC')
if self.host != 'localhost' and self.host is not None:
try:
if self.password:
self.tunnel = SSHTunnelForwarder(self.ssh_host, ssh_username=self.ssh_user, ssh_password=self.ssh_password,
local_bind_address=('localhost', self.port), remote_bind_address=('localhost', self.port))
elif self.pkey is not None:
self.tunnel = SSHTunnelForwarder(self.ssh_host, ssh_username=self.ssh_user, ssh_pkey=self.ssh_pkey,
local_bind_address=('localhost', self.port), remote_bind_address=('localhost', self.port))
except ValueError as error:
logger.error("Either password or private key path must be set in config.")
raise error
self.tunnel.start()
self.client = DelugeRPCClient(self.host, self.port, self.user, self.password)
self.client.connect()
def add(self, url):
logger.info('Adding magnet with url: {}.'.format(url))
if (url.startswith('magnet')):
return self.client.call('core.add_torrent_magnet', url, {})
elif url.startswith('http'):
magnet = self.getMagnetFromFile(url)
return self.client.call('core.add_torrent_magnet', magnet, {})
def get_all(self, _filter=None):
if (type(_filter) is list and len(_filter)):
if ('seeding' in _filter):
response = self.client.call('core.get_torrents_status', {'state': 'Seeding'}, [])
elif ('downloading' in _filter):
response = self.client.call('core.get_torrents_status', {'state': 'Downloading'}, [])
elif ('paused' in _filter):
response = self.client.call('core.get_torrents_status', {'paused': 'true'}, [])
else:
response = self.client.call('core.get_torrents_status', {}, [])
return self.parseResponse(response)
def search(self, query):
allTorrents = self.get_all()
torrentNamesMatchingQuery = []
if len(allTorrents):
for torrent in allTorrents:
if query in torrent.name:
torrentNamesMatchingQuery.append(torrent)
allTorrents = torrentNamesMatchingQuery
return allTorrents
q_list = split_words(query)
return [ t for t in self.get_all() if (set(q_list) <= set(split_words(t.name))) ]
def get(self, id):
response = self.client.call('core.get_torrent_status', id, {})
return Torrent.fromDeluge(response)
def togglePaused(self, id):
torrent = self.get(id)
if (torrent.paused):
response = self.client.call('core.resume_torrent', [id])
else:
response = self.client.call('core.pause_torrent', [id])
return response
def remove(self, name, destroy=False):
matches = list(filter(lambda t: t.name == name, self.get_all()))
logger.info('Matches for {}: {}'.format(name, matches))
if (len(matches) > 1):
raise ValueError('Multiple files found matching key. Unable to remove.')
elif (len(matches) == 1):
torrent = matches[0]
response = self.client.call('core.remove_torrent', torrent.key, destroy)
logger.info('Response: {}'.format(str(response)))
if (response == False):
raise AttributeError('Unable to remove torrent.')
return response
else:
logger.error('ERROR. No torrent found with that name.')
def filterOnValue(self, torrents, value):
filteredTorrents = []
for t in torrents:
value_template = {'key': None, 'name': None, value: None}
value_template['key'] = t.key
value_template['name'] = t.name
value_template[value] = getattr(t, value)
filteredTorrents.append(value_template)
return filteredTorrents
def progress(self):
attributes = ['progress', 'eta', 'state', 'finished']
all_torrents = self.get_all()
torrents = []
for i, attribute in enumerate(attributes):
if i < 1:
torrents = self.filterOnValue(all_torrents, attribute)
continue
torrents = [dict(e, **v) for e,v in zip(torrents, self.filterOnValue(all_torrents, attribute))]
return torrents
def __del__(self):
if hasattr(self, 'tunnel'):
logger.info('Closing ssh tunnel')
self.tunnel.stop()
def getMagnetFromFile(self, url):
logger.info('File url found, fetching magnet.')
r = requests.get(url, allow_redirects=False)
magnet = r.headers['Location']
logger.info('Found magnet: {}.'.format(magnet))
return magnet

81
delugeClient/delugeUtils.py Normal file
View File

@@ -0,0 +1,81 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @Author: kevinmidboe
# @Date: 2018-04-17 19:55:38
# @Last Modified by: KevinMidboe
# @Last Modified time: 2018-05-04 00:04:25
import os
import sys
import json
import shutil
import logging
import colored
import configparser
from pprint import pprint
from colored import stylize
__all__ = ('ColorizeFilter', )
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
logger = logging.getLogger('deluge_cli')
def checkConfigExists():
user_config_dir = os.path.expanduser("~") + "/.config/delugeClient"
config_dir = os.path.join(user_config_dir, 'config.ini')
def getConfig():
"""
Read path and get configuartion file with site settings
:return: config settings read from 'config.ini'
:rtype: configparser.ConfigParser
"""
config = configparser.ConfigParser()
user_config_dir = os.path.expanduser("~") + "/.config/delugeClient"
config_dir = os.path.join(user_config_dir, 'config.ini')
if not os.path.isfile(config_dir):
defaultConfig = os.path.join(BASE_DIR, 'default_config.ini')
logger.error('Missing config! Moved default.config.ini to {}.\nOpen this file and set all varaibles!'.format(config_dir))
os.makedirs(user_config_dir, exist_ok=True)
shutil.copyfile(defaultConfig, config_dir)
config.read(config_dir)
requiredParameters = [('deluge host', config['deluge']['host']), ('deluge port', config['deluge']['port']),
('deluge user', config['deluge']['user']), ('deluge password', config['deluge']['password']),
('ssh password', config['ssh']['user'])]
for key, value in requiredParameters:
if value == '':
logger.error('Missing value for variable: "{}" in config: \
"$HOME/.config/delugeClient/config.ini".'.format(key))
exit(1)
return config
class ColorizeFilter(logging.Filter):
"""
Class for setting specific colors to levels of severity for log output
"""
color_by_level = {
10: 'chartreuse_3b',
20: 'white',
30: 'orange_1',
40: 'red'
}
def filter(self, record):
record.raw_msg = record.msg
color = self.color_by_level.get(record.levelno)
if color:
record.msg = stylize(record.msg, colored.fg(color))
return True
def convert(data):
if isinstance(data, bytes): return data.decode('utf-8')
if isinstance(data, dict): return dict(map(convert, data.items()))
if isinstance(data, tuple): return map(convert, data)
json_data = json.dumps(data)
return json_data

0
deluge_socket.py → delugeClient/deluge_socket.py
View File

48
delugeClient/torrent.py Normal file
View File

@@ -0,0 +1,48 @@
import json
import logging
from distutils.util import strtobool
from utils import convert
logger = logging.getLogger('deluge_cli')
class Torrent(object):
def __init__(self, key, name, progress, eta, save_path, state, paused, finished, files):
super(Torrent, self).__init__()
self.key = key
self.name = name
self.progress = "{0:.2f}".format(float(progress))
self.eta = eta
self.save_path = save_path
self.state = state
self.paused = paused
self.finished = finished
self.files = list(files)
def isFolder(self):
return len(self.files) > 1
def toBool(self, value):
return True if strtobool(value) else False
@classmethod
def fromDeluge(cls, d):
# Receive a dict with byte values, convert all elements to string values
d = convert(d)
toBool = lambda val: True if strtobool(val) else False
return cls(d['hash'], d['name'], d['progress'], d['eta'], d['save_path'], d['state'],
toBool(d['paused']), toBool(d['is_finished']), d['files'])
def toJSON(self, files=False):
torrentDict = {'key': self.key, 'name': self.name, 'progress': self.progress, 'eta': self.eta,
'save_path': self.save_path, 'state': self.state, 'paused': self.paused,
'finished': self.finished, 'files': self.files, 'is_folder': self.isFolder()}
if (files is False):
del torrentDict['files']
return json.dumps(torrentDict)
def __str__(self):
return "Name: {}, Progress: {}%, ETA: {}, State: {}, Paused: {}".format(
self.name, self.progress, self.eta, self.state, self.paused)

309
deluge_cli.py
View File

@@ -1,309 +0,0 @@
#!/usr/bin/env python3.6
"""Custom delugeRPC client
Usage:
deluge_cli add MAGNET [DIR] [--debug | --warning | --error]
deluge_cli search NAME
deluge_cli get TORRENT
deluge_cli ls [--downloading | --seeding | --paused]
deluge_cli toggle TORRENT
deluge_cli progress
deluge_cli rm NAME [--destroy] [--debug | --warning | --error]
deluge_cli (-h | --help)
deluge_cli --version
Arguments:
MAGNET Magnet link to add
DIR Directory to save to
TORRENT A selected torrent
Options:
-h --help Show this screen
--version Show version
--debug Print all debug log
--warning Print only logged warnings
--error Print error messages (Error/Warning)
"""
import argparse
import os
import sys
import re
import signal
import socket
import logging
import logging.config
import configparser
from distutils.util import strtobool
from pprint import pprint
from deluge_client import DelugeRPCClient
from sshtunnel import SSHTunnelForwarder
from docopt import docopt
from utils import ColorizeFilter, convert
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
logger = logging.getLogger('deluge_cli')
logger.setLevel(logging.DEBUG)
fh = logging.FileHandler(os.path.join(BASE_DIR, 'deluge_cli.log'))
fh.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.ERROR)
formatter = logging.Formatter('%(asctime)s %(levelname)8s %(name)s | %(message)s')
fh.setFormatter(formatter)
logger.addHandler(fh)
logger.addHandler(ch)
logger.addFilter(ColorizeFilter())
def getConfig():
"""
Read path and get configuartion file with site settings
:return: config settings read from 'config.ini'
:rtype: configparser.ConfigParser
"""
config = configparser.ConfigParser()
config_dir = os.path.join(BASE_DIR, 'config.ini')
config.read(config_dir)
config_values = list(dict(config.items('Deluge')).values())
config_values.extend(list(dict(config.items('ssh')).values()))
if any(value.startswith('YOUR') for value in config_values):
raise ValueError('Please set variables in config.ini file.')
return config
def split_words(string):
logger.debug('Splitting input: {} (type: {}) with split_words'.format(string, type(string)))
return re.findall(r"[\w\d']+", string.lower())
class Deluge(object):
"""docstring for ClassName"""
def __init__(self):
config = getConfig()
self.host = config['Deluge']['HOST']
self.port = int(config['Deluge']['PORT'])
self.user = config['Deluge']['USER']
self.password = config['Deluge']['PASSWORD']
self.ssh_host = config['ssh']['HOST']
self.ssh_user = config['ssh']['USER']
self.ssh_pkey = config['ssh']['PKEY']
self._connect()
def parseResponse(self, response):
torrents = []
for key in response:
torrent = response[key]
torrents.append(Torrent.fromDeluge(torrent))
return torrents
def _connect(self):
logger.info('Checking if script on same server as deluge RPC')
if (socket.gethostbyname(socket.gethostname()) != self.host):
self.tunnel = SSHTunnelForwarder(self.ssh_host, ssh_username=self.ssh_user, ssh_pkey=self.ssh_pkey,
local_bind_address=('localhost', self.port), remote_bind_address=('localhost', self.port))
self.tunnel.start()
self.client = DelugeRPCClient(self.host, self.port, self.user, self.password)
self.client.connect()
def add(self, url):
if (url.startswith('magnet')):
return self.client.call('core.add_torrent_magnet', url, {})
def get_all(self, _filter=None):
if (type(_filter) is list and len(_filter)):
if ('seeding' in _filter):
response = self.client.call('core.get_torrents_status', {'state': 'Seeding'}, [])
elif ('downloading' in _filter):
response = self.client.call('core.get_torrents_status', {'state': 'Downloading'}, [])
elif ('paused' in _filter):
response = self.client.call('core.get_torrents_status', {'paused': 'true'}, [])
else:
response = self.client.call('core.get_torrents_status', {}, [])
return self.parseResponse(response)
def search(self, query):
q_list = split_words(query)
return [ t for t in self.get_all() if (set(q_list) <= set(split_words(t.name))) ]
def get(self, id):
response = self.client.call('core.get_torrent_status', id, {})
return Torrent.fromDeluge(response)
def togglePaused(self, id):
torrent = self.get(id)
if (torrent.paused):
response = self.client.call('core.resume_torrent', [id])
else:
response = self.client.call('core.pause_torrent', [id])
print('Response:', response)
def remove(self, name, destroy=False):
matches = list(filter(lambda t: t.name == name, self.get_all()))
logger.info('Matches for {}: {}'.format(name, matches))
if (len(matches) > 1):
raise ValueError('Multiple files found matching key. Unable to remove.')
elif (len(matches) == 1):
torrent = matches[0]
response = self.client.call('core.remove_torrent', torrent.key, destroy)
logger.info('Response: {}'.format(str(response)))
if (response == False):
raise AttributeError('Unable to remove torrent.')
return response
else:
logger.error('ERROR. No torrent found with that name.')
def filterOnValue(self, torrents, value):
filteredTorrents = []
for t in torrents:
value_template = {'key': None, 'name': None, value: None}
value_template['key'] = t.key
value_template['name'] = t.name
value_template[value] = getattr(t, value)
filteredTorrents.append(value_template)
return filteredTorrents
def progress(self):
attributes = ['progress', 'eta', 'state', 'finished']
all_torrents = self.get_all()
torrents = []
for i, attribute in enumerate(attributes):
if i < 1:
torrents = self.filterOnValue(all_torrents, attribute)
continue
torrents = [dict(e, **v) for e,v in zip(torrents, self.filterOnValue(all_torrents, attribute))]
return torrents
def __del__(self):
if hasattr(self, 'tunnel'):
logger.info('Closing ssh tunnel')
self.tunnel.stop()
class Torrent(object):
def __init__(self, key, name, progress, eta, save_path, state, paused, finished, files):
super(Torrent, self).__init__()
self.key = key
self.name = name
self.progress = "{0:.2f}".format(float(progress))
self.eta = eta
self.save_path = save_path
self.state = state
self.paused = paused
self.finished = finished
self.files = list(files)
def isFolder(self):
return len(self.files) > 1
def toBool(self, value):
return True if strtobool(value) else False
@classmethod
def fromDeluge(cls, d):
# Receive a dict with byte values, convert all elements to string values
d = convert(d)
toBool = lambda val: True if strtobool(val) else False
return cls(d['hash'], d['name'], d['progress'], d['eta'], d['save_path'], d['state'],
toBool(d['paused']), toBool(d['is_finished']), d['files'])
def toJSON(self):
return {'key': self.key, 'name': self.name, 'progress': self.progress, 'eta': self.eta,
'save_path': self.save_path, 'state': self.state, 'paused': self.paused,
'finished': self.finished, 'files': self.files, 'is_folder': self.isFolder()}
def __str__(self):
return "Name: {}, Progress: {}%, ETA: {}, State: {}, Paused: {}".format(
self.name, self.progress, self.eta, self.state, self.paused)
def signal_handler(signal, frame):
"""
Handle exit by Keyboardinterrupt
"""
logger.info('\nGood bye!')
sys.exit(0)
def main():
"""
Main function, parse the input
"""
signal.signal(signal.SIGINT, signal_handler)
arguments = docopt(__doc__, version='1')
# Set logging level for streamHandler
if arguments['--debug']:
ch.setLevel(logging.DEBUG)
elif arguments['--warning']:
ch.setLevel(logging.WARNING)
elif arguments['--error']:
ch.setLevel(logging.ERROR)
logger.info('Deluge client')
logger.debug(arguments)
# Get config settings
deluge = Deluge()
_id = arguments['TORRENT']
query = arguments['NAME']
magnet = arguments['MAGNET']
name = arguments['NAME']
_filter = [ a[2:] for a in ['--downloading', '--seeding', '--paused'] if arguments[a] ]
print(_id, query, _filter)
if arguments['add']:
logger.info('Add cmd selected with link {}'.format(magnet))
response = deluge.add(magnet)
print('Add response: ', response)
elif arguments['search']:
logger.info('Search cmd selected for query: {}'.format(query))
response = deluge.search(query)
[ pprint(t.toJSON()) for t in response ]
elif arguments['progress']:
logger.info('Progress cmd selected.')
pprint(deluge.progress())
exit(0)
[ pprint(t.toJSON()) for t in deluge.progress() ]
elif arguments['get']:
logger.info('Get cmd selected for id: {}'.format(_id))
response = deluge.get(_id)
pprint(response.toJSON())
elif arguments['ls']:
logger.info('List cmd selected')
[ pprint(t.toJSON()) for t in deluge.get_all(_filter=_filter) ]
elif arguments['toggle']:
logger.info('Toggling id: {}'.format(_id))
deluge.togglePaused(_id)
elif arguments['rm']:
destroy = arguments['--destroy']
logger.info('Remove by name: {}.'.format(name))
if destroy:
logger.info('Destroy set, removing files')
deluge.remove(name, destroy)
if __name__ == '__main__':
main()

16
setup.py
View File

@@ -1,10 +1,14 @@
import setuptools
from setuptools import setup, find_packages
from sys import path
from os.path import dirname
path.append(dirname(__file__))
import delugeClient
with open("README.md", "r", encoding="utf-8") as fh:
long_description = fh.read()
setuptools.setup(
setup(
name="delugeClient",
version=delugeClient.__version__,
author="KevinMidboe",
@@ -27,10 +31,12 @@ setuptools.setup(
],
entry_points={
'console_scripts': [
'delugeClient = delugeClient.__main__:main',
'delugeclient = delugeClient.__main__:main',
],
},
package_dir={"": "delugeClient"},
packages=setuptools.find_packages(where="delugeClient"),
packages=find_packages(),
package_data={
'delugeClient': ['default_config.ini'],
},
python_requires=">=3.6",
)

42
utils.py
View File

@@ -1,42 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @Author: kevinmidboe
# @Date: 2018-04-17 19:55:38
# @Last Modified by: KevinMidboe
# @Last Modified time: 2018-05-04 00:04:25
import logging
import colored
import json
from pprint import pprint
from colored import stylize
__all__ = ('ColorizeFilter', )
class ColorizeFilter(logging.Filter):
"""
Class for setting specific colors to levels of severity for log output
"""
color_by_level = {
10: 'chartreuse_3b',
20: 'white',
30: 'orange_1',
40: 'red'
}
logger = logging.getLogger('deluge_cli')
def filter(self, record):
record.raw_msg = record.msg
color = self.color_by_level.get(record.levelno)
if color:
record.msg = stylize(record.msg, colored.fg(color))
return True
def convert(data):
if isinstance(data, bytes): return data.decode('utf-8')
if isinstance(data, dict): return dict(map(convert, data.items()))
if isinstance(data, tuple): return map(convert, data)
json_data = json.dumps(data)
return json_data