diff options
Diffstat (limited to 'deluge/plugins/AutoAdd/deluge_autoadd/core.py')
| -rw-r--r-- | deluge/plugins/AutoAdd/deluge_autoadd/core.py | 528 |
1 files changed, 528 insertions, 0 deletions
diff --git a/deluge/plugins/AutoAdd/deluge_autoadd/core.py b/deluge/plugins/AutoAdd/deluge_autoadd/core.py new file mode 100644 index 0000000..271d5f0 --- /dev/null +++ b/deluge/plugins/AutoAdd/deluge_autoadd/core.py @@ -0,0 +1,528 @@ +# +# Copyright (C) 2009 GazpachoKing <chase.sterling@gmail.com> +# Copyright (C) 2011 Pedro Algarvio <pedro@algarvio.me> +# +# Basic plugin template created by: +# Copyright (C) 2008 Martijn Voncken <mvoncken@gmail.com> +# Copyright (C) 2007-2009 Andrew Resch <andrewresch@gmail.com> +# Copyright (C) 2009 Damien Churchill <damoxc@gmail.com> +# +# This file is part of Deluge and is licensed under GNU General Public License 3.0, or later, with +# the additional special exception to link portions of this program with the OpenSSL library. +# See LICENSE for more details. +# + +import logging +import os +import shutil +from base64 import b64encode + +from twisted.internet import reactor +from twisted.internet.defer import maybeDeferred +from twisted.internet.task import LoopingCall, deferLater +from twisted.python.failure import Failure + +import deluge.component as component +import deluge.configmanager +from deluge._libtorrent import lt +from deluge.common import AUTH_LEVEL_ADMIN, is_magnet +from deluge.core.rpcserver import export +from deluge.error import AddTorrentError, InvalidTorrentError +from deluge.event import DelugeEvent +from deluge.plugins.pluginbase import CorePluginBase + +log = logging.getLogger(__name__) + + +DEFAULT_PREFS = {'watchdirs': {}, 'next_id': 1} + + +OPTIONS_AVAILABLE = { # option: builtin + 'enabled': False, + 'path': False, + 'append_extension': False, + 'copy_torrent': False, + 'delete_copy_torrent_toggle': False, + 'abspath': False, + 'download_location': True, + 'max_download_speed': True, + 'max_upload_speed': True, + 'max_connections': True, + 'max_upload_slots': True, + 'prioritize_first_last': True, + 'auto_managed': True, + 'stop_at_ratio': True, + 'stop_ratio': True, + 'remove_at_ratio': True, + 'move_completed': True, + 'move_completed_path': True, + 'label': False, + 'add_paused': True, + 'queue_to_top': False, + 'owner': True, + 'seed_mode': True, +} + +MAX_NUM_ATTEMPTS = 10 + + +class AutoaddOptionsChangedEvent(DelugeEvent): + """Emitted when the options for the plugin are changed.""" + + def __init__(self): + pass + + +def check_input(cond, message): + if not cond: + raise Exception(message) + + +class Core(CorePluginBase): + def enable(self): + # reduce typing, assigning some values to self... + self.config = deluge.configmanager.ConfigManager('autoadd.conf', DEFAULT_PREFS) + self.config.run_converter((0, 1), 2, self.__migrate_config_1_to_2) + self.config.save() + self.watchdirs = self.config['watchdirs'] + + self.rpcserver = component.get('RPCServer') + component.get('EventManager').register_event_handler( + 'PreTorrentRemovedEvent', self.__on_pre_torrent_removed + ) + + # Dict of Filename:Attempts + self.invalid_torrents = {} + # Loopingcall timers for each enabled watchdir + self.update_timers = {} + deferLater(reactor, 5, self.enable_looping) + + def enable_looping(self): + # Enable all looping calls for enabled watchdirs here + for watchdir_id, watchdir in self.watchdirs.items(): + if watchdir['enabled']: + self.enable_watchdir(watchdir_id) + + def disable(self): + # disable all running looping calls + component.get('EventManager').deregister_event_handler( + 'PreTorrentRemovedEvent', self.__on_pre_torrent_removed + ) + for loopingcall in self.update_timers.values(): + loopingcall.stop() + self.config.save() + + def update(self): + pass + + @export + def set_options(self, watchdir_id, options): + """Update the options for a watch folder.""" + watchdir_id = str(watchdir_id) + options = self._make_unicode(options) + check_input(watchdir_id in self.watchdirs, _('Watch folder does not exist.')) + if 'path' in options: + options['abspath'] = os.path.abspath(options['path']) + check_input(os.path.isdir(options['abspath']), _('Path does not exist.')) + for w_id, w in self.watchdirs.items(): + if options['abspath'] == w['abspath'] and watchdir_id != w_id: + raise Exception('Path is already being watched.') + for key in options: + if key not in OPTIONS_AVAILABLE: + if key not in [key2 + '_toggle' for key2 in OPTIONS_AVAILABLE]: + raise Exception('autoadd: Invalid options key:%s' % key) + # disable the watch loop if it was active + if watchdir_id in self.update_timers: + self.disable_watchdir(watchdir_id) + + self.watchdirs[watchdir_id].update(options) + # re-enable watch loop if appropriate + if self.watchdirs[watchdir_id]['enabled']: + self.enable_watchdir(watchdir_id) + self.config.save() + component.get('EventManager').emit(AutoaddOptionsChangedEvent()) + + def load_torrent(self, filename, magnet): + log.debug('Attempting to open %s for add.', filename) + file_mode = 'r' if magnet else 'rb' + try: + with open(filename, file_mode) as _file: + filedump = _file.read() + except OSError as ex: + log.warning('Unable to open %s: %s', filename, ex) + raise ex + + if not filedump: + raise EOFError('Torrent is 0 bytes!') + + # Get the info to see if any exceptions are raised + if not magnet: + decoded_torrent = lt.bdecode(filedump) + if decoded_torrent is None: + raise InvalidTorrentError('Torrent file failed decoding.') + lt.torrent_info(decoded_torrent) + + return filedump + + def split_magnets(self, filename): + log.debug('Attempting to open %s for splitting magnets.', filename) + magnets = [] + try: + with open(filename) as _file: + magnets = list(filter(len, _file.read().splitlines())) + except OSError as ex: + log.warning('Unable to open %s: %s', filename, ex) + + if len(magnets) < 2: + return [] + + path = filename.rsplit(os.sep, 1)[0] + for magnet in magnets: + if not is_magnet(magnet): + log.warning('Found line which is not a magnet: %s', magnet) + continue + + for part in magnet.split('&'): + if part.startswith('dn='): + name = part[3:].strip() + if name: + mname = os.sep.join([path, name + '.magnet']) + break + else: + short_hash = magnet.split('btih:')[1][:8] + mname = '.'.join([os.path.splitext(filename)[0], short_hash, 'magnet']) + + try: + with open(mname, 'w') as _mfile: + _mfile.write(magnet) + except OSError as ex: + log.warning('Unable to open %s: %s', mname, ex) + return magnets + + def update_watchdir(self, watchdir_id): + """Check the watch folder for new torrents to add.""" + log.trace('Updating watchdir id: %s', watchdir_id) + watchdir_id = str(watchdir_id) + watchdir = self.watchdirs[watchdir_id] + if not watchdir['enabled']: + # We shouldn't be updating because this watchdir is not enabled + log.debug('Watchdir id %s is not enabled. Disabling it.', watchdir_id) + self.disable_watchdir(watchdir_id) + return + + if not os.path.isdir(watchdir['abspath']): + log.warning('Invalid AutoAdd folder: %s', watchdir['abspath']) + self.disable_watchdir(watchdir_id) + return + + # Generate options dict for watchdir + options = {} + if 'stop_at_ratio_toggle' in watchdir: + watchdir['stop_ratio_toggle'] = watchdir['stop_at_ratio_toggle'] + # We default to True when reading _toggle values, so a config + # without them is valid, and applies all its settings. + for option, value in watchdir.items(): + if OPTIONS_AVAILABLE.get(option): + if watchdir.get(option + '_toggle', True) or option in [ + 'owner', + 'seed_mode', + ]: + options[option] = value + + # Check for .magnet files containing multiple magnet links and + # create a new .magnet file for each of them. + for filename in os.listdir(watchdir['abspath']): + try: + filepath = os.path.join(watchdir['abspath'], filename) + except UnicodeDecodeError as ex: + log.error( + 'Unable to auto add torrent due to improper filename encoding: %s', + ex, + ) + continue + if os.path.isdir(filepath): + # Skip directories + continue + elif os.path.splitext(filename)[1] == '.magnet' and self.split_magnets( + filepath + ): + os.remove(filepath) + + for filename in os.listdir(watchdir['abspath']): + try: + filepath = os.path.join(watchdir['abspath'], filename) + except UnicodeDecodeError as ex: + log.error( + 'Unable to auto add torrent due to improper filename encoding: %s', + ex, + ) + continue + + if os.path.isdir(filepath): + # Skip directories + continue + + ext = os.path.splitext(filename)[1].lower() + magnet = ext == '.magnet' + if not magnet and not ext == '.torrent': + log.debug('File checked for auto-loading is invalid: %s', filename) + continue + + try: + filedump = self.load_torrent(filepath, magnet) + except (OSError, EOFError, RuntimeError, InvalidTorrentError) as ex: + # If torrent is invalid, keep track of it so can try again on the next pass. + # This catches torrent files that may not be fully saved to disk at load time. + log.debug('Torrent is invalid: %s', ex) + if filename in self.invalid_torrents: + self.invalid_torrents[filename] += 1 + if self.invalid_torrents[filename] >= MAX_NUM_ATTEMPTS: + log.warning( + 'Maximum attempts reached while trying to add the ' + 'torrent file with the path %s', + filepath, + ) + os.rename(filepath, filepath + '.invalid') + del self.invalid_torrents[filename] + else: + self.invalid_torrents[filename] = 1 + continue + + def on_torrent_added(torrent_id, filename, filepath): + if 'Label' in component.get('CorePluginManager').get_enabled_plugins(): + if watchdir.get('label_toggle', True) and watchdir.get('label'): + label = component.get('CorePlugin.Label') + if not watchdir['label'] in label.get_labels(): + label.add(watchdir['label']) + try: + label.set_torrent(torrent_id, watchdir['label']) + except Exception as ex: + log.error('Unable to set label: %s', ex) + + if ( + watchdir.get('queue_to_top_toggle', True) + and 'queue_to_top' in watchdir + ): + if watchdir['queue_to_top']: + component.get('TorrentManager').queue_top(torrent_id) + else: + component.get('TorrentManager').queue_bottom(torrent_id) + + # Rename, copy or delete the torrent once added to deluge. + if watchdir.get('append_extension_toggle'): + if not watchdir.get('append_extension'): + watchdir['append_extension'] = '.added' + os.rename(filepath, filepath + watchdir['append_extension']) + elif watchdir.get('copy_torrent_toggle'): + copy_torrent_path = watchdir['copy_torrent'] + copy_torrent_file = os.path.join(copy_torrent_path, filename) + log.debug( + 'Moving added torrent file "%s" to "%s"', + os.path.basename(filepath), + copy_torrent_path, + ) + shutil.move(filepath, copy_torrent_file) + else: + os.remove(filepath) + + def fail_torrent_add(err_msg, filepath, magnet): + if isinstance(err_msg, Failure): + err_msg = err_msg.getErrorMessage() + + # torrent handle is invalid and so is the magnet link + log.error( + 'Cannot Autoadd %s: %s: %s', + 'magnet' if magnet else 'torrent file', + filepath, + err_msg, + ) + os.rename(filepath, filepath + '.invalid') + + try: + # The torrent looks good, so lets add it to the session. + if magnet: + d = maybeDeferred( + component.get('Core').add_torrent_magnet, + filedump.strip(), + options, + ) + else: + d = component.get('Core').add_torrent_file_async( + filename, b64encode(filedump), options + ) + d.addCallback(on_torrent_added, filename, filepath) + d.addErrback(fail_torrent_add, filepath, magnet) + except AddTorrentError as ex: + fail_torrent_add(str(ex), filepath, magnet) + + def on_update_watchdir_error(self, failure, watchdir_id): + """Disables any watch folders with un-handled exceptions.""" + self.disable_watchdir(watchdir_id) + log.error( + 'Disabling "%s", error during update: %s', + self.watchdirs[watchdir_id]['path'], + failure, + ) + + @export + def enable_watchdir(self, watchdir_id): + w_id = str(watchdir_id) + # Enable the looping call + if w_id not in self.update_timers or not self.update_timers[w_id].running: + self.update_timers[w_id] = LoopingCall(self.update_watchdir, w_id) + self.update_timers[w_id].start(5).addErrback( + self.on_update_watchdir_error, w_id + ) + # Update the config + if not self.watchdirs[w_id]['enabled']: + self.watchdirs[w_id]['enabled'] = True + self.config.save() + component.get('EventManager').emit(AutoaddOptionsChangedEvent()) + + @export + def disable_watchdir(self, watchdir_id): + w_id = str(watchdir_id) + # Disable the looping call + if w_id in self.update_timers: + if self.update_timers[w_id].running: + self.update_timers[w_id].stop() + del self.update_timers[w_id] + # Update the config + if self.watchdirs[w_id]['enabled']: + self.watchdirs[w_id]['enabled'] = False + self.config.save() + component.get('EventManager').emit(AutoaddOptionsChangedEvent()) + + @export + def set_config(self, config): + """Sets the config dictionary.""" + config = self._make_unicode(config) + for key in config: + self.config[key] = config[key] + self.config.save() + component.get('EventManager').emit(AutoaddOptionsChangedEvent()) + + @export + def get_config(self): + """Returns the config dictionary.""" + return self.config.config + + @export + def get_watchdirs(self): + session_user = self.rpcserver.get_session_user() + session_auth_level = self.rpcserver.get_session_auth_level() + if session_auth_level == AUTH_LEVEL_ADMIN: + log.debug( + 'Current logged in user %s is an ADMIN, send all ' 'watchdirs', + session_user, + ) + return self.watchdirs + + watchdirs = {} + for watchdir_id, watchdir in self.watchdirs.items(): + if watchdir.get('owner', 'localclient') == session_user: + watchdirs[watchdir_id] = watchdir + + log.debug( + 'Current logged in user %s is not an ADMIN, send only ' + 'their watchdirs: %s', + session_user, + list(watchdirs), + ) + return watchdirs + + def _make_unicode(self, options): + opts = {} + for key in options: + if isinstance(options[key], bytes): + options[key] = options[key].decode('utf8') + opts[key] = options[key] + return opts + + @export + def add(self, options=None): + """Add a watch folder.""" + if options is None: + options = {} + options = self._make_unicode(options) + abswatchdir = os.path.abspath(options['path']) + check_input(os.path.isdir(abswatchdir), _('Path does not exist.')) + check_input( + os.access(abswatchdir, os.R_OK | os.W_OK), + 'You must have read and write access to watch folder.', + ) + if abswatchdir in [wd['abspath'] for wd in self.watchdirs.values()]: + raise Exception('Path is already being watched.') + options.setdefault('enabled', False) + options['abspath'] = abswatchdir + watchdir_id = self.config['next_id'] + self.watchdirs[str(watchdir_id)] = options + if options.get('enabled'): + self.enable_watchdir(watchdir_id) + self.config['next_id'] = watchdir_id + 1 + self.config.save() + component.get('EventManager').emit(AutoaddOptionsChangedEvent()) + return watchdir_id + + @export + def remove(self, watchdir_id): + """Remove a watch folder.""" + watchdir_id = str(watchdir_id) + check_input( + watchdir_id in self.watchdirs, 'Unknown Watchdir: %s' % self.watchdirs + ) + if self.watchdirs[watchdir_id]['enabled']: + self.disable_watchdir(watchdir_id) + del self.watchdirs[watchdir_id] + self.config.save() + component.get('EventManager').emit(AutoaddOptionsChangedEvent()) + + def __migrate_config_1_to_2(self, config): + for watchdir_id in config['watchdirs']: + config['watchdirs'][watchdir_id]['owner'] = 'localclient' + return config + + def __on_pre_torrent_removed(self, torrent_id): + try: + torrent = component.get('TorrentManager')[torrent_id] + except KeyError: + log.warning( + 'Unable to remove torrent file for torrent id %s. It' + 'was already deleted from the TorrentManager', + torrent_id, + ) + return + torrent_fname = torrent.filename + for watchdir in self.watchdirs.values(): + if not watchdir.get('copy_torrent_toggle', False): + # This watchlist does copy torrents + continue + elif not watchdir.get('delete_copy_torrent_toggle', False): + # This watchlist is not set to delete finished torrents + continue + copy_torrent_path = watchdir['copy_torrent'] + torrent_fname_path = os.path.join(copy_torrent_path, torrent_fname) + if os.path.isfile(torrent_fname_path): + try: + os.remove(torrent_fname_path) + log.info( + 'Removed torrent file "%s" from "%s"', + torrent_fname, + copy_torrent_path, + ) + break + except OSError as ex: + log.info( + 'Failed to removed torrent file "%s" from "%s": %s', + torrent_fname, + copy_torrent_path, + ex, + ) + + @export + def is_admin_level(self): + return self.rpcserver.get_session_auth_level() == deluge.common.AUTH_LEVEL_ADMIN + + @export + def get_auth_user(self): + return self.rpcserver.get_session_user() |