From 2f40ca82b93787ecc7c254c8da47dc61056369d0 Mon Sep 17 00:00:00 2001 From: salmonik Date: Mon, 23 Mar 2026 17:43:24 +0100 Subject: triggers --- cogs/trigger.py | 305 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 305 insertions(+) create mode 100644 cogs/trigger.py diff --git a/cogs/trigger.py b/cogs/trigger.py new file mode 100644 index 0000000..daf9e7b --- /dev/null +++ b/cogs/trigger.py @@ -0,0 +1,305 @@ +import os + +import discord +from discord.ext import commands +import random +import sqlite3 +import logging + +logger = logging.getLogger(__name__) + + +DB_PATH = 'data/triggers.db' + + +def get_connection(): + return sqlite3.connect(DB_PATH) + + +class Trigger(commands.Cog): + def __init__(self, bot): + self.bot = bot + self.db_check() + + trigger_group = discord.SlashCommandGroup("trigger", "trigger commands") + + @staticmethod + def db_check(): + if not os.path.exists('data'): + os.makedirs('data') + + db_exists = os.path.exists(DB_PATH) + if not db_exists: + logger.warning("Triggers database not found. Creating new database.") + + try: + conn = sqlite3.connect(DB_PATH) + c = conn.cursor() + + # Enable foreign key constraints + c.execute('PRAGMA foreign_keys = ON') + + # Create triggers table + c.execute(''' + CREATE TABLE IF NOT EXISTS triggers ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL UNIQUE, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )''') + + # Create trigger_keywords table with foreign key constraint + c.execute(''' + CREATE TABLE IF NOT EXISTS trigger_keywords ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + trigger_id INTEGER NOT NULL, + keyword TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (trigger_id) REFERENCES triggers(id) ON DELETE CASCADE, + UNIQUE(trigger_id, keyword) + )''') + + # Create trigger_responses table with foreign key constraint + c.execute(''' + CREATE TABLE IF NOT EXISTS trigger_responses ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + trigger_id INTEGER NOT NULL, + response_text TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (trigger_id) REFERENCES triggers(id) ON DELETE CASCADE + )''') + + # Create indexes for better query performance + c.execute('CREATE INDEX IF NOT EXISTS idx_trigger_keywords_trigger_id ON trigger_keywords(trigger_id)') + c.execute('CREATE INDEX IF NOT EXISTS idx_trigger_keywords_keyword ON trigger_keywords(keyword)') + c.execute('CREATE INDEX IF NOT EXISTS idx_trigger_responses_trigger_id ON trigger_responses(trigger_id)') + + # Run integrity check + integrity_check = c.execute('PRAGMA integrity_check').fetchone() + if integrity_check and integrity_check[0] == 'ok': + logger.info("Triggers database integrity check passed") + else: + logger.warning(f"Triggers database integrity check: {integrity_check}") + + conn.commit() + + except sqlite3.Error as e: + logger.error(f"Triggers database error: {e}") + raise + + finally: + if 'conn' in locals(): + conn.close() + + if not db_exists: + logger.info("Triggers database initialized successfully") + + @commands.Cog.listener() + async def on_message(self, message): + if message.author == self.bot.user: + return + + await self.trigger(message) + + @staticmethod + async def trigger(message): + content = message.content.lower() + with get_connection() as conn: + c = conn.cursor() + c.execute("SELECT id, name FROM triggers") + for trigger_id, name in c.fetchall(): + c.execute("SELECT keyword FROM trigger_keywords WHERE trigger_id=?", (trigger_id,)) + keywords = [row[0] for row in c.fetchall()] + if any(keyword in content for keyword in keywords): + c.execute("SELECT response_text FROM trigger_responses WHERE trigger_id=?", (trigger_id,)) + responses = [row[0] for row in c.fetchall()] + if responses: + await message.channel.send(random.choice(responses).replace('\\n', '\n')) + logger.info(f"{message.author} triggered {name}") + break + + @staticmethod + def get_type(name): + with sqlite3.connect(DB_PATH) as conn: + c = conn.cursor() + c.execute("SELECT name FROM triggers") + return [row[0] for row in c.fetchall()] + + @trigger_group.command( + description="Create a new trigger", + ) + async def create( + self, + ctx: discord.ApplicationContext, + name = discord.Option( + str, + "the name of the trigger", + autocomplete= discord.utils.basic_autocomplete(get_type), + required=True + ) + ): + with get_connection() as conn: + c = conn.cursor() + c.execute("INSERT OR IGNORE INTO triggers (name) VALUES (?)", (name,)) + conn.commit() + + await ctx.respond("trigger added") + + @trigger_group.command( + description="Add a trigger or response to the name", + ) + async def add( + self, + ctx: discord.ApplicationContext, + name = discord.Option( + str, + "the name of the trigger", + autocomplete= discord.utils.basic_autocomplete(get_type), + required=True + ), + trigger = discord.Option( + str, + "the trigger to add to the name", + required=False + ), + response = discord.Option( + str, + "the response to add to the name", + required=False + ) + ): + with get_connection() as conn: + c = conn.cursor() + c.execute("SELECT id FROM triggers WHERE name=?", (name,)) + row = c.fetchone() + if not row: + await ctx.respond("Trigger not found.") + logger.warning(f"Trigger not found: {name}") + return + trigger_id = row[0] + if trigger: + c.execute("INSERT INTO trigger_keywords (trigger_id, keyword) VALUES (?, ?)", (trigger_id, trigger)) + if response: + c.execute("INSERT INTO trigger_responses (trigger_id, response_text) VALUES (?, ?)", (trigger_id, response)) + conn.commit() + + await ctx.respond("trigger updated") + logger.info(f"Trigger updated: {name}") + + @trigger_group.command( + description="Remove a trigger or response from the name", + ) + async def remove( + self, + ctx: discord.ApplicationContext, + name = discord.Option( + str, + "the name of the trigger", + autocomplete= discord.utils.basic_autocomplete(get_type), + required=True + ), + trigger = discord.Option( + str, + "the trigger to remove from the name", + required=False + ), + response = discord.Option( + str, + "the response to remove from the name", + required=False + ) + ): + with get_connection() as conn: + c = conn.cursor() + c.execute("SELECT id FROM triggers WHERE name=?", (name,)) + row = c.fetchone() + if not row: + await ctx.respond("Trigger not found.") + logger.warning(f"Trigger not found: {name}") + return + trigger_id = row[0] + if trigger: + c.execute("DELETE FROM trigger_keywords WHERE trigger_id=? AND keyword=?", (trigger_id, trigger)) + if response: + c.execute("DELETE FROM trigger_responses WHERE trigger_id=? AND response_text=?", (trigger_id, response)) + conn.commit() + + await ctx.respond("trigger removed") + logger.info(f"Trigger removed: {name}") + + @trigger_group.command( + description="Delete the whole trigger", + ) + async def delete( + self, + ctx: discord.ApplicationContext, + name = discord.Option( + str, + "the name of the trigger", + autocomplete= discord.utils.basic_autocomplete(get_type), + required=True + ) + ): + with get_connection() as conn: + c = conn.cursor() + c.execute("SELECT id FROM triggers WHERE name=?", (name,)) + row = c.fetchone() + if not row: + await ctx.respond("Trigger not found.") + return + trigger_id = row[0] + c.execute("DELETE FROM trigger_keywords WHERE trigger_id=?", (trigger_id,)) + c.execute("DELETE FROM trigger_responses WHERE trigger_id=?", (trigger_id,)) + c.execute("DELETE FROM triggers WHERE id=?", (trigger_id,)) + conn.commit() + + await ctx.respond("trigger deleted") + logger.info(f"Trigger deleted: {name}") + + + @trigger_group.command( + description="List all triggers", + ) + async def list( + self, + ctx: discord.ApplicationContext, + name = discord.Option( + str, + "the name of the trigger", + autocomplete=discord.utils.basic_autocomplete(get_type), + required=True + ) + ): + with get_connection() as conn: + c = conn.cursor() + if name: + c.execute("SELECT id FROM triggers WHERE name=?", (name,)) + row = c.fetchone() + if not row: + await ctx.respond("Trigger not found.") + logger.warning(f"Trigger not found: {name}") + return + trigger_id = row[0] + c.execute("SELECT keyword FROM trigger_keywords WHERE trigger_id=?", (trigger_id,)) + triggers = [row[0] for row in c.fetchall()] + c.execute("SELECT response_text FROM trigger_responses WHERE trigger_id=?", (trigger_id,)) + responses = [row[0] for row in c.fetchall()] + trigger_list = "\n\n".join(triggers) + response_list = "\n\n".join(responses) + message = f"**Trigger: {name}**\n**Triggers:**\n{trigger_list}\n**Responses:**\n{response_list}" + await ctx.respond(message) + logger.info(f"Trigger listed: {name}") + else: + c.execute("SELECT id, name FROM triggers") + messages = [] + for trigger_id, name in c.fetchall(): + c.execute("SELECT keyword FROM trigger_keywords WHERE trigger_id=?", (trigger_id,)) + triggers = ", ".join([row[0] for row in c.fetchall()]) + c.execute("SELECT response_text FROM trigger_responses WHERE trigger_id=?", (trigger_id,)) + responses = ", ".join([row[0] for row in c.fetchall()]) + messages.append(f"**{name}**\nTriggers: {triggers}\nResponses: {responses}") + await ctx.respond("\n\n".join(messages)) + logger.info(f"Triggers listed") + +def setup(bot): + bot.add_cog(Trigger(bot)) \ No newline at end of file -- cgit v1.2.3