Firepup650/bitbot-3.11-fork
2
0
Fork 0
Code Issues Pull requests Activity Actions

Hand over regex-based-command responsibilities to commands module

Browse source
This commit is contained in:
jesopo 2019-05-18 18:35:47 +01:00
parent d64d12f8fd
commit 4e0e63e7f8
5 changed files with 182 additions and 191 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

187
modules/commands/__init__.py
View file

@ -90,10 +90,9 @@ class Module(ModuleManager.BaseModule):
return True return True
return False return False
def message(self, event, command, args_index=1): def _find_command_hook(self, server, command, is_channel):
args_split = event["message_split"][args_index:]
if not self.has_command(command): if not self.has_command(command):
aliases = self._get_aliases(event["server"]) aliases = self._get_aliases(server)
if command.lower() in aliases: if command.lower() in aliases:
command, _, new_args = aliases[command.lower()].partition(" ") command, _, new_args = aliases[command.lower()].partition(" ")
@ -102,12 +101,8 @@ class Module(ModuleManager.BaseModule):
except IndexError: except IndexError:
return return
hook = None
if self.has_command(command): if self.has_command(command):
if self._is_ignored(event["server"], event["user"], command):
return
hook = None
target = None
for potential_hook in self.get_hooks(command): for potential_hook in self.get_hooks(command):
alias_of = self._get_alias_of(potential_hook) alias_of = self._get_alias_of(potential_hook)
if alias_of: if alias_of:
@ -118,97 +113,92 @@ class Module(ModuleManager.BaseModule):
"'%s' is an alias of unknown command '%s'" "'%s' is an alias of unknown command '%s'"
% (command.lower(), alias_of.lower())) % (command.lower(), alias_of.lower()))
is_channel = "channel" in event
if not is_channel and potential_hook.kwargs.get("channel_only"): if not is_channel and potential_hook.kwargs.get("channel_only"):
continue continue
if is_channel and potential_hook.kwargs.get("private_only"): if is_channel and potential_hook.kwargs.get("private_only"):
continue continue
hook = potential_hook hook = potential_hook
target = event["user"] if not is_channel else event["channel"]
break break
if not hook: return hook
return
module_name = self._get_prefix(hook) or "" def command(self, server, target, is_channel, user, command, args_split,
if not module_name and hasattr(hook.function, "__self__"): tags, statusmsg, hook, **kwargs):
module_name = hook.function.__self__._name if self._is_ignored(server, user, command):
return False
msgid = MSGID_TAG.get_value(event["tags"]) module_name = self._get_prefix(hook) or ""
statusmsg = "".join(event.get("statusmsg", [])) if not module_name and hasattr(hook.function, "__self__"):
stdout = outs.StdOut(event["server"], module_name, target, msgid, module_name = hook.function.__self__._name
statusmsg)
stderr = outs.StdErr(event["server"], module_name, target, msgid,
statusmsg)
command_method = self._command_method(target, event["server"])
if hook.kwargs.get("remove_empty", True): msgid = MSGID_TAG.get_value(tags)
args_split = list(filter(None, args_split)) stdout = outs.StdOut(server, module_name, target, msgid, statusmsg)
stderr = outs.StdErr(server, module_name, target, msgid, statusmsg)
command_method = self._command_method(target, server)
min_args = hook.kwargs.get("min_args") if hook.kwargs.get("remove_empty", True):
if min_args and len(args_split) < min_args: args_split = list(filter(None, args_split))
command_prefix = ""
if is_channel: target.buffer.skip_next()
command_prefix = self._command_prefix(event["server"],
target) min_args = hook.kwargs.get("min_args")
usage = self._get_usage(hook, command, command_prefix) if min_args and len(args_split) < min_args:
if usage: command_prefix = ""
stderr.write("Not enough arguments, usage: %s" % if is_channel:
usage).send(command_method) command_prefix = self._command_prefix(server, target)
else: usage = self._get_usage(hook, command, command_prefix)
stderr.write("Not enough arguments (minimum: %d)" % if usage:
min_args).send(command_method) stderr.write("Not enough arguments, usage: %s" %
usage).send(command_method)
else: else:
returns = self.events.on("preprocess.command").call_unsafe( stderr.write("Not enough arguments (minimum: %d)" %
hook=hook, user=event["user"], server=event["server"], min_args).send(command_method)
target=target, is_channel=is_channel, tags=event["tags"], else:
args_split=args_split) returns = self.events.on("preprocess.command").call_unsafe(
hook=hook, user=user, server=server, target=target,
is_channel=is_channel, tags=tags, args_split=args_split,
command=command)
hard_fail = False hard_fail = False
force_success = False force_success = False
error = None error = None
for returned in returns: for returned in returns:
if returned == utils.consts.PERMISSION_HARD_FAIL: if returned == utils.consts.PERMISSION_HARD_FAIL:
hard_fail = True hard_fail = True
break break
elif returned == utils.consts.PERMISSION_FORCE_SUCCESS: elif returned == utils.consts.PERMISSION_FORCE_SUCCESS:
force_success = True force_success = True
elif returned: elif returned:
error = returned error = returned
if hard_fail or (not force_success and error): if hard_fail or (not force_success and error):
if error: if error:
stderr.write(error).send(command_method) stderr.write(error).send(command_method)
target.buffer.skip_next() target.buffer.skip_next()
return return True
args = " ".join(args_split) args = " ".join(args_split)
server = event["server"]
user = event["user"]
new_event = self.events.on("received.command").on(command new_event = self.events.on(hook.event_name).make_event(user=user,
).make_event(user=user, server=server, target=target, server=server, target=target, args=args, tags=tags,
args=args, tags=event["tags"], args_split=args_split, args_split=args_split, stdout=stdout, stderr=stderr,
stdout=stdout, stderr=stderr, command=command.lower(), is_channel=is_channel, command=command, **kwargs)
is_channel=is_channel)
self.log.trace("calling command '%s': %s", self.log.trace("calling command '%s': %s",
[command, new_event.kwargs]) [command, new_event.kwargs])
try: try:
hook.call(new_event) hook.call(new_event)
except utils.EventError as e: except utils.EventError as e:
stderr.write(str(e)) stderr.write(str(e))
if not hook.kwargs.get("skip_out", False): if not hook.kwargs.get("skip_out", False):
command_method = self._command_method( command_method = self._command_method(target, server)
target, event["server"]) stdout.send(command_method)
stdout.send(command_method) stderr.send(command_method)
stderr.send(command_method) target.last_stdout = stdout
target.last_stdout = stdout target.last_stderr = stderr
target.last_stderr = stderr return new_event.eaten
target.buffer.skip_next()
event.eat()
def _command_prefix(self, server, channel): def _command_prefix(self, server, channel):
return channel.get_setting("command-prefix", return channel.get_setting("command-prefix",
@ -225,22 +215,53 @@ class Module(ModuleManager.BaseModule):
prefixed_commands = event["channel"].get_setting("prefixed-commands", True) prefixed_commands = event["channel"].get_setting("prefixed-commands", True)
command_prefix = self._command_prefix(event["server"], event["channel"]) command_prefix = self._command_prefix(event["server"], event["channel"])
command = None
args_split = None
if event["message_split"][0].startswith(command_prefix): if event["message_split"][0].startswith(command_prefix):
if not prefixed_commands: if not prefixed_commands:
return return
command = event["message_split"][0].replace( command = event["message_split"][0].replace(
command_prefix, "", 1).lower() command_prefix, "", 1).lower()
self.message(event, command) args_split = event["message_split"][1:]
elif len(event["message_split"]) > 1 and self.is_highlight( elif len(event["message_split"]) > 1 and self.is_highlight(
event["server"], event["message_split"][0]): event["server"], event["message_split"][0]):
command = event["message_split"][1].lower() command = event["message_split"][1].lower()
self.message(event, command, 2) args_split = event["message_split"][2:]
if command:
hook = self._find_command_hook(event["server"], command, True)
if hook:
self.command(event["server"], event["channel"], True,
event["user"], command, args_split, event["tags"],
"".join(event["statusmsg"]), hook)
else:
regex_hook = self.events.on("command.regex").get_hooks()
for hook in regex_hook:
pattern = hook.get_kwarg("pattern", None)
if not pattern and hook.get_kwarg("pattern-url", None) == "1":
pattern = utils.http.REGEX_URL
if pattern:
match = re.match(pattern, event["message"])
if match:
command = hook.get_kwarg("command", "")
res = self.command(event["server"], event["channel"],
True, event["user"], command, "", event["tags"],
"".join(event["statusmsg"]), hook, match=match,
message=event["message"])
if res:
break
@utils.hook("received.message.private", priority=EventManager.PRIORITY_LOW) @utils.hook("received.message.private", priority=EventManager.PRIORITY_LOW)
def private_message(self, event): def private_message(self, event):
if event["message_split"] and not event["action"]: if event["message_split"] and not event["action"]:
command = event["message_split"][0].lower() command = event["message_split"][0].lower()
self.message(event, command) hook = self._find_command_hook(event["server"], command, False)
if hook:
self.command(event["server"], event["user"], False,
event["user"], command, event["message_split"][1:],
event["tags"], "", hook)
def _get_help(self, hook): def _get_help(self, hook):
return hook.get_kwarg("help", None) or hook.docstring.description return hook.get_kwarg("help", None) or hook.docstring.description

87
modules/karma.py
View file

@ -20,68 +20,53 @@ class Module(ModuleManager.BaseModule):
return utils.irc.color(str(karma), utils.consts.LIGHTGREEN) return utils.irc.color(str(karma), utils.consts.LIGHTGREEN)
return str(karma) return str(karma)
@utils.hook("new.user") @utils.hook("new.user")
def new_user(self, event): def new_user(self, event):
event["user"].last_karma = None event["user"].last_karma = None
@utils.hook("received.message.channel") @utils.hook("command.regex")
def channel_message(self, event): def channel_message(self, event):
match = re.match(REGEX_KARMA, event["message"].strip()) """
if match and not event["action"]: :command: karma
is_ignored_f = self.exports.get_one("is-ignored", :pattern: ^(.*[^-+])[-+]*(\+{2,}|\-{2,})$
lambda _1, _2: False) """
if is_ignored_f(event["server"], event["user"], "karma"): verbose = event["target"].get_setting("karma-verbose", False)
nickname_only = event["server"].get_setting("karma-nickname-only",
False)
if not event["user"].last_karma or (time.time()-event["user"
].last_karma) >= KARMA_DELAY_SECONDS:
target = event["match"].group(1).strip().rstrip("".join(WORD_STOP))
if event["server"].irc_lower(target) == event["user"].name:
if verbose:
event["stdout"].write("You cannot change your own karma")
return return
is_silenced_f = self.exports.get_one("is-silenced", lambda _: False) setting = "karma-%s" % target
if is_silenced_f(event["channel"]): setting_target = event["server"]
return if nickname_only:
user = event["server"].get_user(target)
verbose = event["channel"].get_setting("karma-verbose", False) setting = "karma"
nickname_only = event["server"].get_setting("karma-nickname-only", setting_target = user
False) if not event["target"].has_user(user):
if not event["user"].last_karma or (time.time()-event["user"
].last_karma) >= KARMA_DELAY_SECONDS:
target = match.group(1).strip().rstrip("".join(WORD_STOP))
if event["server"].irc_lower(target) == event["user"].name:
if verbose:
self.events.on("send.stderr").call(
module_name="Karma", target=event["channel"],
message="You cannot change your own karma",
server=event["server"])
return return
setting = "karma-%s" % target positive = event["match"].group(2)[0] == "+"
setting_target = event["server"] karma = setting_target.get_setting(setting, 0)
if nickname_only: karma += 1 if positive else -1
user = event["server"].get_user(target)
setting = "karma"
setting_target = user
if not event["channel"].has_user(user):
return
positive = match.group(2)[0] == "+" if not karma == 0:
karma = setting_target.get_setting(setting, 0) setting_target.set_setting(setting, karma)
karma += 1 if positive else -1 else:
setting_target.del_setting(setting)
if not karma == 0: karma_str = self._karma_str(karma)
setting_target.set_setting(setting, karma) if verbose:
else: event["stdout"].write(
setting_target.del_setting(setting) "%s now has %s karma" % (target, karma_str))
event["user"].last_karma = time.time()
karma_str = self._karma_str(karma) elif verbose:
if verbose: event["stderr"].write("Try again in a couple of seconds")
self.events.on("send.stdout").call(
module_name="Karma", target=event["channel"],
message="%s now has %s karma" % (target, karma_str),
server=event["server"])
event["user"].last_karma = time.time()
elif verbose:
self.events.on("send.stderr").call(module_name="Karma",
target=event["channel"], server=event["server"],
message="Try again in a couple of seconds")
@utils.hook("received.command.karma") @utils.hook("received.command.karma")
def karma(self, event): def karma(self, event):

25
modules/sed.py
View file

@ -12,19 +12,18 @@ REGEX_SED = re.compile("^s/")
"validate": utils.bool_or_none}) "validate": utils.bool_or_none})
class Module(ModuleManager.BaseModule): class Module(ModuleManager.BaseModule):
def _closest_setting(self, event, setting, default): def _closest_setting(self, event, setting, default):
return event["channel"].get_setting(setting, return event["target"].get_setting(setting,
event["server"].get_setting(setting, default)) event["server"].get_setting(setting, default))
@utils.hook("received.message.channel") @utils.hook("command.regex")
def channel_message(self, event): def channel_message(self, event):
"""
:command: sed
:pattern: ^s/
"""
sed_split = re.split(REGEX_SPLIT, event["message"], 3) sed_split = re.split(REGEX_SPLIT, event["message"], 3)
if event["message"].startswith("s/") and len(sed_split) > 2: if event["message"].startswith("s/") and len(sed_split) > 2:
if event["action"] or not self._closest_setting(event, "sed", if not self._closest_setting(event, "sed", False):
False):
return
is_ignored_f = short_url = self.exports.get_one("is-ignored",
lambda _1, _2: False)
if is_ignored_f(event["server"], event["user"], "sed"):
return return
regex_flags = 0 regex_flags = 0
@ -50,15 +49,13 @@ class Module(ModuleManager.BaseModule):
pattern = re.compile(sed_split[1], regex_flags) pattern = re.compile(sed_split[1], regex_flags)
except: except:
traceback.print_exc() traceback.print_exc()
self.events.on("send.stderr").call(target=event["channel"], event["stderr"].write("Invalid regex in pattern")
module_name="Sed", server=event["server"],
message="Invalid regex in pattern")
return return
replace = utils.irc.bold(sed_split[2].replace("\\/", "/")) replace = utils.irc.bold(sed_split[2].replace("\\/", "/"))
for_user = event["user"].nickname if self._closest_setting(event, for_user = event["user"].nickname if self._closest_setting(event,
"sed-sender-only", False) else None "sed-sender-only", False) else None
line = event["channel"].buffer.find(pattern, from_self=False, line = event["target"].buffer.find(pattern, from_self=False,
for_user=for_user, not_pattern=REGEX_SED) for_user=for_user, not_pattern=REGEX_SED)
if line: if line:
new_message = re.sub(pattern, replace, line.message, count) new_message = re.sub(pattern, replace, line.message, count)
@ -66,6 +63,4 @@ class Module(ModuleManager.BaseModule):
prefix = "* %s" % line.sender prefix = "* %s" % line.sender
else: else:
prefix = "<%s>" % line.sender prefix = "<%s>" % line.sender
self.events.on("send.stdout").call(target=event[ event["stdout"].write("%s %s" % (prefix, new_message))
"channel"], module_name="Sed", server=event["server"],
message="%s %s" % (prefix, new_message))

