PrivateCoffee/matrix-reportbot
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Created RSSBot

Browse source
This commit is contained in:
Kumi 2024-03-02 15:31:14 +01:00
parent d5a96aebb6
commit 967d738e64
Signed by: kumi
GPG key ID: ECBCC9082395383F
80 changed files with 830 additions and 4002 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.gitignore vendored
View file

@ -1,10 +1,6 @@
*.db
*.db.wal
*.db-journal
config.ini config.ini
venv/ .venv/
*.pyc *.pyc
__pycache__/ __pycache__/
*.bak
dist/ dist/
pantalaimon.conf pantalaimon.conf

21
.gitlab-ci.yml
View file

@ -1,21 +0,0 @@
image: python:3.10
stages:
- test
- publish
before_script:
- python -V
- python -m venv venv
- source venv/bin/activate
- pip install -U pip
- pip install .[all]
publish:
stage: publish
script:
- pip install -U hatchling twine build
- python -m build .
- python -m twine upload --username __token__ --password ${PYPI_TOKEN} dist/*
only:
- tags

15
.vscode/launch.json vendored
View file

@ -1,15 +0,0 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Python: Module",
"type": "python",
"request": "launch",
"module": "gptbot",
"justMyCode": true
}
]
}

3
.vscode/settings.json vendored
View file

@ -1,3 +0,0 @@
{
"python.formatting.provider": "black"
}

21
CHANGELOG.md
View file

@ -1,21 +0,0 @@
# Changelog
### 0.3.3 (2024-01-26)
* Implement recursion check in response generation (e6bc23e564e51aa149432fc67ce381a9260ee5f5)
* Implement tool emulation for models without tool support (0acc1456f9e4efa09e799f6ce2ec9a31f439fe4a)
* Allow selection of chat model by room (87173ae284957f66594e66166508e4e3bd60c26b)
### 0.3.2 (2023-12-14)
* Removed key upload from room event handler
* Fixed output of `python -m gptbot -v` to display currently installed version
* Workaround for bug preventing bot from responding when files are uploaded to an encrypted room
#### Known Issues
* When using Pantalaimon: Bot is unable to download/use files uploaded to encrypted rooms
### 0.3.1 (2023-12-07)
* Fixed issue in newroom task causing it to be called over and over again

2
LICENSE
View file

@ -1,4 +1,4 @@
Copyright (c) 2023 Kumi Mitterer <gptbot@kumi.email> Copyright (c) 2024 Private.coffee Team <support@private.coffee>
Permission is hereby granted, free of charge, to any person obtaining a copy Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal of this software and associated documentation files (the "Software"), to deal

218
README.md
View file

@ -1,217 +1,3 @@
# GPTbot # Matrix-RSSBot
GPTbot is a simple bot that uses different APIs to generate responses to This is a simple, no-database RSS to Matrix bridge bot.
messages in a Matrix room.
## Features
- AI-generated responses to text, image and voice messages in a Matrix room
(chatbot)
- Currently supports OpenAI (`gpt-3.5-turbo` and `gpt-4`, including vision
preview, `whisper` and `tts`)
- Able to generate pictures using OpenAI `dall-e-2`/`dall-e-3` models
- Able to browse the web to find information
- Able to use OpenWeatherMap to get weather information (requires separate
API key)
- Even able to roll dice!
- Mathematical calculations via the `!gptbot calculate` command
- Currently supports WolframAlpha (requires separate API key)
- Really useful commands like `!gptbot help` and `!gptbot coin`
- sqlite3 database to store room settings
## Installation
To run the bot, you will need Python 3.10 or newer.
The bot has been tested with Python 3.11 on Arch, but should work with any
current version, and should not require any special dependencies or operating
system features.
### Production
The easiest way to install the bot is to use pip to install it from pypi.
```shell
# If desired, activate a venv first
python -m venv venv
. venv/bin/activate
# Install the bot
pip install matrix-gptbot[all]
```
This will install the latest release of the bot and all required dependencies
for all available features.
You can also use `pip install git+https://kumig.it/kumitterer/matrix-gptbot.git`
to install the latest version from the Git repository.
#### End-to-end encryption
WARNING: Using end-to-end encryption seems to sometimes cause problems with
file attachments, especially in rooms that are not encrypted, if the same
user also uses the bot in encrypted rooms.
The bot itself does not implement end-to-end encryption. However, it can be
used in conjunction with [pantalaimon](https://github.com/matrix-org/pantalaimon),
which is actually installed as a dependency of the bot.
To use pantalaimon, create a `pantalaimon.conf` following the example in
`pantalaimon.example.conf`, making sure to change the homeserver URL to match
your homeserver. Then, start pantalaimon with `pantalaimon -c pantalaimon.conf`.
You first have to log in to your homeserver using `python pantalaimon_first_login.py`,
and can then use the returned access token in your bot's `config.ini` file.
Make sure to also point the bot to your pantalaimon instance by setting
`homeserver` to your pantalaimon instance instead of directly to your
homeserver in your `config.ini`.
Note: If you don't use pantalaimon, the bot will still work, but it will not
be able to decrypt or encrypt messages. This means that you cannot use it in
rooms with end-to-end encryption enabled.
### Development
Clone the repository and install the requirements to a virtual environment.
```shell
# Clone the repository
git clone https://kumig.it/kumitterer/matrix-gptbot.git
cd matrix-gptbot
# If desired, activate a venv first
python -m venv venv
. venv/bin/activate
# Install the bot in editable mode
pip install -e .[dev]
# Go to the bot directory and start working
cd src/gptbot
```
Of course, you can also fork the repository on [GitHub](https://github.com/kumitterer/matrix-gptbot/)
and work on your own copy.
### Configuration
The bot requires a configuration file to be present in the working directory.
Copy the provided `config.dist.ini` to `config.ini` and edit it to your needs.
## Running
The bot can be run with `python -m gptbot`. If required, activate a venv first.
You may want to run the bot in a screen or tmux session, or use a process
manager like systemd. The repository contains a sample systemd service file
(`gptbot.service`) that you can use as a starting point. You will need to
adjust the paths in the file to match your setup, then copy it to
`/etc/systemd/system/gptbot.service`. You can then start the bot with
`systemctl start gptbot` and enable it to start automatically on boot with
`systemctl enable gptbot`.
Analogously, you can use the provided `gptbot-pantalaimon.service` file to run
pantalaimon as a systemd service.
## Usage
Once it is running, just invite the bot to a room and it will start responding
to messages. If you want to create a new room, you can use the `!gptbot newroom`
command at any time, which will cause the bot to create a new room and invite
you to it. You may also specify a room name, e.g. `!gptbot newroom My new room`.
### Reply generation
Note that the bot will respond to _all_ messages in the room by default. If you
don't want this, for example because you want to use the bot in a room with
other people, you can use the `!gptbot roomsettings` command to change the
settings for the current room. For example, you can disable response generation
with `!gptbot roomsettings always_reply false`.
With this setting, the bot will only be triggered if a message begins with
`!gptbot chat`. For example, `!gptbot chat Hello, how are you?` will cause the
bot to generate a response to the message `Hello, how are you?`. The bot will
still get previous messages in the room as context for generating the response.
### Tools
The bot has a selection of tools at its disposal that it will automatically use
to generate responses. For example, if you send a message like "Draw me a
picture of a cat", the bot will automatically use DALL-E to generate an image
of a cat.
Note that this only works if the bot is configured to use a model that supports
tools. This currently is only the case for OpenAI's `gpt-3.5-turbo` model. If
you wish to use `gpt-4` instead, you can set the `ForceTools` option in the
`[OpenAI]` section of the config file to `1`. This will cause the bot to use
`gpt-3.5-turbo` for tool generation and `gpt-4` for generating the final text
response.
Similarly, it will attempt to use the `gpt-4-vision-preview` model to "read"
the contents of images if a non-vision model is used.
### Commands
There are a few commands that you can use to explicitly call a certain feature
of the bot. For example, if you want to generate an image from a text prompt,
you can use the `!gptbot imagine` command. For example, `!gptbot imagine a cat`
will cause the bot to generate an image of a cat.
To learn more about the available commands, `!gptbot help` will print a list of
available commands.
### Voice input and output
The bot supports voice input and output, but it is disabled by default. To
enable it, use the `!gptbot roomsettings` command to change the settings for
the current room. `!gptbot roomsettings stt true` will enable voice input using
OpenAI's `whisper` model, and `!gptbot roomsettings tts true` will enable voice
output using the `tts` model.
Note that this currently only works for audio messages and .mp3 file uploads.
## Troubleshooting
**Help, the bot is not responding!**
First of all, make sure that the bot is actually running. (Okay, that's not
really troubleshooting, but it's a good start.)
If the bot is running, check the logs. The first few lines should contain
"Starting bot...", "Syncing..." and "Bot started". If you don't see these
lines, something went wrong during startup. Fortunately, the logs should
contain more information about what went wrong.
If you need help figuring out what went wrong, feel free to open an issue.
**Help, the bot is flooding the room with responses!**
The bot will respond to _all_ messages in the room, with two exceptions:
- If you turn off response generation for the room, the bot will only respond
to messages that begin with `!gptbot <command>`.
- Messages sent by the bot itself will not trigger a response.
There is a good chance that you are seeing the bot responding to its own
messages. First, stop the bot, or it will keep responding to its own messages,
consuming tokens.
Check that the UserID provided in the config file matches the UserID of the bot.
If it doesn't, change the config file and restart the bot. Note that the UserID
is optional, so you can also remove it from the config file altogether and the
bot will try to figure out its own User ID.
If the User ID is correct or not set, something else is going on. In this case,
please check the logs and open an issue if you can't figure out what's going on.
## License
This project is licensed under the terms of the MIT license. See the [LICENSE](LICENSE)
file for details.

153
config.dist.ini
View file

@ -5,7 +5,7 @@
############################################################################### ###############################################################################
[GPTBot] [RSSBot]
# Some way for the user to contact you. # Some way for the user to contact you.
# Ideally, either your personal user ID or a support room # Ideally, either your personal user ID or a support room
@ -20,29 +20,13 @@ Operator = Contact details not set
# Debug = 1 # Debug = 1
# The default room name used by the !newroom command # The default room name used by the !newroom command
# Defaults to GPTBot if not set # Defaults to RSSBot if not set
# #
# DefaultRoomName = GPTBot # DefaultRoomName = RSSBot
# Contents of a special message sent to the GPT API with every request.
# Can be used to give the bot some context about the environment it's running in
#
# SystemMessage = You are a helpful bot.
# Force inclusion of the SystemMessage defined above if one is defined on per-room level
# If no custom message is defined for the room, SystemMessage is always included
#
# ForceSystemMessage = 0
# Path to a custom logo
# Used as room/space image and profile picture
# Defaults to logo.png in assets directory
#
# Logo = assets/logo.png
# Display name for the bot # Display name for the bot
# #
# DisplayName = GPTBot # DisplayName = RSSBot
# A list of allowed users # A list of allowed users
# If not defined, everyone is allowed to use the bot # If not defined, everyone is allowed to use the bot
@ -58,81 +42,6 @@ LogLevel = info
############################################################################### ###############################################################################
[OpenAI]
# The Chat Completion model you want to use.
#
# Unless you are in the GPT-4 beta (if you don't know - you aren't),
# leave this as the default value (gpt-3.5-turbo)
#
# Model = gpt-3.5-turbo
# The Image Generation model you want to use.
#
# ImageModel = dall-e-2
# Your OpenAI API key
#
# Find this in your OpenAI account:
# https://platform.openai.com/account/api-keys
#
APIKey = sk-yoursecretkey
# The maximum amount of input sent to the API
#
# In conjunction with MaxMessage, this determines how much context (= previous
# messages) you can send with your query.
#
# If you set this too high, the responses you receive will become shorter the
# longer the conversation gets.
#
# https://help.openai.com/en/articles/4936856-what-are-tokens-and-how-to-count-them
#
# MaxTokens = 3000
# The maximum number of messages in the room that will be considered as context
#
# By default, the last (up to) 20 messages will be sent as context, in addition
# to the system message and the current query itself.
#
# MaxMessages = 20
# The base URL of the OpenAI API
#
# Setting this allows you to use a self-hosted AI model for chat completions
# using something like https://github.com/abetlen/llama-cpp-python
#
# BaseURL = https://openai.local/v1
# Whether to force the use of tools in the chat completion model
#
# Currently, only gpt-3.5-turbo supports tools. If you set this to 1, the bot
# will use that model for tools even if you have a different model set as the
# default. It will only generate the final result using the default model.
#
# ForceTools = 0
# Whether to emulate tools in the chat completion model
#
# This will make the bot use the default model to *emulate* tools. This is
# useful if you want to use a model that doesn't support tools, but still want
# to be able to use tools. However, this may cause all kinds of weird results.
#
# EmulateTools = 0
###############################################################################
[WolframAlpha]
# An API key for Wolfram|Alpha
# Request one at https://developer.wolframalpha.com
#
# Leave unset to disable Wolfram|Alpha integration (`!gptbot calculate`)
#
#APIKey = YOUR-APIKEY
###############################################################################
[Matrix] [Matrix]
# The URL to your Matrix homeserver # The URL to your Matrix homeserver
@ -153,58 +62,6 @@ AccessToken = syt_yoursynapsetoken
# The Matrix user ID of the bot (@local:domain.tld) # The Matrix user ID of the bot (@local:domain.tld)
# Only specify this if the bot fails to figure it out by itself # Only specify this if the bot fails to figure it out by itself
# #
# UserID = @gptbot:matrix.local # UserID = @rssbot:matrix.local
############################################################################### ###############################################################################
[Database]
# Path of the main database
# Used to "remember" settings, etc.
#
Path = database.db
# Path of the Crypto Store - required to support encrypted rooms
# (not tested/supported yet)
#
CryptoStore = store.db
###############################################################################
[TrackingMore]
# API key for TrackingMore
# If not defined, the bot will not be able to provide parcel tracking
#
# APIKey = abcde-fghij-klmnop
###############################################################################
[Replicate]
# API key for replicate.com
# Can be used to run lots of different AI models
# If not defined, the features that depend on it are not available
#
# APIKey = r8_alotoflettersandnumbershere
###############################################################################
[HuggingFace]
# API key for Hugging Face
# Can be used to run lots of different AI models
# If not defined, the features that depend on it are not available
#
# APIKey = __________________________
###############################################################################
[OpenWeatherMap]
# API key for OpenWeatherMap
# If not defined, the bot will be unable to provide weather information
#
# APIKey = __________________________
###############################################################################

15
gptbot-pantalaimon.service
View file

@ -1,15 +0,0 @@
[Unit]
Description=Pantalaimon for GPTbot
Requires=network.target
[Service]
Type=simple
User=gptbot
Group=gptbot
WorkingDirectory=/opt/gptbot
ExecStart=/opt/gptbot/venv/bin/python3 -um pantalaimon.main -c pantalaimon.conf
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target

15
gptbot.service
View file

@ -1,15 +0,0 @@
[Unit]
Description=GPTbot - A multifunctional Chatbot for Matrix
Requires=network.target
[Service]
Type=simple
User=gptbot
Group=gptbot
WorkingDirectory=/opt/gptbot
ExecStart=/opt/gptbot/venv/bin/python3 -um gptbot
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target

56
pyproject.toml
View file

@ -6,20 +6,20 @@ build-backend = "hatchling.build"
allow-direct-references = true allow-direct-references = true
[project] [project]
name = "matrix-gptbot" name = "matrix-rssbot"
version = "0.3.5" version = "0.1.0"
authors = [ authors = [
{ name="Kumi Mitterer", email="gptbot@kumi.email" }, { name="Private.coffee Team", email="support@private.coffee" },
] ]
description = "Multifunctional Chatbot for Matrix" description = "Simple RSS feed bridge for Matrix"
readme = "README.md" readme = "README.md"
license = { file="LICENSE" } license = { file="LICENSE" }
requires-python = ">=3.10" requires-python = ">=3.10"
packages = [ packages = [
"src/gptbot" "src/matrix_rssbot"
] ]
classifiers = [ classifiers = [
@ -29,47 +29,19 @@ classifiers = [
] ]
dependencies = [ dependencies = [
"matrix-nio[e2e]", "matrix-nio",
"markdown2[all]", "pantalaimon",
"tiktoken", "setuptools",
"python-magic", "markdown2",
"pillow", "feedparser",
] ]
[project.optional-dependencies]
openai = [
"openai>=1.2",
"pydub",
]
wolframalpha = [
"wolframalpha",
]
e2ee = [
"pantalaimon>=0.10.5",
]
all = [
"matrix-gptbot[openai,wolframalpha,e2ee]",
"geopy",
"beautifulsoup4",
]
dev = [
"matrix-gptbot[all]",
"black",
"hatchling",
"twine",
"build",
]
[project.urls] [project.urls]
"Homepage" = "https://kumig.it/kumitterer/matrix-gptbot" "Homepage" = "https://git.private.coffee/PrivateCoffee/matrix-rssbot"
"Bug Tracker" = "https://kumig.it/kumitterer/matrix-gptbot/issues" "Bug Tracker" = "https://git.private.coffee/PrivateCoffee/matrix-rssbot/issues"
[project.scripts] [project.scripts]
gptbot = "gptbot:main" rssbot = "matrix_rssbot:main"
[tool.hatch.build.targets.wheel] [tool.hatch.build.targets.wheel]
packages = ["src/gptbot"] packages = ["src/matrix_rssbot"]

15
rssbot-pantalaimon.service Normal file
View file

@ -0,0 +1,15 @@
[Unit]
Description=Pantalaimon for RSSbot
Requires=network.target
[Service]
Type=simple
User=rssbot
Group=rssbot
WorkingDirectory=/opt/rssbot
ExecStart=/opt/rssbot/venv/bin/python3 -um pantalaimon.main -c pantalaimon.conf
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target

15
rssbot.service Normal file
View file

@ -0,0 +1,15 @@
[Unit]
Description=RSSbot
Requires=network.target
[Service]
Type=simple
User=rssbot
Group=rssbot
WorkingDirectory=/opt/rssbot
ExecStart=/opt/rssbot/.venv/bin/python3 -um rssbot
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target

BIN
src/gptbot/assets/logo.png
View file

Binary file not shown.
Side by side

Before

Width:  |  Height:  |  Size: 186 KiB

11
src/gptbot/callbacks/test.py
View file

@ -1,11 +0,0 @@
from nio import MatrixRoom, Event
async def test_callback(room: MatrixRoom, event: Event, bot):
"""Test callback for debugging purposes.
Args:
room (MatrixRoom): The room the event was sent in.
event (Event): The event that was sent.
"""
bot.logger.log(f"Test callback called: {room.room_id} {event.event_id} {event.sender} {event.__class__}")

11
src/gptbot/callbacks/test_response.py
View file

@ -1,11 +0,0 @@
from nio import ErrorResponse
async def test_response_callback(response, bot):
if isinstance(response, ErrorResponse):
bot.logger.log(
f"Error response received ({response.__class__.__name__}): {response.message}",
"warning",
)
else:
bot.logger.log(f"{response.__class__} response received", "debug")

1351
src/gptbot/classes/bot.py
View file

File diff suppressed because it is too large Load diff

8
src/gptbot/classes/dict.py
View file

@ -1,8 +0,0 @@
class AttrDict(dict):
def __getattr__(self, key):
if key in self:
return self[key]
raise AttributeError(f"'{type(self).__name__}' object has no attribute '{key}'")
def __setattr__(self, key, value):
self[key] = value

676
src/gptbot/classes/openai.py
View file

@ -1,676 +0,0 @@
import openai
import requests
import tiktoken
import asyncio
import json
import base64
import inspect
from functools import partial
from contextlib import closing
from typing import Dict, List, Tuple, Generator, AsyncGenerator, Optional, Any
from io import BytesIO
from pydub import AudioSegment
from .logging import Logger
from ..tools import TOOLS, Handover, StopProcessing
ASSISTANT_CODE_INTERPRETER = [
{
"type": "code_interpreter",
},
]
class AttributeDictionary(dict):
def __init__(self, *args, **kwargs):
super(AttributeDictionary, self).__init__(*args, **kwargs)
self.__dict__ = self
class OpenAI:
api_key: str
chat_model: str = "gpt-3.5-turbo"
logger: Logger
api_code: str = "openai"
@property
def chat_api(self) -> str:
return self.chat_model
classification_api = chat_api
image_model: str = "dall-e-2"
tts_model: str = "tts-1-hd"
tts_voice: str = "alloy"
stt_model: str = "whisper-1"
operator: str = "OpenAI ([https://openai.com](https://openai.com))"
def __init__(
self,
bot,
api_key,
chat_model=None,
image_model=None,
tts_model=None,
tts_voice=None,
stt_model=None,
base_url=None,
logger=None,
):
self.bot = bot
self.api_key = api_key
self.chat_model = chat_model or self.chat_model
self.image_model = image_model or self.image_model
self.logger = logger or bot.logger or Logger()
self.base_url = base_url or openai.base_url
self.openai_api = openai.AsyncOpenAI(
api_key=self.api_key, base_url=self.base_url
)
self.tts_model = tts_model or self.tts_model
self.tts_voice = tts_voice or self.tts_voice
self.stt_model = stt_model or self.stt_model
def supports_chat_images(self):
return "vision" in self.chat_model
def json_decode(self, data):
if data.startswith("```json\n"):
data = data[8:]
elif data.startswith("```\n"):
data = data[4:]
if data.endswith("```"):
data = data[:-3]
try:
return json.loads(data)
except:
return False
async def _request_with_retries(
self, request: partial, attempts: int = 5, retry_interval: int = 2
) -> AsyncGenerator[Any | list | Dict, None]:
"""Retry a request a set number of times if it fails.
Args:
request (partial): The request to make with retries.
attempts (int, optional): The number of attempts to make. Defaults to 5.
retry_interval (int, optional): The interval in seconds between attempts. Defaults to 2 seconds.
Returns:
AsyncGenerator[Any | list | Dict, None]: The OpenAI response for the request.
"""
# call the request function and return the response if it succeeds, else retry
current_attempt = 1
while current_attempt <= attempts:
try:
response = await request()
return response
except Exception as e:
self.logger.log(f"Request failed: {e}", "error")
self.logger.log(f"Retrying in {retry_interval} seconds...")
await asyncio.sleep(retry_interval)
current_attempt += 1
# if all attempts failed, raise an exception
raise Exception("Request failed after all attempts.")
async def generate_chat_response(
self,
messages: List[Dict[str, str]],
user: Optional[str] = None,
room: Optional[str] = None,
allow_override: bool = True,
use_tools: bool = True,
model: Optional[str] = None,
) -> Tuple[str, int]:
"""Generate a response to a chat message.
Args:
messages (List[Dict[str, str]]): A list of messages to use as context.
user (Optional[str], optional): The user to use the assistant for. Defaults to None.
room (Optional[str], optional): The room to use the assistant for. Defaults to None.
allow_override (bool, optional): Whether to allow the chat model to be overridden. Defaults to True.
use_tools (bool, optional): Whether to use tools. Defaults to True.
model (Optional[str], optional): The model to use. Defaults to None, which uses the default chat model.
Returns:
Tuple[str, int]: The response text and the number of tokens used.
"""
self.logger.log(
f"Generating response to {len(messages)} messages for user {user} in room {room}..."
)
original_model = chat_model = model or self.chat_model
# Check current recursion depth to prevent infinite loops
if use_tools:
frames = inspect.stack()
current_function = inspect.getframeinfo(frames[0][0]).function
count = sum(
1
for frame in frames
if inspect.getframeinfo(frame[0]).function == current_function
)
self.logger.log(
f"{current_function} appears {count} times in the call stack"
)
if count > 5:
self.logger.log(f"Recursion depth exceeded, aborting.")
return await self.generate_chat_response(
messsages=messages,
user=user,
room=room,
allow_override=False, # TODO: Could this be a problem?
use_tools=False,
model=original_model,
)
tools = [
{
"type": "function",
"function": {
"name": tool_name,
"description": tool_class.DESCRIPTION,
"parameters": tool_class.PARAMETERS,
},
}
for tool_name, tool_class in TOOLS.items()
]
original_messages = messages
if allow_override and not "gpt-3.5-turbo" in original_model:
if self.bot.config.getboolean("OpenAI", "ForceTools", fallback=False):
self.logger.log(f"Overriding chat model to use tools")
chat_model = "gpt-3.5-turbo-0125"
out_messages = []
for message in messages:
if isinstance(message, dict):
if isinstance(message["content"], str):
out_messages.append(message)
else:
message_content = []
for content in message["content"]:
if content["type"] == "text":
message_content.append(content)
if message_content:
message["content"] = message_content
out_messages.append(message)
else:
out_messages.append(message)
messages = out_messages
self.logger.log(f"Generating response with model {chat_model}...")
if (
use_tools
and self.bot.config.getboolean("OpenAI", "EmulateTools", fallback=False)
and not self.bot.config.getboolean("OpenAI", "ForceTools", fallback=False)
and not "gpt-3.5-turbo" in chat_model
):
self.bot.logger.log("Using tool emulation mode.", "debug")
messages = (
[
{
"role": "system",
"content": """You are a tool dispatcher for an AI chat model. You decide which tools to use for the current conversation. You DO NOT RESPOND DIRECTLY TO THE USER. Instead, respond with a JSON object like this:
{ "type": "tool", "tool": tool_name, "parameters": { "name": "value" } }
- tool_name is the name of the tool you want to use.
- parameters is an object containing the parameters for the tool. The parameters are defined in the tool's description.
The following tools are available:
"""
+ "\n".join(
[
f"- {tool_name}: {tool_class.DESCRIPTION} ({tool_class.PARAMETERS})"
for tool_name, tool_class in TOOLS.items()
]
)
+ """
If no tool is required, or all information is already available in the message thread, respond with an empty JSON object: {}
Otherwise, respond with a single required tool call. Remember that you DO NOT RESPOND to the user. You MAY ONLY RESPOND WITH JSON OBJECTS CONTAINING TOOL CALLS! DO NOT RESPOND IN NATURAL LANGUAGE.
DO NOT include any other text or syntax in your response, only the JSON object. DO NOT surround it in code tags (```). DO NOT, UNDER ANY CIRCUMSTANCES, ASK AGAIN FOR INFORMATION ALREADY PROVIDED IN THE MESSAGES YOU RECEIVED! DO NOT REQUEST MORE INFORMATION THAN ABSOLUTELY REQUIRED TO RESPOND TO THE USER'S MESSAGE! Remind the user that they may ask you to search for additional information if they need it.
""",
}
]
+ messages
)
self.logger.log(f"{messages[0]['content']}")
kwargs = {
"model": chat_model,
"messages": messages,
"user": room,
}
if "gpt-3.5-turbo" in chat_model and use_tools:
kwargs["tools"] = tools
if "gpt-4" in chat_model:
kwargs["max_tokens"] = self.bot.config.getint(
"OpenAI", "MaxTokens", fallback=4000
)
chat_partial = partial(self.openai_api.chat.completions.create, **kwargs)
response = await self._request_with_retries(chat_partial)
choice = response.choices[0]
result_text = choice.message.content
self.logger.log(f"Generated response: {result_text}")
additional_tokens = 0
if (not result_text) and choice.message.tool_calls:
tool_responses = []
for tool_call in choice.message.tool_calls:
try:
tool_response = await self.bot.call_tool(
tool_call, room=room, user=user
)
if tool_response != False:
tool_responses.append(
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(tool_response),
}
)
except StopProcessing as e:
return (e.args[0] if e.args else False), 0
except Handover:
return await self.generate_chat_response(
messages=original_messages,
user=user,
room=room,
allow_override=False,
use_tools=False,
model=original_model,
)
if not tool_responses:
self.logger.log(f"No more responses received, aborting.")
result_text = False
else:
try:
messages = (
original_messages[:-1]
+ [choice.message]
+ tool_responses
+ original_messages[-1:]
)
result_text, additional_tokens = await self.generate_chat_response(
messages=messages, user=user, room=room, model=original_model
)
except openai.APIError as e:
if e.code == "max_tokens":
self.logger.log(
f"Max tokens exceeded, falling back to no-tools response."
)
try:
new_messages = []
for message in original_messages:
new_message = message
if isinstance(message, dict):
if message["role"] == "tool":
new_message["role"] = "system"
del new_message["tool_call_id"]
else:
continue
new_messages.append(new_message)
(
result_text,
additional_tokens,
) = await self.generate_chat_response(
messages=new_messages,
user=user,
room=room,
allow_override=False,
use_tools=False,
model=original_model,
)
except openai.APIError as e:
if e.code == "max_tokens":
(
result_text,
additional_tokens,
) = await self.generate_chat_response(
messages=original_messages,
user=user,
room=room,
allow_override=False,
use_tools=False,
model=original_model,
)
else:
raise e
elif isinstance((tool_object := self.json_decode(result_text)), dict):
if "tool" in tool_object:
tool_name = tool_object["tool"]
tool_class = TOOLS[tool_name]
tool_parameters = (
tool_object["parameters"] if "parameters" in tool_object else {}
)
self.logger.log(
f"Using tool {tool_name} with parameters {tool_parameters}..."
)
tool_call = AttributeDictionary(
{
"function": AttributeDictionary(
{
"name": tool_name,
"arguments": json.dumps(tool_parameters),
}
),
}
)
try:
tool_response = await self.bot.call_tool(
tool_call, room=room, user=user
)
if tool_response != False:
tool_responses = [
{
"role": "system",
"content": str(tool_response),
}
]
except StopProcessing as e:
return (e.args[0] if e.args else False), 0
except Handover:
return await self.generate_chat_response(
messages=original_messages,
user=user,
room=room,
allow_override=False,
use_tools=False,
model=original_model,
)
if not tool_responses:
self.logger.log(f"No response received, aborting.")
result_text = False
else:
try:
messages = (
original_messages[:-1]
+ [choice.message]
+ tool_responses
+ original_messages[-1:]
)
(
result_text,
additional_tokens,
) = await self.generate_chat_response(
messages=messages,
user=user,
room=room,
model=original_model,
)
except openai.APIError as e:
if e.code == "max_tokens":
(
result_text,
additional_tokens,
) = await self.generate_chat_response(
messages=original_messages,
user=user,
room=room,
allow_override=False,
use_tools=False,
model=original_model,
)
else:
raise e
else:
result_text, additional_tokens = await self.generate_chat_response(
messages=original_messages,
user=user,
room=room,
allow_override=False,
use_tools=False,
model=original_model,
)
elif not original_model == chat_model:
new_messages = []
for message in original_messages:
new_message = message
if isinstance(message, dict):
if message["role"] == "tool":
new_message["role"] = "system"
del new_message["tool_call_id"]
else:
continue
new_messages.append(new_message)
result_text, additional_tokens = await self.generate_chat_response(
messages=new_messages,
user=user,
room=room,
allow_override=False,
model=original_model,
)
try:
tokens_used = response.usage.total_tokens
except:
tokens_used = 0
self.logger.log(f"Generated response with {tokens_used} tokens.")
return result_text, tokens_used + additional_tokens
async def classify_message(
self, query: str, user: Optional[str] = None
) -> Tuple[Dict[str, str], int]:
system_message = """You are a classifier for different types of messages. You decide whether an incoming message is meant to be a prompt for an AI chat model, or meant for a different API. You respond with a JSON object like this:
{ "type": event_type, "prompt": prompt }
- If the message you received is meant for the AI chat model, the event_type is "chat", and the prompt is the literal content of the message you received. This is also the default if none of the other options apply.
- If it is a prompt for a calculation that can be answered better by WolframAlpha than an AI chat bot, the event_type is "calculate". Optimize the message you received for input to WolframAlpha, and return it as the prompt attribute.
- If it is a prompt for an AI image generation, the event_type is "imagine". Optimize the message you received for use with DALL-E, and return it as the prompt attribute.
- If the user is asking you to create a new room, the event_type is "newroom", and the prompt is the name of the room, if one is given, else an empty string.
- If the user is asking you to throw a coin, the event_type is "coin". The prompt is an empty string.
- If the user is asking you to roll a dice, the event_type is "dice". The prompt is an string containing an optional number of sides, if one is given, else an empty string.
- If for any reason you are unable to classify the message (for example, if it infringes on your terms of service), the event_type is "error", and the prompt is a message explaining why you are unable to process the message.
Only the event_types mentioned above are allowed, you must not respond in any other way."""
messages = [
{"role": "system", "content": system_message},
{"role": "user", "content": query},
]
self.logger.log(f"Classifying message '{query}'...")
chat_partial = partial(
self.openai_api.chat.completions.create,
model=self.chat_model,
messages=messages,
user=str(user),
)
response = await self._request_with_retries(chat_partial)
try:
result = json.loads(response.choices[0].message["content"])
except:
result = {"type": "chat", "prompt": query}
tokens_used = response.usage["total_tokens"]
self.logger.log(
f"Classified message as {result['type']} with {tokens_used} tokens."
)
return result, tokens_used
async def text_to_speech(
self, text: str, user: Optional[str] = None
) -> Generator[bytes, None, None]:
"""Generate speech from text.
Args:
text (str): The text to use.
Yields:
bytes: The audio data.
"""
self.logger.log(
f"Generating speech from text of length: {len(text.split())} words..."
)
speech = await self.openai_api.audio.speech.create(
model=self.tts_model, input=text, voice=self.tts_voice
)
return speech.content
async def speech_to_text(
self, audio: bytes, user: Optional[str] = None
) -> Tuple[str, int]:
"""Generate text from speech.
Args:
audio (bytes): The audio data.
Returns:
Tuple[str, int]: The text and the number of tokens used.
"""
self.logger.log(f"Generating text from speech...")
audio_file = BytesIO()
AudioSegment.from_file(BytesIO(audio)).export(audio_file, format="mp3")
audio_file.name = "audio.mp3"
response = await self.openai_api.audio.transcriptions.create(
model=self.stt_model,
file=audio_file,
)
text = response.text
self.logger.log(f"Recognized text: {len(text.split())} words.")
return text
async def generate_image(
self, prompt: str, user: Optional[str] = None, orientation: str = "square"
) -> Generator[bytes, None, None]:
"""Generate an image from a prompt.
Args:
prompt (str): The prompt to use.
user (Optional[str], optional): The user to use the assistant for. Defaults to None.
orientation (str, optional): The orientation of the image. Defaults to "square".
Yields:
bytes: The image data.
"""
self.logger.log(f"Generating image from prompt '{prompt}'...")
split_prompt = prompt.split()
delete_first = False
size = "1024x1024"
if self.image_model == "dall-e-3":
if orientation == "portrait" or (
delete_first := split_prompt[0] == "--portrait"
):
size = "1024x1792"
elif orientation == "landscape" or (
delete_first := split_prompt[0] == "--landscape"
):
size = "1792x1024"
if delete_first:
prompt = " ".join(split_prompt[1:])
self.logger.log(
f"Generating image with size {size} using model {self.image_model}..."
)
args = {
"model": self.image_model,
"quality": "standard" if self.image_model != "dall-e-3" else "hd",
"prompt": prompt,
"n": 1,
"size": size,
}
if user:
args["user"] = user
image_partial = partial(self.openai_api.images.generate, **args)
response = await self._request_with_retries(image_partial)
images = []
for image in response.data:
image = requests.get(image.url).content
images.append(image)
return images, len(images)
async def describe_images(
self, messages: list, user: Optional[str] = None
) -> Tuple[str, int]:
"""Generate a description for an image.
Args:
image (bytes): The image data.
Returns:
Tuple[str, int]: The description and the number of tokens used.
"""
self.logger.log(f"Generating description for images in conversation...")
system_message = "You are an image description generator. You generate descriptions for all images in the current conversation, one after another."
messages = [{"role": "system", "content": system_message}] + messages[1:]
if not "vision" in (chat_model := self.chat_model):
chat_model = self.chat_model + "gpt-4-vision-preview"
chat_partial = partial(
self.openai_api.chat.completions.create,
model=self.chat_model,
messages=messages,
user=str(user),
)
response = await self._request_with_retries(chat_partial)
return response.choices[0].message.content, response.usage.total_tokens

32
src/gptbot/classes/trackingmore.py
View file

@ -1,32 +0,0 @@
import trackingmore
import requests
from .logging import Logger
from typing import Dict, List, Tuple, Generator, Optional
class TrackingMore:
api_key: str
logger: Logger
client: trackingmore.TrackingMore
api_code: str = "trackingmore"
parcel_api: str = "trackingmore"
operator: str = "TrackingMore ([https://www.trackingmore.com](https://www.trackingmore.com))"
def __init__(self, api_key: str, logger: Optional[Logger] = None):
self.api_key: str = api_key
self.logger: Logger = logger or Logger()
self.client = trackingmore.TrackingMore(self.api_key)
def lookup_parcel(self, query: str, carrier: Optional[str] = None, user: Optional[str] = None) -> Tuple[str, int]:
self.logger.log(f"Querying TrackingMore for {query}")
if query == "carriers":
response = "\n".join(f"* {carrier['courier_name']} - {carrier['courier_code']}" for carrier in self.client.get_carriers())
return response, 1
response = self.client.track_shipment(query)
return response, 1

51
src/gptbot/classes/wolframalpha.py
View file

@ -1,51 +0,0 @@
import wolframalpha
import requests
from .logging import Logger
from typing import Dict, List, Tuple, Generator, Optional
class WolframAlpha:
api_key: str
logger: Logger
client: wolframalpha.Client
api_code: str = "wolfram"
calculation_api: str = "alpha"
operator: str = "Wolfram ([https://www.wolfram.com](https://www.wolfram.com))"
def __init__(self, api_key: str, logger: Optional[Logger] = None):
self.api_key: str = api_key
self.logger: Logger = logger or Logger()
self.client = wolframalpha.Client(self.api_key)
def generate_calculation_response(self, query: str, text: Optional[bool] = False, results_only: Optional[bool] = False, user: Optional[str] = None) -> Generator[str | bytes, None, None]:
self.logger.log(f"Querying WolframAlpha for {query}")
response: wolframalpha.Result = self.client.query(query)
if not response.success:
yield "Could not process your query."
if response.didyoumeans:
yield "Did you mean: " + response.didyoumeans["didyoumean"][0]["#text"]
return
if response.error:
self.logger.log("Error in query to WolframAlpha: " + response.error, "error")
for pod in response.pods if not results_only else (response.results if len(list(response.results)) else response.pods):
self.logger.log(pod.keys(), "debug")
if pod.title:
yield "*" + pod.title + "*"
for subpod in pod.subpods:
self.logger.log(subpod.keys(), "debug")
if subpod.title:
yield "*" + subpod.title + "*"
if subpod.img and not text:
image = requests.get(subpod.img.src).content
yield image
elif subpod.plaintext:
yield "```\n" + subpod.plaintext + "\n```"

31
src/gptbot/commands/__init__.py
View file

@ -1,31 +0,0 @@
from importlib import import_module
from .unknown import command_unknown
COMMANDS = {}
for command in [
"help",
"newroom",
"stats",
"botinfo",
"coin",
"ignoreolder",
"systemmessage",
"imagine",
"calculate",
"classify",
"chat",
"custom",
"privacy",
"roomsettings",
"dice",
"parcel",
"space",
"tts",
]:
function = getattr(import_module(
"." + command, "gptbot.commands"), "command_" + command)
COMMANDS[command] = function
COMMANDS[None] = command_unknown

36
src/gptbot/commands/calculate.py
View file

@ -1,36 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_calculate(room: MatrixRoom, event: RoomMessageText, bot):
prompt = event.body.split()[2:]
text = False
results_only = True
if "--text" in prompt:
text = True
delete = prompt.index("--text")
del prompt[delete]
if "--details" in prompt:
results_only = False
delete = prompt.index("--details")
del prompt[delete]
prompt = " ".join(prompt)
if prompt:
bot.logger.log("Querying calculation API...")
for subpod in bot.calculation_api.generate_calculation_response(prompt, text, results_only, user=room.room_id):
bot.logger.log(f"Sending subpod...")
if isinstance(subpod, bytes):
await bot.send_image(room, subpod)
else:
await bot.send_message(room, subpod, True)
bot.log_api_usage(event, room, f"{bot.calculation_api.api_code}-{bot.calculation_api.calculation_api}", tokens_used)
return
await bot.send_message(room, "You need to provide a prompt.", True)

15
src/gptbot/commands/chat.py
View file

@ -1,15 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_chat(room: MatrixRoom, event: RoomMessageText, bot):
prompt = " ".join(event.body.split()[2:])
if prompt:
bot.logger.log("Sending chat message...")
event.body = prompt
await bot.process_query(room, event, from_chat_command=True)
return
await bot.send_message(room, "You need to provide a prompt.", True)

29
src/gptbot/commands/classify.py
View file

@ -1,29 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_classify(room: MatrixRoom, event: RoomMessageText, bot):
prompt = " ".join(event.body.split()[2:])
if prompt:
bot.logger.log("Classifying message...")
try:
response, tokens_used = await bot.classification_api.classify_message(prompt, user=room.room_id)
except Exception as e:
bot.logger.log(f"Error classifying message: {e}", "error")
await bot.send_message(room, "Sorry, I couldn't classify the message. Please try again later.", True)
return
message = f"The message you provided seems to be of type: {response['type']}."
if not prompt == response["prompt"]:
message += f"\n\nPrompt: {response['prompt']}."
await bot.send_message(room, message, True)
bot.log_api_usage(event, room, f"{bot.classification_api.api_code}-{bot.classification_api.classification_api}", tokens_used)
return
await bot.send_message(room, "You need to provide a prompt.", True)

13
src/gptbot/commands/coin.py
View file

@ -1,13 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
from random import SystemRandom
async def command_coin(room: MatrixRoom, event: RoomMessageText, bot):
bot.logger.log("Flipping a coin...")
heads = SystemRandom().choice([True, False])
body = "Flipping a coin... It's " + ("heads!" if heads else "tails!")
await bot.send_message(room, body, True)

9
src/gptbot/commands/custom.py
View file

@ -1,9 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_custom(room: MatrixRoom, event: RoomMessageText, bot):
bot.logger.log("Forwarding custom command to room...")
await bot.process_query(room, event)
return

22
src/gptbot/commands/dice.py
View file

@ -1,22 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
from random import SystemRandom
async def command_dice(room: MatrixRoom, event: RoomMessageText, bot):
bot.logger.log("Rolling a dice...")
try:
sides = int(event.body.split()[2])
except ValueError:
sides = 6
if sides < 2:
await bot.send_message(room, f"A dice with {sides} sides? How would that work?", True)
else:
result = SystemRandom().randint(1, sides)
body = f"Rolling a {sides}-sided dice... It's a {result}!"
await bot.send_message(room, body, True)

26
src/gptbot/commands/help.py
View file

@ -1,26 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_help(room: MatrixRoom, event: RoomMessageText, bot):
body = """Available commands:
- !gptbot help - Show this message
- !gptbot botinfo - Show information about the bot
- !gptbot privacy - Show privacy information
- !gptbot newroom \<room name\> - Create a new room and invite yourself to it
- !gptbot stats - Show usage statistics for this room
- !gptbot systemmessage \<message\> - Get or set the system message for this room
- !gptbot space [enable|disable|update|invite] - Enable, disable, force update, or invite yourself to your space
- !gptbot coin - Flip a coin (heads or tails)
- !gptbot dice [number] - Roll a dice with the specified number of sides (default: 6)
- !gptbot imagine \<prompt\> - Generate an image from a prompt
- !gptbot calculate [--text] [--details] \<query\> - Calculate a result to a calculation, optionally forcing text output instead of an image, and optionally showing additional details like the input interpretation
- !gptbot chat \<message\> - Send a message to the chat API
- !gptbot classify \<message\> - Classify a message using the classification API
- !gptbot custom \<message\> - Used for custom commands handled by the chat model and defined through the room's system message
- !gptbot roomsettings [use_classification|use_timing|always_reply|system_message|tts] [true|false|\<message\>] - Get or set room settings
- !gptbot ignoreolder - Ignore messages before this point as context
"""
await bot.send_message(room, body, True)

9
src/gptbot/commands/ignoreolder.py
View file

@ -1,9 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_ignoreolder(room: MatrixRoom, event: RoomMessageText, bot):
body = """Alright, messages before this point will not be processed as context anymore.
If you ever reconsider, you can simply delete your message and I will start processing messages before it again."""
await bot.send_message(room, body, True)

26
src/gptbot/commands/imagine.py
View file

@ -1,26 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_imagine(room: MatrixRoom, event: RoomMessageText, bot):
prompt = " ".join(event.body.split()[2:])
if prompt:
bot.logger.log("Generating image...")
try:
images, tokens_used = await bot.image_api.generate_image(prompt, user=room.room_id)
except Exception as e:
bot.logger.log(f"Error generating image: {e}", "error")
await bot.send_message(room, "Sorry, I couldn't generate an image. Please try again later.", True)
return
for image in images:
bot.logger.log(f"Sending image...")
await bot.send_image(room, image)
bot.log_api_usage(event, room, f"{bot.image_api.api_code}-{bot.image_api.image_model}", tokens_used)
return
await bot.send_message(room, "You need to provide a prompt.", True)

19
src/gptbot/commands/parcel.py
View file

@ -1,19 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_parcel(room: MatrixRoom, event: RoomMessageText, bot):
prompt = event.body.split()[2:]
if prompt:
bot.logger.log("Looking up parcels...")
for parcel in prompt:
status, tokens_used = bot.parcel_api.lookup_parcel(parcel, user=room.room_id)
await bot.send_message(room, status, True)
bot.log_api_usage(event, room, f"{bot.parcel_api.api_code}-{bot.parcel_api.parcel_api}", tokens_used)
return
await bot.send_message(room, "You need to provide tracking numbers.", True)

17
src/gptbot/commands/privacy.py
View file

@ -1,17 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_privacy(room: MatrixRoom, event: RoomMessageText, bot):
body = "**Privacy**\n\nIf you use this bot, note that your messages will be sent to the following recipients:\n\n"
body += "- The bot's operator" + (f"({bot.operator})" if bot.operator else "") + "\n"
if bot.chat_api:
body += "- For chat requests: " + f"{bot.chat_api.operator}" + "\n"
if bot.image_api:
body += "- For image generation requests (!gptbot imagine): " + f"{bot.image_api.operator}" + "\n"
if bot.calculate_api:
body += "- For calculation requests (!gptbot calculate): " + f"{bot.calculate_api.operator}" + "\n"
await bot.send_message(room, body, True)

130
src/gptbot/commands/roomsettings.py
View file

@ -1,130 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
from contextlib import closing
async def command_roomsettings(room: MatrixRoom, event: RoomMessageText, bot):
setting = event.body.split()[2] if len(event.body.split()) > 2 else None
value = " ".join(event.body.split()[3:]) if len(
event.body.split()) > 3 else None
if setting in ("classification", "timing"):
setting = f"use_{setting}"
if setting == "systemmessage":
setting = "system_message"
if setting == "system_message":
if value:
bot.logger.log("Adding system message...")
with closing(bot.database.cursor()) as cur:
cur.execute(
"""INSERT INTO room_settings (room_id, setting, value) VALUES (?, ?, ?)
ON CONFLICT (room_id, setting) DO UPDATE SET value = ?;""",
(room.room_id, "system_message", value, value)
)
bot.database.commit()
await bot.send_message(room, f"Alright, I've stored the system message: '{value}'.", True)
return
bot.logger.log("Retrieving system message...")
system_message = bot.get_system_message(room)
await bot.send_message(room, f"The current system message is: '{system_message}'.", True)
return
if setting in ("use_classification", "always_reply", "use_timing", "tts", "stt"):
if value:
if value.lower() in ["true", "false"]:
value = value.lower() == "true"
bot.logger.log(f"Setting {setting} status for {room.room_id} to {value}...")
with closing(bot.database.cursor()) as cur:
cur.execute(
"""INSERT INTO room_settings (room_id, setting, value) VALUES (?, ?, ?)
ON CONFLICT (room_id, setting) DO UPDATE SET value = ?;""",
(room.room_id, setting, "1" if value else "0", "1" if value else "0")
)
bot.database.commit()
await bot.send_message(room, f"Alright, I've set {setting} to: '{value}'.", True)
return
await bot.send_message(room, "You need to provide a boolean value (true/false).", True)
return
bot.logger.log(f"Retrieving {setting} status for {room.room_id}...")
with closing(bot.database.cursor()) as cur:
cur.execute(
"""SELECT value FROM room_settings WHERE room_id = ? AND setting = ?;""",
(room.room_id, setting)
)
value = cur.fetchone()[0]
if not value:
if setting in ("use_classification", "use_timing"):
value = False
elif setting == "always_reply":
value = True
else:
value = bool(int(value))
await bot.send_message(room, f"The current {setting} status is: '{value}'.", True)
return
if bot.allow_model_override and setting == "model":
if value:
bot.logger.log(f"Setting chat model for {room.room_id} to {value}...")
with closing(bot.database.cursor()) as cur:
cur.execute(
"""INSERT INTO room_settings (room_id, setting, value) VALUES (?, ?, ?)
ON CONFLICT (room_id, setting) DO UPDATE SET value = ?;""",
(room.room_id, "model", value, value)
)
bot.database.commit()
await bot.send_message(room, f"Alright, I've set the chat model to: '{value}'.", True)
return
bot.logger.log(f"Retrieving chat model for {room.room_id}...")
with closing(bot.database.cursor()) as cur:
cur.execute(
"""SELECT value FROM room_settings WHERE room_id = ? AND setting = ?;""",
(room.room_id, "model")
)
value = cur.fetchone()[0]
if not value:
value = bot.chat_api.chat_model
else:
value = str(value)
await bot.send_message(room, f"The current chat model is: '{value}'.", True)
return
message = f"""The following settings are available:
- system_message [message]: Get or set the system message to be sent to the chat model
- classification [true/false]: Get or set whether the room uses classification
- always_reply [true/false]: Get or set whether the bot should reply to all messages (if false, only reply to mentions and commands)
- tts [true/false]: Get or set whether the bot should generate audio files instead of sending text
- stt [true/false]: Get or set whether the bot should attempt to process information from audio files
- timing [true/false]: Get or set whether the bot should return information about the time it took to generate a response
"""
if bot.allow_model_override:
message += "- model [model]: Get or set the chat model to be used for this room"
await bot.send_message(room, message, True)

148
src/gptbot/commands/space.py
View file

@ -1,148 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
from nio.responses import RoomInviteError
from contextlib import closing
async def command_space(room: MatrixRoom, event: RoomMessageText, bot):
if len(event.body.split()) == 3:
request = event.body.split()[2]
if request.lower() == "enable":
bot.logger.log("Enabling space...")
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"SELECT space_id FROM user_spaces WHERE user_id = ? AND active = TRUE", (event.sender,))
space = cursor.fetchone()
if not space:
space = await bot.create_space("GPTBot")
bot.logger.log(
f"Created space {space} for user {event.sender}")
if bot.logo_uri:
await bot.matrix_client.room_put_state(space, "m.room.avatar", {
"url": bot.logo_uri
}, "")
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"INSERT INTO user_spaces (space_id, user_id) VALUES (?, ?)", (space, event.sender))
else:
space = space[0]
response = await bot.matrix_client.room_invite(space, event.sender)
if isinstance(response, RoomInviteError):
bot.logger.log(
f"Failed to invite user {event.sender} to space {space}", "error")
await bot.send_message(
room, "Sorry, I couldn't invite you to the space. Please try again later.", True)
return
bot.database.commit()
await bot.send_message(room, "Space enabled.", True)
request = "update"
elif request.lower() == "disable":
bot.logger.log("Disabling space...")
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"SELECT space_id FROM user_spaces WHERE user_id = ? AND active = TRUE", (event.sender,))
space = cursor.fetchone()[0]
if not space:
bot.logger.log(f"User {event.sender} does not have a space")
await bot.send_message(room, "You don't have a space enabled.", True)
return
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"UPDATE user_spaces SET active = FALSE WHERE user_id = ?", (event.sender,))
bot.database.commit()
await bot.send_message(room, "Space disabled.", True)
return
if request.lower() == "update":
bot.logger.log("Updating space...")
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"SELECT space_id FROM user_spaces WHERE user_id = ? AND active = TRUE", (event.sender,))
space = cursor.fetchone()[0]
if not space:
bot.logger.log(f"User {event.sender} does not have a space")
await bot.send_message(
room, "You don't have a space enabled. Create one first using `!gptbot space enable`.", True)
return
rooms = bot.matrix_client.rooms
join_rooms = []
for room in rooms.values():
if event.sender in room.users.keys():
bot.logger.log(
f"Adding room {room.room_id} to space {space}")
join_rooms.append(room.room_id)
await bot.add_rooms_to_space(space, join_rooms)
if bot.logo_uri:
await bot.matrix_client.room_put_state(space, "m.room.avatar", {
"url": bot.logo_uri
}, "")
await bot.send_message(room, "Space updated.", True)
return
if request.lower() == "invite":
bot.logger.log("Inviting user to space...")
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"SELECT space_id FROM user_spaces WHERE user_id = ?", (event.sender,))
space = cursor.fetchone()[0]
if not space:
bot.logger.log(f"User {event.sender} does not have a space")
await bot.send_message(
room, "You don't have a space enabled. Create one first using `!gptbot space enable`.", True)
return
response = await bot.matrix_client.room_invite(space, event.sender)
if isinstance(response, RoomInviteError):
bot.logger.log(
f"Failed to invite user {user} to space {space}", "error")
await bot.send_message(
room, "Sorry, I couldn't invite you to the space. Please try again later.", True)
return
await bot.send_message(room, "Invited you to the space.", True)
return
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"SELECT active FROM user_spaces WHERE user_id = ?", (event.sender,))
status = cursor.fetchone()
if not status:
await bot.send_message(
room, "You don't have a space enabled. Create one using `!gptbot space enable`.", True)
return
if not status[0]:
await bot.send_message(
room, "Your space is disabled. Enable it using `!gptbot space enable`.", True)
return
await bot.send_message(
room, "Your space is enabled. Rooms will be added to it automatically.", True)
return

20
src/gptbot/commands/stats.py
View file

@ -1,20 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
from contextlib import closing
async def command_stats(room: MatrixRoom, event: RoomMessageText, bot):
bot.logger.log("Showing stats...")
if not bot.database:
bot.logger.log("No database connection - cannot show stats")
bot.send_message(room, "Sorry, I'm not connected to a database, so I don't have any statistics on your usage.", True)
return
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"SELECT SUM(tokens) FROM token_usage WHERE room_id = ?", (room.room_id,))
total_tokens = cursor.fetchone()[0] or 0
bot.send_message(room, f"Total tokens used: {total_tokens}", True)

29
src/gptbot/commands/systemmessage.py
View file

@ -1,29 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
from contextlib import closing
async def command_systemmessage(room: MatrixRoom, event: RoomMessageText, bot):
system_message = " ".join(event.body.split()[2:])
if system_message:
bot.logger.log("Adding system message...")
with closing(bot.database.cursor()) as cur:
cur.execute(
"""
INSERT INTO room_settings (room_id, setting, value) VALUES (?, ?, ?)
ON CONFLICT (room_id, setting) DO UPDATE SET value = ?;
""",
(room.room_id, "system_message", system_message, system_message)
)
await bot.send_message(room, f"Alright, I've stored the system message: '{system_message}'.", True)
return
bot.logger.log("Retrieving system message...")
system_message = bot.get_system_message(room)
await bot.send_message(room, f"The current system message is: '{system_message}'.", True)

23
src/gptbot/commands/tts.py
View file

@ -1,23 +0,0 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_tts(room: MatrixRoom, event: RoomMessageText, bot):
prompt = " ".join(event.body.split()[2:])
if prompt:
bot.logger.log("Generating speech...")
try:
content = await bot.tts_api.text_to_speech(prompt, user=room.room_id)
except Exception as e:
bot.logger.log(f"Error generating speech: {e}", "error")
await bot.send_message(room, "Sorry, I couldn't generate an audio file. Please try again later.", True)
return
bot.logger.log(f"Sending audio file...")
await bot.send_file(room, content, "audio.mp3", "audio/mpeg", "m.audio")
return
await bot.send_message(room, "You need to provide a prompt.", True)

50
src/gptbot/migrations/__init__.py
View file

@ -1,50 +0,0 @@
from collections import OrderedDict
from typing import Optional
from importlib import import_module
from sqlite3 import Connection as SQLiteConnection
MAX_MIGRATION = 8
MIGRATIONS = OrderedDict()
for i in range(1, MAX_MIGRATION + 1):
MIGRATIONS[i] = import_module(f".migration_{i}", __package__).migration
def get_version(db: SQLiteConnection) -> int:
"""Get the current database version.
Args:
db (SQLiteConnection): Database connection.
Returns:
int: Current database version.
"""
try:
return int(db.execute("SELECT MAX(id) FROM migrations").fetchone()[0])
except:
return 0
def migrate(db: SQLiteConnection, from_version: Optional[int] = None, to_version: Optional[int] = None) -> None:
"""Migrate the database to a specific version.
Args:
db (SQLiteConnection): Database connection.
from_version (Optional[int]): Version to migrate from. If None, the current version is used.
to_version (Optional[int]): Version to migrate to. If None, the latest version is used.
"""
if from_version is None:
from_version = get_version(db)
if to_version is None:
to_version = max(MIGRATIONS.keys())
if from_version > to_version:
raise ValueError("Cannot migrate from a higher version to a lower version.")
for version in range(from_version, to_version):
if version + 1 in MIGRATIONS:
MIGRATIONS[version + 1](db)
return from_version, to_version

33
src/gptbot/migrations/migration_1.py
View file

@ -1,33 +0,0 @@
# Initial migration, token usage logging
from datetime import datetime
from contextlib import closing
def migration(conn):
with closing(conn.cursor()) as cursor:
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS token_usage (
message_id TEXT PRIMARY KEY,
room_id TEXT NOT NULL,
tokens INTEGER NOT NULL,
timestamp TIMESTAMP NOT NULL
)
"""
)
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS migrations (
id INTEGER NOT NULL,
timestamp TIMESTAMP NOT NULL
)
"""
)
cursor.execute(
"INSERT INTO migrations (id, timestamp) VALUES (1, ?)",
(datetime.now(),)
)
conn.commit()

7
src/gptbot/migrations/migration_2.py
View file

@ -1,7 +0,0 @@
# Migration for Matrix Store - No longer used
from datetime import datetime
from contextlib import closing
def migration(conn):
pass

25
src/gptbot/migrations/migration_3.py
View file

@ -1,25 +0,0 @@
# Migration for custom system messages
from datetime import datetime
from contextlib import closing
def migration(conn):
with closing(conn.cursor()) as cursor:
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS system_messages (
room_id TEXT NOT NULL,
message_id TEXT NOT NULL,
user_id TEXT NOT NULL,
body TEXT NOT NULL,
timestamp BIGINT NOT NULL
)
"""
)
cursor.execute(
"INSERT INTO migrations (id, timestamp) VALUES (3, ?)",
(datetime.now(),)
)
conn.commit()

19
src/gptbot/migrations/migration_4.py
View file

@ -1,19 +0,0 @@
# Migration to add API column to token usage table
from datetime import datetime
from contextlib import closing
def migration(conn):
with closing(conn.cursor()) as cursor:
cursor.execute(
"""
ALTER TABLE token_usage ADD COLUMN api TEXT DEFAULT 'openai'
"""
)
cursor.execute(
"INSERT INTO migrations (id, timestamp) VALUES (4, ?)",
(datetime.now(),)
)
conn.commit()

49
src/gptbot/migrations/migration_5.py
View file

@ -1,49 +0,0 @@
# Migration to add room settings table
from datetime import datetime
from contextlib import closing
def migration(conn):
with closing(conn.cursor()) as cursor:
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS room_settings (
room_id TEXT NOT NULL,
setting TEXT NOT NULL,
value TEXT NOT NULL,
PRIMARY KEY (room_id, setting)
)
"""
)
cursor.execute("SELECT * FROM system_messages")
system_messages = cursor.fetchall()
# Get latest system message for each room
cursor.execute(
"""
SELECT system_messages.room_id, system_messages.message_id, system_messages.user_id, system_messages.body, system_messages.timestamp
FROM system_messages
INNER JOIN (
SELECT room_id, MAX(timestamp) AS timestamp FROM system_messages GROUP BY room_id
) AS latest_system_message ON system_messages.room_id = latest_system_message.room_id AND system_messages.timestamp = latest_system_message.timestamp
"""
)
system_messages = cursor.fetchall()
for message in system_messages:
cursor.execute(
"INSERT INTO room_settings (room_id, setting, value) VALUES (?, ?, ?)",
(message[0], "system_message", message[1])
)
cursor.execute("DROP TABLE system_messages")
cursor.execute(
"INSERT INTO migrations (id, timestamp) VALUES (5, ?)",
(datetime.now(),)
)
conn.commit()

33
src/gptbot/migrations/migration_6.py
View file

@ -1,33 +0,0 @@
# Migration to drop primary key constraint from token_usage table
from datetime import datetime
from contextlib import closing
def migration(conn):
with closing(conn.cursor()) as cursor:
cursor.execute(
"""
CREATE TABLE token_usage_temp (
message_id TEXT NOT NULL,
room_id TEXT NOT NULL,
api TEXT NOT NULL,
tokens INTEGER NOT NULL,
timestamp TIMESTAMP NOT NULL
)
"""
)
cursor.execute(
"INSERT INTO token_usage_temp SELECT message_id, room_id, api, tokens, timestamp FROM token_usage"
)
cursor.execute("DROP TABLE token_usage")
cursor.execute("ALTER TABLE token_usage_temp RENAME TO token_usage")
cursor.execute(
"INSERT INTO migrations (id, timestamp) VALUES (6, ?)",
(datetime.now(),)
)
conn.commit()

24
src/gptbot/migrations/migration_7.py
View file

@ -1,24 +0,0 @@
# Migration to add user_spaces table
from datetime import datetime
from contextlib import closing
def migration(conn):
with closing(conn.cursor()) as cursor:
cursor.execute(
"""
CREATE TABLE user_spaces (
space_id TEXT NOT NULL,
user_id TEXT NOT NULL,
active BOOLEAN NOT NULL DEFAULT TRUE,
PRIMARY KEY (space_id, user_id)
)
"""
)
cursor.execute(
"INSERT INTO migrations (id, timestamp) VALUES (7, ?)",
(datetime.now(),)
)
conn.commit()

23
src/gptbot/migrations/migration_8.py
View file

@ -1,23 +0,0 @@
# Migration to add settings table
from datetime import datetime
from contextlib import closing
def migration(conn):
with closing(conn.cursor()) as cursor:
cursor.execute(
"""
CREATE TABLE IF NOT EXISTS settings (
setting TEXT NOT NULL,
value TEXT NOT NULL,
PRIMARY KEY (setting)
)
"""
)
cursor.execute(
"INSERT INTO migrations (id, timestamp) VALUES (8, ?)",
(datetime.now(),)
)
conn.commit()

21
src/gptbot/tools/__init__.py
View file

@ -1,21 +0,0 @@
from importlib import import_module
from .base import BaseTool, StopProcessing, Handover
TOOLS = {}
for tool in [
"weather",
"geocode",
"dice",
"websearch",
"webrequest",
"imagine",
"imagedescription",
"wikipedia",
"datetime",
"newroom",
]:
tool_class = getattr(import_module(
"." + tool, "gptbot.tools"), tool.capitalize())
TOOLS[tool] = tool_class

21
src/gptbot/tools/base.py
View file

@ -1,21 +0,0 @@
class BaseTool:
DESCRIPTION: str
PARAMETERS: dict
def __init__(self, **kwargs):
self.kwargs = kwargs
self.bot = kwargs.get("bot")
self.room = kwargs.get("room")
self.user = kwargs.get("user")
self.messages = kwargs.get("messages", [])
async def run(self):
raise NotImplementedError()
class StopProcessing(Exception):
"""Stop processing the message."""
pass
class Handover(Exception):
"""Handover to the original model, if applicable. Stop using tools."""
pass

16
src/gptbot/tools/datetime.py
View file

@ -1,16 +0,0 @@
from .base import BaseTool
from datetime import datetime
class Datetime(BaseTool):
DESCRIPTION = "Get the current date and time."
PARAMETERS = {
"type": "object",
"properties": {
},
}
async def run(self):
"""Get the current date and time."""
return f"""**Current date and time (UTC)**
{datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S")}"""

26
src/gptbot/tools/dice.py
View file

@ -1,26 +0,0 @@
from .base import BaseTool
from random import SystemRandom
class Dice(BaseTool):
DESCRIPTION = "Roll dice."
PARAMETERS = {
"type": "object",
"properties": {
"dice": {
"type": "string",
"description": "The number of sides on the dice.",
"default": "6",
},
},
"required": [],
}
async def run(self):
"""Roll dice."""
dice = int(self.kwargs.get("dice", 6))
return f"""**Dice roll**
Used dice: {dice}
Result: {SystemRandom().randint(1, dice)}
"""

34
src/gptbot/tools/geocode.py
View file

@ -1,34 +0,0 @@
from geopy.geocoders import Nominatim
from .base import BaseTool
class Geocode(BaseTool):
DESCRIPTION = "Get location information (latitude, longitude) for a given location name."
PARAMETERS = {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The location name.",
},
},
"required": ["location"],
}
async def run(self):
"""Get location information for a given location."""
if not (location := self.kwargs.get("location")):
raise Exception('No location provided.')
geolocator = Nominatim(user_agent=self.bot.USER_AGENT)
location = geolocator.geocode(location)
if location:
return f"""**Location information for {location.address}**
Latitude: {location.latitude}
Longitude: {location.longitude}
"""
raise Exception('Could not find location data for that location.')

15
src/gptbot/tools/imagedescription.py
View file

@ -1,15 +0,0 @@
from .base import BaseTool, Handover
class Imagedescription(BaseTool):
DESCRIPTION = "Describe the content of the images in the conversation."
PARAMETERS = {
"type": "object",
"properties": {
},
}
async def run(self):
"""Describe images in the conversation."""
image_api = self.bot.image_api
return (await image_api.describe_images(self.messages, self.user))[0]

34
src/gptbot/tools/imagine.py
View file

@ -1,34 +0,0 @@
from .base import BaseTool, StopProcessing
class Imagine(BaseTool):
DESCRIPTION = "Use generative AI to create images from text prompts."
PARAMETERS = {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": "The prompt to use.",
},
"orientation": {
"type": "string",
"description": "The orientation of the image.",
"enum": ["square", "landscape", "portrait"],
"default": "square",
},
},
"required": ["prompt"],
}
async def run(self):
"""Use generative AI to create images from text prompts."""
if not (prompt := self.kwargs.get("prompt")):
raise Exception('No prompt provided.')
api = self.bot.image_api
orientation = self.kwargs.get("orientation", "square")
images, tokens = await api.generate_image(prompt, self.room, orientation=orientation)
for image in images:
await self.bot.send_image(self.room, image, prompt)
raise StopProcessing()

57
src/gptbot/tools/newroom.py
View file

@ -1,57 +0,0 @@
from .base import BaseTool, StopProcessing
from nio import RoomCreateError, RoomInviteError
from contextlib import closing
class Newroom(BaseTool):
DESCRIPTION = "Create a new Matrix room"
PARAMETERS = {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "The name of the room to create.",
"default": "GPTBot"
}
},
}
async def run(self):
"""Create a new Matrix room"""
name = self.kwargs.get("name", "GPTBot")
self.bot.logger.log("Creating new room...")
new_room = await self.bot.matrix_client.room_create(name=name)
if isinstance(new_room, RoomCreateError):
self.bot.logger.log(f"Failed to create room: {new_room.message}")
raise
self.bot.logger.log(f"Inviting {self.user} to new room...")
invite = await self.bot.matrix_client.room_invite(new_room.room_id, self.user)
if isinstance(invite, RoomInviteError):
self.bot.logger.log(f"Failed to invite user: {invite.message}")
raise
await self.bot.send_message(new_room.room_id, "Welcome to your new room! What can I do for you?")
with closing(self.bot.database.cursor()) as cursor:
cursor.execute(
"SELECT space_id FROM user_spaces WHERE user_id = ? AND active = TRUE", (self.user,))
space = cursor.fetchone()
if space:
self.bot.logger.log(f"Adding new room to space {space[0]}...")
await self.bot.add_rooms_to_space(space[0], [new_room.room_id])
if self.bot.logo_uri:
await self.bot.matrix_client.room_put_state(room, "m.room.avatar", {
"url": self.bot.logo_uri
}, "")
await self.bot.matrix_client.room_put_state(
new_room.room_id, "m.room.power_levels", {"users": {self.user: 100, self.bot.matrix_client.user_id: 100}})
raise StopProcessing("Created new Matrix room with ID " + new_room.room_id + " and invited user.")

52
src/gptbot/tools/weather.py
View file

@ -1,52 +0,0 @@
import aiohttp
from datetime import datetime
from .base import BaseTool
class Weather(BaseTool):
DESCRIPTION = "Get weather information for a given location."
PARAMETERS = {
"type": "object",
"properties": {
"latitude": {
"type": "string",
"description": "The latitude of the location.",
},
"longitude": {
"type": "string",
"description": "The longitude of the location.",
},
},
"required": ["latitude", "longitude"],
}
async def run(self):
"""Get weather information for a given location."""
if not (latitude := self.kwargs.get("latitude")) or not (longitude := self.kwargs.get("longitude")):
raise Exception('No location provided.')
weather_api_key = self.bot.config.get("OpenWeatherMap", "APIKey")
if not weather_api_key:
raise Exception('Weather API key not found.')
url = f'https://api.openweathermap.org/data/3.0/onecall?lat={latitude}&lon={longitude}&appid={weather_api_key}&units=metric'
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
return f"""**Weather report**
Current: {data['current']['temp']}°C, {data['current']['weather'][0]['description']}
Feels like: {data['current']['feels_like']}°C
Humidity: {data['current']['humidity']}%
Wind: {data['current']['wind_speed']}m/s
Sunrise: {datetime.fromtimestamp(data['current']['sunrise']).strftime('%H:%M')}
Sunset: {datetime.fromtimestamp(data['current']['sunset']).strftime('%H:%M')}
Today: {data['daily'][0]['temp']['day']}°C, {data['daily'][0]['weather'][0]['description']}, {data['daily'][0]['summary']}
Tomorrow: {data['daily'][1]['temp']['day']}°C, {data['daily'][1]['weather'][0]['description']}, {data['daily'][1]['summary']}
"""
else:
raise Exception(f'Could not connect to weather API: {response.status} {response.reason}')

59
src/gptbot/tools/webrequest.py
View file

@ -1,59 +0,0 @@
from .base import BaseTool
import aiohttp
from bs4 import BeautifulSoup
import re
class Webrequest(BaseTool):
DESCRIPTION = "Browse an external website by URL."
PARAMETERS = {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "The URL to request.",
},
},
"required": ["url"],
}
async def html_to_text(self, html):
# Parse the HTML content of the response
soup = BeautifulSoup(html, 'html.parser')
# Format the links within the text
for link in soup.find_all('a'):
link_text = link.get_text()
link_href = link.get('href')
new_link_text = f"{link_text} ({link_href})"
link.replace_with(new_link_text)
# Extract the plain text content of the website
plain_text_content = soup.get_text()
# Remove extra whitespace
plain_text_content = re.sub('\s+', ' ', plain_text_content).strip()
# Return the formatted text content of the website
return plain_text_content
async def run(self):
"""Make a web request to a given URL."""
if not (url := self.kwargs.get("url")):
raise Exception('No URL provided.')
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.text()
output = await self.html_to_text(data)
return f"""**Web request**
URL: {url}
Status: {response.status} {response.reason}
{output}
"""

37
src/gptbot/tools/websearch.py
View file

@ -1,37 +0,0 @@
from .base import BaseTool
import aiohttp
from urllib.parse import quote_plus
class Websearch(BaseTool):
DESCRIPTION = "Search the web for a given query."
PARAMETERS = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The query to search for.",
},
},
"required": ["query"],
}
async def run(self):
"""Search the web for a given query."""
if not (query := self.kwargs.get("query")):
raise Exception('No query provided.')
query = quote_plus(query)
url = f'https://librey.private.coffee/api.php?q={query}'
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
response_text = "**Search results for {query}**"
for result in data:
response_text += f"\n{result['title']}\n{result['url']}\n{result['description']}\n"
return response_text

79
src/gptbot/tools/wikipedia.py
View file

@ -1,79 +0,0 @@
from .base import BaseTool
from urllib.parse import urlencode
import aiohttp
class Wikipedia(BaseTool):
DESCRIPTION = "Get information from Wikipedia."
PARAMETERS = {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The query to search for.",
},
"language": {
"type": "string",
"description": "The language to search in.",
"default": "en",
},
"extract": {
"type": "string",
"description": "What information to extract from the page. If not provided, the full page will be returned."
},
"summarize": {
"type": "boolean",
"description": "Whether to summarize the page or not.",
"default": False,
}
},
"required": ["query"],
}
async def run(self):
"""Get information from Wikipedia."""
if not (query := self.kwargs.get("query")):
raise Exception('No query provided.')
language = self.kwargs.get("language", "en")
extract = self.kwargs.get("extract")
summarize = self.kwargs.get("summarize", False)
args = {
"action": "query",
"format": "json",
"titles": query,
}
args["prop"] = "revisions"
args["rvprop"] = "content"
url = f'https://{language}.wikipedia.org/w/api.php?{urlencode(args)}'
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
data = await response.json()
try:
pages = data['query']['pages']
page = list(pages.values())[0]
content = page['revisions'][0]['*']
except KeyError:
raise Exception(f'No results for {query} found in Wikipedia.')
if extract:
chat_messages = [{"role": "system", "content": f"Extract the following from the provided content: {extract}"}]
chat_messages.append({"role": "user", "content": content})
content, _ = await self.bot.chat_api.generate_chat_response(chat_messages, room=self.room, user=self.user, allow_override=False, use_tools=False)
if summarize:
chat_messages = [{"role": "system", "content": "Summarize the following content:"}]
chat_messages.append({"role": "user", "content": content})
content, _ = await self.bot.chat_api.generate_chat_response(chat_messages, room=self.room, user=self.user, allow_override=False, use_tools=False)
return f"**Wikipedia: {page['title']}**\n{content}"
else:
raise Exception(f'Could not connect to Wikipedia API: {response.status} {response.reason}')

0
src/gptbot/__init__.py → src/matrix_rssbot/__init__.py
View file

6
src/gptbot/__main__.py → src/matrix_rssbot/__main__.py
View file

@ -1,4 +1,4 @@
from .classes.bot import GPTBot from .classes.bot import RSSBot
from argparse import ArgumentParser from argparse import ArgumentParser
from configparser import ConfigParser from configparser import ConfigParser
@ -12,7 +12,7 @@ def sigterm_handler(_signo, _stack_frame):
def get_version(): def get_version():
try: try:
package_version = pkg_resources.get_distribution("matrix_gptbot").version package_version = pkg_resources.get_distribution("matrix_rssbot").version
except pkg_resources.DistributionNotFound: except pkg_resources.DistributionNotFound:
return None return None
return package_version return package_version
@ -41,7 +41,7 @@ if __name__ == "__main__":
config.read(args.config) config.read(args.config)
# Create bot # Create bot
bot = GPTBot.from_config(config) bot = RSSBot.from_config(config)
# Listen for SIGTERM # Listen for SIGTERM
signal.signal(signal.SIGTERM, sigterm_handler) signal.signal(signal.SIGTERM, sigterm_handler)

0
src/gptbot/classes/__init__.py → src/matrix_rssbot/classes/__init__.py
View file

580
src/matrix_rssbot/classes/bot.py Normal file
View file

@ -0,0 +1,580 @@
import asyncio
import functools
from nio import (
AsyncClient,
AsyncClientConfig,
WhoamiResponse,
DevicesResponse,
Event,
Response,
MatrixRoom,
Api,
RoomMessagesError,
GroupEncryptionError,
EncryptionError,
RoomMessageText,
RoomSendResponse,
SyncResponse,
RoomMessageNotice,
JoinError,
RoomLeaveError,
RoomSendError,
RoomVisibility,
RoomCreateError,
RoomMessageMedia,
RoomMessageImage,
RoomMessageFile,
RoomMessageAudio,
DownloadError,
DownloadResponse,
ToDeviceEvent,
ToDeviceError,
RoomPutStateError,
RoomGetStateError,
)
from typing import Optional, List
from configparser import ConfigParser
from datetime import datetime
from io import BytesIO
from pathlib import Path
from contextlib import closing
import base64
import uuid
import traceback
import json
import importlib.util
import sys
import traceback
import markdown2
from .logging import Logger
from .callbacks import RESPONSE_CALLBACKS, EVENT_CALLBACKS
from .commands import COMMANDS
class RSSBot:
# Default values
matrix_client: Optional[AsyncClient] = None
sync_token: Optional[str] = None
sync_response: Optional[SyncResponse] = None
logger: Optional[Logger] = Logger()
room_ignore_list: List[str] = [] # List of rooms to ignore invites from
config: ConfigParser = ConfigParser()
# Properties
@property
def sync_token(self) -> Optional[str]:
if self.sync_response:
return self.sync_response.next_batch
@property
def allowed_users(self) -> List[str]:
"""List of users allowed to use the bot.
Returns:
List[str]: List of user IDs. Defaults to [], which means all users are allowed.
"""
try:
return json.loads(self.config["RSSBot"]["AllowedUsers"])
except:
return []
@property
def display_name(self) -> str:
"""Display name of the bot user.
Returns:
str: The display name of the bot user. Defaults to "RSSBot".
"""
return self.config["RSSBot"].get("DisplayName", "RSSBot")
@property
def default_room_name(self) -> str:
"""Default name of rooms created by the bot.
Returns:
str: The default name of rooms created by the bot. Defaults to the display name of the bot.
"""
return self.config["RSSBot"].get("DefaultRoomName", self.display_name)
@property
def debug(self) -> bool:
"""Whether to enable debug logging.
Returns:
bool: Whether to enable debug logging. Defaults to False.
"""
return self.config["RSSBot"].getboolean("Debug", False)
# User agent to use for HTTP requests
USER_AGENT = (
"matrix-rssbot/dev (+https://git.private.coffee/PrivateCoffee/matrix-rssbot)"
)
@classmethod
def from_config(cls, config: ConfigParser):
"""Create a new RSSBot instance from a config file.
Args:
config (ConfigParser): ConfigParser instance with the bot's config.
Returns:
RSSBot: The new RSSBot instance.
"""
# Create a new RSSBot instance
bot = cls()
bot.config = config
# Override default values
if "RSSBot" in config:
if "LogLevel" in config["RSSBot"]:
bot.logger = Logger(config["RSSBot"]["LogLevel"])
# Set up the Matrix client
assert "Matrix" in config, "Matrix config not found"
homeserver = config["Matrix"]["Homeserver"]
bot.matrix_client = AsyncClient(homeserver)
bot.matrix_client.access_token = config["Matrix"]["AccessToken"]
bot.matrix_client.user_id = config["Matrix"].get("UserID")
bot.matrix_client.device_id = config["Matrix"].get("DeviceID")
# Return the new RSSBot instance
return bot
async def _get_user_id(self) -> str:
"""Get the user ID of the bot from the whoami endpoint.
Requires an access token to be set up.
Returns:
str: The user ID of the bot.
"""
assert self.matrix_client, "Matrix client not set up"
user_id = self.matrix_client.user_id
if not user_id:
assert self.matrix_client.access_token, "Access token not set up"
response = await self.matrix_client.whoami()
if isinstance(response, WhoamiResponse):
user_id = response.user_id
else:
raise Exception(f"Could not get user ID: {response}")
return user_id
async def _get_device_id(self) -> str:
"""Guess the device ID of the bot.
Requires an access token to be set up.
Returns:
str: The guessed device ID.
"""
assert self.matrix_client, "Matrix client not set up"
device_id = self.matrix_client.device_id
if not device_id:
assert self.matrix_client.access_token, "Access token not set up"
devices = await self.matrix_client.devices()
if isinstance(devices, DevicesResponse):
device_id = devices.devices[0].id
return device_id
async def process_command(self, room: MatrixRoom, event: RoomMessageText):
"""Process a command. Called from the event_callback() method.
Delegates to the appropriate command handler.
Args:
room (MatrixRoom): The room the command was sent in.
event (RoomMessageText): The event containing the command.
"""
self.logger.log(
f"Received command {event.body} from {event.sender} in room {room.room_id}",
"debug",
)
if event.body.startswith("* "):
event.body = event.body[2:]
command = event.body.split()[1] if event.body.split()[1:] else None
await COMMANDS.get(command, COMMANDS[None])(room, event, self)
async def _event_callback(self, room: MatrixRoom, event: Event):
self.logger.log("Received event: " + str(event.event_id), "debug")
try:
for eventtype, callback in EVENT_CALLBACKS.items():
if isinstance(event, eventtype):
await callback(room, event, self)
except Exception as e:
self.logger.log(
f"Error in event callback for {event.__class__}: {e}", "error"
)
if self.debug:
await self.send_message(
room, f"Error: {e}\n\n```\n{traceback.format_exc()}\n```", True
)
def user_is_allowed(self, user_id: str) -> bool:
"""Check if a user is allowed to use the bot.
Args:
user_id (str): The user ID to check.
Returns:
bool: Whether the user is allowed to use the bot.
"""
return (
(
user_id in self.allowed_users
or f"*:{user_id.split(':')[1]}" in self.allowed_users
or f"@*:{user_id.split(':')[1]}" in self.allowed_users
)
if self.allowed_users
else True
)
async def event_callback(self, room: MatrixRoom, event: Event):
"""Callback for events.
Args:
room (MatrixRoom): The room the event was sent in.
event (Event): The event.
"""
if event.sender == self.matrix_client.user_id:
return
if not self.user_is_allowed(event.sender):
if len(room.users) == 2:
await self.matrix_client.room_send(
room.room_id,
"m.room.message",
{
"msgtype": "m.notice",
"body": f"You are not allowed to use this bot. Please contact {self.operator} for more information.",
},
)
return
task = asyncio.create_task(self._event_callback(room, event))
async def _response_callback(self, response: Response):
for response_type, callback in RESPONSE_CALLBACKS.items():
if isinstance(response, response_type):
await callback(response, self)
async def response_callback(self, response: Response):
task = asyncio.create_task(self._response_callback(response))
async def accept_pending_invites(self):
"""Accept all pending invites."""
assert self.matrix_client, "Matrix client not set up"
invites = self.matrix_client.invited_rooms
for invite in [k for k in invites.keys()]:
if invite in self.room_ignore_list:
self.logger.log(
f"Ignoring invite to room {invite} (room is in ignore list)",
"debug",
)
continue
self.logger.log(f"Accepting invite to room {invite}")
response = await self.matrix_client.join(invite)
if isinstance(response, JoinError):
self.logger.log(
f"Error joining room {invite}: {response.message}. Not trying again.",
"error",
)
leave_response = await self.matrix_client.room_leave(invite)
if isinstance(leave_response, RoomLeaveError):
self.logger.log(
f"Error leaving room {invite}: {leave_response.message}",
"error",
)
self.room_ignore_list.append(invite)
else:
self.send_message(invite, "Thank you for inviting me to your room!")
async def upload_file(
self,
file: bytes,
filename: str = "file",
mime: str = "application/octet-stream",
) -> str:
"""Upload a file to the homeserver.
Args:
file (bytes): The file to upload.
filename (str, optional): The name of the file. Defaults to "file".
mime (str, optional): The MIME type of the file. Defaults to "application/octet-stream".
Returns:
str: The MXC URI of the uploaded file.
"""
bio = BytesIO(file)
bio.seek(0)
response, _ = await self.matrix_client.upload(
bio, content_type=mime, filename=filename, filesize=len(file)
)
return response.content_uri
async def send_image(
self, room: MatrixRoom, image: bytes, message: Optional[str] = None
):
"""Send an image to a room.
Args:
room (MatrixRoom|str): The room to send the image to.
image (bytes): The image to send.
message (str, optional): The message to send with the image. Defaults to None.
"""
if isinstance(room, MatrixRoom):
room = room.room_id
self.logger.log(
f"Sending image of size {len(image)} bytes to room {room}", "debug"
)
bio = BytesIO(image)
img = Image.open(bio)
mime = Image.MIME[img.format]
(width, height) = img.size
self.logger.log(
f"Uploading - Image size: {width}x{height} pixels, MIME type: {mime}",
"debug",
)
content_uri = await self.upload_file(image, "image", mime)
self.logger.log("Uploaded image - sending message...", "debug")
content = {
"body": message or "",
"info": {
"mimetype": mime,
"size": len(image),
"w": width,
"h": height,
},
"msgtype": "m.image",
"url": content_uri,
}
status = await self.matrix_client.room_send(room, "m.room.message", content)
self.logger.log("Sent image", "debug")
async def send_file(
self, room: MatrixRoom, file: bytes, filename: str, mime: str, msgtype: str
):
"""Send a file to a room.
Args:
room (MatrixRoom|str): The room to send the file to.
file (bytes): The file to send.
filename (str): The name of the file.
mime (str): The MIME type of the file.
"""
if isinstance(room, MatrixRoom):
room = room.room_id
self.logger.log(
f"Sending file of size {len(file)} bytes to room {room}", "debug"
)
content_uri = await self.upload_file(file, filename, mime)
self.logger.log("Uploaded file - sending message...", "debug")
content = {
"body": filename,
"info": {"mimetype": mime, "size": len(file)},
"msgtype": msgtype,
"url": content_uri,
}
status = await self.matrix_client.room_send(room, "m.room.message", content)
self.logger.log("Sent file", "debug")
async def send_message(
self,
room: MatrixRoom | str,
message: str,
notice: bool = False,
msgtype: Optional[str] = None,
):
"""Send a message to a room.
Args:
room (MatrixRoom): The room to send the message to.
message (str): The message to send.
notice (bool): Whether to send the message as a notice. Defaults to False.
"""
if isinstance(room, str):
room = self.matrix_client.rooms[room]
markdowner = markdown2.Markdown(extras=["fenced-code-blocks"])
formatted_body = markdowner.convert(message)
msgtype = msgtype if msgtype else "m.notice" if notice else "m.text"
if not msgtype.startswith("rssbot."):
msgcontent = {
"msgtype": msgtype,
"body": message,
"format": "org.matrix.custom.html",
"formatted_body": formatted_body,
}
else:
msgcontent = {
"msgtype": msgtype,
"content": message,
}
content = None
if not content:
msgtype = "m.room.message"
content = msgcontent
method, path, data = Api.room_send(
self.matrix_client.access_token,
room.room_id,
msgtype,
content,
uuid.uuid4(),
)
response = await self.matrix_client._send(
RoomSendResponse, method, path, data, (room.room_id,)
)
if isinstance(response, RoomSendError):
self.logger.log(f"Error sending message: {response.message}", "error")
async def send_state_event(
self,
room: MatrixRoom | str,
event_type: str,
content: dict,
state_key: str = "",
):
if isinstance(room, MatrixRoom):
room = room.room_id
response = await self.matrix_client.room_put_state(
room, event_type, content, state_key
)
if isinstance(response, RoomPutStateError):
self.logger.log(f"Error putting state in {room}")
async def get_state_event(
self, room: MatrixRoom | str, event_type: str, state_key: Optional[str] = None
):
if isinstance(room, MatrixRoom):
room = room.room_id
state = await self.matrix_client.room_get_state(room)
if isinstance(state, RoomGetStateError):
self.logger.log(f"Could not get state for room {room}")
for event in state.events:
if event["type"] == event_type:
if state_key is None or event["state_key"] == state_key:
return event
async def run(self):
"""Start the bot."""
# Set up the Matrix client
assert self.matrix_client, "Matrix client not set up"
assert self.matrix_client.access_token, "Access token not set up"
if not self.matrix_client.user_id:
self.matrix_client.user_id = await self._get_user_id()
if not self.matrix_client.device_id:
self.matrix_client.device_id = await self._get_device_id()
client_config = AsyncClientConfig(
store_sync_tokens=False, encryption_enabled=False, store=None
)
self.matrix_client.config = client_config
# Run initial sync (includes joining rooms)
self.logger.log("Running initial sync...", "debug")
sync = await self.matrix_client.sync(timeout=30000, full_state=True)
if isinstance(sync, SyncResponse):
await self.response_callback(sync)
else:
self.logger.log(f"Initial sync failed, aborting: {sync}", "critical")
exit(1)
# Set up callbacks
self.logger.log("Setting up callbacks...", "debug")
self.matrix_client.add_event_callback(self.event_callback, Event)
self.matrix_client.add_response_callback(self.response_callback, Response)
# Set custom name
if self.display_name:
self.logger.log(f"Setting display name to {self.display_name}", "debug")
asyncio.create_task(self.matrix_client.set_displayname(self.display_name))
# Start syncing events
self.logger.log("Starting sync loop...", "warning")
try:
await self.matrix_client.sync_forever(timeout=30000, full_state=True)
finally:
self.logger.log("Syncing one last time...", "warning")
await self.matrix_client.sync(timeout=30000, full_state=True)
def __del__(self):
"""Close the bot."""
if self.matrix_client:
asyncio.run(self.matrix_client.close())

4
src/gptbot/callbacks/__init__.py → src/matrix_rssbot/classes/callbacks/__init__.py
View file

@ -10,22 +10,18 @@ from nio import (
KeysQueryResponse KeysQueryResponse
) )
from .test import test_callback
from .sync import sync_callback from .sync import sync_callback
from .invite import room_invite_callback from .invite import room_invite_callback
from .join import join_callback from .join import join_callback
from .message import message_callback from .message import message_callback
from .roommember import roommember_callback from .roommember import roommember_callback
from .test_response import test_response_callback
RESPONSE_CALLBACKS = { RESPONSE_CALLBACKS = {
#Response: test_response_callback,
SyncResponse: sync_callback, SyncResponse: sync_callback,
JoinResponse: join_callback, JoinResponse: join_callback,
} }
EVENT_CALLBACKS = { EVENT_CALLBACKS = {
#Event: test_callback,
InviteEvent: room_invite_callback, InviteEvent: room_invite_callback,
RoomMessageText: message_callback, RoomMessageText: message_callback,
RoomMemberEvent: roommember_callback, RoomMemberEvent: roommember_callback,

0
src/gptbot/callbacks/invite.py → src/matrix_rssbot/classes/callbacks/invite.py
View file

0
src/gptbot/callbacks/join.py → src/matrix_rssbot/classes/callbacks/join.py
View file

13
src/gptbot/callbacks/message.py → src/matrix_rssbot/classes/callbacks/message.py
View file

@ -10,21 +10,18 @@ async def message_callback(room: MatrixRoom | str, event: RoomMessageText, bot):
latency = received - sent latency = received - sent
if event.sender == bot.matrix_client.user_id: if event.sender == bot.matrix_client.user_id:
bot.logger.log("Message is from bot itself - ignoring") bot.logger.log("Message is from bot itself - ignoring", "debug")
elif event.body.startswith("!gptbot") or event.body.startswith("* !gptbot"): elif event.body.startswith("!rssbot") or event.body.startswith("* !rssbot"):
await bot.process_command(room, event) await bot.process_command(room, event)
elif event.body.startswith("!"): elif event.body.startswith("!"):
bot.logger.log(f"Received {event.body} - might be a command, but not for this bot - ignoring") bot.logger.log(f"Received {event.body} - might be a command, but not for this bot - ignoring", "debug")
else: else:
await bot.process_query(room, event) bot.logger.log(f"Received regular message - ignoring", "debug")
processed = datetime.now() processed = datetime.now()
processing_time = processed - received processing_time = processed - received
bot.logger.log(f"Message processing took {processing_time.total_seconds()} seconds (latency: {latency.total_seconds()} seconds)") bot.logger.log(f"Message processing took {processing_time.total_seconds()} seconds (latency: {latency.total_seconds()} seconds)", "debug")
if bot.room_uses_timing(room):
await bot.send_message(room, f"Message processing took {processing_time.total_seconds()} seconds (latency: {latency.total_seconds()} seconds)", True)

4
src/gptbot/callbacks/roommember.py → src/matrix_rssbot/classes/callbacks/roommember.py
View file

@ -2,9 +2,9 @@ from nio import RoomMemberEvent, MatrixRoom, KeysUploadError
async def roommember_callback(room: MatrixRoom, event: RoomMemberEvent, bot): async def roommember_callback(room: MatrixRoom, event: RoomMemberEvent, bot):
if event.membership == "leave": if event.membership == "leave":
bot.logger.log(f"User {event.state_key} left room {room.room_id} - am I alone now?") bot.logger.log(f"User {event.state_key} left room {room.room_id} - am I alone now?", "debug")
if len(room.users) == 1: if len(room.users) == 1:
bot.logger.log("Yes, I was abandoned - leaving...") bot.logger.log("Yes, I was abandoned - leaving...", "debug")
await bot.matrix_client.room_leave(room.room_id) await bot.matrix_client.room_leave(room.room_id)
return return

3
src/gptbot/callbacks/sync.py → src/matrix_rssbot/classes/callbacks/sync.py
View file

@ -1,8 +1,7 @@
async def sync_callback(response, bot): async def sync_callback(response, bot):
bot.logger.log( bot.logger.log(
f"Sync response received (next batch: {response.next_batch})", "debug") f"Sync response received (next batch: {response.next_batch})", "debug")
SYNC_TOKEN = response.next_batch
bot.sync_token = SYNC_TOKEN bot.sync_response = response
await bot.accept_pending_invites() await bot.accept_pending_invites()

20
src/matrix_rssbot/classes/commands/__init__.py Normal file
View file

@ -0,0 +1,20 @@
from importlib import import_module
from .unknown import command_unknown
COMMANDS = {}
for command in [
"help",
"newroom",
"botinfo",
"privacy",
"addfeed",
"listfeeds",
"processfeeds",
]:
function = getattr(import_module(
"." + command, "matrix_rssbot.classes.commands"), "command_" + command)
COMMANDS[command] = function
COMMANDS[None] = command_unknown

47
src/matrix_rssbot/classes/commands/addfeed.py Normal file
View file

@ -0,0 +1,47 @@
from nio.events.room_events import RoomMessageText
from nio import RoomPutStateError
from nio.rooms import MatrixRoom
from datetime import datetime
from urllib.request import urlopen
import feedparser
async def command_addfeed(room: MatrixRoom, event: RoomMessageText, bot):
url = event.body.split()[2]
bot.logger.log(f"Adding new feed to room {room.room_id}")
state = await bot.get_state_event(room, "rssbot.feeds")
if not state:
feeds = []
else:
feeds = state["content"]["feeds"]
feeds.append(url)
try:
feed = feedparser.parse(url)
for entry in feed.entries:
print(entry)
except:
await bot.send_state_event(
f"Could not access or parse feed at {url}. Please ensure that you got the URL right, and that it is actually an RSS/Atom feed.",
True,
)
try:
await bot.send_state_event(
room,
"rssbot.feed_state",
{"timestamp": int(datetime.now().timestamp())},
url,
)
await bot.send_state_event(room, "rssbot.feeds", {"feeds": feeds})
await bot.send_message(room, f"Added {url} to this room's feeds.", True)
except:
await bot.send_message(
room, "Sorry, something went wrong. Please try again.", true
)

10
src/gptbot/commands/botinfo.py → src/matrix_rssbot/classes/commands/botinfo.py
View file

@ -5,19 +5,11 @@ from nio.rooms import MatrixRoom
async def command_botinfo(room: MatrixRoom, event: RoomMessageText, bot): async def command_botinfo(room: MatrixRoom, event: RoomMessageText, bot):
logging("Showing bot info...") logging("Showing bot info...")
body = f"""GPT Info: body = f"""
Model: {bot.model}
Maximum context tokens: {bot.max_tokens}
Maximum context messages: {bot.max_messages}
Room info: Room info:
Bot user ID: {bot.matrix_client.user_id} Bot user ID: {bot.matrix_client.user_id}
Current room ID: {room.room_id} Current room ID: {room.room_id}
System message: {bot.get_system_message(room)}
For usage statistics, run !gptbot stats
""" """
await bot.send_message(room, body, True) await bot.send_message(room, body, True)

17
src/matrix_rssbot/classes/commands/help.py Normal file
View file

@ -0,0 +1,17 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_help(room: MatrixRoom, event: RoomMessageText, bot):
body = """Available commands:
- !rssbot help - Show this message
- !rssbot botinfo - Show information about the bot
- !rssbot privacy - Show privacy information
- !rssbot addfeed \<url\> - Bridges a new feed to the current room
- !rssbot listfeeds - Lists all bridged feeds
- !rssbot removefeed \<index|url\> - Removes a bridged feed given the numeric index from the listfeeds command or the URL of the feed
- !rssbot newroom \<room name\> - Create a new room and invite yourself to it
"""
await bot.send_message(room, body, True)

16
src/matrix_rssbot/classes/commands/listfeeds.py Normal file
View file

@ -0,0 +1,16 @@
from nio.events.room_events import RoomMessageText
from nio import RoomPutStateError
from nio.rooms import MatrixRoom
async def command_listfeeds(room: MatrixRoom, event: RoomMessageText, bot):
state = await bot.get_state_event(room, "rssbot.feeds")
if (not state) or (not state["content"]["feeds"]):
message = f"There are currently no feeds associated with this room."
else:
message = "This room is currently bridged to the following feeds:\n\n"
for key, value in enumerate(state["content"]["feeds"]):
message += f"- {key}: {value}"
await bot.send_message(room, message, True)

15
src/gptbot/commands/newroom.py → src/matrix_rssbot/classes/commands/newroom.py
View file

@ -24,23 +24,8 @@ async def command_newroom(room: MatrixRoom, event: RoomMessageText, bot):
await bot.send_message(room, f"Sorry, I was unable to invite you to the new room. Please try again later, or create a room manually.", True) await bot.send_message(room, f"Sorry, I was unable to invite you to the new room. Please try again later, or create a room manually.", True)
return return
with closing(bot.database.cursor()) as cursor:
cursor.execute(
"SELECT space_id FROM user_spaces WHERE user_id = ? AND active = TRUE", (event.sender,))
space = cursor.fetchone()
if space:
bot.logger.log(f"Adding new room to space {space[0]}...")
await bot.add_rooms_to_space(space[0], [new_room.room_id])
if bot.logo_uri:
await bot.matrix_client.room_put_state(room, "m.room.avatar", {
"url": bot.logo_uri
}, "")
await bot.matrix_client.room_put_state( await bot.matrix_client.room_put_state(
new_room.room_id, "m.room.power_levels", {"users": {event.sender: 100, bot.matrix_client.user_id: 100}}) new_room.room_id, "m.room.power_levels", {"users": {event.sender: 100, bot.matrix_client.user_id: 100}})
await bot.matrix_client.joined_rooms() await bot.matrix_client.joined_rooms()
await bot.send_message(room, f"Alright, I've created a new room called '{room_name}' and invited you to it. You can find it at {new_room.room_id}", True) await bot.send_message(room, f"Alright, I've created a new room called '{room_name}' and invited you to it. You can find it at {new_room.room_id}", True)
await bot.send_message(bot.matrix_client.rooms[new_room.room_id], f"Welcome to the new room! What can I do for you?")

11
src/matrix_rssbot/classes/commands/privacy.py Normal file
View file

@ -0,0 +1,11 @@
from nio.events.room_events import RoomMessageText
from nio.rooms import MatrixRoom
async def command_privacy(room: MatrixRoom, event: RoomMessageText, bot):
body = "**Privacy**\n\nIf you use this bot, note that your messages will be sent to the following recipients:\n\n"
body += "- The bot's operator" + (f"({bot.operator})" if bot.operator else "") + "\n"
body += "- The operator(s) of the involved Matrix homeservers\n"
await bot.send_message(room, body, True)

45
src/matrix_rssbot/classes/commands/processfeeds.py Normal file
View file

@ -0,0 +1,45 @@
from nio.events.room_events import RoomMessageText
from nio import RoomPutStateError
from nio.rooms import MatrixRoom
from datetime import datetime
import feedparser
async def command_processfeeds(room: MatrixRoom, event: RoomMessageText, bot):
bot.logger.log(f"Processing feeds for room {room.room_id}")
state = await bot.get_state_event(room, "rssbot.feeds")
if not state:
feeds = []
else:
feeds = state["content"]["feeds"]
for feed in feeds:
feed_state = await bot.get_state_event(room, "rssbot.feed_state", feed)
if feed_state:
bot.logger.log(f"Identified timestamp as {feed_state['content']['timestamp']}")
timestamp = int(feed_state["content"]["timestamp"])
else:
timestamp = 0
try:
feed_content = feedparser.parse(feed)
new_timestamp = timestamp
for entry in feed_content.entries:
entry_timestamp = int(datetime(*entry.published_parsed[:6]).timestamp())
if entry_timestamp > timestamp:
entry_message = f"__{feed_content.feed.title}: {entry.title}__\n\n{entry.description}\n\n{entry.link}"
await bot.send_message(room, entry_message)
new_timestamp = max(entry_timestamp, new_timestamp)
await bot.send_state_event(room, "rssbot.feed_state", {"timestamp": new_timestamp}, feed)
except:
raise
await bot.send_message(
room,
f"Could not access or parse RSS feed at {feed}. Please ensure that you got the URL right, and that it is actually an RSS feed.",
True
)

28
src/matrix_rssbot/classes/commands/removefeed.py Normal file
View file

@ -0,0 +1,28 @@
from nio.events.room_events import RoomMessageText
from nio import RoomPutStateError
from nio.rooms import MatrixRoom
async def command_removefeed(room: MatrixRoom, event: RoomMessageText, bot):
identifier = event.body.split()[2]
state = await bot.get_state_event(room, "rssbot.feeds")
if (not state) or (not state["content"]["feeds"]):
feeds = []
else:
feeds = state["content"]["feeds"]
if identifier.isnumeric():
try:
feed = feeds.pop(int(identifier))
except IndexError:
await bot.send_message(room, f"There is no feed with index {identifier}.")
return
else:
try:
feeds.remove(identifier)
except ValueError:
await bot.send_message(room, f"There is no bridged feed with the provided URL.")
return
await bot.send_state_event(room, "rssbot.feeds", {"feeds": feeds})

2
src/gptbot/commands/unknown.py → src/matrix_rssbot/classes/commands/unknown.py
View file

@ -5,4 +5,4 @@ from nio.rooms import MatrixRoom
async def command_unknown(room: MatrixRoom, event: RoomMessageText, bot): async def command_unknown(room: MatrixRoom, event: RoomMessageText, bot):
bot.logger.log("Unknown command") bot.logger.log("Unknown command")
await bot.send_message(room, "Unknown command - try !gptbot help", True) await bot.send_message(room, "Unknown command - try !rssbot help", True)

0
src/gptbot/classes/logging.py → src/matrix_rssbot/classes/logging.py
View file