class AngelBot(ClientXMPP):
- commands = {}
- muc_commands = {}
-
messages = defaultdict(
lambda: {
"messages": Lifo(100),
uri = urlparse(u)
await self.parse_uri(uri, sender, mtype)
- def muc_word(self, func):
- name = func.__name__
- self.muc_commands[name] = func
- return func
-
- def muc_command(self, func):
- name = self.prefix + func.__name__
- self.muc_commands[name] = func
- return func
-
- def word(self, func):
- name = func.__name__
- self.commands[name] = func
- return func
-
- def command(self, func):
- name = self.prefix + func.__name__
- self.commands[name] = func
- return func
-
def s(self, msg, sender, mtype):
try:
text = msg["body"]
if msg["type"] in ("chat", "normal"):
mtype = "chat"
sender = msg["from"].bare
- message = msg["body"]
- ctx = message.strip().split()
+ edit = "urn:xmpp:message-correct:0" in str(msg)
+ if edit:
+ return
+
try:
if not msg["oob"]["url"]:
if urls := self.get_urls(msg):
except Exception:
pass
- edit = "urn:xmpp:message-correct:0" in str(msg)
-
- cm = ctx.pop(0)
- if cm in self.commands:
- if not edit:
- self.commands[cm](msg, ctx, sender)
- else:
- self.s(msg, sender, mtype)
+ self.s(msg, sender, mtype)
async def muc_message(self, msg):
if msg["type"] in ("groupchat", "normal"):
else:
self.s(msg, sender, mtype)
- @self.muc_word
- def ping(msg, ctx, sender):
- if ctx:
- return
- bot.send_message(
- mto=sender, mbody=f"pong: {msg['mucnick']}", mtype="groupchat"
- )
-
- @self.word
- def ping(msg, ctx, sender):
- if ctx:
- return
- bot.send_message(
- mto=sender,
- mbody=f"pong: {msg.get_from().bare}",
- mtype="chat",
- )
-
- @self.command
- def join(msg, ctx, sender):
- if sender == bot.owner:
- for channel in ctx:
- bot.plugin["xep_0045"].join_muc(channel, bot.nick)
-
if __name__ == "__main__":
config = configparser.ConfigParser()