lanonna

lanonna/commands.py

@@ -0,0 +1,17 @@
+from string import ascii_lowercase as ascii_lowercase_
+
+commands = {
+    'asino',
+}
+
+ascii_lowercase = set(ascii_lowercase_)
+
+
+def parse(src: str) -> None | tuple[str, str]:
+    '''The command format in PEG format is: `![a-z] ?whitespace ?(.*)'''
+    if src.startswith('!') and src[1:2] in ascii_lowercase:
+        cmd = src[1:].split(' ', 1)[0]
+        idx = len(cmd) + 2
+        return (cmd, src[idx:])
+    else:
+        return None

lanonna/config.py

@@ -0,0 +1,43 @@
+from typing import NamedTuple, Union
+import os
+import tomllib
+
+CONF = '/etc/lanonna.toml'
+
+
+class Configuration(NamedTuple):
+    matrix_url: str
+    matrix_username: str
+    matrix_password: str
+    mq_url: str
+
+
+def init() -> Union[str | Configuration]:
+
+    def missing_key(k):
+        return f'Missing configuration value in configuration file: {k}.'
+
+
+    if not os.path.exists(CONF):
+        return f"Can't read configuration file. Please create {CONF}."
+
+    with open('/etc/lanonna.toml', 'rb') as f:
+        conf = tomllib.load(f)
+
+    if 'lanonna' not in conf: # check that there is a toml section for this
+        return missing_key('lanonna')
+
+    conf = conf['lanonna']
+    keys = ('matrix_url', 'matrix_username', 'matrix_password', 'mq_url')
+
+    for key in keys:
+        if key not in conf:
+            return missing_key(key)
+        else:
+            continue
+
+    rest = set(conf.keys()) - set(keys)
+    if len(rest) > 0:
+        return f"There are unknown configuration values in the config file: '{rest}'. Please remove those."
+    else:
+        return Configuration(**conf)

lanonna/lanonna.py

@@ -0,0 +1,64 @@
+import asyncio
+import logging
+import sys
+
+
+import matrix
+import mq
+import protocol
+import config
+
+logger = logging.getLogger('lanonna')
+logger.setLevel(logging.WARNING)
+logging.basicConfig(level=logging.INFO)
+
+async def matrixmain(matrix_client: matrix.MatrixClient, mq_client: mq.MQClient):
+    from nio import RoomMessageText
+
+    client = matrix_client.client
+
+    callback = lambda r, e: matrix.message_received_cb(matrix_client, mq_client, logger, r, e)
+    client.add_event_callback(callback, RoomMessageText)
+
+    try:
+        await client.sync_forever(timeout=30000, full_state=True)  # milliseconds
+        logging.info('Exiting from the matrix loop')
+        client.logout()
+    except Exception as e:
+        logging.exception(f'Exception in matrix loop: {e}')
+
+
+async def mqmain(rabbit_client: mq.MQClient, matrix_client: matrix.MatrixClient):
+    async def loop():
+        async with rabbit_client.queue.iterator() as queue_iter:
+            async for message in queue_iter:
+                async with message.process():
+                    body = message.body.decode()
+                    br = protocol.json_to_bot_response(body)
+                    # logging.info(f'New message from MQ: {str(br)[:24]}...')
+                    await matrix.send_text(matrix_client, br)
+
+    try:
+        async with rabbit_client.connection:
+            await loop()
+    except Exception as e:
+        logging.exception(f'Exception in mq loop: {e}')
+
+
+async def main(conf: config.Configuration):
+    mq_client = await mq.initialize(conf)
+    matrix_client = await matrix.initialize(conf)
+
+    await asyncio.gather(
+        mqmain(mq_client, matrix_client),
+        matrixmain(matrix_client, mq_client)
+    )
+
+if __name__ == '__main__':
+    configuration = config.init()
+    if isinstance(configuration, str): # it's an error!
+        print(configuration, file=sys.stderr)
+        sys.exit(2)
+    else:
+        loop = asyncio.get_event_loop()
+        loop.run_until_complete(main(configuration))

lanonna/matrix.py

