#298 Add ability to chose matched rule
This commit is contained in:
nvbn
@@ -0,0 +1,77 @@
|
||||
from imp import load_source
|
||||
from pathlib import Path
|
||||
from . import conf, types, logs
|
||||
import sys
|
||||
|
||||
|
||||
def load_rule(rule, settings):
|
||||
"""Imports rule module and returns it."""
|
||||
name = rule.name[:-3]
|
||||
rule_module = load_source(name, str(rule))
|
||||
priority = getattr(rule_module, 'priority', conf.DEFAULT_PRIORITY)
|
||||
return types.Rule(name, rule_module.match,
|
||||
rule_module.get_new_command,
|
||||
getattr(rule_module, 'enabled_by_default', True),
|
||||
getattr(rule_module, 'side_effect', None),
|
||||
settings.priority.get(name, priority),
|
||||
getattr(rule_module, 'requires_output', True))
|
||||
|
||||
|
||||
def get_loaded_rules(rules, settings):
|
||||
"""Yields all available rules."""
|
||||
for rule in rules:
|
||||
if rule.name != '__init__.py':
|
||||
loaded_rule = load_rule(rule, settings)
|
||||
if loaded_rule in settings.rules:
|
||||
yield loaded_rule
|
||||
|
||||
|
||||
def get_rules(user_dir, settings):
|
||||
"""Returns all enabled rules."""
|
||||
bundled = Path(__file__).parent \
|
||||
.joinpath('rules') \
|
||||
.glob('*.py')
|
||||
user = user_dir.joinpath('rules').glob('*.py')
|
||||
return get_loaded_rules(sorted(bundled) + sorted(user), settings)
|
||||
|
||||
|
||||
def get_matched_rules(command, rules, settings):
|
||||
"""Returns first matched rule for command."""
|
||||
script_only = command.stdout is None and command.stderr is None
|
||||
|
||||
for rule in rules:
|
||||
if script_only and rule.requires_output:
|
||||
continue
|
||||
|
||||
try:
|
||||
with logs.debug_time(u'Trying rule: {};'.format(rule.name),
|
||||
settings):
|
||||
if rule.match(command, settings):
|
||||
yield rule
|
||||
except Exception:
|
||||
logs.rule_failed(rule, sys.exc_info(), settings)
|
||||
|
||||
|
||||
def make_corrected_commands(command, rules, settings):
|
||||
for rule in rules:
|
||||
new_commands = rule.get_new_command(command, settings)
|
||||
if not isinstance(new_commands, list):
|
||||
new_commands = [new_commands]
|
||||
for n, new_command in enumerate(new_commands):
|
||||
yield types.CorrectedCommand(script=new_command,
|
||||
side_effect=rule.side_effect,
|
||||
priority=(n + 1) * rule.priority)
|
||||
|
||||
|
||||
def get_corrected_commands(command, user_dir, settings):
|
||||
rules = list(get_rules(user_dir, settings))
|
||||
logs.debug(
|
||||
u'Loaded rules: {}'.format(', '.join(rule.name for rule in rules)),
|
||||
settings)
|
||||
matched = list(get_matched_rules(command, rules, settings))
|
||||
logs.debug(
|
||||
u'Matched rules: {}'.format(', '.join(rule.name for rule in matched)),
|
||||
settings)
|
||||
corrected_commands = make_corrected_commands(command, matched, settings)
|
||||
return sorted(corrected_commands,
|
||||
key=lambda corrected_command: corrected_command.priority)
|
||||
+21
-21
@@ -28,27 +28,6 @@ def rule_failed(rule, exc_info, settings):
|
||||
exception('Rule {}'.format(rule.name), exc_info, settings)
|
||||
|
||||
|
||||
def show_command(new_command, side_effect, settings):
|
||||
sys.stderr.write('{bold}{command}{reset}{side_effect}\n'.format(
|
||||
command=new_command,
|
||||
side_effect=' (+side effect)' if side_effect else '',
|
||||
bold=color(colorama.Style.BRIGHT, settings),
|
||||
reset=color(colorama.Style.RESET_ALL, settings)))
|
||||
|
||||
|
||||
def confirm_command(new_command, side_effect, settings):
|
||||
sys.stderr.write(
|
||||
'{bold}{command}{reset}{side_effect} '
|
||||
'[{green}enter{reset}/{red}ctrl+c{reset}]'.format(
|
||||
command=new_command,
|
||||
side_effect=' (+side effect)' if side_effect else '',
|
||||
bold=color(colorama.Style.BRIGHT, settings),
|
||||
green=color(colorama.Fore.GREEN, settings),
|
||||
red=color(colorama.Fore.RED, settings),
|
||||
reset=color(colorama.Style.RESET_ALL, settings)))
|
||||
sys.stderr.flush()
|
||||
|
||||
|
||||
def failed(msg, settings):
|
||||
sys.stderr.write('{red}{msg}{reset}\n'.format(
|
||||
msg=msg,
|
||||
@@ -56,6 +35,27 @@ def failed(msg, settings):
|
||||
reset=color(colorama.Style.RESET_ALL, settings)))
|
||||
|
||||
|
||||
def show_corrected_command(corrected_command, settings):
|
||||
sys.stderr.write('{bold}{script}{reset}{side_effect}\n'.format(
|
||||
script=corrected_command.script,
|
||||
side_effect=' (+side effect)' if corrected_command.side_effect else '',
|
||||
bold=color(colorama.Style.BRIGHT, settings),
|
||||
reset=color(colorama.Style.RESET_ALL, settings)))
|
||||
|
||||
|
||||
def confirm_text(corrected_command, settings):
|
||||
sys.stderr.write(
|
||||
'\033[1K\r{bold}{script}{reset}{side_effect} '
|
||||
'[{green}enter{reset}/{blue}↑{reset}/{blue}↓{reset}/{red}ctrl+c{reset}]'.format(
|
||||
script=corrected_command.script,
|
||||
side_effect=' (+side effect)' if corrected_command.side_effect else '',
|
||||
bold=color(colorama.Style.BRIGHT, settings),
|
||||
green=color(colorama.Fore.GREEN, settings),
|
||||
red=color(colorama.Fore.RED, settings),
|
||||
reset=color(colorama.Style.RESET_ALL, settings),
|
||||
blue=color(colorama.Fore.BLUE, settings)))
|
||||
|
||||
|
||||
def debug(msg, settings):
|
||||
if settings.debug:
|
||||
sys.stderr.write(u'{blue}{bold}DEBUG:{reset} {msg}\n'.format(
|
||||
|
||||
+11
-83
@@ -1,4 +1,3 @@
|
||||
from imp import load_source
|
||||
from pathlib import Path
|
||||
from os.path import expanduser
|
||||
from pprint import pformat
|
||||
@@ -9,6 +8,8 @@ from psutil import Process, TimeoutExpired
|
||||
import colorama
|
||||
import six
|
||||
from . import logs, conf, types, shells
|
||||
from .corrector import get_corrected_commands
|
||||
from .ui import select_command
|
||||
|
||||
|
||||
def setup_user_dir():
|
||||
@@ -21,37 +22,6 @@ def setup_user_dir():
|
||||
return user_dir
|
||||
|
||||
|
||||
def load_rule(rule):
|
||||
"""Imports rule module and returns it."""
|
||||
rule_module = load_source(rule.name[:-3], str(rule))
|
||||
return types.Rule(rule.name[:-3], rule_module.match,
|
||||
rule_module.get_new_command,
|
||||
getattr(rule_module, 'enabled_by_default', True),
|
||||
getattr(rule_module, 'side_effect', None),
|
||||
getattr(rule_module, 'priority', conf.DEFAULT_PRIORITY),
|
||||
getattr(rule_module, 'requires_output', True))
|
||||
|
||||
|
||||
def _get_loaded_rules(rules, settings):
|
||||
"""Yields all available rules."""
|
||||
for rule in rules:
|
||||
if rule.name != '__init__.py':
|
||||
loaded_rule = load_rule(rule)
|
||||
if loaded_rule in settings.rules:
|
||||
yield loaded_rule
|
||||
|
||||
|
||||
def get_rules(user_dir, settings):
|
||||
"""Returns all enabled rules."""
|
||||
bundled = Path(__file__).parent \
|
||||
.joinpath('rules') \
|
||||
.glob('*.py')
|
||||
user = user_dir.joinpath('rules').glob('*.py')
|
||||
rules = _get_loaded_rules(sorted(bundled) + sorted(user), settings)
|
||||
return sorted(rules, key=lambda rule: settings.priority.get(
|
||||
rule.name, rule.priority))
|
||||
|
||||
|
||||
def wait_output(settings, popen):
|
||||
"""Returns `True` if we can get output of the command in the
|
||||
`wait_command` time.
|
||||
@@ -100,46 +70,12 @@ def get_command(settings, args):
|
||||
return types.Command(script, None, None)
|
||||
|
||||
|
||||
def get_matched_rule(command, rules, settings):
|
||||
"""Returns first matched rule for command."""
|
||||
script_only = command.stdout is None and command.stderr is None
|
||||
|
||||
for rule in rules:
|
||||
if script_only and rule.requires_output:
|
||||
continue
|
||||
|
||||
try:
|
||||
with logs.debug_time(u'Trying rule: {};'.format(rule.name),
|
||||
settings):
|
||||
if rule.match(command, settings):
|
||||
return rule
|
||||
except Exception:
|
||||
logs.rule_failed(rule, sys.exc_info(), settings)
|
||||
|
||||
|
||||
def confirm(new_command, side_effect, settings):
|
||||
"""Returns `True` when running of new command confirmed."""
|
||||
if not settings.require_confirmation:
|
||||
logs.show_command(new_command, side_effect, settings)
|
||||
return True
|
||||
|
||||
logs.confirm_command(new_command, side_effect, settings)
|
||||
try:
|
||||
sys.stdin.read(1)
|
||||
return True
|
||||
except KeyboardInterrupt:
|
||||
logs.failed('Aborted', settings)
|
||||
return False
|
||||
|
||||
|
||||
def run_rule(rule, command, settings):
|
||||
def run_command(command, settings):
|
||||
"""Runs command from rule for passed command."""
|
||||
new_command = shells.to_shell(rule.get_new_command(command, settings))
|
||||
if confirm(new_command, rule.side_effect, settings):
|
||||
if rule.side_effect:
|
||||
rule.side_effect(command, settings)
|
||||
shells.put_to_history(new_command)
|
||||
print(new_command)
|
||||
if command.side_effect:
|
||||
command.side_effect(command, settings)
|
||||
shells.put_to_history(command.script)
|
||||
print(command.script)
|
||||
|
||||
|
||||
# Entry points:
|
||||
@@ -152,18 +88,10 @@ def main():
|
||||
logs.debug(u'Run with settings: {}'.format(pformat(settings)), settings)
|
||||
|
||||
command = get_command(settings, sys.argv)
|
||||
rules = get_rules(user_dir, settings)
|
||||
logs.debug(
|
||||
u'Loaded rules: {}'.format(', '.join(rule.name for rule in rules)),
|
||||
settings)
|
||||
|
||||
matched_rule = get_matched_rule(command, rules, settings)
|
||||
if matched_rule:
|
||||
logs.debug(u'Matched rule: {}'.format(matched_rule.name), settings)
|
||||
run_rule(matched_rule, command, settings)
|
||||
return
|
||||
|
||||
logs.failed('No fuck given', settings)
|
||||
corrected_commands = get_corrected_commands(command, user_dir, settings)
|
||||
selected_command = select_command(corrected_commands, settings)
|
||||
if selected_command:
|
||||
run_command(selected_command, settings)
|
||||
|
||||
|
||||
def print_alias():
|
||||
|
||||
@@ -3,6 +3,8 @@ from collections import namedtuple
|
||||
|
||||
Command = namedtuple('Command', ('script', 'stdout', 'stderr'))
|
||||
|
||||
CorrectedCommand = namedtuple('CorrectedCommand', ('script', 'side_effect', 'priority'))
|
||||
|
||||
Rule = namedtuple('Rule', ('name', 'match', 'get_new_command',
|
||||
'enabled_by_default', 'side_effect',
|
||||
'priority', 'requires_output'))
|
||||
|
||||
@@ -0,0 +1,88 @@
|
||||
import sys
|
||||
from getch import getch
|
||||
from . import logs
|
||||
|
||||
SELECT = 0
|
||||
ABORT = 1
|
||||
PREVIOUS = 2
|
||||
NEXT = 3
|
||||
|
||||
|
||||
def read_actions():
|
||||
"""Yields actions for pressed keys."""
|
||||
buffer = []
|
||||
ch = None
|
||||
while True:
|
||||
try:
|
||||
try:
|
||||
ch = getch()
|
||||
except OverflowError: # Ctrl+C, KeyboardInterrupt will be reraised
|
||||
pass
|
||||
except KeyboardInterrupt:
|
||||
yield ABORT
|
||||
|
||||
if ch in ('\n', '\r'): # Enter
|
||||
yield SELECT
|
||||
|
||||
buffer.append(ch)
|
||||
buffer = buffer[-3:]
|
||||
|
||||
if buffer == ['\x1b', '[', 'A']: # ↑
|
||||
yield PREVIOUS
|
||||
|
||||
if buffer == ['\x1b', '[', 'B']: # ↓
|
||||
yield NEXT
|
||||
|
||||
|
||||
class CommandSelector(object):
|
||||
def __init__(self, commands):
|
||||
self._commands = commands
|
||||
self._index = 0
|
||||
self._on_change = lambda x: x
|
||||
|
||||
def next(self):
|
||||
self._index = (self._index + 1) % len(self._commands)
|
||||
self._on_change(self.value)
|
||||
|
||||
def previous(self):
|
||||
self._index = (self._index - 1) % len(self._commands)
|
||||
self._on_change(self.value)
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
return self._commands[self._index]
|
||||
|
||||
def on_change(self, fn):
|
||||
self._on_change = fn
|
||||
fn(self.value)
|
||||
|
||||
|
||||
def select_command(corrected_commands, settings):
|
||||
"""Returns:
|
||||
|
||||
- the first command when confirmation disabled;
|
||||
- None when ctrl+c pressed;
|
||||
- selected command.
|
||||
|
||||
"""
|
||||
if not corrected_commands:
|
||||
logs.failed('No fuck given', settings)
|
||||
return
|
||||
|
||||
selector = CommandSelector(corrected_commands)
|
||||
if not settings.require_confirmation:
|
||||
logs.show_corrected_command(selector.value, settings)
|
||||
return selector.value
|
||||
|
||||
selector.on_change(lambda val: logs.confirm_text(val, settings))
|
||||
for key in read_actions():
|
||||
if key == SELECT:
|
||||
sys.stderr.write('\n')
|
||||
return selector.value
|
||||
elif key == ABORT:
|
||||
logs.failed('\nAborted', settings)
|
||||
return
|
||||
elif key == PREVIOUS:
|
||||
selector.previous()
|
||||
elif key == NEXT:
|
||||
selector.next()
|
||||
Reference in New Issue
Block a user