path: root/cogs/trigger.py

# 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 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:
                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,
            "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".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 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")

    @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))