(Mostly) restore bot back to pre data loss

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

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:
+		if len(content.split(" ")) >= 3:
-			self.last_message = message.content
+			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()

requirements.txt

@@ -1,2 +1,3 @@
-twitchio==2.4.0
+twitchio==2.10.0
-python-dotenv==0.21.0
+python-dotenv==1.0.1
+requests==2.32.3