83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
|
import re
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from moonchat import *
|
||
|
|
|
||
|
|
class Bot:
|
||
|
|
def __init__(self, chat: Moonchat, command_matcher: re.Pattern):
|
||
|
|
self.chat = chat
|
||
|
|
self.command_matcher = command_matcher
|
||
|
|
self.commands = dict()
|
||
|
|
self.last_annoyed = datetime.now()
|
||
|
|
self.seen = dict()
|
||
|
|
|
||
|
|
async def handle_incoming(self):
|
||
|
|
async for message in self.chat.messages():
|
||
|
|
now = datetime.now()
|
||
|
|
if message.nickname == 'Server':
|
||
|
|
continue # ignore the server
|
||
|
|
last_seen = self.seen.get(message.nickname, None)
|
||
|
|
if last_seen:
|
||
|
|
seen_delta = now - last_seen
|
||
|
|
if seen_delta > timedelta(hours=2):
|
||
|
|
last_seen = None
|
||
|
|
if not last_seen:
|
||
|
|
if (now - self.last_annoyed) > timedelta(minutes=10):
|
||
|
|
await self.chat.send_message(f"hello {message.nickname}! i am a robot. say [help]")
|
||
|
|
self.last_annoyed = now
|
||
|
|
self.seen[message.nickname] = datetime.now()
|
||
|
|
match = self.command_matcher.search(message.content)
|
||
|
|
if not match:
|
||
|
|
continue # ignore not our messages
|
||
|
|
command = match.groupdict().get('command', None)
|
||
|
|
if not command:
|
||
|
|
continue # ????
|
||
|
|
split = command.split()
|
||
|
|
if not len(split):
|
||
|
|
continue # ????????????
|
||
|
|
exector = split[0]
|
||
|
|
command_function = self.commands.get(exector, None)
|
||
|
|
if command_function:
|
||
|
|
await command_function(self, message, split)
|
||
|
|
else:
|
||
|
|
await self.chat.send_message(f"{message.nickname}: sorry that's not a valid command")
|
||
|
|
|
||
|
|
async def who_command(bot: Bot, message: MoonchatMessage, args):
|
||
|
|
"""See recent users"""
|
||
|
|
now = datetime.now()
|
||
|
|
result = "Users from last 1hour: "
|
||
|
|
for username, last_seen in bot.seen.items():
|
||
|
|
delta: timedelta = (now - last_seen)
|
||
|
|
if delta < timedelta(hours=1):
|
||
|
|
minutes, seconds = divmod(delta.seconds, 60)
|
||
|
|
result += f"{username}({minutes}m{seconds}s), "
|
||
|
|
await bot.chat.send_message(result)
|
||
|
|
|
||
|
|
async def whoami(bot: Bot, message: MoonchatMessage, args):
|
||
|
|
"""Print your nickname"""
|
||
|
|
await bot.chat.send_message(message.nickname)
|
||
|
|
|
||
|
|
async def help(bot: Bot, message: MoonchatMessage, args):
|
||
|
|
command = args[1] if len(args) > 1 else None
|
||
|
|
command_function = bot.commands.get(command, None)
|
||
|
|
if command_function:
|
||
|
|
await bot.chat.send_message(f"{command}: {command_function.__doc__}")
|
||
|
|
return
|
||
|
|
command_list = ', '.join(bot.commands.keys())
|
||
|
|
await bot.chat.send_message(f"Commands available: {command_list}")
|
||
|
|
|
||
|
|
matcher = re.compile(r"\[(?P<command>[\w\s]+)\]")
|
||
|
|
|
||
|
|
async def main():
|
||
|
|
chat = await Moonchat.connect("7ks473deh6ggtwqsvbqdurepv5i6iblpbkx33b6cydon3ajph73sssad.onion", 50000)
|
||
|
|
bot = Bot(chat, matcher)
|
||
|
|
bot.commands["help"] = help
|
||
|
|
bot.commands['who'] = who_command
|
||
|
|
bot.commands['whoami'] = whoami
|
||
|
|
await chat.send_message("i am a robot! do [help]")
|
||
|
|
await bot.handle_incoming()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
import asyncio
|
||
|
|
asyncio.run(main())
|