Coverage for deluge/ui/console/main : 0%

# -*- coding: utf-8 -*- 

# 

# Copyright (C) 2008-2009 Ido Abramovich <ido.deluge@gmail.com> 

# Copyright (C) 2009 Andrew Resch <andrewresch@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. 

# 

from __future__ import print_function 

import locale 

import logging 

import optparse 

import os 

import re 

import shlex 

import sys 

from twisted.internet import defer, reactor 

import deluge.common 

import deluge.component as component 

from deluge.ui.client import client 

from deluge.ui.console import UI_PATH, colors 

from deluge.ui.console.eventlog import EventLog 

from deluge.ui.console.statusbars import StatusBars 

from deluge.ui.coreconfig import CoreConfig 

from deluge.ui.sessionproxy import SessionProxy 

from deluge.ui.ui import _UI 

log = logging.getLogger(__name__) 

class Console(_UI): 

    help = """Starts the Deluge console interface""" 

    def __init__(self): 

        super(Console, self).__init__("console") 

        group = optparse.OptionGroup(self.parser, "Console Options", "These daemon connect options will be " 

                                     "used for commands, or if console ui autoconnect is enabled.") 

        group.add_option("-d", "--daemon", dest="daemon_addr") 

        group.add_option("-p", "--port", dest="daemon_port", type="int") 

        group.add_option("-u", "--username", dest="daemon_user") 

        group.add_option("-P", "--password", dest="daemon_pass") 

        self.parser.add_option_group(group) 

        self.parser.disable_interspersed_args() 

        self.console_cmds = load_commands(os.path.join(UI_PATH, "commands")) 

        class CommandOptionGroup(optparse.OptionGroup): 

            def __init__(self, parser, title, description=None, cmds=None): 

                optparse.OptionGroup.__init__(self, parser, title, description) 

                self.cmds = cmds 

            def format_help(self, formatter): 

                result = formatter.format_heading(self.title) 

                formatter.indent() 

                if self.description: 

                    result += "%s\n" % formatter.format_description(self.description) 

                for cname in self.cmds: 

                    cmd = self.cmds[cname] 

                    if cmd.interactive_only or cname in cmd.aliases: 

                        continue 

                    allnames = [cname] 

                    allnames.extend(cmd.aliases) 

                    cname = "/".join(allnames) 

                    result += formatter.format_heading(" - ".join([cname, cmd.__doc__])) 

                    formatter.indent() 

                    result += "%*s%s\n" % (formatter.current_indent, "", cmd.usage.split("\n")[0]) 

                    formatter.dedent() 

                formatter.dedent() 

                return result 

        cmd_group = CommandOptionGroup(self.parser, "Console Commands", 

                                       description="""These commands can be issued from the command line. 

                                                    They require quoting and multiple commands separated by ';' 

                                                    e.g. Pause torrent with id 'abcd' and get information for id 'efgh': 

                                                    `%s \"pause abcd; info efgh\"`""" 

                                       % os.path.basename(sys.argv[0]), cmds=self.console_cmds) 

        self.parser.add_option_group(cmd_group) 

    def start(self): 

        super(Console, self).start() 

        ConsoleUI(self.args, self.console_cmds, (self.options.daemon_addr, self.options.daemon_port, 

                                                 self.options.daemon_user, self.options.daemon_pass)) 

def start(): 

    Console().start() 

class DelugeHelpFormatter (optparse.IndentedHelpFormatter): 

    """ 

    Format help in a way suited to deluge Legacy mode - colors, format, indentation... 

    """ 

    replace_dict = { 

        "<torrent-id>": "{!green!}%s{!input!}", 

        "<state>": "{!yellow!}%s{!input!}", 

        "\.\.\.": "{!yellow!}%s{!input!}", 

        "\s\*\s": "{!blue!}%s{!input!}", 

        "(?<![\-a-z])(-[a-zA-Z0-9])": "{!red!}%s{!input!}", 

        # "(\-[a-zA-Z0-9])": "{!red!}%s{!input!}", 

        "--[_\-a-zA-Z0-9]+": "{!green!}%s{!input!}", 

        "(\[|\])": "{!info!}%s{!input!}", 

        "<tab>": "{!white!}%s{!input!}", 

        "[_A-Z]{3,}": "{!cyan!}%s{!input!}", 

        "<download-folder>": "{!yellow!}%s{!input!}", 

        "<torrent-file>": "{!green!}%s{!input!}" 

    } 

    def __init__(self, 

                 indent_increment=2, 

                 max_help_position=24, 

                 width=None, 

                 short_first=1): 

        optparse.IndentedHelpFormatter.__init__( 

            self, indent_increment, max_help_position, width, short_first) 

    def _format_colors(self, string): 

        def r(repl): 

            return lambda s: repl % s.group() 

        for key, replacement in self.replace_dict.items(): 

            string = re.sub(key, r(replacement), string) 

        return string 

    def format_usage(self, usage): 

        return _("{!info!}Usage{!input!}: %s\n") % self._format_colors(usage) 

    def format_option(self, option): 

        result = [] 

        opts = self.option_strings[option] 

        opt_width = self.help_position - self.current_indent - 2 

        if len(opts) > opt_width: 

            opts = "%*s%s\n" % (self.current_indent, "", opts) 

            opts = self._format_colors(opts) 

            indent_first = self.help_position 

        else:  # start help on same line as opts 

            opts = "%*s%-*s  " % (self.current_indent, "", opt_width, opts) 

            opts = self._format_colors(opts) 

            indent_first = 0 

        result.append(opts) 

        if option.help: 

            help_text = self.expand_default(option) 

            help_text = self._format_colors(help_text) 

            help_lines = optparse.textwrap.wrap(help_text, self.help_width) 

            result.append("%*s%s\n" % (indent_first, "", help_lines[0])) 

            result.extend(["%*s%s\n" % (self.help_position, "", line) 

                           for line in help_lines[1:]]) 

        elif opts[-1] != "\n": 

            result.append("\n") 

        return "".join(result) 

