From 636ea209d18aec99e29687674bcb27949ded8b2f Mon Sep 17 00:00:00 2001 From: retoor Date: Mon, 10 Feb 2025 14:16:55 +0100 Subject: [PATCH] The snek bot updated! --- README.md | 92 +++++++++++++++++++++++++++++++-------------- example.py | 62 ++++++++++++++++++------------ src/snekbot/bot.py | 94 ++++++++++++++++++++++------------------------ 3 files changed, 146 insertions(+), 102 deletions(-) diff --git a/README.md b/README.md index cf4b585..e3fb191 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Snekbot API -This is the Snekbot API. This document describes how to create a bot responding to "hey", "hello", "bye" and "@username-of-bot". +This is the Snekbot API. This document describes how to create a bot responding to "hello", "bye" and "@username-of-bot". ## 5 minute tutorial @@ -32,52 +32,73 @@ Make sure you have this information right now: Open a file ending with the `.py` extension and pase this content. ```python +import asyncio -import asyncio from snekbot.bot import Bot class ExampleBot(Bot): - async def on_join(self, data): - print(f"I joined {data.channel_uid}!") - - async def on_leave(self, data): - print(f"I left {data.channel_uid}!") - - async def on_ping(self, data): - print(f"Ping from {data.user_nick}") + async def on_join(self, channel_uid): + print(f"I joined!") await self.send_message( - data.channel_uid, - "I should respond with Bong according to BordedDev. So here, bong!", + channel_uid, + f"Hello, i'm actively part of the conversation in channel {channel_uid} now, you don't have to mention me anymore. " + ) + + async def on_leave(self, channel_uid): + print(f"I left!!") + await self.send_message( + channel_uid, + "I stop actively being part of the conversation now. Bye!" + ) + + async def on_ping(self,username, user_nick, channel_uid, message): + print(f"Ping from {user_nick} in channel {channel_uid}: {message}") + await self.send_message( + channel_uid, + "pong " + message ) async def on_own_message(self, data): print(f"Received my own message: {data.message}") - async def on_mention(self, data): - message = data.message[len(self.username) + 2 :] - print(f"Mention from {data.user_nick}: {message}") + async def on_mention(self, username, user_nick, channel_uid, message): - result = f'Hey {data.user_nick}, Thanks for mentioning me "{message}".' - await self.send_message(data.channel_uid, result) + message = message[len(self.username) + 2 :] + print(f"Mention from {user_nick}: {message}") - async def on_message(self, data): - print(f"Message from {data.user_nick}: {data.message}") - - message = data.message.lower() + if "source" in message: + with open(__file__) as f: + result = f.read() + result = result.replace(f'"{self.username}"', '"example username"') + result = result.replace(self.password, "example password") + result = ( + "This is the actual source code running me now. Fresh from the bakery:\n\n```python\n" + + result + + "\n```" + ) + await self.send_message(channel_uid, result) + else: + await self.send_message(channel_uid, f'Hey {user_nick}, Thanks for mentioning me "{message}".') + + async def on_message(self, sender_username, sender_nick, channel_uid, message): + print(f"Message from {sender_nick}: {message}") + if not self.has_joined(channel_uid): + print(f"Probably not for me since i'm not mentioned and not joined yet") + return + message = message.lower() result = None - - if "hey" in message or "hello" in message: - result = f"Hi {data.user_nick}" + if "hello" in message: + result = f"Hi @{sender_nick}" elif "bye" in message: - result = f"Bye {data.user_nick}" + result = f"Bye @{sender_nick}" if result: - await self.send_message(data.channel_uid, result) + await self.send_message(channel_uid, result) -bot = ExampleBot(username="Your username", password="Your password",url="wss://your-snek-instance.com/rpc.ws") +bot = ExampleBot(url="ws://snek.molodetz.nl/rpc.ws", username="example", password="example") asyncio.run(bot.run()) ``` @@ -86,4 +107,19 @@ Make sure you have (still) activated your virtual env. ```bash python [your-script].py ``` -If you get the error 'python not found' or 'aiohttp not found', run `source .venv/bin/activate` again and run `python [your script].py` again. +If you get the error 'python not found' or 'aiohttp not found', run `source .venv/bin/activate` again and run `python [your script].py` again. + +#### Debugging +Add `import logging` and `logging.BasicConfig(level=logging.DEBUG)`. + +#### Summary +The `ExampleBot` class inherits from a base `Bot` class and implements several event handlers: + - `on_join`: Sends a welcome message when the bot joins a channel. + - `on_leave`: Sends a goodbye message when the bot leaves. + - `on_ping`: Responds with "pong" when it receives a ping message. + - `on_own_message`: Logs messages sent by the bot itself. + - `on_mention`: Handles mentions; if "source" is in the message, it replies with its own source code, with sensitive data disguised. + - `on_message`: Responds to "hello" and "bye" messages if the bot has joined the channel. +The bot will be instantiated and runs asynchronously. + + diff --git a/example.py b/example.py index f478650..74fc2a8 100644 --- a/example.py +++ b/example.py @@ -1,32 +1,42 @@ import asyncio +import logging + +logging.basicConfig(level=logging.DEBUG) from snekbot.bot import Bot class ExampleBot(Bot): - async def on_join(self, data): - print(f"I joined {data.channel_uid}!") - - async def on_leave(self, data): - print(f"I left {data.channel_uid}!") - - async def on_ping(self, data): - print(f"Ping from {data.user_nick}") + async def on_join(self, channel_uid): + print(f"I joined!") await self.send_message( - data.channel_uid, - "I should respond with Bong according to BordedDev. So here, bong!", + channel_uid, + f"Hello, i'm actively part of the conversation in channel {channel_uid} now, you don't have to mention me anymore. " + ) + + async def on_leave(self, channel_uid): + print(f"I left!!") + await self.send_message( + channel_uid, + "I stop actively being part of the conversation now. Bye!" + ) + + async def on_ping(self,username, user_nick, channel_uid, message): + print(f"Ping from {user_nick} in channel {channel_uid}: {message}") + await self.send_message( + channel_uid, + "pong " + message ) async def on_own_message(self, data): print(f"Received my own message: {data.message}") - async def on_mention(self, data): + async def on_mention(self, username, user_nick, channel_uid, message): - message = data.message[len(self.username) + 2 :] - print(f"Mention from {data.user_nick}: {message}") + message = message[len(self.username) + 2 :] + print(f"Mention from {user_nick}: {message}") - result = f'Hey {data.user_nick}, Thanks for mentioning me "{message}".' if "source" in message: with open(__file__) as f: result = f.read() @@ -37,21 +47,25 @@ class ExampleBot(Bot): + result + "\n```" ) + await self.send_message(channel_uid, result) + else: + await self.send_message(channel_uid, f'Hey {user_nick}, Thanks for mentioning me "{message}".') - await self.send_message(data.channel_uid, result) - - async def on_message(self, data): - print(f"Message from {data.user_nick}: {data.message}") - message = data.message.lower() + async def on_message(self, sender_username, sender_nick, channel_uid, message): + print(f"Message from {sender_nick}: {message}") + if not self.has_joined(channel_uid): + print(f"Probably not for me since i'm not mentioned and not joined yet") + return + message = message.lower() result = None - if "hey" in message or "hello" in message: - result = f"Hi {data.user_nick}" + if "hello" in message: + result = f"Hi @{sender_nick}" elif "bye" in message: - result = f"Bye {data.user_nick}" + result = f"Bye @{sender_nick}" if result: - await self.send_message(data.channel_uid, result) + await self.send_message(channel_uid, result) -bot = ExampleBot(username="example", password="example") +bot = ExampleBot(url="ws://snek.molodetz.nl/rpc.ws", username="example", password="example") asyncio.run(bot.run()) diff --git a/src/snekbot/bot.py b/src/snekbot/bot.py index 5bc1528..1245b8a 100644 --- a/src/snekbot/bot.py +++ b/src/snekbot/bot.py @@ -17,6 +17,9 @@ import aiohttp from snekbot.rpc import RPC +import logging + +logger = logging.getLogger("snekbot") class Bot: @@ -28,7 +31,7 @@ class Bot: self.channels = None self.rpc = None self.ws = None - self.join_conversation = False + self.joined = set() async def run(self, reconnect=True): while True: @@ -39,6 +42,9 @@ class Bot: await asyncio.sleep(1) if not reconnect: break + + def has_joined(self, channel_uid): + return channel_uid in self.joined async def send_message(self, channel_uid, message): await self.rpc.send_message(channel_uid, message) @@ -50,60 +56,48 @@ class Bot: self.ws = ws self.rpc = RPC(ws) rpc = self.rpc - await (await rpc.login(self.username, self.password))() - try: - raise Exception(self.login_result.exception) - except: - pass + success = await (await rpc.login(self.username, self.password))() + print(success) self.channels = await (await rpc.get_channels())() + for channel in self.channels: + logger.debug("Found channel: " + channel["name"]) self.user = (await (await rpc.get_user(None))()).data + logger.debug("Logged in as: " + self.user["username"]) self.join_conversation = False while True: - print("Waiting for message...") - data = await rpc.receive() - - if not data: - break - try: - message = data.message.strip() - except: - continue + try: + await self.on_idle() + except AttributeError: + pass + + data = await rpc.receive() - if data.username == self.user["username"]: try: + message = data.message.strip() + except AttributeError: + continue + + if data.username == self.user["username"]: await self.on_own_message(data) - except Exception as ex: - print("Error", ex) - continue - elif message.startswith("ping"): - try: - await self.on_ping(data) - except Exception as ex: - print("Error:", ex) - continue - elif "@" + self.user["nick"] in data.message: - try: - await self.on_mention(data) - except Exception as ex: - print("Error:", ex) - continue - elif "@" + self.user["nick"] + " join" in data.message: - self.join_conversation = True - try: - await self.on_join(data) - except: - print("Error:", ex) - continue - elif "@" + self.user["nick"] + " leave" in data.message: - self.join_conversation = False - try: - await self.on_leave(data) - except: - print("Error:", ex) - continue - else: - try: - await self.on_message(data) - except Exception as ex: - print("Error:", ex) + elif message.startswith("ping"): + await self.on_ping(data.username, data.user_nick, data.channel_uid, data.message.lstrip("ping ").strip()) + elif any([ + "@" + self.user["nick"] + " join" in data.message, + "@" + self.user["username"] + " join" in data.message]): + self.joined.add(data.channel_uid) + await self.on_join(data.channel_uid) + elif any([ + "@" + self.user["nick"] + " leave" in data.message, + "@" + self.user["username"] + " leave" in data.message]): + self.joined.remove(data.channel_uid) + await self.on_leave(data.channel_uid) + elif "@" + self.user["nick"] in data.message or "@" + self.user["username"] in data.message: + await self.on_mention(data.username, data.user_nick, data.channel_uid, data.message) + else: + await self.on_message(data.username, data.user_nick, data.channel_uid, data.message) + except AttributeError: + logger.debug("Event unhandled regarding message: " + data.message) + except Exception as ex: + logger.exception(ex) + break