@@ -0,0 +1,62 @@
+import logging
+from collections import namedtuple
+from typing import Any
+from nio import AsyncClient, InviteEvent, RoomMessageText, MatrixRoom
+from markdown import markdown
+
+import protocol
+import mq
+import config
+import commands
+
+MatrixClient = namedtuple('MatrixClient', ('client'))
+
+async def initialize(c: config.Configuration):
+    client = AsyncClient(c.matrix_url, c.matrix_username)
+
+    logging.info(await client.login(c.matrix_password))
+    # always accept every room invite
+    client.add_event_callback(lambda room, _: client.join(room.room_id), InviteEvent)
+
+    return MatrixClient(client)
+
+
+async def send_text(matrix_client: MatrixClient, response: protocol.BotResponse):
+    client = matrix_client.client
+    content: dict[str, Any] = {"msgtype": "m.text", "body": response.content}
+
+    if response.as_markdown:
+        content["formatted_body"] = markdown(response.content)
+        content["format"] = "org.matrix.custom.html"
+
+    if response.as_reply and response.source_message_id:
+        content["m.relates_to"] = {"m.in_reply_to": {"event_id": response.source_message_id}}
+
+    try:
+        await client.room_send(response.room_id, 'm.room.message', content)
+        logging.info(f'Replied in matrix: {response}')
+    except Exception as e:
+        logging.exception(f"Unable to send message response {response}|{e}")
+
+async def message_received_cb(matrix_client: MatrixClient,
+                              rabbit_client: mq.MQClient,
+                              logger: logging.Logger,
+                              room: MatrixRoom,
+                              event: RoomMessageText):
+    import mq
+    cmd = commands.parse(event.body)
+    if cmd:
+        logging.info(f'got new matrix command: {cmd}')
+        swm = protocol.SwitchboardMessage(command=cmd[0],
+                                          content=cmd[1],
+                                          source_message_id=event.event_id,
+                                          sender_nick=room.user_name(event.sender),
+                                          room_id=room.room_id)
+        try:
+            if swm.command in commands.commands:
+                await mq.route_to_exchange(rabbit_client, swm)
+            else:
+                help_ = protocol.unknown_cmd_help_reply(swm)
+                await send_text(matrix_client, help_)
+        except Exception as e:
+            logging.exception(f"Can't route switchboard message: {swm}|{str(e)}")

lanonna/mq.py

@@ -0,0 +1,30 @@
+import aio_pika
+from collections import namedtuple
+
+import protocol
+import config
+
+MQClient = namedtuple('MQClient', ('queue', 'exchange',
+                                   'connection', 'channel'))
+
+
+async def initialize(c: config.Configuration):
+    connection = await aio_pika.connect_robust(c.mq_url)
+    channel = await connection.channel()
+
+    queue = await channel.declare_queue('lanonna')
+    exchange = await channel.declare_exchange('lanonna', durable=True)
+    await queue.bind(exchange)
+
+    return MQClient(queue, exchange, connection, channel)
+
+
+async def route_to_exchange(rabbit_client: MQClient, msg: protocol.SwitchboardMessage):
+    import json
+
+    routing_key = msg.command
+
+    btes = json.dumps(msg._asdict()).encode()
+    mqmsg = aio_pika.Message(body=btes)
+
+    await rabbit_client.channel.default_exchange.publish(mqmsg, routing_key)

lanonna/protocol.py

@@ -0,0 +1,49 @@
+from typing import NamedTuple, Optional
+import json
+
+
+class SwitchboardMessage(NamedTuple):
+    command: str
+    content: str
+    source_message_id: str
+    sender_nick: str
+    room_id: str
+
+
+class BotResponse(NamedTuple):
+    content: str
+    source_message_id: str | None
+    room_id: str
+    as_reply: bool  # requires source_message_id
+    as_markdown: bool
+
+
+def unknown_cmd_help_reply(swm: SwitchboardMessage):
+    '''Returns the standard help text
+       because the user requested an unknown command'''
+    import commands
+    WRONG_CMD_TEXT = f'Hai sbagliato commando, lezzo! Prova uno di questi: {commands.commands}'
+    return BotResponse(content=WRONG_CMD_TEXT,
+                       source_message_id=swm.source_message_id,
+                       room_id=swm.room_id,
+                       as_reply=True,
+                       as_markdown=False)
+
+
+def json_to_bot_response(json_str: str) -> Optional[BotResponse]:
+    import logging
+    try:
+        data = json.loads(json_str)
+
+        content = data['content']
+        room_id = data['room_id']
+
+        as_reply = data.get('as_reply', False)
+        as_markdown = data.get('as_markdown', False)
+        source_message_id = data.get('source_message_id', None)
+
+        return BotResponse(content, source_message_id, room_id, as_reply, as_markdown)
+
+    except Exception as e:
+        logging.exception(f"Error parsing JSON or missing required fields: {str(e)}")
+        return None

lanonna/pyproject.toml

@@ -0,0 +1,23 @@
+[tool.poetry]
+name = "lanonna"
+version = "1.0"
+description = "La Nonna è un bot per lezzo e matrix che sa tutto ma anche niente."
+authors = ["bparodi"]
+license = "GPLv3"
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.10"
+matrix-nio = "^0.23.0"
+pika = "^1.3.2"
+aio-pika = "^9.3.1"
+markdown = "^3.5.1"
+
+
+[tool.poetry.group.dev.dependencies]
+mypy = "^1.8.0"
+pytest = "^7.4.3"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

lanonna/tests/test_cmd_parser.py

@@ -0,0 +1,27 @@
+import commands
+
+def test_parser():
+    yes = [
+        ('!alert', ''),
+        ('!alert 23', '23'),
+        ('!alert @9090132-', '@9090132-'),
+        ('!alert ciao', 'ciao'),
+        ('!alert ciao ciaone', 'ciao ciaone'),
+        ('!alert ciao ciaone ancora', 'ciao ciaone ancora'),
+    ]
+
+    nope = [
+        '!',
+        '!123',
+        '! ',
+        'alert',
+        ' '
+    ]
+
+    for c, exp in yes:
+        res = commands.parse(c)
+        assert res[0] == 'alert'
+        assert res[1] == exp
+
+    for c in nope:
+        assert commands.parse(c) is None