infrastructure/modules/dgn-chatops/takumi.py
Ryan Lahfa 6b30a95fbb
All checks were successful
build configuration / build_vault01 (push) Successful in 1m13s
build configuration / build_storage01 (push) Successful in 1m16s
build configuration / build_web02 (push) Successful in 1m18s
build configuration / build_compute01 (push) Successful in 1m28s
build configuration / build_web01 (push) Successful in 1m43s
lint / check (push) Successful in 24s
build configuration / build_geo01 (push) Successful in 1m2s
build configuration / build_rescue01 (push) Successful in 1m6s
build configuration / build_bridge01 (push) Successful in 1m1s
build configuration / build_geo02 (push) Successful in 1m4s
build configuration / push_to_cache_storage01 (push) Successful in 1m25s
build configuration / push_to_cache_web02 (push) Successful in 1m21s
build configuration / push_to_cache_rescue01 (push) Successful in 1m17s
build configuration / push_to_cache_geo01 (push) Successful in 1m29s
build configuration / push_to_cache_compute01 (push) Successful in 1m55s
build configuration / push_to_cache_geo02 (push) Successful in 1m20s
build configuration / push_to_cache_bridge01 (push) Successful in 1m25s
build configuration / push_to_cache_web01 (push) Successful in 2m16s
feat(dgn-chatops): !pm shorthand for !probe_meet
Bumped minor version of Takumi for non-breaking changes.

Signed-off-by: Ryan Lahfa <ryan@dgnum.eu>
2024-09-21 00:52:13 +02:00

121 lines
4.6 KiB
Python

#!/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())