forked from DGNum/infrastructure
Ryan Lahfa
f40323bfc0
Those are not superpowers and should be used sparringly and responsibly. LLMs are not bulletproof. They are mostly bullshit generators. But even a bullshit generator has usefulness. Signed-off-by: Ryan Lahfa <ryan@dgnum.eu>
134 lines
5.1 KiB
Python
134 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
|
import asyncio
|
|
|
|
from ircrobots.interface import IBot
|
|
from ollama import Client as OllamaClient
|
|
from loadcredential import Credentials
|
|
import base64
|
|
|
|
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 __init__(self, bot: IBot, name: str, llm_client: OllamaClient):
|
|
super().__init__(bot, name)
|
|
self.llm_client = llm_client
|
|
|
|
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):
|
|
credentials = Credentials()
|
|
base64_encoded_password = base64.b64encode(credentials["OLLAMA_PROXY_PASSWORD"])
|
|
token = f"takumi:{base64_encoded_password}"
|
|
llm_client = OllamaClient(host='https://ollama01.beta.dgnum.eu', headers={'Authorization': f'Basic {token}'})
|
|
return Server(self, name, llm_client)
|
|
|
|
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())
|