from typing import Optional import random import twitchio from twitchio.ext import commands as cmds from meta import Bot, Context from . import logger from ..koans import KoanRegistry from ..data import KoansData from ..lib import minify class KoanComponent(cmds.Component): def __init__(self, bot: Bot): self.bot = bot self.data = bot.dbconn.load_registry(KoansData()) self.koans = KoanRegistry(self.data) async def component_load(self): await self.data.init() await self.bot.version_check(*self.data.VERSION) await self.koans.init() @cmds.Component.listener() async def event_message(self, payload: twitchio.ChatMessage) -> None: print(f"[{payload.broadcaster.name}] - {payload.chatter.name}: {payload.text}") async def component_command_error(self, payload): try: raise payload.exception except cmds.ArgumentError: if cmd := payload.context.command: usage = f"{cmd.qualified_name} {cmd.signature}" await payload.context.reply(f"USAGE: {usage}") return False except Exception: pass @cmds.group(invoke_fallback=True) async def koans(self, ctx: cmds.Context) -> None: """ List the (names of the) koans in this channel. !koans """ community = await self.bot.profiles.fetch_community(ctx.broadcaster) cid = community.communityid koans = await self.koans.get_community_koans(cid) if koans: count = len(koans) parts = [] for koan in koans[-10:]: minified = minify(koan.content, 20) formatted = f"#{koan.koanlabel}: {minified}" parts.append(formatted) partstr = '; '.join(parts) laststr = "Last 10: " if count > 10 else "" message = f"We have {count} koans! {laststr}{partstr}" await ctx.reply(message) else: await ctx.reply("No koans have been made in this channel!") @koans.command(name='add', aliases=['new', 'create']) @cmds.is_moderator() async def koans_add(self, ctx: cmds.Context, *, content: str): """ Add or overwrite a koan to this channel. !koans add This is a wind koan """ community = await self.bot.profiles.fetch_community(ctx.broadcaster) cid = community.communityid profile = await self.bot.profiles.fetch_profile(ctx.chatter) pid = profile.profileid koan = await self.koans.create_koan( cid, content, created_by=pid, ) await ctx.reply(f"Koan #{koan.koanlabel} created!") @koans.command(name='edit', aliases=['update',]) @cmds.is_moderator() async def koans_edit(self, ctx: cmds.Context, label: int, *, new_content: str): community = await self.bot.profiles.fetch_community(ctx.broadcaster) cid = community.communityid koan = await self.koans.get_koan_label(cid, label) if koan: await self.data.koans.update_where(koanid=koan.koanid).set(content=new_content) await ctx.reply(f"Updated koan #{label}") else: await ctx.reply(f"Koan #{label}' does not exist to delete!") @koans.command(name='del', aliases=['delete', 'rm', 'remove']) @cmds.is_moderator() async def koans_del(self, ctx: cmds.Context, label: int): """ Remove a koan from this channel by number. !koans del 5 """ community = await self.bot.profiles.fetch_community(ctx.broadcaster) cid = community.communityid koan = await self.koans.get_koan_label(cid, label) if koan: await koan.delete() await ctx.reply(f"Deleted koan #{label}") else: await ctx.reply(f"Koan #{label}' does not exist to delete!") @cmds.command(name='koan') async def koan(self, ctx: cmds.Context, label: Optional[int] = None): """ Show a koan from this channel. Optionally by number. !koan !koan 5 """ community = await self.bot.profiles.fetch_community(ctx.broadcaster) cid = community.communityid koan = None if label is not None: koan = await self.koans.get_koan_label(cid, label) if not koan: await ctx.reply(f"Koan #{label} does not exist!") else: koans = await self.koans.get_community_koans(communityid=cid) if koans: koan = random.choice(koans) else: await ctx.reply("This channel doesn't have any koans!") if koan: formatted = f"Koan #{koan.koanlabel}: {koan.content}" parts = [formatted[i : i+500] for i in range(0, len(formatted), 500)] for part in parts: await ctx.reply(part)