forked from HoloTech/customcmd-plugin
feat: Data and client framework
This commit is contained in:
@@ -0,0 +1,79 @@
|
||||
import twitchio
|
||||
|
||||
from customcmd.client import ExecClient
|
||||
from customcmd.data import CustomCmd, CustomCmdData
|
||||
from meta import Bot
|
||||
|
||||
from .context import CustomContext, CustomCmdUser
|
||||
|
||||
class TwitchCmdClient(ExecClient):
|
||||
def __init__(self, bot: Bot, data: CustomCmdData):
|
||||
super().__init__()
|
||||
self.bot = bot
|
||||
self.data = data
|
||||
|
||||
async def process_message(self, message: twitchio.ChatMessage):
|
||||
# Check the prefixes in this community
|
||||
# Or get the 'possible' prefixes in this context, which should be mostly cached
|
||||
prefixes = await self.get_prefixes(message)
|
||||
prefixes = sorted(prefixes, key=len, reverse=True)
|
||||
|
||||
content = message.text
|
||||
if message.reply:
|
||||
# Strip the hidden mention prefix from the start of the message
|
||||
prefix = message.reply.parent_user.mention
|
||||
content = content[len(prefix) + 1:]
|
||||
|
||||
# If we start with a valid prefix, parse out a command name
|
||||
if prefix := next((p for p in prefixes if content.startswith(p)), None):
|
||||
# Strip the valid prefix
|
||||
content = content[len(prefix):].strip()
|
||||
if not content:
|
||||
# Exit early, no command
|
||||
return
|
||||
|
||||
# Extract possible command word
|
||||
maybe_command, *args = content.split()
|
||||
maybe_command = maybe_command.lower()
|
||||
|
||||
# Lookup command words in this community (cache?)
|
||||
comm = await self.bot.profiles.fetch_community(message.broadcaster)
|
||||
cmdmap = await self.get_commands_in(comm.communityid)
|
||||
|
||||
if cmd := cmdmap.get(maybe_command):
|
||||
# We now finally have a command.
|
||||
# Read the arguments, build the context, run the command.
|
||||
# TODO: Check the permissions and log etc.
|
||||
# TODO: Exception handling, as usual
|
||||
# TODO: Nice string reprs
|
||||
ctx = CustomContext(
|
||||
bot=self.bot,
|
||||
args=args,
|
||||
alias=maybe_command,
|
||||
used_prefix=prefix,
|
||||
message=message,
|
||||
initial_globals={
|
||||
'author': CustomCmdUser(message.chatter),
|
||||
'channel': CustomCmdUser(message.broadcaster),
|
||||
}
|
||||
)
|
||||
result = await self.run_command(ctx, cmd.response)
|
||||
if result:
|
||||
await message.broadcaster.send_message(
|
||||
result,
|
||||
sender=self.bot.user
|
||||
)
|
||||
|
||||
async def get_prefixes(self, message: twitchio.ChatMessage):
|
||||
# TODO:
|
||||
return ('!',)
|
||||
|
||||
async def get_commands_in(self, communityid: int) -> dict[str, CustomCmd]:
|
||||
"""
|
||||
Get the custom commands in this community.
|
||||
|
||||
Returns a map of {alias -> CustomCommand}
|
||||
"""
|
||||
# TODO: Alias support
|
||||
commands = await CustomCmd.fetch_where(communityid=communityid)
|
||||
return {cmd.name: cmd for cmd in commands}
|
||||
@@ -0,0 +1,98 @@
|
||||
from typing import Optional
|
||||
|
||||
import twitchio
|
||||
from twitchio import PartialUser
|
||||
from twitchio.ext import commands as cmds
|
||||
|
||||
from customcmd.parser.context import ExecContext
|
||||
from meta import Bot
|
||||
from twitch.client import TwitchCmdClient
|
||||
from utils.lib import utc_now
|
||||
|
||||
from . import logger
|
||||
from ..customcmd.data import CustomCmd, CustomCmdData
|
||||
from ..customcmd.client import ExecClient
|
||||
|
||||
# TODO: Make an alphacroc or croccyalpha client for testing
|
||||
# TODO: Fix the tracker not actually guarding the commands properly
|
||||
|
||||
|
||||
|
||||
|
||||
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(args=[], content=''), response)
|
||||
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)
|
||||
|
||||
# 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!")
|
||||
@@ -0,0 +1,68 @@
|
||||
import twitchio
|
||||
|
||||
from customcmd.client import ExecClient
|
||||
from customcmd.data import CustomCmd, CustomCmdData
|
||||
from meta import Bot
|
||||
from utils.lib import utc_now, strfdur
|
||||
|
||||
from ..customcmd.parser import ExecContext
|
||||
from .ctxvars import stuff
|
||||
|
||||
|
||||
class CustomCmdUser:
|
||||
def __init__(self, user: twitchio.PartialUser):
|
||||
self.user = user
|
||||
|
||||
@property
|
||||
def _exec_attr_id(self):
|
||||
return self.user.id
|
||||
|
||||
@property
|
||||
def _exec_attr_name(self):
|
||||
return self.user.display_name
|
||||
|
||||
def __str__(self):
|
||||
return self.user.display_name or self.user.name or self.user.id
|
||||
|
||||
async def _exec_attr_whisper(self, ctx, message: str):
|
||||
await self.user.send_whisper(
|
||||
to_user=self.user,
|
||||
message=message
|
||||
)
|
||||
|
||||
async def _exec_attr_followage(self, ctx, channel: twitchio.PartialUser):
|
||||
followed_at = None
|
||||
query = await channel.fetch_followers(user=self.user)
|
||||
async for follower in query.followers:
|
||||
followed_at = follower.followed_at
|
||||
|
||||
if followed_at:
|
||||
durstr = strfdur(utc_now() - followed_at)
|
||||
resp = f"{self.user.mention} has been following {channel.mention} for {durstr}"
|
||||
else:
|
||||
return f"{self.user.mention} is not following {channel.mention}"
|
||||
|
||||
@classmethod
|
||||
def from_user(cls, user: twitchio.PartialUser):
|
||||
return cls(user)
|
||||
|
||||
|
||||
class CustomContext(ExecContext):
|
||||
"""
|
||||
CustomCommand ExecContext instantiated from a Twitch message
|
||||
"""
|
||||
def __init__(
|
||||
self,
|
||||
bot: Bot,
|
||||
message: twitchio.ChatMessage,
|
||||
alias: str,
|
||||
used_prefix: str,
|
||||
**kwargs):
|
||||
super().__init__(**kwargs)
|
||||
self.bot = bot
|
||||
self.message = message
|
||||
self.alias = alias
|
||||
self.used_prefix = used_prefix
|
||||
|
||||
self.enter_scope(stuff)
|
||||
# TODO: Need a nice way to define some global functions. Registry pattern?
|
||||
@@ -0,0 +1,30 @@
|
||||
import random
|
||||
from typing import Optional
|
||||
|
||||
|
||||
stuff = {}
|
||||
|
||||
def register_var(name: str | None = None):
|
||||
def wrapper(coro):
|
||||
varname = name or coro.__name__
|
||||
stuff[varname] = coro
|
||||
return coro
|
||||
return wrapper
|
||||
|
||||
|
||||
@register_var('random')
|
||||
async def random_func(ctx, a: int, b: Optional[int] = None):
|
||||
if b is not None:
|
||||
low, high = a, b
|
||||
else:
|
||||
low, high = 0, a
|
||||
return random.randint(low, high)
|
||||
|
||||
@register_var('user')
|
||||
async def user_func(ctx, userstr: int | str):
|
||||
"""
|
||||
Attempt to turn the given userstr into a user.
|
||||
Currently just looks up the user directly.
|
||||
Some fuzzy matching may be appropriate in the future.
|
||||
"""
|
||||
...
|
||||
Reference in New Issue
Block a user