from typing import Optional import twitchio from twitchio import PartialUser from twitchio.ext import commands as cmds from meta import Bot from utils.lib import utc_now from ..customcmd.data import CustomCmd, CustomCmdData from ..customcmd.client import ExecClient from ..customcmd.parser import ExecContext from . import logger from .client import TwitchCmdClient # TODO: Make an alphacroc or croccyalpha client for testing # TODO: Fix the tracker not actually guarding the commands properly # TODO: Fix suffix of result not printing # TODO When command produces None don't stringify to 'None' class CustomCmdComponent(cmds.Component): def __init__(self, bot: Bot): self.bot = bot self.data = bot.dbconn.load_registry(CustomCmdData()) self.cmdclient = TwitchCmdClient(self.bot, self.data) # ----- API ----- async def component_load(self): await self.data.init() await self.bot.version_check(*self.data.VERSION) async def component_teardown(self): pass # ----- Methods ----- # ----- Event handlers ----- @cmds.Component.listener() async def event_message(self, payload: twitchio.ChatMessage): await self.cmdclient.process_message(payload) # ----- Commands ----- @cmds.group(name="commands") async def cmd_grp(self, ctx: cmds.Context): # TODO: Probably attempt to list commands or something here. pass @cmd_grp.command(name="add", aliases=("create", "new")) @cmds.is_broadcaster() # TODO: Better command creation perm checks async def cmd_add(self, ctx: cmds.Context, name: str, *, response: str): comm = await self.bot.profiles.fetch_community(ctx.broadcaster) cid = comm.communityid # Make sure the command doesn't already exist cmdmap = await self.cmdclient.get_commands_in(cid) if name.lower() in cmdmap: await ctx.reply(f"The command {name} is already defined!") return # Make sure the command parses (can't guarantee it runs, but at least it can parse) try: await self.cmdclient.run_command( ExecContext(alias="testing", args=[], content=""), response, dryrun=True ) except Exception: # TODO: Nicer errors for pinpointing problem await ctx.reply(f"Sorry, could not parse your command!") logger.info( f"Failed to parse attempted customcmd '{response}'", exc_info=True ) return # If everything is good, add it to the table # TODO: Again, permission resolution author_profile = await self.bot.profiles.fetch_profile(ctx.chatter) await CustomCmd.create( communityid=cid, name=name.lower(), response=response, permlevel=0, creatorid=author_profile.profileid, ) await ctx.reply(f"Created a new command !{name}") @cmd_grp.command(name="del", aliases=("delete", "remove", "rm")) @cmds.is_broadcaster() async def cmd_del(self, ctx: cmds.Context, name: str): # Lookup the name to make sure a command does exist # Delete the command (again, consider permissions later) # Confirm with user. comm = await self.bot.profiles.fetch_community(ctx.broadcaster) cid = comm.communityid # Make sure the command already exists cmdmap = await self.cmdclient.get_commands_in(cid) if name.lower() in cmdmap: cmd = cmdmap[name.lower()] await cmd.delete() await ctx.reply(f"Deleted the command !{name}") else: await ctx.reply(f"The command {name} is already defined!")