53
modules/title.py
View file

@ -44,40 +44,33 @@ class Module(ModuleManager.BaseModule):
else: else:
return None return None
@utils.hook("received.message.channel", @utils.hook("command.regex",
priority=EventManager.PRIORITY_MONITOR) priority=EventManager.PRIORITY_MONITOR)
def channel_message(self, event): def channel_message(self, event):
match = re.search(utils.http.REGEX_URL, event["message"]) """
if match and event["channel"].get_setting("auto-title", False): :command: title
is_ignored_f = short_url = self.exports.get_one("is-ignored", :pattern-url: 1
lambda _1, _2: False) """
if is_ignored_f(event["server"], event["user"], "title"): url = event["match"].group(0)
return title = self._get_title(event["target"], event["match"].group(0))
url = match.group(0) if title:
title = self._get_title(event["channel"], match.group(0)) message = title
if event["target"].get_setting("auto-title-first", False):
setting = "url-last-%s" % self._url_hash(url)
first_details = event["target"].get_setting(setting, None)
if title: if first_details:
message = title first_nickname, first_timestamp, _ = first_details
if event["channel"].get_setting("auto-title-first", False): timestamp_parsed = utils.iso8601_parse(first_timestamp)
setting = "url-last-%s" % self._url_hash(url) timestamp_human = utils.datetime_human(timestamp_parsed)
first_details = event["channel"].get_setting(setting, None) message = "%s (first posted by %s at %s)" % (title,
first_nickname, timestamp_human)
if first_details: else:
first_nickname, first_timestamp, _ = first_details event["target"].set_setting(setting,
timestamp_parsed = utils.iso8601_parse(first_timestamp) [event["user"].nickname, utils.iso8601_format_now(),
timestamp_human = utils.datetime_human(timestamp_parsed) url])
message = "%s (first posted by %s at %s)" % (title, event["stdout"].write(message)
first_nickname, timestamp_human)
else:
event["channel"].set_setting(setting,
[event["user"].nickname, utils.iso8601_format_now(),
url])
self.events.on("send.stdout").call(target=event["channel"],
message=message, module_name="Title",
server=event["server"])
@utils.hook("received.command.t", alias_of="title") @utils.hook("received.command.t", alias_of="title")
@utils.hook("received.command.title", usage="[URL]") @utils.hook("received.command.title", usage="[URL]")