class OptionParser(optparse.OptionParser): 

    """subclass from optparse.OptionParser so exit() won't exit.""" 

    def __init__(self, **kwargs): 

        optparse.OptionParser.__init__(self, **kwargs) 

        self.formatter = DelugeHelpFormatter() 

    def exit(self, status=0, msg=None): 

        self.values._exit = True 

        if msg: 

            print(msg) 

    def error(self, msg): 

        """error(msg : string) 

           Print a usage message incorporating 'msg' to stderr and exit. 

           If you override this in a subclass, it should not return -- it 

           should either exit or raise an exception. 

        """ 

        raise Exception(msg) 

    def print_usage(self, file=None): 

        console = component.get("ConsoleUI") 

        if self.usage: 

            for line in self.get_usage().splitlines(): 

                console.write(line) 

    def print_help(self, file=None): 

        console = component.get("ConsoleUI") 

        console.set_batch_write(True) 

        for line in self.format_help().splitlines(): 

            console.write(line) 

        console.set_batch_write(False) 

    def format_option_help(self, formatter=None): 

        if formatter is None: 

            formatter = self.formatter 

        formatter.store_option_strings(self) 

        result = [] 

        result.append(formatter.format_heading(_("{!info!}Options{!input!}"))) 

        formatter.indent() 

        if self.option_list: 

            result.append(optparse.OptionContainer.format_option_help(self, formatter)) 

            result.append("\\n") 

        for group in self.option_groups: 

            result.append(group.format_help(formatter)) 

            result.append("\\n") 

        formatter.dedent() 

        # Drop the last "\\n", or the header if no options or option groups: 

        return "".join(result[:-1]) 

class BaseCommand(object): 

    usage = "usage" 

    interactive_only = False 

    option_list = tuple() 

    aliases = [] 

    def complete(self, text, *args): 

        return [] 

    def handle(self, *args, **options): 

        pass 

    @property 

    def name(self): 

        return "base" 

    @property 

    def epilog(self): 

        return self.__doc__ 

    def split(self, text): 

        if deluge.common.windows_check(): 

            text = text.replace("\\", "\\\\") 

        result = shlex.split(text) 

        for i, s in enumerate(result): 

            result[i] = s.replace(r"\ ", " ") 

        result = filter(lambda s: s != "", result) 

        return result 

    def create_parser(self): 

        return OptionParser(prog=self.name, usage=self.usage, epilog=self.epilog, option_list=self.option_list) 

def load_commands(command_dir, exclude=[]): 

    def get_command(name): 

        return getattr(__import__("deluge.ui.console.commands.%s" % name, {}, {}, ["Command"]), "Command")() 

    try: 

        commands = [] 

        for filename in os.listdir(command_dir): 

            if filename.split(".")[0] in exclude or filename.startswith("_"): 

                continue 

            if not (filename.endswith(".py") or filename.endswith(".pyc")): 

                continue 

            cmd = get_command(filename.split(".")[len(filename.split(".")) - 2]) 

            aliases = [filename.split(".")[len(filename.split(".")) - 2]] 

            aliases.extend(cmd.aliases) 

            for a in aliases: 

                commands.append((a, cmd)) 

        return dict(commands) 

    except OSError: 

        return {} 

