The snek bot updated!

Browse Source
This commit is contained in:
retoor 2025-02-10 14:16:55 +01:00
parent 0bde0cedd4
commit 636ea209d1
3 changed files with 146 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
README.md
View File

@ -1,6 +1,6 @@
# Snekbot API # 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 ## 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. Open a file ending with the `.py` extension and pase this content.
```python ```python
import asyncio import asyncio
from snekbot.bot import Bot from snekbot.bot import Bot
class ExampleBot(Bot): class ExampleBot(Bot):
async def on_join(self, data): async def on_join(self, channel_uid):
print(f"I joined {data.channel_uid}!") print(f"I joined!")
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}")
await self.send_message( await self.send_message(
data.channel_uid, channel_uid,
"I should respond with Bong according to BordedDev. So here, bong!", 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): async def on_own_message(self, data):
print(f"Received my own message: {data.message}") 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}")
result = f'Hey {data.user_nick}, Thanks for mentioning me "{message}".' message = message[len(self.username) + 2 :]
await self.send_message(data.channel_uid, result) print(f"Mention from {user_nick}: {message}")
async def on_message(self, data): if "source" in message:
print(f"Message from {data.user_nick}: {data.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}".')
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 result = None
if "hello" in message:
if "hey" in message or "hello" in message: result = f"Hi @{sender_nick}"
result = f"Hi {data.user_nick}"
elif "bye" in message: elif "bye" in message:
result = f"Bye {data.user_nick}" result = f"Bye @{sender_nick}"
if result: 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()) asyncio.run(bot.run())
``` ```
@ -87,3 +108,18 @@ Make sure you have (still) activated your virtual env.
python [your-script].py 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.

62
example.py
View File

@ -1,32 +1,42 @@
import asyncio import asyncio
import logging
logging.basicConfig(level=logging.DEBUG)
from snekbot.bot import Bot from snekbot.bot import Bot
class ExampleBot(Bot): class ExampleBot(Bot):
async def on_join(self, data): async def on_join(self, channel_uid):
print(f"I joined {data.channel_uid}!") print(f"I joined!")
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}")
await self.send_message( await self.send_message(
data.channel_uid, channel_uid,
"I should respond with Bong according to BordedDev. So here, bong!", 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): async def on_own_message(self, data):
print(f"Received my own message: {data.message}") 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 :] message = message[len(self.username) + 2 :]
print(f"Mention from {data.user_nick}: {message}") print(f"Mention from {user_nick}: {message}")
result = f'Hey {data.user_nick}, Thanks for mentioning me "{message}".'
if "source" in message: if "source" in message:
with open(__file__) as f: with open(__file__) as f:
result = f.read() result = f.read()
@ -37,21 +47,25 @@ class ExampleBot(Bot):
+ result + result
+ "\n```" + "\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, sender_username, sender_nick, channel_uid, message):
print(f"Message from {sender_nick}: {message}")
async def on_message(self, data): if not self.has_joined(channel_uid):
print(f"Message from {data.user_nick}: {data.message}") print(f"Probably not for me since i'm not mentioned and not joined yet")
message = data.message.lower() return
message = message.lower()
result = None result = None
if "hey" in message or "hello" in message: if "hello" in message:
result = f"Hi {data.user_nick}" result = f"Hi @{sender_nick}"
elif "bye" in message: elif "bye" in message:
result = f"Bye {data.user_nick}" result = f"Bye @{sender_nick}"
if result: 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()) asyncio.run(bot.run())

94
src/snekbot/bot.py
View File

@ -17,6 +17,9 @@ import aiohttp
from snekbot.rpc import RPC from snekbot.rpc import RPC
import logging
logger = logging.getLogger("snekbot")
class Bot: class Bot:
@ -28,7 +31,7 @@ class Bot:
self.channels = None self.channels = None
self.rpc = None self.rpc = None
self.ws = None self.ws = None
self.join_conversation = False self.joined = set()
async def run(self, reconnect=True): async def run(self, reconnect=True):
while True: while True:
@ -40,6 +43,9 @@ class Bot:
if not reconnect: if not reconnect:
break break
def has_joined(self, channel_uid):
return channel_uid in self.joined
async def send_message(self, channel_uid, message): async def send_message(self, channel_uid, message):
await self.rpc.send_message(channel_uid, message) await self.rpc.send_message(channel_uid, message)
return True return True
@ -50,60 +56,48 @@ class Bot:
self.ws = ws self.ws = ws
self.rpc = RPC(ws) self.rpc = RPC(ws)
rpc = self.rpc rpc = self.rpc
await (await rpc.login(self.username, self.password))() success = await (await rpc.login(self.username, self.password))()
try: print(success)
raise Exception(self.login_result.exception)
except:
pass
self.channels = await (await rpc.get_channels())() 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 self.user = (await (await rpc.get_user(None))()).data
logger.debug("Logged in as: " + self.user["username"])
self.join_conversation = False self.join_conversation = False
while True: while True:
print("Waiting for message...")
data = await rpc.receive()
if not data:
break
try: try:
message = data.message.strip() try:
except: await self.on_idle()
continue except AttributeError:
pass
data = await rpc.receive()
if data.username == self.user["username"]:
try: try:
message = data.message.strip()
except AttributeError:
continue
if data.username == self.user["username"]:
await self.on_own_message(data) await self.on_own_message(data)
except Exception as ex: elif message.startswith("ping"):
print("Error", ex) await self.on_ping(data.username, data.user_nick, data.channel_uid, data.message.lstrip("ping ").strip())
continue elif any([
elif message.startswith("ping"): "@" + self.user["nick"] + " join" in data.message,
try: "@" + self.user["username"] + " join" in data.message]):
await self.on_ping(data) self.joined.add(data.channel_uid)
except Exception as ex: await self.on_join(data.channel_uid)
print("Error:", ex) elif any([
continue "@" + self.user["nick"] + " leave" in data.message,
elif "@" + self.user["nick"] in data.message: "@" + self.user["username"] + " leave" in data.message]):
try: self.joined.remove(data.channel_uid)
await self.on_mention(data) await self.on_leave(data.channel_uid)
except Exception as ex: elif "@" + self.user["nick"] in data.message or "@" + self.user["username"] in data.message:
print("Error:", ex) await self.on_mention(data.username, data.user_nick, data.channel_uid, data.message)
continue else:
elif "@" + self.user["nick"] + " join" in data.message: await self.on_message(data.username, data.user_nick, data.channel_uid, data.message)
self.join_conversation = True except AttributeError:
try: logger.debug("Event unhandled regarding message: " + data.message)
await self.on_join(data) except Exception as ex:
except: logger.exception(ex)
print("Error:", ex) break
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)