diff options
| -rw-r--r-- | modules/channel_op.py | 12 | ||||
| -rw-r--r-- | modules/echo.py | 6 | ||||
| -rw-r--r-- | src/core_modules/command_spec.py | 86 | ||||
| -rw-r--r-- | src/utils/parse.py | 67 |
4 files changed, 103 insertions, 68 deletions
diff --git a/modules/channel_op.py b/modules/channel_op.py index e8fe8095..f274961c 100644 --- a/modules/channel_op.py +++ b/modules/channel_op.py @@ -49,7 +49,7 @@ class Module(ModuleManager.BaseModule): @utils.kwarg("require_mode", "o") @utils.kwarg("require_access", "kick") @utils.kwarg("usage", "<nickname> [reason]") - @utils.spec("!r~channel !cuser ?...") + @utils.spec("!r~channel !cuser ?string") def kick(self, event): self._kick(event["server"], event["target"], event["spec"][0], event["spec"][1]) @@ -99,7 +99,7 @@ class Module(ModuleManager.BaseModule): @utils.kwarg("require_mode", "o") @utils.kwarg("require_access", "ban") @utils.kwarg("usage", "[+time] <target>") - @utils.spec("!r~channel ?time !user|text") + @utils.spec("!r~channel ?time !user|word") def ban(self, event): self._ban(event["server"], event["spec"][0], event["spec"][2], True, event["spec"][1], True) @@ -118,7 +118,7 @@ class Module(ModuleManager.BaseModule): @utils.kwarg("require_mode", "o") @utils.kwarg("require_access", "kickban") @utils.kwarg("usage", "[+time] <nickname> [reason]") - @utils.spec("!r~channel ?time !cuser| ?...") + @utils.spec("!r~channel ?time !cuser| ?string") def kickban(self, event): self._ban(event["server"], event["spec"][0], event["spec"][2], False, event["spec"][1], True) @@ -161,7 +161,7 @@ class Module(ModuleManager.BaseModule): @utils.kwarg("require_access", "topic") @utils.kwarg("remove_empty", False) @utils.kwarg("usage", "<topic>") - @utils.spec("!r~channel !...") + @utils.spec("!r~channel !string") def topic(self, event): event["spec"][0].send_topic(event["spec"][1]) @@ -170,7 +170,7 @@ class Module(ModuleManager.BaseModule): @utils.kwarg("require_access", "topic") @utils.kwarg("remove_empty", False) @utils.kwarg("usage", "<topic>") - @utils.spec("!r~channel !...") + @utils.spec("!r~channel !string") def tappend(self, event): event["spec"][0].send_topic(event["spec"][0].topic + event["spec"][1]) @@ -264,7 +264,7 @@ class Module(ModuleManager.BaseModule): @utils.kwarg("require_mode", "o") @utils.kwarg("require_access", "flags") @utils.kwarg("usage", "<nickname> [flags]") - @utils.spec("!r~channel !ouser ?...") + @utils.spec("!r~channel !ouser ?string") def flags(self, event): target = event["spec"][1] current_flags = event["spec"][0].get_user_setting(target.get_id(), diff --git a/modules/echo.py b/modules/echo.py index 8276bd02..7c36a480 100644 --- a/modules/echo.py +++ b/modules/echo.py @@ -6,14 +6,14 @@ class Module(ModuleManager.BaseModule): @utils.hook("received.command.echo") @utils.kwarg("remove_empty", False) @utils.kwarg("help", "Echo a string back") - @utils.spec("!...") + @utils.spec("!string") def echo(self, event): event["stdout"].write(event["spec"][0]) @utils.hook("received.command.action") @utils.kwarg("remove_empty", False) @utils.kwarg("help", "Make the bot send a /me") - @utils.spec("!...") + @utils.spec("!string") def action(self, event): event["target"].send_message("\x01ACTION %s\x01" % event["spec"][0]) @@ -21,6 +21,6 @@ class Module(ModuleManager.BaseModule): @utils.kwarg("permission", "say") @utils.kwarg("remove_empty", False) @utils.kwarg("help", "Send a message to a target") - @utils.spec("!word !...") + @utils.spec("!word !string") def msg(self, event): event["server"].send_message(event["spec"][0], event["spec"][1]) diff --git a/src/core_modules/command_spec.py b/src/core_modules/command_spec.py index 3ce10845..340f6fee 100644 --- a/src/core_modules/command_spec.py +++ b/src/core_modules/command_spec.py @@ -27,79 +27,71 @@ from src import EventManager, ModuleManager, utils # - "user" - an argument of a user's nickname # - "ouser" - an argument of a potentially offline user's nickname # - "word" - one word from arguments -# - "..." - collect all remaining args in to a string +# - "string" - collect all remaining args in to a string class Module(ModuleManager.BaseModule): - def _spec_chunk(self, server, channel, user, spec_types, args): + def _spec_value(self, server, channel, user, argument_types, args): options = [] first_error = None - for spec_type in spec_types: - chunk = None + for argument_type in argument_types: + value = None n = 0 error = None - if spec_type.type == "time" and args: - time, _ = utils.parse.timed_args(args) - chunk = time - n = 1 - error = "Invalid timeframe" - elif spec_type.type == "rchannel": + simple_value, simple_count = argument_type.simple(args) + if not simple_count == -1: + value = simple_value + n = simple_count + error = argument_type.error() + elif argument_type.type == "rchannel": if channel: - chunk = channel + value = channel elif args: n = 1 if args[0] in server.channels: - chunk = server.channels.get(args[0]) + value = server.channels.get(args[0]) error = "No such channel" else: error = "No channel provided" - elif spec_type.type == "channel" and args: + elif argument_type.type == "channel" and args: if args[0] in server.channels: - chunk = server.channels.get(args[0]) + value = server.channels.get(args[0]) n = 1 error = "No such channel" - elif spec_type.type == "cuser" and args: + elif argument_type.type == "cuser" and args: tuser = server.get_user(args[0], create=False) if tuser and channel.has_user(tuser): - chunk = tuser + value = tuser n = 1 error = "That user is not in this channel" - elif spec_type.type == "ruser": + elif argument_type.type == "ruser": if args: - chunk = server.get_user(args[0], create=False) + value = server.get_user(args[0], create=False) n = 1 else: - chunk = user + value = user error = "No such user" - elif spec_type.type == "user": + elif argument_type.type == "user": if args: - chunk = server.get_user(args[0], create=False) + value = server.get_user(args[0], create=False) n = 1 error = "No such user" else: error = "No user provided" - elif spec_type.type == "ouser" and args: + elif argument_type.type == "ouser" and args: if server.has_user_id(args[0]): - chunk = server.get_user(args[0]) + value = server.get_user(args[0]) n = 1 error = "Unknown nickname" - elif spec_type.type == "word": - if args: - chunk = args[0] - n = 1 - elif spec_type.type == "...": - if args: - chunk = " ".join(args) - n = max(1, len(args)) - options.append([spec_type, chunk, n, error]) + options.append([argument_type, value, n, error]) return options @utils.hook("preprocess.command") @utils.kwarg("priority", EventManager.PRIORITY_HIGH) def preprocess(self, event): - spec_types = event["hook"].get_kwarg("spec", None) - if not spec_types == None: + spec_arguments = event["hook"].get_kwarg("spec", None) + if not spec_arguments == None: server = event["server"] channel = event["target"] if event["is_channel"] else None user = event["user"] @@ -108,22 +100,22 @@ class Module(ModuleManager.BaseModule): out = [] kwargs = {"channel": channel} - for item in spec_types: - options = self._spec_chunk(server, kwargs["channel"], user, - item.types, args) + for spec_argument in spec_arguments: + options = self._spec_value(server, kwargs["channel"], user, + spec_argument.types, args) found = None first_error = None - for spec_type, chunk, n, error in options: - if not chunk == None: - if spec_type.exported: - kwargs[spec_type.exported] = chunk + for argument_type, value, n, error in options: + if not value == None: + if argument_type.exported: + kwargs[argument_type.exported] = value found = True args = args[n:] - if len(item.types) > 1: - chunk = [spec_type, chunk] - found = chunk + if len(spec_argument.types) > 1: + value = [argument_type.type, value] + found = value break elif not error and n > 0: error = "Not enough arguments" @@ -131,11 +123,11 @@ class Module(ModuleManager.BaseModule): if error and not first_error: first_error = error - out.append(found) - - if not item.optional and not found: + if not spec_argument.optional and not found: error = first_error or "Invalid arguments" return utils.consts.PERMISSION_HARD_FAIL, error + out.append(found) + kwargs["spec"] = out event["kwargs"].update(kwargs) diff --git a/src/utils/parse.py b/src/utils/parse.py index aa7a9ee1..3af092f1 100644 --- a/src/utils/parse.py +++ b/src/utils/parse.py @@ -156,36 +156,79 @@ def format_token_replace(s: str, vars: typing.Dict[str, str], s = s[:i] + vars[token.replace(sigil, "", 1)] + s[i+len(token):] return s -class ArgumentSpecType(object): - def __init__(self, type_name: str, name: str, exported: str): +class SpecArgumentType(object): + def __init__(self, type_name: str, name: typing.Optional[str], exported: str): self.type = type_name - self.name = name + self._name = name self.exported = exported -class ArgumentSpec(object): - def __init__(self, optional: bool, types: typing.List[ArgumentSpecType]): + def name(self) -> typing.Optional[str]: + return self._name + def simple(self, args: typing.List[str]) -> typing.Tuple[typing.Any, int]: + return None, -1 + def error(self) -> typing.Optional[str]: + return None + +class SpecArgumentTypeWord(SpecArgumentType): + def simple(self, args: typing.List[str]) -> typing.Tuple[typing.Any, int]: + if args: + return args[0], 1 + return None, 1 +class SpecArgumentTypeWordLower(SpecArgumentTypeWord): + def simple(self, args: typing.List[str]) -> typing.Tuple[typing.Any, int]: + out = SpecArgumentTypeWord.simple(self, args) + if out[0]: + return out[0].lower(), out[1] + return out + +class SpecArgumentTypeString(SpecArgumentType): + def name(self): + return "%s ..." % SpecArgumentType.name(self) + def simple(self, args: typing.List[str]) -> typing.Tuple[typing.Any, int]: + return " ".join(args), len(args) +class SpecArgumentTypeTime(SpecArgumentType): + def name(self): + return "+%s" % (SpecArgumentType.name(self) or "time") + def simple(self, args: typing.List[str]) -> typing.Tuple[typing.Any, int]: + time, _ = timed_args(args) + return time, 1 + def error(self) -> typing.Optional[str]: + return "Invalid timeframe" + +SPEC_ARGUMENT_TYPES = { + "word": SpecArgumentTypeWord, + "wordlower": SpecArgumentTypeWordLower, + "string": SpecArgumentTypeString, + "time": SpecArgumentTypeTime +} + +class SpecArgument(object): + def __init__(self, optional: bool, types: typing.List[SpecArgumentType]): self.optional = optional self.types = types -def argument_spec(spec: str) -> typing.List[ArgumentSpec]: - out: typing.List[ArgumentSpec] = [] +def argument_spec(spec: str) -> typing.List[SpecArgument]: + out: typing.List[SpecArgument] = [] for spec_argument in spec.split(" "): optional = spec_argument[0] == "?" - argument_types: typing.List[ArgumentSpecType] = [] + argument_types: typing.List[SpecArgumentType] = [] for argument_type in spec_argument[1:].split("|"): exported = "" if "~" in argument_type: exported = argument_type.split("~", 1)[1] argument_type = argument_type.replace("~", "", 1) - argument_type_name = argument_type + argument_type_name: typing.Optional[str] = None name_end = argument_type.find(">") - if argument_type[0] == "<" and name_end > 0: + if argument_type.startswith("<") and name_end > 0: argument_type_name = argument_type[1:name_end] argument_type = argument_type[name_end+1:] - argument_types.append(ArgumentSpecType(argument_type, + argument_type_class = SpecArgumentType + if argument_type in SPEC_ARGUMENT_TYPES: + argument_type_class = SPEC_ARGUMENT_TYPES[argument_type] + argument_types.append(argument_type_class(argument_type, argument_type_name, exported)) - out.append(ArgumentSpec(optional, argument_types)) + out.append(SpecArgument(optional, argument_types)) return out |