infrastructure/modules/nixos/dgn-chatops/takumi.py

#!/usr/bin/env python3
import asyncio

from irctokens.line import build, Line
from ircrobots.bot import Bot as BaseBot
from ircrobots.server import Server as BaseServer
from ircrobots.params import ConnectionParams

import aiohttp

BRIDGE_NICKNAME = "hermes"

SERVERS = [
    ("dgnum", "irc.dgnum.eu")
]

TEAMS = {
    "fai": ("tomate", "elias", "JeMaGius", "Luj", "catvayor", "Raito"),
    "marketing": ("cst1", "elias"),
    "bureau": ("Raito", "JeMaGius", "Luj", "gdd")
}

# times format is 0700-29092024
TRIGGER = '!'
async def create_meet(title: str, times: list[str], timezone: str = "UTC") -> str:
    async with aiohttp.ClientSession() as session:
        payload = {
            'name': title,
            'times': times,
            'timezone': timezone
        }
        async with session.post('https://api.meet.dgnum.eu/event', json=payload) as response:
            response.raise_for_status()
            id = (await response.json()).get('id')
            if not id:
                raise RuntimeError('No ID attributed to a meet')
            return f'https://meet.dgnum.eu/{id}'

def expand_times(times: list[str]) -> list[str]:
    expanded = []
    # TODO: verify the date exist in the calendar
    # TODO: verify that we don't write any duplicates.
    for time in times:
        if '-' not in time:
            for i in range(7, 20):
                expanded.append(f'{i:02}00-{time}')
        else:
            expanded.append(time)
    return expanded

def bridge_stripped(possible_command: str, origin_nick: str) -> str | None:
    if origin_nick.lower() == BRIDGE_NICKNAME:
        stripped_user = possible_command.split(':')[1].lstrip()
        return stripped_user if stripped_user.startswith(TRIGGER) else None
    else:
        return possible_command if possible_command.startswith(TRIGGER) else None

class Server(BaseServer):
    def extract_valid_command(self, line: Line) -> str | None:
        me = self.nickname_lower
        if line.command == "PRIVMSG" and \
                self.has_channel(line.params[0]) and \
                line.hostmask is not None and \
                self.casefold(line.hostmask.nickname) != me and \
                self.has_user(line.hostmask.nickname):
            return bridge_stripped(line.params[1], line.hostmask.nickname)


    async def line_read(self, line: Line):
        print(f"{self.name} < {line.format()}")
        if line.command == "001":
            print(f"connected to {self.isupport.network}")
            await self.send(build("JOIN", ["#dgnum-bridge-test"]))

        # In case `!probe_meet <title> <team> <time_1> <time_2> … <time_N> [<timezone>]`
        if (command := self.extract_valid_command(line)) is not None:
            text = command.lstrip(TRIGGER)
            if text.startswith('probe_meet') or text.startswith('pm'):
                args = text.split(' ')
                if len(args) < 4:
                    await self.send(build("PRIVMSG", [line.params[0], "usage is !probe_meet <title> <team> <time_1> [<time_2> <time_3> … <time_N>] ; time is in [00-hour-]DDMMYYYY format."]))
                    return

                title, team = args[1], args[2]
                print(f"creating meet '{title}' for team '{team}'")
                try:
                    times = expand_times(args[3:])
                    link = await create_meet(title, times)
                    if team not in TEAMS:
                        await self.send(build("PRIVMSG", [line.params[0], f"team {team} does not exist"]))
                        return

                    targets = TEAMS[team]
                    ping_mentions = ', '.join(targets)
                    await self.send(build("PRIVMSG", [line.params[0], f'{ping_mentions} {link}']))
                except ValueError as e:
                    print(e)
                    await self.send(build("PRIVMSG", [line.params[0], "time format is [00-hour-]DDMMYYYY, hour is optional, by default it's 07:00 to 19:00 in Europe/Paris timezone"]))
                except aiohttp.ClientError as e:
                    print(e)
                    await self.send(build("PRIVMSG", [line.params[0], "failed to create the meet on meet.dgnum.eu, API error, check the logs"]))


    async def line_send(self, line: Line):
        print(f"{self.name} > {line.format()}")

class Bot(BaseBot):
    def create_server(self, name: str):
        return Server(self, name)

async def main():
    bot = Bot()
    for name, host in SERVERS:
        # For IPv4-only connections.
        params = ConnectionParams("Takumi", host, 6698)
        await bot.add_server(name, params)

    await bot.run()

if __name__ == "__main__":
    asyncio.run(main())