From cd8ed8bc5793985df2c1692072a932ceee517b95 Mon Sep 17 00:00:00 2001 From: salmonik Date: Wed, 8 Jul 2026 02:13:35 +0200 Subject: OPEN THE RUST GATES Signed-off-by: salmonik --- cogs/trigger.py | 260 -------------------------------------------------------- 1 file changed, 260 deletions(-) delete mode 100644 cogs/trigger.py (limited to 'cogs/trigger.py') diff --git a/cogs/trigger.py b/cogs/trigger.py deleted file mode 100644 index ce80d43..0000000 --- a/cogs/trigger.py +++ /dev/null @@ -1,260 +0,0 @@ -# This file is part of Senator Sperma -# Licensed under the GPL-3.0 License -# Copyright (C) 2026 Salmonik @ salmonik@tuta.io -# https://codeberg.org/salmonik/senator-sperma - -import os -import discord -from discord.ext import commands -import random -import logging - -from cogs.utils.database import get_connection, init_database - -logger = logging.getLogger(__name__) - -init_database() - -_SELECT_KEYWORD_SQL = "SELECT keyword FROM trigger_keywords WHERE trigger_id=?" -_SELECT_RESPONSE_SQL = "SELECT response_text FROM trigger_responses WHERE trigger_id=?" -_SELECT_TRIGGER_ID_SQL = "SELECT id FROM triggers WHERE name=?" -_TRIGGER_NOT_FOUND_MSG = "Trigger not found." -_TRIGGER_NAME_OPTION = "the name of the trigger" - - -class Trigger(commands.Cog): - def __init__(self, bot): - self.bot = bot - - trigger_group = discord.SlashCommandGroup("trigger", "trigger commands") - - @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_SQL, (trigger_id,)) - keywords = [row[0] for row in c.fetchall()] - if any(keyword in content for keyword in keywords): - c.execute(_SELECT_RESPONSE_SQL, (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 get_connection() 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, - _TRIGGER_NAME_OPTION, - 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, - _TRIGGER_NAME_OPTION, - 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_TRIGGER_ID_SQL, (name,)) - row = c.fetchone() - if not row: - await ctx.respond(_TRIGGER_NOT_FOUND_MSG) - logger.warning(f"Trigger not found: {name}") - return - trigger_id = row[0] - if trigger: - for keyword in trigger.split('\\'): - keyword = keyword.strip() - if keyword: - c.execute("INSERT INTO trigger_keywords (trigger_id, keyword) VALUES (?, ?)", (trigger_id, keyword)) - if response: - for resp in response.split('\\'): - resp = resp.strip() - if resp: - c.execute("INSERT INTO trigger_responses (trigger_id, response_text) VALUES (?, ?)", (trigger_id, resp)) - 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, - _TRIGGER_NAME_OPTION, - 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_TRIGGER_ID_SQL, (name,)) - row = c.fetchone() - if not row: - await ctx.respond(_TRIGGER_NOT_FOUND_MSG) - 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, - _TRIGGER_NAME_OPTION, - autocomplete= discord.utils.basic_autocomplete(get_type), - required=True - ) - ): - with get_connection() as conn: - c = conn.cursor() - c.execute(_SELECT_TRIGGER_ID_SQL, (name,)) - row = c.fetchone() - if not row: - await ctx.respond(_TRIGGER_NOT_FOUND_MSG) - 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, - _TRIGGER_NAME_OPTION, - 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_MSG) - logger.warning(f"Trigger not found: {name}") - return - trigger_id = row[0] - c.execute(_SELECT_KEYWORD_SQL, (trigger_id,)) - triggers = [row[0] for row in c.fetchall()] - c.execute(_SELECT_RESPONSE_SQL, (trigger_id,)) - responses = [row[0] for row in c.fetchall()] - trigger_list = "\n".join(triggers) - response_list = "\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_SQL, (trigger_id,)) - triggers = ", ".join([row[0] for row in c.fetchall()]) - c.execute(_SELECT_RESPONSE_SQL, (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("Triggers listed") - - @trigger_group.command( - description="how to use trigger group?" - ) - async def help( - self, - ctx: discord.ApplicationContext, - ): - await ctx.respond("Trigger create - creates a group for triggers/responses\n" - "Trigger add - adds triggers/responses to a group, \\ is the delimeter\n" - "Trigger remove - removes triggers/responses from a group, \\ is the delimeter\n" - "Trigger delete - deletes the whole group\n" - "Trigger list - lists triggers/responses of a group\n" - "Trigger help - display this message\n") - - -def setup(bot): - bot.add_cog(Trigger(bot)) -- cgit v1.2.3