Compare commits
38
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
92f536d407 | ||
|
|
a322fbae84 | ||
|
|
bd6a55cd85 | ||
|
|
4fd5e06eaa | ||
|
|
1f7aa0a5ee | ||
|
|
3b6fc28d4c | ||
|
|
c024d85db5 | ||
|
|
a5bfccb19a | ||
|
|
7dc1d485b1 | ||
|
|
8255d54e56 | ||
|
|
37beca0dd6 | ||
|
|
087566811a | ||
|
|
095bd828cf | ||
|
|
0387dafef8 | ||
|
|
0e2becc212 | ||
|
|
b69f169892 | ||
|
|
40b3b900ac | ||
|
|
73bf08843f | ||
|
|
d2edf0e60d | ||
|
|
bd4b05d2d4 | ||
|
|
eb7733be69 | ||
|
|
131999816b | ||
|
|
ff70196756 | ||
|
|
180c167a43 | ||
|
|
75af912bf5 | ||
|
|
a64a07511d | ||
|
|
9209cc568d | ||
|
|
7056315d5f | ||
|
|
b77dc31ea5 | ||
|
|
ad59a695af | ||
|
|
89b98fc5af | ||
|
|
97252a8062 | ||
|
|
0534ec6b15 | ||
|
|
11f39bb673 | ||
|
|
247e4417f5 | ||
|
|
bfe22e40ee | ||
|
|
59902085b3 | ||
|
|
3d59e45e7e |
@@ -2,7 +2,6 @@
|
|||||||
cogs made for our Royal butt.
|
cogs made for our Royal butt.
|
||||||
|
|
||||||
## Our Cogs:
|
## Our Cogs:
|
||||||
- [FitnessCog](./fitnessCog)
|
|
||||||
- [RecruitmentCog](./recruitmentCog/)
|
- [RecruitmentCog](./recruitmentCog/)
|
||||||
- [ReginaldCog](./reginaldCog/)
|
- [ReginaldCog](./reginaldCog/)
|
||||||
- [TrafficCog](./trafficCog)
|
- [TrafficCog](./trafficCog)
|
||||||
|
|||||||
@@ -1,6 +0,0 @@
|
|||||||
from redbot.core.bot import Red
|
|
||||||
from .fitness import FitnessCog
|
|
||||||
|
|
||||||
async def setup(bot: Red):
|
|
||||||
cog = FitnessCog(bot)
|
|
||||||
await bot.add_cog(cog)
|
|
||||||
@@ -1,48 +0,0 @@
|
|||||||
{
|
|
||||||
"convertable": true,
|
|
||||||
"units": {
|
|
||||||
"meter": {
|
|
||||||
"factor": 1,
|
|
||||||
"aliases": [
|
|
||||||
"meter",
|
|
||||||
"meters",
|
|
||||||
"m",
|
|
||||||
"m."
|
|
||||||
]
|
|
||||||
},
|
|
||||||
"kilometer": {
|
|
||||||
"factor": 1000,
|
|
||||||
"aliases": [
|
|
||||||
"kilometer",
|
|
||||||
"kilometers",
|
|
||||||
"km",
|
|
||||||
"km."
|
|
||||||
]
|
|
||||||
},
|
|
||||||
"foot": {
|
|
||||||
"factor": 0.3048,
|
|
||||||
"aliases": [
|
|
||||||
"foot",
|
|
||||||
"feet",
|
|
||||||
"ft",
|
|
||||||
"ft."
|
|
||||||
]
|
|
||||||
},
|
|
||||||
"mile": {
|
|
||||||
"factor": 1609.344,
|
|
||||||
"aliases": [
|
|
||||||
"mile",
|
|
||||||
"miles",
|
|
||||||
"mi",
|
|
||||||
"mi."
|
|
||||||
]
|
|
||||||
},
|
|
||||||
"step": {
|
|
||||||
"factor": 0.18,
|
|
||||||
"aliases": [
|
|
||||||
"step",
|
|
||||||
"steps"
|
|
||||||
]
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,3 +0,0 @@
|
|||||||
{
|
|
||||||
"convertable": false
|
|
||||||
}
|
|
||||||
@@ -1,119 +0,0 @@
|
|||||||
import json
|
|
||||||
import csv
|
|
||||||
import os
|
|
||||||
|
|
||||||
|
|
||||||
def activity_log_path(activity: str) -> str:
|
|
||||||
"""
|
|
||||||
Returns the path of the activity log file in .csv format
|
|
||||||
:param activity:
|
|
||||||
:return:
|
|
||||||
"""
|
|
||||||
filename = os.path.join("activity_logs", f"{activity}.csv")
|
|
||||||
return filename
|
|
||||||
|
|
||||||
|
|
||||||
def activity_units_path(activity: str) -> str:
|
|
||||||
"""
|
|
||||||
Returns the path of the activity activities file in .json format
|
|
||||||
:param activity:
|
|
||||||
:return:
|
|
||||||
"""
|
|
||||||
filename = os.path.join("activities", f"{activity}.json")
|
|
||||||
return filename
|
|
||||||
|
|
||||||
|
|
||||||
def is_convertable(activity: str) -> bool:
|
|
||||||
"""
|
|
||||||
Returns True if the activity has multiple units that needed to be converted.
|
|
||||||
i.e. distance may be sent in feet, but would require to be converted in meters.
|
|
||||||
:param activity:
|
|
||||||
:return:
|
|
||||||
"""
|
|
||||||
filename = activity_units_path(activity)
|
|
||||||
raw_data = json.load(open(filename))
|
|
||||||
return raw_data.get("transformable")
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def create_activity_log_file(activity: str) -> None:
|
|
||||||
"""
|
|
||||||
Creates the activity log file in .csv format
|
|
||||||
:param activity:
|
|
||||||
:return:
|
|
||||||
"""
|
|
||||||
if not os.path.exists(activity_units_path(activity)):
|
|
||||||
raise ValueError(f"{activity} is not a valid activity")
|
|
||||||
if not os.path.exists("activity_logs"):
|
|
||||||
os.makedirs("activity_logs")
|
|
||||||
filename = activity_log_path(activity)
|
|
||||||
with open(filename, 'a') as f:
|
|
||||||
writer = csv.writer(f)
|
|
||||||
writer.writerow(["timestamp", "username", "value"])
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def convert_units(activity: str, value: int | float, unit: str) -> float:
|
|
||||||
"""
|
|
||||||
Returns the value of activity in a factor of 1, i.e. converting feet to meters
|
|
||||||
:param activity:
|
|
||||||
:param value:
|
|
||||||
:param unit:
|
|
||||||
:return:
|
|
||||||
"""
|
|
||||||
filename = activity_units_path(activity)
|
|
||||||
if not os.path.exists(filename):
|
|
||||||
raise ValueError(f"{activity} is not a valid activity")
|
|
||||||
|
|
||||||
# Consider None unit value as factor of 1
|
|
||||||
if unit is None:
|
|
||||||
return value
|
|
||||||
|
|
||||||
raw_data = json.load(open(filename))
|
|
||||||
units_data = raw_data.get("units")
|
|
||||||
|
|
||||||
for unit_name, data in units_data.items():
|
|
||||||
if unit in data["aliases"]:
|
|
||||||
return value * data["factor"]
|
|
||||||
|
|
||||||
raise ValueError(f"{unit} is not a valid unit")
|
|
||||||
|
|
||||||
|
|
||||||
def log_activity(
|
|
||||||
timestamp: int,
|
|
||||||
username: str,
|
|
||||||
activity: str,
|
|
||||||
value: int | float,
|
|
||||||
unit: str | None
|
|
||||||
) -> None:
|
|
||||||
"""
|
|
||||||
Logs the activity in .csv file
|
|
||||||
:param timestamp:
|
|
||||||
:param username:
|
|
||||||
:param activity:
|
|
||||||
:param value:
|
|
||||||
:param unit:
|
|
||||||
:return:
|
|
||||||
"""
|
|
||||||
if value <= 0 or not isinstance(value, int | float):
|
|
||||||
raise ValueError(f"{value} is not a valid value")
|
|
||||||
filename = activity_log_path(activity)
|
|
||||||
if not os.path.exists(filename):
|
|
||||||
create_activity_log_file(activity)
|
|
||||||
if is_convertable(activity):
|
|
||||||
converted_value = round(convert_units(activity, value, unit), 2)
|
|
||||||
else:
|
|
||||||
converted_value = round(value, 2)
|
|
||||||
with open(filename, 'a') as f:
|
|
||||||
writer = csv.writer(f)
|
|
||||||
writer.writerow([timestamp, username, converted_value])
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
log_activity(
|
|
||||||
timestamp=0,
|
|
||||||
username="test",
|
|
||||||
activity="climb",
|
|
||||||
value=1,
|
|
||||||
unit="m"
|
|
||||||
)
|
|
||||||
@@ -1,78 +0,0 @@
|
|||||||
import os
|
|
||||||
import discord
|
|
||||||
from redbot.core import Config, commands
|
|
||||||
from .activity_logger import log_activity, activity_log_path, activity_units_path
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class FitnessCog(commands.Cog):
|
|
||||||
def __init__(self, bot):
|
|
||||||
self.bot = bot
|
|
||||||
# Some hardcoded values, this probably should be changed with redbot config. IDK how to use it or what's in it.
|
|
||||||
default_threads = "1457402451530350727,1328363409648910356"
|
|
||||||
self.threads_id = list(map(int, os.getenv("THREADS_ID", default_threads).split(",")))
|
|
||||||
self.confirm_reactions = os.getenv("CONFIRMATION_REACTIONS", "🐈💨")
|
|
||||||
|
|
||||||
@commands.Cog.listener()
|
|
||||||
async def on_message(self, message):
|
|
||||||
if message.author.bot:
|
|
||||||
return
|
|
||||||
if message.channel.id not in self.threads_id:
|
|
||||||
return
|
|
||||||
if not message.content.startswith("!"):
|
|
||||||
return
|
|
||||||
|
|
||||||
# region Fitness commands
|
|
||||||
if message.content.startswith("!getunits"):
|
|
||||||
await self.get_units_file(message)
|
|
||||||
return
|
|
||||||
if message.content.startswith("!getlog"):
|
|
||||||
await self.get_log_file(message)
|
|
||||||
return
|
|
||||||
# Make sure this one is past every other ! commands
|
|
||||||
await self.fitness_activity_log(message)
|
|
||||||
return
|
|
||||||
# endregion Fitness commands
|
|
||||||
|
|
||||||
# region Fitness commands functions
|
|
||||||
@staticmethod
|
|
||||||
async def get_units_file(message: discord.Message):
|
|
||||||
try:
|
|
||||||
raw = message.content[1:].lower().strip().split(" ", 1)
|
|
||||||
filename = activity_units_path(raw[1])
|
|
||||||
await message.channel.send(file=discord.File(fp=filename))
|
|
||||||
except Exception as e:
|
|
||||||
await message.channel.send(str(e))
|
|
||||||
|
|
||||||
@staticmethod
|
|
||||||
async def get_log_file(message: discord.Message):
|
|
||||||
try:
|
|
||||||
raw = message.content[1:].lower().strip().split(" ", 1)
|
|
||||||
filename = activity_log_path(raw[1])
|
|
||||||
await message.channel.send(file=discord.File(fp=filename))
|
|
||||||
except Exception as e:
|
|
||||||
await message.channel.send(str(e))
|
|
||||||
|
|
||||||
async def log_fitness(self, message):
|
|
||||||
try:
|
|
||||||
raw = message.content[1:].lower().strip().split(" ", 2)
|
|
||||||
|
|
||||||
timestamp = int(message.created_at.timestamp())
|
|
||||||
username = message.author.name
|
|
||||||
activity = raw[0]
|
|
||||||
value = float(raw[1])
|
|
||||||
unit = raw[2] if len(raw) > 2 else "m"
|
|
||||||
|
|
||||||
log_activity(timestamp, username, activity, value, unit)
|
|
||||||
|
|
||||||
for r in self.confirm_reactions:
|
|
||||||
await message.add_reaction(r)
|
|
||||||
|
|
||||||
except Exception as e:
|
|
||||||
await message.reply(str(e))
|
|
||||||
# endregion Fitness commands functions
|
|
||||||
|
|
||||||
|
|
||||||
def setup(bot):
|
|
||||||
bot.add_cog(FitnessCog(bot))
|
|
||||||
|
|
||||||
@@ -0,0 +1,211 @@
|
|||||||
|
from abc import ABC, abstractmethod
|
||||||
|
from openai import OpenAI
|
||||||
|
from reginaldCog.messenger_clients.messenger_client import ClientMessage
|
||||||
|
|
||||||
|
|
||||||
|
class ILLMContent(ABC):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAIContent(ILLMContent):
|
||||||
|
def __init__(self):
|
||||||
|
self._content_items: list[dict[str, str]] = []
|
||||||
|
|
||||||
|
@property
|
||||||
|
def content_items(self) -> list[dict]:
|
||||||
|
return self._content_items
|
||||||
|
|
||||||
|
@content_items.setter
|
||||||
|
def content_items(self, value: list[dict[str, str]]):
|
||||||
|
self._content_items = value
|
||||||
|
|
||||||
|
|
||||||
|
class ILLMContentBuilder(ABC):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAIContentBuilder(ILLMContentBuilder):
|
||||||
|
def __init__(self, content: OpenAIContent):
|
||||||
|
self.content = content
|
||||||
|
|
||||||
|
def add_output_text(self, text: str):
|
||||||
|
item = {"type": "output_text", "text": text}
|
||||||
|
self.content.content_items.append(item)
|
||||||
|
return self
|
||||||
|
|
||||||
|
def add_input_text(self, text: str):
|
||||||
|
item = {"type": "input_text", "text": text}
|
||||||
|
self.content.content_items.append(item)
|
||||||
|
return self
|
||||||
|
|
||||||
|
def add_input_image(self, image_url: str):
|
||||||
|
item = {"type": "input_image", "image_url": image_url}
|
||||||
|
self.content.content_items.append(item)
|
||||||
|
return self
|
||||||
|
|
||||||
|
def add_from_dict(self, item: dict):
|
||||||
|
self.content.content_items.append(item)
|
||||||
|
return self
|
||||||
|
|
||||||
|
|
||||||
|
class ILLMMessage(ABC):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAIMessage(ILLMMessage):
|
||||||
|
def __init__(self):
|
||||||
|
self.content = OpenAIContent()
|
||||||
|
self.role = ""
|
||||||
|
|
||||||
|
@property
|
||||||
|
def to_dict(self):
|
||||||
|
return {"role": self.role, "content": self.content.content_items}
|
||||||
|
|
||||||
|
|
||||||
|
class ILLMMessageBuilder(ABC):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAIMessageBuilder(ILLMMessageBuilder):
|
||||||
|
def __init__(self, message: OpenAIMessage):
|
||||||
|
self.message = message
|
||||||
|
|
||||||
|
def set_role(self, role: str):
|
||||||
|
self.message.role = role
|
||||||
|
return self
|
||||||
|
|
||||||
|
def set_content(self, content: OpenAIContent):
|
||||||
|
self.message.content = content
|
||||||
|
return self
|
||||||
|
|
||||||
|
|
||||||
|
class ILLMPrompt(ABC):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAIPrompt(ILLMPrompt):
|
||||||
|
def __init__(self):
|
||||||
|
self.messages = []
|
||||||
|
|
||||||
|
@property
|
||||||
|
def to_list(self):
|
||||||
|
return [i_message.to_dict for i_message in self.messages]
|
||||||
|
|
||||||
|
|
||||||
|
class ILLMPromptBuilder(ABC):
|
||||||
|
@abstractmethod
|
||||||
|
def __init__(self, prompt: ILLMPrompt):
|
||||||
|
pass
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def add_message(self, message: ILLMMessage):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAIPromptBuilder(ILLMPromptBuilder):
|
||||||
|
def __init__(self, prompt: OpenAIPrompt):
|
||||||
|
self.prompt = prompt
|
||||||
|
|
||||||
|
def add_message(self, message: OpenAIMessage):
|
||||||
|
self.prompt.messages.append(message)
|
||||||
|
|
||||||
|
|
||||||
|
class ILLMClient(ABC):
|
||||||
|
@abstractmethod
|
||||||
|
def get_response(self, prompt: ILLMPrompt):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAIClient(ILLMClient):
|
||||||
|
def __init__(self):
|
||||||
|
self.model = 'gpt-4.1-mini'
|
||||||
|
self.client = OpenAI()
|
||||||
|
|
||||||
|
def get_response(self, prompt: OpenAIPrompt):
|
||||||
|
response_input = {"model": self.model, "input": prompt.to_list}
|
||||||
|
return self.client.responses.create(**response_input)
|
||||||
|
|
||||||
|
|
||||||
|
class IMessageAdapter(ABC):
|
||||||
|
@abstractmethod
|
||||||
|
def to_message(self) -> ILLMMessage:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAIResponseAdapter(IMessageAdapter):
|
||||||
|
def __init__(self, response):
|
||||||
|
self.response = response
|
||||||
|
self.response_output = response.output[0]
|
||||||
|
|
||||||
|
def to_message(self) -> OpenAIMessage:
|
||||||
|
content = OpenAIContent()
|
||||||
|
content_builder = OpenAIContentBuilder(content)
|
||||||
|
message = OpenAIMessage()
|
||||||
|
message_builder = OpenAIMessageBuilder(message)
|
||||||
|
|
||||||
|
message_builder.set_role(self.response_output.role)\
|
||||||
|
.set_content(content)
|
||||||
|
|
||||||
|
for i_content_item in self.response_output.content:
|
||||||
|
item = i_content_item.to_dict()
|
||||||
|
content_builder.add_from_dict(item)
|
||||||
|
|
||||||
|
return message
|
||||||
|
|
||||||
|
|
||||||
|
class MessengerClientMessageAdapter(IMessageAdapter):
|
||||||
|
def __init__(self, message: ClientMessage): # Dependency Inversion violation, figure out later
|
||||||
|
self.message = message
|
||||||
|
|
||||||
|
def to_message(self) -> ILLMMessage:
|
||||||
|
content = OpenAIContent()
|
||||||
|
content_builder = OpenAIContentBuilder(content)
|
||||||
|
message = OpenAIMessage()
|
||||||
|
message_builder = OpenAIMessageBuilder(message)
|
||||||
|
|
||||||
|
message_builder.set_role("user")\
|
||||||
|
.set_content(content)
|
||||||
|
|
||||||
|
if self.message:
|
||||||
|
content_builder.add_input_text(self.message.content)
|
||||||
|
for i_image_url in self.message.image_urls:
|
||||||
|
content_builder.add_input_image(i_image_url)
|
||||||
|
|
||||||
|
return message
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
dev_message_text = "You are a butler named Reginald. Give your answers in a refined manners of the English butler."
|
||||||
|
user_message_text = "Tell me what's on this picture"
|
||||||
|
pic_url = "https://media.discordapp.net/attachments/913835221814763550/1408534120421654669/image.png?ex=68ad62df&is=68ac115f&hm=bdfcf878f20e154a575a528b9cb274a8c2162f84ce12fc3876856e821ef9cc63&=&format=webp&quality=lossless&width=1342&height=755"
|
||||||
|
|
||||||
|
# Create developer message content and message
|
||||||
|
dev_content = OpenAIContent()
|
||||||
|
dev_content_builder = OpenAIContentBuilder(dev_content).add_input_text(dev_message_text)
|
||||||
|
dev_message = OpenAIMessage()
|
||||||
|
OpenAIMessageBuilder(dev_message).set_role("developer").set_content(dev_content)
|
||||||
|
|
||||||
|
# Create user message content and message
|
||||||
|
user_content = OpenAIContent()
|
||||||
|
OpenAIContentBuilder(user_content).add_input_text(user_message_text).add_input_image(pic_url)
|
||||||
|
user_message = OpenAIMessage()
|
||||||
|
OpenAIMessageBuilder(user_message).set_role("user").set_content(user_content)
|
||||||
|
|
||||||
|
# Create prompt and add messages
|
||||||
|
test_prompt = OpenAIPrompt()
|
||||||
|
prompt_builder = OpenAIPromptBuilder(test_prompt)
|
||||||
|
prompt_builder.add_message(dev_message)
|
||||||
|
prompt_builder.add_message(user_message)
|
||||||
|
|
||||||
|
# Create client and get response
|
||||||
|
client = OpenAIClient()
|
||||||
|
test_response = client.get_response(test_prompt)
|
||||||
|
|
||||||
|
# Create response adapter and add response to prompt
|
||||||
|
response_adapter = OpenAIResponseAdapter(test_response)
|
||||||
|
response_message = response_adapter.to_message()
|
||||||
|
prompt_builder.add_message(response_message)
|
||||||
|
|
||||||
|
# Print messages
|
||||||
|
print(test_prompt.to_list)
|
||||||
|
|
||||||
@@ -0,0 +1,71 @@
|
|||||||
|
from dataclasses import dataclass, field
|
||||||
|
from abc import ABC
|
||||||
|
|
||||||
|
|
||||||
|
# region Content classes
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Content(ABC):
|
||||||
|
type: str = field(init=False, default='')
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class InputText(Content):
|
||||||
|
type: str = field(init=False, default='input_text')
|
||||||
|
text: str
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class OutputText(Content):
|
||||||
|
type: str = field(init=False, default='output_text')
|
||||||
|
text: str
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class InputImage(Content):
|
||||||
|
type: str = field(init=False, default='input_image')
|
||||||
|
image_url: str | None = field(default=None)
|
||||||
|
file_id: str | None = field(default=None)
|
||||||
|
|
||||||
|
def __post_init__(self):
|
||||||
|
if self.image_url is None and self.file_id is None:
|
||||||
|
raise ValueError('Either `image_url` or `file_id` must be provided.')
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class UrlCitation(Content):
|
||||||
|
type: str = field(init=False, default='url_citation')
|
||||||
|
# To be done
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class FunctionCall(Content):
|
||||||
|
type: str = field(init=False, default='function_call')
|
||||||
|
id: str
|
||||||
|
call_id: str
|
||||||
|
name: str
|
||||||
|
arguments: dict
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class FunctionCallOutput(Content):
|
||||||
|
type: str = field(init=False, default='function_call_output')
|
||||||
|
call_id: str
|
||||||
|
output: str
|
||||||
|
# endregion Content classes
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Message:
|
||||||
|
role: str
|
||||||
|
content: list[Content]
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Prompt:
|
||||||
|
model: str
|
||||||
|
input: list[Message]
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Response:
|
||||||
|
output: list[Message]
|
||||||
@@ -0,0 +1,65 @@
|
|||||||
|
from dataclasses import dataclass, field
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
|
||||||
|
# region dataclasses
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Attachment:
|
||||||
|
content_type: str
|
||||||
|
filename: str
|
||||||
|
id: int
|
||||||
|
size: int
|
||||||
|
url: str
|
||||||
|
ephemeral: bool = False
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Channel:
|
||||||
|
created_at: datetime
|
||||||
|
id: int
|
||||||
|
name: str
|
||||||
|
|
||||||
|
def __str__(self) -> str:
|
||||||
|
return self.name
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Role:
|
||||||
|
id: int
|
||||||
|
name: str
|
||||||
|
|
||||||
|
def __str__(self) -> str:
|
||||||
|
return self.name
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Member:
|
||||||
|
bot: bool
|
||||||
|
created_at: datetime
|
||||||
|
display_name: str # For regular users this is just their global name or their username, but if they have a guild specific nickname then that is returned instead.
|
||||||
|
global_name: str # The user's global nickname, taking precedence over the username in display.
|
||||||
|
id: int
|
||||||
|
joined_at: datetime
|
||||||
|
mention: str
|
||||||
|
name: str # The user's username.
|
||||||
|
nick: str # The guild specific nickname of the user. Takes precedence over the global name.
|
||||||
|
roles: list[Role] = field(default_factory=list)
|
||||||
|
|
||||||
|
def __str__(self) -> str:
|
||||||
|
return self.name
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True)
|
||||||
|
class Message:
|
||||||
|
author: Member
|
||||||
|
channel: Channel
|
||||||
|
created_at: datetime
|
||||||
|
id: int
|
||||||
|
attachments: list[Attachment] = field(default_factory=list)
|
||||||
|
channel_mentions: list[Channel] = field(default_factory=list)
|
||||||
|
content: str = ''
|
||||||
|
mentions: list[Member] = field(default_factory=list)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
pass
|
||||||
@@ -0,0 +1,77 @@
|
|||||||
|
from abc import ABC, abstractmethod
|
||||||
|
from enum import Enum
|
||||||
|
from discord import Message, Attachment
|
||||||
|
|
||||||
|
|
||||||
|
class IClientMessage(ABC):
|
||||||
|
@abstractmethod
|
||||||
|
def __init__(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class ClientMessage(IClientMessage):
|
||||||
|
def __init__(self):
|
||||||
|
self.author_name: str = ''
|
||||||
|
self.content: str = ''
|
||||||
|
self.image_urls: list = []
|
||||||
|
|
||||||
|
|
||||||
|
class IMessageBuilder(ABC):
|
||||||
|
@abstractmethod
|
||||||
|
def __init__(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def set_content(self, value: str):
|
||||||
|
pass
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def set_author_name(self, value: str):
|
||||||
|
pass
|
||||||
|
|
||||||
|
@abstractmethod
|
||||||
|
def set_image_urls(self, value: list[str]):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class DiscordMessageBuilder(IMessageBuilder):
|
||||||
|
def __init__(self, message: IClientMessage):
|
||||||
|
self.message = message
|
||||||
|
|
||||||
|
def set_author_name(self, value: str):
|
||||||
|
self.message.author_name = value
|
||||||
|
return self
|
||||||
|
|
||||||
|
def set_content(self, value: str):
|
||||||
|
self.message.content = value
|
||||||
|
return self
|
||||||
|
|
||||||
|
def set_image_urls(self, value: list[str]):
|
||||||
|
self.message.image_urls = value
|
||||||
|
return self
|
||||||
|
|
||||||
|
|
||||||
|
class IMessageAdapter(ABC):
|
||||||
|
@abstractmethod
|
||||||
|
def create_message(self, message: object) -> IClientMessage:
|
||||||
|
pass
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def validate_image_urls(urls_list: list[Attachment]) -> list[str]:
|
||||||
|
supported_image_formats = ('image/jpeg', 'image/png', 'image/webp', 'image/gif')
|
||||||
|
return [i_attachment.url for i_attachment in urls_list if i_attachment.content_type in supported_image_formats]
|
||||||
|
|
||||||
|
|
||||||
|
class DiscordMessageAdapter(IMessageAdapter):
|
||||||
|
def create_message(self, message: Message) -> IClientMessage:
|
||||||
|
client_message = ClientMessage()
|
||||||
|
message_builder = DiscordMessageBuilder(client_message)
|
||||||
|
urls_list = self.validate_image_urls(message.attachments)
|
||||||
|
message_builder.set_content(message.content)\
|
||||||
|
.set_author_name(message.author.name)\
|
||||||
|
.set_image_urls(urls_list)
|
||||||
|
return client_message
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
pass
|
||||||
+25
-25
@@ -16,7 +16,6 @@ from .weather import time_now, get_current_weather, get_weather_forecast
|
|||||||
from .tools_description import TOOLS
|
from .tools_description import TOOLS
|
||||||
from .debug_stuff import debug
|
from .debug_stuff import debug
|
||||||
|
|
||||||
|
|
||||||
CALLABLE_FUNCTIONS = {
|
CALLABLE_FUNCTIONS = {
|
||||||
# Dictionary with functions to call.
|
# Dictionary with functions to call.
|
||||||
# You can use globals()[func_name](**args) instead, but that's too implicit.
|
# You can use globals()[func_name](**args) instead, but that's too implicit.
|
||||||
@@ -80,7 +79,6 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
if not (await self.is_admin(message) or await self.has_access(message.author)):
|
if not (await self.is_admin(message) or await self.has_access(message.author)):
|
||||||
return # Ignore message if user has no permissions
|
return # Ignore message if user has no permissions
|
||||||
|
|
||||||
|
|
||||||
guild = message.guild
|
guild = message.guild
|
||||||
channel_id = str(message.channel.id)
|
channel_id = str(message.channel.id)
|
||||||
user_id = str(message.author.id)
|
user_id = str(message.author.id)
|
||||||
@@ -116,7 +114,7 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
return
|
return
|
||||||
explicit_invocation = True
|
explicit_invocation = True
|
||||||
|
|
||||||
# ✅ Passive Listening: Check if the message contains relevant keywords
|
# ✅ Passive Listening: Check if the message contains relevant keywords
|
||||||
elif self.should_reginald_interject(message_content):
|
elif self.should_reginald_interject(message_content):
|
||||||
prompt = message_content
|
prompt = message_content
|
||||||
explicit_invocation = False
|
explicit_invocation = False
|
||||||
@@ -124,11 +122,11 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
else:
|
else:
|
||||||
return # Ignore irrelevant messages
|
return # Ignore irrelevant messages
|
||||||
|
|
||||||
# ✅ Context Handling: Maintain conversation flow
|
# ✅ Context Handling: Maintain conversation flow
|
||||||
if memory and memory[-1]["user"] == user_name:
|
if memory and memory[-1]["user"] == user_name:
|
||||||
prompt = f"Continuation of the discussion:\n{prompt}"
|
prompt = f"Continuation of the discussion:\n{prompt}"
|
||||||
|
|
||||||
# ✅ Prepare context messages
|
# ✅ Prepare context messages
|
||||||
formatted_messages = [{"role": "system", "content": self.get_reginald_persona()}]
|
formatted_messages = [{"role": "system", "content": self.get_reginald_persona()}]
|
||||||
|
|
||||||
if user_profile:
|
if user_profile:
|
||||||
@@ -145,10 +143,10 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
"content": f"[{summary['timestamp']}] Topics: {', '.join(summary['topics'])}\n{summary['summary']}"
|
"content": f"[{summary['timestamp']}] Topics: {', '.join(summary['topics'])}\n{summary['summary']}"
|
||||||
})
|
})
|
||||||
|
|
||||||
formatted_messages += [{"role": "user", "content": f"{entry['user']}: {entry['content']}"} for entry in memory]
|
formatted_messages += [{"role": "user", "content": f"{entry['user']}: {entry['content']}"} for entry in
|
||||||
|
memory]
|
||||||
formatted_messages.append({"role": "user", "content": f"{user_name}: {prompt}"})
|
formatted_messages.append({"role": "user", "content": f"{user_name}: {prompt}"})
|
||||||
|
|
||||||
|
|
||||||
##################################################
|
##################################################
|
||||||
# #
|
# #
|
||||||
## Generate AI Response, put into response_text ##
|
## Generate AI Response, put into response_text ##
|
||||||
@@ -166,7 +164,8 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
memory.append({"user": "Reginald", "content": response_text})
|
memory.append({"user": "Reginald", "content": response_text})
|
||||||
|
|
||||||
if len(memory) > self.short_term_memory_limit:
|
if len(memory) > self.short_term_memory_limit:
|
||||||
summary = await self.summarize_memory(message, memory[:int(self.short_term_memory_limit * self.summary_retention_ratio)])
|
summary = await self.summarize_memory(message, memory[:int(
|
||||||
|
self.short_term_memory_limit * self.summary_retention_ratio)])
|
||||||
mid_memory.setdefault(channel_id, []).append({
|
mid_memory.setdefault(channel_id, []).append({
|
||||||
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
|
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
|
||||||
"topics": self.extract_topics_from_summary(summary),
|
"topics": self.extract_topics_from_summary(summary),
|
||||||
@@ -174,18 +173,18 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
})
|
})
|
||||||
if len(mid_memory[channel_id]) > self.summary_retention_limit:
|
if len(mid_memory[channel_id]) > self.summary_retention_limit:
|
||||||
mid_memory[channel_id].pop(0)
|
mid_memory[channel_id].pop(0)
|
||||||
memory = memory[-(self.short_term_memory_limit - int(self.short_term_memory_limit * self.summary_retention_ratio)):]
|
memory = memory[-(self.short_term_memory_limit - int(
|
||||||
|
self.short_term_memory_limit * self.summary_retention_ratio)):]
|
||||||
|
|
||||||
short_memory[channel_id] = memory
|
short_memory[channel_id] = memory
|
||||||
|
|
||||||
await self.send_split_message(message.channel, response_text)
|
await self.send_split_message(message.channel, response_text)
|
||||||
|
|
||||||
|
|
||||||
def should_reginald_interject(self, message_content: str) -> bool:
|
def should_reginald_interject(self, message_content: str) -> bool:
|
||||||
"""Determines if Reginald should respond to a message based on keywords."""
|
"""Determines if Reginald should respond to a message based on keywords."""
|
||||||
direct_invocation = {
|
direct_invocation = {
|
||||||
"reginald,"
|
"reginald,"
|
||||||
}
|
}
|
||||||
message_lower = message_content.lower()
|
message_lower = message_content.lower()
|
||||||
|
|
||||||
return any(message_lower.startswith(invocation) for invocation in direct_invocation)
|
return any(message_lower.startswith(invocation) for invocation in direct_invocation)
|
||||||
@@ -261,7 +260,8 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
await self.config.guild(ctx.guild).openai_api_key.set(api_key)
|
await self.config.guild(ctx.guild).openai_api_key.set(api_key)
|
||||||
await ctx.send("OpenAI API key set successfully.")
|
await ctx.send("OpenAI API key set successfully.")
|
||||||
|
|
||||||
@commands.command(name="reginald_set_listening_channel", help="Set the channel where Reginald listens for messages.")
|
@commands.command(name="reginald_set_listening_channel",
|
||||||
|
help="Set the channel where Reginald listens for messages.")
|
||||||
@commands.has_permissions(administrator=True)
|
@commands.has_permissions(administrator=True)
|
||||||
async def set_listening_channel(self, ctx, channel: discord.TextChannel):
|
async def set_listening_channel(self, ctx, channel: discord.TextChannel):
|
||||||
"""Sets the channel where Reginald will listen for passive responses."""
|
"""Sets the channel where Reginald will listen for passive responses."""
|
||||||
@@ -273,7 +273,8 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
await self.config.guild(ctx.guild).listening_channel.set(channel.id)
|
await self.config.guild(ctx.guild).listening_channel.set(channel.id)
|
||||||
await ctx.send(f"✅ Reginald will now listen only in {channel.mention}.")
|
await ctx.send(f"✅ Reginald will now listen only in {channel.mention}.")
|
||||||
|
|
||||||
@commands.command(name="reginald_get_listening_channel", help="Check which channel Reginald is currently listening in.")
|
@commands.command(name="reginald_get_listening_channel",
|
||||||
|
help="Check which channel Reginald is currently listening in.")
|
||||||
@commands.has_permissions(administrator=True)
|
@commands.has_permissions(administrator=True)
|
||||||
async def get_listening_channel(self, ctx):
|
async def get_listening_channel(self, ctx):
|
||||||
"""Displays the current listening channel."""
|
"""Displays the current listening channel."""
|
||||||
@@ -288,17 +289,16 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
|
|||||||
else:
|
else:
|
||||||
await ctx.send("❌ No listening channel has been set.")
|
await ctx.send("❌ No listening channel has been set.")
|
||||||
|
|
||||||
|
# async def send_long_message(self, ctx, message, prefix: str = ""):
|
||||||
async def send_long_message(self, ctx, message, prefix: str = ""):
|
# """Splits and sends a long message to avoid Discord's 2000-character limit."""
|
||||||
"""Splits and sends a long message to avoid Discord's 2000-character limit."""
|
# chunk_size = 1900 # Leave some space for formatting
|
||||||
chunk_size = 1900 # Leave some space for formatting
|
# if prefix:
|
||||||
if prefix:
|
# prefix_length = len(prefix)
|
||||||
prefix_length = len(prefix)
|
# chunk_size -= prefix_length
|
||||||
chunk_size -= prefix_length
|
#
|
||||||
|
# for i in range(0, len(message), chunk_size):
|
||||||
for i in range(0, len(message), chunk_size):
|
# chunk = message[i:i + chunk_size]
|
||||||
chunk = message[i:i + chunk_size]
|
# await ctx.send(f"{prefix}{chunk}")
|
||||||
await ctx.send(f"{prefix}{chunk}")
|
|
||||||
|
|
||||||
async def send_split_message(self, ctx, content: str, prefix: str = ""):
|
async def send_split_message(self, ctx, content: str, prefix: str = ""):
|
||||||
"""
|
"""
|
||||||
|
|||||||
Reference in New Issue
Block a user