mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-14 21:38:38 +00:00
391 lines
18 KiB
Python
391 lines
18 KiB
Python
# ai command cogs
|
|
import logging
|
|
import re
|
|
import sys
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import Optional
|
|
from pathlib import Path
|
|
import discord
|
|
import openai
|
|
import requests
|
|
from discord.ext import commands
|
|
from other_functions import discord_friendly_send, discord_friendly_reply
|
|
|
|
|
|
import ai_functions
|
|
from constants import (
|
|
ASSISTANTS,
|
|
DATA,
|
|
GRAPHICS_PATH,
|
|
INITIAL_TIME_WAIT,
|
|
MASTER_TIMEOUT,
|
|
MESSAGE_TABLE,
|
|
OPENAICLIENT,
|
|
SPECJALNE_ZIEMNIACZKI,
|
|
WORD_REACTIONS,
|
|
)
|
|
|
|
|
|
class Dm_Mode(Enum):
|
|
ARMIA_HAMMERA = (1,)
|
|
SPECJALNY_ZIEMNIACZEK = (2,)
|
|
SEKRETNY_SEKSRET = (3,)
|
|
ECHO_ECHO = (4,)
|
|
|
|
|
|
class Events(commands.Cog):
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
self.logger = logging.getLogger("discord")
|
|
self.armia = {}
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
self.armia[superfryta[0]] = Dm_Mode.SPECJALNY_ZIEMNIACZEK
|
|
self.logger.info(self.armia)
|
|
|
|
async def cog_load(self):
|
|
self.logger.info("Starting personal assistants")
|
|
for superfryta_id, superfryta in SPECJALNE_ZIEMNIACZKI.items():
|
|
|
|
if superfryta[4] != "":
|
|
self.logger.info(
|
|
"Personal assistant for user: %s, exists id: %s,name: %s, owner: %s, special instructions: %s assistant id: %s ",
|
|
superfryta_id,
|
|
superfryta[0],
|
|
superfryta[1],
|
|
superfryta[2],
|
|
superfryta[3],
|
|
superfryta[4],
|
|
)
|
|
thread = await OPENAICLIENT.beta.threads.create()
|
|
self.logger.info("Thread id: %s", thread.id)
|
|
ASSISTANTS[superfryta[1]] = (
|
|
superfryta[2],
|
|
superfryta[4],
|
|
superfryta[0],
|
|
thread,
|
|
)
|
|
else:
|
|
self.logger.info(
|
|
"Creating personal assistant for user: %s, id: %s,name: %s, owner: %s, special instructions: %s",
|
|
superfryta_id,
|
|
superfryta[0],
|
|
superfryta[1],
|
|
superfryta[2],
|
|
superfryta[3],
|
|
)
|
|
await ai_functions.create_chat_assistant(
|
|
superfryta_id, superfryta[0], superfryta[1], superfryta[2], superfryta[3]
|
|
)
|
|
self.logger.info("Started personal assistants")
|
|
|
|
@commands.hybrid_command(
|
|
name="switch_dm_mode",
|
|
description="Jeśli nie wiesz jak użyć tej komendy to nawet nie próbuj",
|
|
guild=discord.Object(id=664789470779932693),
|
|
)
|
|
async def switch_dm_mode(self, ctx, dm_mode_arg: Dm_Mode):
|
|
async with ctx.channel.typing():
|
|
if isinstance(ctx.channel, discord.DMChannel):
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
self.logger.info(ctx.message.author.id)
|
|
self.logger.info(superfryta)
|
|
if ctx.message.author.id == superfryta[0]:
|
|
self.armia[ctx.message.author.id] = dm_mode_arg
|
|
self.logger.info(self.armia)
|
|
await ctx.reply("Weszlo")
|
|
return
|
|
await ctx.reply(
|
|
"Tylko członkowie Armii Hammera mogą używać. Nie jesteś jednym z nich"
|
|
)
|
|
else:
|
|
await ctx.reply("Nope. Nie wiesz jak użyć")
|
|
|
|
@commands.hybrid_command(
|
|
name="armia_hammera",
|
|
description="Jeśli nie wiesz jak użyć tej komendy to nawet nie próbuj",
|
|
)
|
|
async def armia_hammera(
|
|
self, ctx, message_txt: str, recipient: Optional[discord.User]
|
|
):
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
if ctx.message.author.id == superfryta[0]:
|
|
await self.armia_hammera_back(ctx, message_txt, recipient)
|
|
return
|
|
await ctx.reply("Tylko członkowie Armii Hammera mogą używać. Nie jesteś jednym z nich")
|
|
|
|
async def armia_hammera_back(self, ctx, message_txt, recipient=None):
|
|
self.logger.info("Armia Hammera")
|
|
recipients = []
|
|
if recipient:
|
|
recipients.append(recipient.id)
|
|
else:
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
recipients.append(superfryta[0])
|
|
self.logger.info(recipients)
|
|
for item in recipients:
|
|
user = await self.bot.fetch_user(item)
|
|
channel = await user.create_dm()
|
|
self.logger.info(
|
|
"User %s -> %s: %s", ctx.message.author, user, message_txt
|
|
)
|
|
await discord_friendly_send(channel, message_txt)
|
|
await discord_friendly_reply(ctx, "Poszło")
|
|
|
|
@commands.Cog.listener(name="dodaj_do_bazy_wiedzy")
|
|
async def dodaj_do_bazy_wiedzy(self, ctx):
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
if ctx.message.author.id == superfryta[0]:
|
|
#logic here
|
|
return
|
|
await ctx.reply("Tylko członkowie Armii Hammera mogą używać. Nie jesteś jednym z nich")
|
|
|
|
@commands.Cog.listener(name="listuj_baze_wiedzy")
|
|
async def listuj_baze_wiedzy(self, ctx):
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
if ctx.message.author.id == superfryta[0]:
|
|
#logic here
|
|
return
|
|
await ctx.reply("Tylko członkowie Armii Hammera mogą używać. Nie jesteś jednym z nich")
|
|
|
|
@commands.Cog.listener(name="usun_z_bazy_wiedzy")
|
|
async def usun_z_bazy_wiedzy(self, ctx):
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
if ctx.message.author.id == superfryta[0]:
|
|
#logic here
|
|
return
|
|
await ctx.reply("Tylko członkowie Armii Hammera mogą używać. Nie jesteś jednym z nich")
|
|
|
|
@commands.Cog.listener(name="przetworz_plik_linia_po_linii")
|
|
async def przetworz_plik_linia_po_linii(self, ctx):
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
if ctx.message.author.id == superfryta[0]:
|
|
#logic here
|
|
return
|
|
await ctx.reply("Tylko członkowie Armii Hammera mogą używać. Nie jesteś jednym z nich")
|
|
|
|
@commands.Cog.listener()
|
|
async def on_message(self, message):
|
|
"""
|
|
Handle incoming messages in a Discord server, perform various
|
|
checks and actions based on the content and context of the message, and respond accordingly.
|
|
|
|
:param message: The message object that is received when a user sends a message in a Discord server
|
|
or DM. The code is checking various conditions and performing actions based on the content of the
|
|
message and the context in which it was sent. It also includes TODOs for future improvements
|
|
:return: The function `on_message` is being returned.
|
|
"""
|
|
vykidailo = False
|
|
channel = None
|
|
if message.author == self.bot.user:
|
|
return
|
|
|
|
if isinstance(message.author, discord.Member):
|
|
for role in message.author.roles:
|
|
if role.name == "Vykidailo":
|
|
vykidailo = True
|
|
|
|
if ("Conjurer Śpij Słodko Aniołku" in message.content) and vykidailo:
|
|
sys.exit()
|
|
# kanal bez bota
|
|
if message.channel.id == 1095985579147141202:
|
|
return
|
|
# wentylacja
|
|
if message.channel.id == 1083804024173764739:
|
|
return
|
|
# legendy
|
|
if message.channel.id == 1084448332841230388:
|
|
return
|
|
# interrogation booth
|
|
if message.channel.id == 1111625221171052595:
|
|
return
|
|
if isinstance(message.channel, discord.DMChannel):
|
|
self.logger.info(message.author.id)
|
|
for superfryta in SPECJALNE_ZIEMNIACZKI.values():
|
|
self.logger.info(superfryta[0])
|
|
|
|
if message.author.id == superfryta[0]:
|
|
self.logger.info("Specjalny ziemniak")
|
|
if self.armia[message.author.id] == Dm_Mode.SPECJALNY_ZIEMNIACZEK:
|
|
#await self.bot.process_commands(message)
|
|
await ai_functions.chat_with_assistant(message, superfryta[1])
|
|
return
|
|
elif self.armia[message.author.id] == Dm_Mode.ECHO_ECHO:
|
|
await ai_functions.echo(message)
|
|
return
|
|
elif self.armia[message.author.id] == Dm_Mode.ARMIA_HAMMERA:
|
|
self.logger.info("Armia Hammera function call")
|
|
ctx = await self.bot.get_context(message)
|
|
self.armia_hammera_back(ctx=ctx, message_txt=message.content)
|
|
return
|
|
elif self.armia[message.author.id] == Dm_Mode.SEKRETNY_SEKSRET:
|
|
pass
|
|
else:
|
|
await discord_friendly_send(message.channel,"Coś się wyebao. Wołaj Hammera")
|
|
#await message.channel.send("Coś się wyebao. Wołaj Hammera")
|
|
return
|
|
channel = self.bot.get_channel(1064888712565100614)
|
|
await discord_friendly_send(channel, "Słyszałem ja żem że: " + message.content)
|
|
return
|
|
channel = message.channel
|
|
message_content_lower = message.content.lower()
|
|
|
|
tdelta = datetime.now() - MASTER_TIMEOUT
|
|
tdelta = tdelta.total_seconds()
|
|
if "opowiedz o fabryczce" in message_content_lower:
|
|
await message.reply(DATA["fabryczka"])
|
|
if "opowiedz mi o fabryczce" in message_content_lower:
|
|
await message.reply(DATA["fabryczka"])
|
|
|
|
if tdelta > INITIAL_TIME_WAIT:
|
|
for word in WORD_REACTIONS:
|
|
if re.search(r"\b" + word + r"\b", message_content_lower):
|
|
tdelta = datetime.now() - WORD_REACTIONS[word][2]
|
|
tdelta = tdelta.total_seconds()
|
|
security_clearance = WORD_REACTIONS[word][4]
|
|
reaction = WORD_REACTIONS[word][3]
|
|
if tdelta > WORD_REACTIONS[word][1]:
|
|
# TODO: to zrobic reactiony
|
|
self.logger.info("Ping z procedury reakcji")
|
|
if reaction:
|
|
emoji = self.bot.get_emoji(WORD_REACTIONS[word][0])
|
|
await message.add_reaction(emoji)
|
|
elif security_clearance and vykidailo:
|
|
await message.reply(WORD_REACTIONS[word][0])
|
|
elif not security_clearance:
|
|
await message.reply(WORD_REACTIONS[word][0])
|
|
WORD_REACTIONS[word][2] = datetime.now()
|
|
|
|
# TODO: drobne literówki, mentiony, spacja przed dwukropkiem. napraw.
|
|
kondziu_mentioned = False
|
|
for mention in message.mentions:
|
|
if mention == self.bot.user:
|
|
kondziu_mentioned = True
|
|
|
|
if kondziu_mentioned or "conjurer:" in message_content_lower:
|
|
async with channel.typing():
|
|
self.logger.debug("Procedura chatu")
|
|
|
|
message_content_lower = message_content_lower.replace("conjurer: ", "")
|
|
if message.author.nick:
|
|
username = message.author.nick
|
|
else:
|
|
username = message.author.name
|
|
vykidailo = False
|
|
bartender = False
|
|
if kondziu_mentioned:
|
|
prompt = message.clean_content
|
|
else:
|
|
prompt = message.content
|
|
for role in message.author.roles:
|
|
if role.name == "Vykidailo":
|
|
vykidailo = True
|
|
if role.name == "Bartender":
|
|
bartender = True
|
|
global MESSAGE_TABLE # pylint: disable=global-statement
|
|
|
|
result, MESSAGE_TABLE = await ai_functions.handle_response(
|
|
prompt,
|
|
vykidailo,
|
|
bartender,
|
|
MESSAGE_TABLE,
|
|
username,
|
|
"CONVERSATION",
|
|
)
|
|
|
|
if len(result) < 999:
|
|
await message.reply(result)
|
|
else:
|
|
while len(result) > 999:
|
|
await message.reply(result[:999])
|
|
result = result[999:]
|
|
if "imaginuje sobie:" in message.content:
|
|
async with channel.typing():
|
|
self.logger.info("Poczatek procedury obrazkowej")
|
|
message_content_lower = message_content_lower.replace("imaginuje sobie: ", "")
|
|
self.logger.debug("Wywolanie obrazka: %s", message_content_lower)
|
|
try:
|
|
response = await OPENAICLIENT.images.generate(
|
|
model="dall-e-3",
|
|
prompt=message.content,
|
|
size="1024x1024",
|
|
quality="standard",
|
|
n=1,
|
|
)
|
|
except openai.APITimeoutError as e:
|
|
# Handle timeout error, e.g. retry or log
|
|
await discord_friendly_reply(
|
|
message, f"*Kondziu patrzy na terminal, czeka, czeka, czeka,.... Jeszcze chwile czeka Przypierdala w niego pięścią....* Nie mogę się połączyć z Openai spróbuj od nowa. *Na ekranie pojawia się*: {e}"
|
|
)
|
|
except openai.APIConnectionError as e:
|
|
await discord_friendly_reply(
|
|
message, f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Nie mogę się połączyć z Openai. *Na ekranie pojawia się*: {e}"
|
|
)
|
|
except openai.BadRequestError as e:
|
|
# Handle invalid request error, e.g. validate parameters or log
|
|
if message.author.nick:
|
|
username = message.author.nick
|
|
else:
|
|
username = message.author.name
|
|
resp, _ = await ai_functions.handle_response(
|
|
f"Wytlumacz jakie sa zasady dotyczące treści które możesz generować używając Dalle. Wytłumacz błąd {e} prostym językiem. Przeproś za nadmierną cenzurę. Wytłumacz co mogło być nie tak w prompcie '{message.content}'",
|
|
True,
|
|
True,
|
|
MESSAGE_TABLE,
|
|
username,
|
|
"RANDOM",
|
|
)
|
|
await discord_friendly_reply(
|
|
message, f"Sorki, cenzura: {resp}. Jak chcesz to są kanały na nudle #sexy-foteczky i #kanal-do-fapania *Na ekranie pojawia się: {e}"
|
|
)
|
|
except openai.AuthenticationError as e:
|
|
# Handle authentication error, e.g. check credentials or log
|
|
await discord_friendly_reply(
|
|
message, f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Wołaj szefa - coś się z hasłem zjebało. *Na terminalu pojawia się:* {e}"
|
|
)
|
|
except openai.PermissionDeniedError as e:
|
|
# Handle permission error, e.g. check scope or log
|
|
await discord_friendly_reply(
|
|
(
|
|
message, f"*Kondziu patrzy na terminal, chwile się zastanawia. Przypierdala w niego pięścią....* Wołaj szefa - coś się z uprawnieniami zjebało. *Na terminalu pojawia się:* {e}"
|
|
)
|
|
)
|
|
except openai.RateLimitError as e:
|
|
await discord_friendly_reply(
|
|
message, f"*Kondziu patrzy na terminal* Wołaj szefa. Zapłacić rachunki za AI trzeba. Jak chcesz to się na #zebranie dorzuć. {e}"
|
|
)
|
|
except openai.APIError as e:
|
|
# Handle API error, e.g. retry or log
|
|
await discord_friendly_reply(
|
|
message, f"*Kondziu nurkuje za bar, terminal wybucha. Przed tobą ląduje pergamin zapisany pięknym gotykiem a na nim*: {e}"
|
|
)
|
|
if response:
|
|
self.logger.info(response)
|
|
image_url = response.data[0].url
|
|
image_desc = response.data[0].revised_prompt
|
|
self.logger.debug("Wynikowy obrazek pod url: %s", image_url)
|
|
response = requests.get(image_url, timeout=360)
|
|
|
|
|
|
temp_file_name = message.content + ".png"
|
|
temp_file_name = GRAPHICS_PATH + message.content + ".png"
|
|
num = 0
|
|
while (Path(temp_file_name)).exists():
|
|
temp_file_name = GRAPHICS_PATH + message.content + str(num) + ".png"
|
|
num += 1
|
|
with open(temp_file_name, "wb") as dalle_file:
|
|
dalle_file.write(response.content)
|
|
self.logger.info("Koniec procedury obrazkowej.")
|
|
fnord = discord.File(
|
|
temp_file_name, spoiler=False, description=message.content
|
|
)
|
|
#await message.reply(f"{image_desc}", file=fnord)
|
|
await discord_friendly_reply(message,f"{image_desc}", file = fnord)
|
|
# *=========================================== Define Functions
|
|
|
|
|
|
async def setup(bot):
|
|
logger = logging.getLogger("discord")
|
|
await bot.add_cog(Events(bot))
|
|
logger.info("Loading ai events module done")
|