21
modules/youtube.py
View file

@ -125,20 +125,17 @@ class Module(ModuleManager.BaseModule):
else: else:
event["stderr"].write("No search phrase provided") event["stderr"].write("No search phrase provided")
@utils.hook("received.message.channel", @utils.hook("command.regex",
priority=EventManager.PRIORITY_LOW) priority=EventManager.PRIORITY_LOW)
def channel_message(self, event): def channel_message(self, event):
match = re.search(REGEX_YOUTUBE, event["message"]) """
if match and event["channel"].get_setting("auto-youtube", False): :command: youtube
is_ignored_f = short_url = self.exports.get_one("is-ignored", :-pattern: https?://(?:www.)?
lambda _1, _2: False) (?:youtu.be/|youtube.com/watch\?[\S]*v=)([\w\-]{11})
if is_ignored_f(event["server"], event["user"], "youtube"): """
return if event["target"].get_setting("auto-youtube", False):
youtube_id = event["match"].group(1)
youtube_id = match.group(1)
video_details = self.video_details(youtube_id) video_details = self.video_details(youtube_id)
if video_details: if video_details:
self.events.on("send.stdout").call(target=event["channel"], event["stdout"].write(video_details)
message=video_details, module_name="Youtube",
server=event["server"])
event.eat() event.eat()