From a6a9ab5ebbd2685941fd9adc14d76773ca539934 Mon Sep 17 00:00:00 2001 From: Terry Hearst Date: Fri, 27 Dec 2024 01:40:14 -0500 Subject: [PATCH] (Mostly) restore bot back to pre data loss --- README.md | 10 ++++ bot.py | 125 +++++++++++++++++++++++++++++++++++++++++++++-- requirements.txt | 5 +- 3 files changed, 133 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index a3099a0..add0a76 100644 --- a/README.md +++ b/README.md @@ -25,3 +25,13 @@ python3 bot.py ``` Ask me for help if none of this makes sense lol + +## Generating token + +Link to generate twitch bot token: + +https://twitchtokengenerator.com/quick/n5zYtaZFY8 + +Link to generate pubsub token: + +https://twitchtokengenerator.com/quick/NWDIPn68zM diff --git a/bot.py b/bot.py index 4bc788e..818e073 100644 --- a/bot.py +++ b/bot.py @@ -3,6 +3,7 @@ import os import random import sqlite3 +import re from twitchio.ext import commands from twitchio.ext import pubsub from dotenv import load_dotenv @@ -10,6 +11,7 @@ import requests PREFIX = "!" DB_PATH = "storage.db" +CHESS_RE = re.compile("\\b(([KQRBN]?[a-h]?[1-8]?x?[a-h][1-8](=?[QRBNqrbn])?)|(O-O(-O)?))[+#]?\\b") class Bot(commands.Bot): @@ -24,7 +26,12 @@ class Bot(commands.Bot): commands = self.db_cur.execute("SELECT count(*) FROM custom_commands").fetchone() print(f"{commands[0]} custom commands available") - self.last_message = "cubes are just balls with corners lmao" + self.shuffle_message = "cubes are just balls with corners lmao" + + self.peer_pressure_message = "" + self.peer_pressure_users = [] + + self.hydrate_count = 0 # Adds a custom command from a string async def add_custom_command(self, ctx, command_str, author): @@ -64,6 +71,8 @@ class Bot(commands.Bot): @self.event() async def event_pubsub_channel_points(event: pubsub.PubSubChannelPointsMessage): reward_title = event.reward.title + if reward_title == "Hydrate!": # Hack for hydrate count... + self.hydrate_count += 1 result = self.db_cur.execute("SELECT output_text FROM channel_reward_messages WHERE title=?", [reward_title]).fetchone() if result is not None: message = result[0] @@ -76,6 +85,9 @@ class Bot(commands.Bot): if message.echo: return + if "CHESS_ADDR" in os.environ and "CHESS_TOKEN" in os.environ: + await self.handle_chess(message) + # I really want to do this with the add_command function and have no need for this event_message override, but # I cannot for the life of me figure out how to make "anonymous" coroutines (like, async lambda or something), # so I'm just manually handling the command here before passing to the command handler @@ -83,13 +95,50 @@ class Bot(commands.Bot): first_token = message.content[len(PREFIX):].strip().split()[0] res = self.db_cur.execute("SELECT output_text FROM custom_commands WHERE command=?", [first_token]).fetchone() if res is not None: - await message.channel.send(res[0]) + output = res[0] + output = output.replace("%USER%", message.author.name) + await message.channel.send(output) return + content = message.content.replace(u"\U000E0000", "").strip() + if content == self.peer_pressure_message: + if not message.author.name in self.peer_pressure_users: + self.peer_pressure_users.append(message.author.name) + if len(self.peer_pressure_users) >= 3: + await message.channel.send(content) + self.peer_pressure_users = [] + else: + self.peer_pressure_message = content + self.peer_pressure_users = [message.author.name] + await self.handle_commands(message) - if len(message.content.split(" ")) > 1: - self.last_message = message.content + if len(content.split(" ")) >= 3: + self.shuffle_message = content + + async def handle_chess(self, message): + # Don't do this if this is already a chess command (e.g. likely to have false positives when entering FEN) + if message.content.startswith(PREFIX) and message.content[len(PREFIX):].strip().split()[0] == "chess": + return + + # Check if the current message contains a chess move and play it + move_match = CHESS_RE.search(message.content) + if move_match is not None: + move = move_match.group() + headers = {"Authorization": f"Bearer {os.environ['CHESS_TOKEN']}"} + params = {"chatter": message.author.name, "fullMessage": message.content} + raw_response = requests.post(f"{os.environ['CHESS_ADDR']}/api/move/{move}", params=params, headers=headers) + response = raw_response.json() + if "error" in response: + await message.channel.send(f"Error: {response['error']}") + else: + output = f"Playing move {response['move']['san']}" + if response["draw"]: + output += ", game ends in a draw!" + elif response["checkmate"]: + output += f", {'black' if response['move']['color'] == 'b' else 'white'} wins by checkmate!" + output += " !chess" + await message.channel.send(output) @commands.command() async def hello(self, ctx: commands.Context): @@ -168,10 +217,76 @@ class Bot(commands.Bot): @commands.command() async def shuffle(self, ctx: commands.Context): - shuffled_message_list = self.last_message.split(" ") + shuffled_message_list = self.shuffle_message.split(" ") random.shuffle(shuffled_message_list) await ctx.send(" ".join(shuffled_message_list)) + # Requested by linguini15 + @commands.command() + async def shape(self, ctx: commands.Context): + platonic_solids = ["tetrahedron", "hexahedron", "octahedron", "dodecahedron", "icosahedron"] + solid = random.choice(platonic_solids) + message = f"did you guys know the marble is a{'n' if solid[0] in ['i', 'o'] else ''} {solid}!?!?!?" + await ctx.send(message) + + @commands.command() + async def dot(self, ctx: commands.Context): + responses = [ + "Oh, no way! It's another door!", + "You found a... What is this? I don't even...", + "Hey, listen!", + "Owls creep me out.", + "I can't remember what that is. I forget... My memory isn't what it used to be. I can't remember...", + "Hello, what's this? You found a secret passage! Who knows where it leads? I don't.", + "I'm as confused as you are. I have no idea.", + "It looks like a book. Yup! it's a book!", + "You have found a treasure map! A map of what? to where? I don't know! Figure it out yourself!", + "I wonder what this is...", + "I wonder what this means...", + "Something like, oh, I don't know...", + "...", + "Uh?", + "What does it mean?" + ] + await ctx.send(random.choice(responses)) + + # Requested by friendsupporter3829 + @commands.command() + async def hydratecount(self, ctx: commands.Context): + await ctx.send(f"Number of hydrates this stream: {self.hydrate_count}") + + # Requested by linguini15 + @commands.command() + async def calc(self, ctx: commands.Context): + args = ctx.message.content.split(" ", 1) + if len(args) < 2: + await ctx.send("Requires argument to calculate") # No it doesn't lmfao + return + + components_seed = random.random() + has_real = components_seed <= 0.85 + has_imaginary = components_seed >= 0.6 + + message = "" + if has_real: + real = random.triangular(-20, 20) + message += f"{real:.2f}" + if random.random() > 0.75: + message += "π" + if has_imaginary: + imag = random.triangular(-20, 20) + if has_real: + message += f" {'+' if imag >= 0 else '-'} " + imag = abs(imag) + message += f"{imag:.2f}" + if random.random() > 0.75: + message += "π" + message += "i" + + await ctx.send(message) + + # TODO - !weather (requested by linguini15), !chess + def main(): load_dotenv() diff --git a/requirements.txt b/requirements.txt index 38ecb64..5f4374b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ -twitchio==2.4.0 -python-dotenv==0.21.0 +twitchio==2.10.0 +python-dotenv==1.0.1 +requests==2.32.3