class ConsoleUI(component.Component): 

    def __init__(self, args=None, cmds=None, daemon=None): 

        component.Component.__init__(self, "ConsoleUI", 2) 

        # keep track of events for the log view 

        self.events = [] 

        try: 

            locale.setlocale(locale.LC_ALL, "") 

            self.encoding = locale.getpreferredencoding() 

        except: 

            self.encoding = sys.getdefaultencoding() 

        log.debug("Using encoding: %s", self.encoding) 

        # start up the session proxy 

        self.sessionproxy = SessionProxy() 

        client.set_disconnect_callback(self.on_client_disconnect) 

        # Set the interactive flag to indicate where we should print the output 

        self.interactive = True 

        self._commands = cmds 

        if args: 

            args = " ".join(args) 

            self.interactive = False 

            if not cmds: 

                print("Sorry, couldn't find any commands") 

                return 

            else: 

                from deluge.ui.console.commander import Commander 

                cmdr = Commander(cmds) 

                if daemon: 

                    cmdr.exec_args(args, *daemon) 

                else: 

                    cmdr.exec_args(args, None, None, None, None) 

        self.coreconfig = CoreConfig() 

        if self.interactive and not deluge.common.windows_check(): 

            # We use the curses.wrapper function to prevent the console from getting 

            # messed up if an uncaught exception is experienced. 

            import curses.wrapper 

            curses.wrapper(self.run) 

        elif self.interactive and deluge.common.windows_check(): 

            print("""\nDeluge-console does not run in interactive mode on Windows. \n 

Please use commands from the command line, eg:\n 

    deluge-console.exe help 

    deluge-console.exe info 

    deluge-console.exe "add --help" 

    deluge-console.exe "add -p c:\\mytorrents c:\\new.torrent" 

            """) 

        else: 

            reactor.run() 

    def run(self, stdscr): 

        """ 

        This method is called by the curses.wrapper to start the mainloop and 

        screen. 

        :param stdscr: curses screen passed in from curses.wrapper 

        """ 

        # We want to do an interactive session, so start up the curses screen and 

        # pass it the function that handles commands 

        colors.init_colors() 

        self.statusbars = StatusBars() 

        from deluge.ui.console.modes.connectionmanager import ConnectionManager 

        self.stdscr = stdscr 

        self.screen = ConnectionManager(stdscr, self.encoding) 

        self.eventlog = EventLog() 

        self.screen.topbar = "{!status!}Deluge " + deluge.common.get_version() + " Console" 

        self.screen.bottombar = "{!status!}" 

        self.screen.refresh() 

        # The Screen object is designed to run as a twisted reader so that it 

        # can use twisted's select poll for non-blocking user input. 

        reactor.addReader(self.screen) 

        # Start the twisted mainloop 

        reactor.run() 

    def start(self): 

        # Maintain a list of (torrent_id, name) for use in tab completion 

        self.torrents = [] 

        if not self.interactive: 

            self.started_deferred = defer.Deferred() 

            def on_session_state(result): 

                def on_torrents_status(torrents): 

                    for torrent_id, status in torrents.items(): 

                        self.torrents.append((torrent_id, status["name"])) 

                    self.started_deferred.callback(True) 

                client.core.get_torrents_status({"id": result}, ["name"]).addCallback(on_torrents_status) 

            client.core.get_session_state().addCallback(on_session_state) 

    def match_torrent(self, string): 

        """ 

        Returns a list of torrent_id matches for the string.  It will search both 

        torrent_ids and torrent names, but will only return torrent_ids. 

        :param string: str, the string to match on 

        :returns: list of matching torrent_ids. Will return an empty list if 

            no matches are found. 

        """ 

        if self.interactive and isinstance(self.screen, deluge.ui.console.modes.legacy.Legacy): 

            return self.screen.match_torrent(string) 

        matches = [] 

        string = string.decode(self.encoding) 

        for tid, name in self.torrents: 

            if tid.startswith(string) or name.startswith(string): 

                matches.append(tid) 

        return matches 

    def get_torrent_name(self, torrent_id): 

        if self.interactive and hasattr(self.screen, "get_torrent_name"): 

            return self.screen.get_torrent_name(torrent_id) 

        for tid, name in self.torrents: 

            if torrent_id == tid: 

                return name 

        return None 

    def set_batch_write(self, batch): 

        if self.interactive and isinstance(self.screen, deluge.ui.console.modes.legacy.Legacy): 

            return self.screen.set_batch_write(batch) 

    def tab_complete_torrent(self, line): 

        if self.interactive and isinstance(self.screen, deluge.ui.console.modes.legacy.Legacy): 

            return self.screen.tab_complete_torrent(line) 

    def tab_complete_path(self, line, type="file", ext="", sort="name", dirs_first=True): 

        if self.interactive and isinstance(self.screen, deluge.ui.console.modes.legacy.Legacy): 

            return self.screen.tab_complete_path(line, type=type, ext=ext, sort=sort, dirs_first=dirs_first) 

    def set_mode(self, mode): 

        reactor.removeReader(self.screen) 

        self.screen = mode 

        self.statusbars.screen = self.screen 

        reactor.addReader(self.screen) 

        mode.refresh() 

    def on_client_disconnect(self): 

        component.stop() 

    def write(self, s): 

        if self.interactive: 

            if isinstance(self.screen, deluge.ui.console.modes.legacy.Legacy): 

                self.screen.write(s) 

            else: 

                component.get("LegacyUI").add_line(s, False) 

                self.events.append(s) 

        else: 

            print(colors.strip_colors(s.encode(self.encoding))) 

    def write_event(self, s): 

        if self.interactive: 

            if isinstance(self.screen, deluge.ui.console.modes.legacy.Legacy): 

                self.events.append(s) 

                self.screen.write(s) 

            else: 

                component.get("LegacyUI").add_line(s, False) 

                self.events.append(s) 

        else: 

            print(colors.strip_colors(s.encode(self.encoding)))