153 lines
4.3 KiB
Python
Executable file
153 lines
4.3 KiB
Python
Executable file
"""
|
|
This module implements an IRC bot that aims to be compatible with the discord.py API in order to
|
|
facilitate re-use of the commands for that bot.
|
|
|
|
The supported features are the subset of what is needed for this project and not more than that.
|
|
"""
|
|
|
|
import asyncio
|
|
import inspect
|
|
import os
|
|
import shlex
|
|
import sys
|
|
|
|
import pydle
|
|
from discord_webhook import DiscordEmbed
|
|
|
|
import commands as botcommands
|
|
|
|
|
|
def command_args(handler, sig):
|
|
parse = []
|
|
for par in list(sig.parameters.values())[1:]:
|
|
if par.annotation in (str, inspect._empty):
|
|
parse.append(lambda v: v)
|
|
elif par.annotation is bool:
|
|
parse.append(lambda v: bool(v))
|
|
else:
|
|
raise Exception(f"unsuported type {par.annotation}")
|
|
|
|
async def _proxy(ctx):
|
|
args = shlex.split(ctx._message)[1:]
|
|
args = [parse[i](v) for i, v in enumerate(args)]
|
|
await handler(ctx, *args)
|
|
|
|
return _proxy
|
|
|
|
|
|
class DiscordAuthor:
|
|
def __init__(self, nick):
|
|
self.nick = nick
|
|
self.global_name = nick
|
|
|
|
|
|
class DiscordContext:
|
|
def __init__(self, bot, target, source, message):
|
|
self._bot = bot
|
|
self._target = target
|
|
self._source = source
|
|
self._message = message
|
|
self.author = DiscordAuthor(source)
|
|
|
|
def typing(self):
|
|
class NilTyping:
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *exc):
|
|
return None
|
|
|
|
return NilTyping()
|
|
|
|
async def reply(self, m, file=None):
|
|
if file:
|
|
# TODO: Host the file somewhere so it can be downloaded.
|
|
m += f" {file.filename}"
|
|
|
|
lines = m.strip().split("\n")
|
|
lines = [f"{self._source}: {line}" for line in lines]
|
|
await self._bot.message(self._target, "\n".join(lines))
|
|
|
|
|
|
class DiscordImplBot(pydle.Client):
|
|
def __init__(self, channel, nickname, *, push_messages=None, prefix="!"):
|
|
super().__init__(nickname, realname=nickname)
|
|
self._channel = channel
|
|
self._cmd_prefix = prefix
|
|
self._cmds = {}
|
|
self._push_messages = push_messages
|
|
self._push_messages_task = None
|
|
|
|
async def on_connect(self):
|
|
await self.join(self._channel)
|
|
if self._push_messages is not None:
|
|
self._push_messages_task = asyncio.create_task(self._handle_push_messages())
|
|
|
|
async def on_disconnect(self):
|
|
if self._push_messages_task is not None:
|
|
self._push_messages_task.cancel()
|
|
self._push_messages_task = None
|
|
|
|
async def _handle_push_messages(self):
|
|
async for message in self._push_messages():
|
|
if isinstance(message, DiscordEmbed):
|
|
await self.notice(self._channel, f"{message.title} {message.url}")
|
|
else:
|
|
await self.notice(self._channel, message)
|
|
|
|
async def on_message(self, target, source, message):
|
|
# Don't respond to our own messages, as this leads to a positive feedback loop.
|
|
if source == self.nickname:
|
|
return
|
|
if not message.startswith(self._cmd_prefix):
|
|
return
|
|
|
|
name = message.removeprefix(self._cmd_prefix).split(" ", 1)[0]
|
|
cmd_fn = self._cmds.get(name)
|
|
if not cmd_fn:
|
|
return
|
|
|
|
ctx = DiscordContext(self, target, source, message)
|
|
await cmd_fn(ctx)
|
|
|
|
# Discord API: Register a new command
|
|
def command(self, *, name=None, description=None):
|
|
def _reg_cmd(handler):
|
|
nonlocal name
|
|
name = name or handler.__name__
|
|
|
|
sig = inspect.signature(handler)
|
|
if len(sig.parameters) == 1:
|
|
# Just the DiscordContext argument.
|
|
self._cmds[name] = handler
|
|
else:
|
|
self._cmds[name] = command_args(handler, sig)
|
|
|
|
return _reg_cmd
|
|
|
|
|
|
def main(*, server, channel, nick, mqtt_host):
|
|
def push_messages():
|
|
return botcommands.run_events(mqtt_host)
|
|
|
|
bot = DiscordImplBot(
|
|
channel,
|
|
nick,
|
|
push_messages=push_messages,
|
|
)
|
|
botcommands.setup(bot, mqtt_host)
|
|
bot.run(server, tls=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mqtt_host = os.getenv("MQTT_HOST")
|
|
if not mqtt_host:
|
|
print("MQTT_HOST unset")
|
|
sys.exit(1)
|
|
|
|
main(
|
|
server="irc.libera.chat",
|
|
channel="#bitlair-bot-test",
|
|
nick="Bitlair",
|
|
mqtt_host=mqtt_host,
|
|
)
|