337 lines
No EOL
10 KiB
Python
337 lines
No EOL
10 KiB
Python
import os
|
|
import re
|
|
import json
|
|
import time
|
|
import random
|
|
import traceback
|
|
import subprocess
|
|
import tempfile
|
|
import shutil
|
|
import sys
|
|
from typing import (
|
|
List,
|
|
Dict,
|
|
Literal,
|
|
Set,
|
|
Optional,
|
|
Tuple,
|
|
Any,
|
|
TypedDict,
|
|
Union,
|
|
Callable,
|
|
Coroutine,
|
|
TypeVar,
|
|
Type,
|
|
)
|
|
import logging
|
|
from modules.prestartchecks import start_checks
|
|
from modules.logger import GooberFormatter
|
|
from modules.volta.main import *
|
|
import logging
|
|
from modules.settings import Settings as SettingsManager
|
|
from modules.permission import requires_admin
|
|
from modules.volta.main import _
|
|
|
|
logger = logging.getLogger("goober")
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(logging.DEBUG)
|
|
console_handler.setFormatter(GooberFormatter())
|
|
|
|
file_handler = logging.FileHandler("log.txt", mode="w+", encoding="UTF-8")
|
|
file_handler.setLevel(logging.DEBUG)
|
|
file_handler.setFormatter(GooberFormatter(colors=False))
|
|
|
|
logger.addHandler(console_handler)
|
|
logger.addHandler(file_handler)
|
|
|
|
settings_manager = SettingsManager()
|
|
settings = settings_manager.settings
|
|
|
|
splash_text: str = ""
|
|
|
|
with open(settings["splash_text_loc"], "r", encoding="UTF-8") as f:
|
|
splash_text = "".join(f.readlines())
|
|
print(splash_text)
|
|
|
|
start_checks()
|
|
|
|
import discord
|
|
from discord.ext import commands
|
|
from discord import app_commands
|
|
from discord import Colour, Message
|
|
|
|
from better_profanity import profanity
|
|
from discord.ext import commands
|
|
|
|
from modules.markovmemory import *
|
|
from modules.sentenceprocessing import *
|
|
from modules.unhandledexception import handle_exception
|
|
|
|
sys.excepthook = handle_exception
|
|
|
|
|
|
class MessageMetadata(TypedDict):
|
|
user_id: str
|
|
user_name: str
|
|
guild_id: str | Literal["DM"]
|
|
guild_name: str | Literal["DM"]
|
|
channel_id: str
|
|
channel_name: str
|
|
message: str
|
|
timestamp: float
|
|
|
|
|
|
# Constants with type hints
|
|
positive_gifs: List[str] = settings["bot"]["misc"]["positive_gifs"]
|
|
currenthash: str = ""
|
|
launched: bool = False
|
|
slash_commands_enabled: bool = False
|
|
|
|
# Set up Discord bot intents and create bot instance
|
|
intents: discord.Intents = discord.Intents.default()
|
|
intents.messages = True
|
|
intents.message_content = True
|
|
|
|
bot: commands.Bot = commands.Bot(
|
|
command_prefix=settings["bot"]["prefix"],
|
|
intents=intents,
|
|
allowed_mentions=discord.AllowedMentions(
|
|
everyone=False, roles=False, users=False, replied_user=True
|
|
),
|
|
)
|
|
|
|
# Load memory and Markov model for text generation
|
|
memory: List[str | Dict[Literal["_meta"], MessageMetadata]] = load_memory()
|
|
markov_model: Optional[markovify.Text] = load_markov_model()
|
|
if not markov_model:
|
|
logger.error(_('markov_model_not_found'))
|
|
memory = load_memory()
|
|
markov_model = train_markov_model(memory)
|
|
|
|
generated_sentences: Set[str] = set()
|
|
used_words: Set[str] = set()
|
|
|
|
|
|
async def load_cogs_from_folder(bot: commands.Bot, folder_name="assets/cogs"):
|
|
for filename in [file for file in os.listdir(folder_name) if file.endswith(".py")]:
|
|
cog_name: str = filename[:-3]
|
|
|
|
if (
|
|
"internal" not in folder_name
|
|
and cog_name not in settings["bot"]["enabled_cogs"]
|
|
):
|
|
logger.debug(f"Skipping cog {cog_name} (not in enabled cogs)")
|
|
continue
|
|
|
|
module_path = folder_name.replace("/", ".").replace("\\", ".") + f".{cog_name}"
|
|
|
|
try:
|
|
await bot.load_extension(module_path)
|
|
logger.info(f"{_('loaded_cog')} {cog_name}")
|
|
except Exception as e:
|
|
logger.error(f"{_('cog_fail')} {cog_name} {e}")
|
|
traceback.print_exc()
|
|
|
|
|
|
# Event: Called when the bot is ready
|
|
@bot.event
|
|
async def on_ready() -> None:
|
|
global launched
|
|
|
|
folder_name: str = "cogs"
|
|
if launched:
|
|
return
|
|
|
|
await load_cogs_from_folder(bot)
|
|
await load_cogs_from_folder(bot, "assets/cogs/internal")
|
|
try:
|
|
synced: List[discord.app_commands.AppCommand] = await bot.tree.sync()
|
|
|
|
logger.info(f"{_('synced_commands')} {len(synced)} {_('synced_commands2')}")
|
|
logger.info(_('started'))
|
|
|
|
except discord.errors.Forbidden as perm_error:
|
|
logger.error(f"Permission error while syncing commands: {perm_error}")
|
|
logger.error(
|
|
"Make sure the bot has the 'applications.commands' scope and is invited with the correct permissions."
|
|
)
|
|
quit()
|
|
except Exception as e:
|
|
logger.error(f"{_('fail_commands_sync')} {e}")
|
|
traceback.print_exc()
|
|
quit()
|
|
|
|
if not settings["bot"]["misc"]["active_song"]:
|
|
return
|
|
await bot.change_presence(
|
|
activity=discord.Activity(
|
|
type=discord.ActivityType.listening,
|
|
name=settings["bot"]["misc"]["active_song"],
|
|
)
|
|
)
|
|
launched = True
|
|
|
|
bot.remove_command('help')
|
|
# Command: Show help information
|
|
@bot.hybrid_command(description=f"{(_('command_desc_help'))}")
|
|
async def help(ctx: commands.Context) -> None:
|
|
embed: discord.Embed = discord.Embed(
|
|
title=f"{(_('command_help_embed_title'))}",
|
|
description=f"{(_('command_help_embed_desc'))}",
|
|
color=Colour(0x000000)
|
|
)
|
|
|
|
command_categories: Dict[str, List[str]] = {
|
|
f"{(_('command_help_categories_general'))}": ["mem", "talk", "about", "ping", "impact", "demotivator", "help"],
|
|
f"{(_('command_help_categories_admin'))}": ["stats", "retrain", "setlanguage"]
|
|
}
|
|
|
|
custom_commands: List[str] = []
|
|
for cog_name, cog in bot.cogs.items():
|
|
for command in cog.get_commands():
|
|
if command.name not in command_categories[f"{(_('command_help_categories_general'))}"] and command.name not in command_categories[f"{(_('command_help_categories_admin'))}"]:
|
|
custom_commands.append(command.name)
|
|
|
|
if custom_commands:
|
|
embed.add_field(name=f"{(_('command_help_categories_custom'))}", value="\n".join([f"{PREFIX}{command}" for command in custom_commands]), inline=False)
|
|
|
|
for category, commands_list in command_categories.items():
|
|
commands_in_category: str = "\n".join([f"{PREFIX}{command}" for command in commands_list])
|
|
embed.add_field(name=category, value=commands_in_category, inline=False)
|
|
|
|
await send_message(ctx, embed=embed)
|
|
|
|
@bot.event
|
|
async def on_command_error(ctx: commands.Context, error: commands.CommandError) -> None:
|
|
from modules.unhandledexception import handle_exception
|
|
|
|
if isinstance(error, commands.CommandInvokeError):
|
|
original: Exception = error.original
|
|
handle_exception(
|
|
type(original),
|
|
original,
|
|
original.__traceback__,
|
|
context=f"Command: {ctx.command} | User: {ctx.author}",
|
|
)
|
|
else:
|
|
handle_exception(
|
|
type(error),
|
|
error,
|
|
error.__traceback__,
|
|
context=f"Command: {ctx.command} | User: {ctx.author}",
|
|
)
|
|
|
|
# Event: Called on every message
|
|
@bot.event
|
|
async def on_message(message: discord.Message) -> None:
|
|
global memory, markov_model
|
|
EMOJIS = [
|
|
"\U0001f604",
|
|
"\U0001f44d",
|
|
"\U0001f525",
|
|
"\U0001f4af",
|
|
"\U0001f389",
|
|
"\U0001f60e",
|
|
] # originally was emojis but it would probably shit itself on systems without unicode so....
|
|
if message.author.bot:
|
|
return
|
|
|
|
if str(message.author.id) in settings["bot"]["blacklisted_users"]:
|
|
return
|
|
|
|
commands = [
|
|
settings["bot"]["prefix"] + command.name for command in bot.tree.get_commands()
|
|
]
|
|
|
|
if message.content.startswith(tuple(commands)):
|
|
logger.info(f"{(_('command_ran')).format(message=message)}")
|
|
await bot.process_commands(message)
|
|
return
|
|
|
|
if (
|
|
profanity.contains_profanity(message.content)
|
|
and settings["bot"]["misc"]["block_profanity"]
|
|
):
|
|
return
|
|
|
|
if message.content:
|
|
if not settings["bot"]["user_training"]:
|
|
return
|
|
|
|
formatted_message: str = message.content
|
|
cleaned_message: str = preprocess_message(formatted_message)
|
|
if cleaned_message:
|
|
memory.append(cleaned_message)
|
|
|
|
message_metadata: MessageMetadata = {
|
|
"user_id": str(message.author.id),
|
|
"user_name": str(message.author),
|
|
"guild_id": str(message.guild.id) if message.guild else "DM",
|
|
"guild_name": str(message.guild.name) if message.guild else "DM",
|
|
"channel_id": str(message.channel.id),
|
|
"channel_name": str(message.channel),
|
|
"message": message.content,
|
|
"timestamp": time.time(),
|
|
}
|
|
try:
|
|
if isinstance(memory, list):
|
|
memory.append({"_meta": message_metadata})
|
|
else:
|
|
logger.warning("Memory is not a list; can't append metadata")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to append metadata to memory: {e}")
|
|
|
|
save_memory(memory)
|
|
|
|
sentiment_score = is_positive(
|
|
message.content
|
|
) # doesnt work but im scared to change the logic now please ignore
|
|
if sentiment_score > 0.8:
|
|
if not settings["bot"]["react_to_messages"]:
|
|
return
|
|
emoji = random.choice(EMOJIS)
|
|
try:
|
|
await message.add_reaction(emoji)
|
|
except Exception as e:
|
|
logger.info(f"Failed to react with emoji: {e}")
|
|
|
|
await bot.process_commands(message)
|
|
|
|
|
|
# Event: Called on every interaction (slash command, etc.)
|
|
@bot.event
|
|
async def on_interaction(interaction: discord.Interaction) -> None:
|
|
logger.info(f"{(_('command_ran_s')).format(interaction=interaction)}{name}")
|
|
|
|
|
|
# Global check: Block blacklisted users from running commands
|
|
@bot.check
|
|
async def block_blacklisted(ctx: commands.Context) -> bool:
|
|
if ctx.author.id not in settings["bot"]["blacklisted_users"]:
|
|
return True
|
|
|
|
try:
|
|
if isinstance(ctx, discord.Interaction):
|
|
if not ctx.response.is_done():
|
|
await ctx.response.send_message(_('blacklisted'), ephemeral=True)
|
|
else:
|
|
await ctx.followup.send(_('blacklisted'), ephemeral=True)
|
|
else:
|
|
await ctx.send(_('blacklisted_user'), ephemeral=True)
|
|
except:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
# Helper: Improve sentence coherence (simple capitalization fix)
|
|
def improve_sentence_coherence(sentence: str) -> str:
|
|
# Capitalizes "i" to "I" in the sentence
|
|
sentence = sentence.replace(" i ", " I ")
|
|
return sentence
|
|
|
|
# Start the bot
|
|
if __name__ == "__main__":
|
|
bot.run(os.environ.get("DISCORDBOTTOKEN", "")) |