124 lines
4.0 KiB
Python
124 lines
4.0 KiB
Python
from typing import Optional
|
|
import random
|
|
import twitchio
|
|
from twitchio.ext import commands as cmds
|
|
|
|
from datamodels import Koan, Communities
|
|
|
|
from . import logger
|
|
|
|
|
|
class KoanComponent(cmds.Component):
|
|
def __init__(self, bot):
|
|
self.bot = bot
|
|
|
|
@cmds.Component.listener()
|
|
async def event_message(self, payload: twitchio.ChatMessage) -> None:
|
|
print(f"[{payload.broadcaster.name}] - {payload.chatter.name}: {payload.text}")
|
|
|
|
@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 Communities.fetch_or_create(twitchid=ctx.channel.id, name=ctx.channel.name)
|
|
cid = community.communityid
|
|
|
|
koans = await Koan.fetch_where(communityid=cid)
|
|
if koans:
|
|
names = ', '.join(koan.name for koan in koans)
|
|
await ctx.reply(
|
|
f"Koans: {names}"
|
|
)
|
|
else:
|
|
await ctx.reply("No koans have been made in this channel!")
|
|
|
|
@koans.command(name='add', aliases=['new', 'create'])
|
|
async def koans_add(self, ctx: cmds.Context, name: str, *, text: str):
|
|
"""
|
|
Add or overwrite a koan to this channel.
|
|
|
|
!koans add wind This is a wind koan
|
|
"""
|
|
community = await Communities.fetch_or_create(twitchid=ctx.channel.id, name=ctx.channel.name)
|
|
cid = community.communityid
|
|
|
|
name = name.lower()
|
|
|
|
assert isinstance(ctx.author, twitchio.Chatter)
|
|
if (ctx.author.moderator or ctx.author.broadcaster):
|
|
# Delete the koan with this name if it exists
|
|
existing = await Koan.table.delete_where(
|
|
communityid=cid,
|
|
name=name,
|
|
)
|
|
|
|
# Insert the new koan
|
|
await Koan.create(
|
|
communityid=cid,
|
|
name=name,
|
|
message=text
|
|
)
|
|
|
|
# Ack
|
|
if existing:
|
|
await ctx.reply(f"Updated the koan '{name}'")
|
|
else:
|
|
await ctx.reply(f"Created the new koan '{name}'")
|
|
|
|
@koans.command(name='del', aliases=['delete', 'rm', 'remove'])
|
|
async def koans_del(self, ctx: cmds.Context, name: str):
|
|
"""
|
|
Remove a koan from this channel by name.
|
|
|
|
!koans del wind
|
|
"""
|
|
community = await Communities.fetch_or_create(twitchid=ctx.channel.id, name=ctx.channel.name)
|
|
cid = community.communityid
|
|
|
|
name = name.lower()
|
|
|
|
assert isinstance(ctx.author, twitchio.Chatter)
|
|
if (ctx.author.moderator or ctx.author.broadcaster):
|
|
# Delete the koan with this name if it exists
|
|
existing = await Koan.table.delete_where(
|
|
communityid=cid,
|
|
name=name,
|
|
)
|
|
if existing:
|
|
await ctx.reply(f"Deleted the koan '{name}'")
|
|
else:
|
|
await ctx.reply(f"The koan '{name}' does not exist to delete!")
|
|
|
|
@cmds.command(name='koan')
|
|
async def koan(self, ctx: cmds.Context, name: Optional[str] = None):
|
|
"""
|
|
Show a koan from this channel. Optionally by name.
|
|
|
|
!koan
|
|
!koan wind
|
|
"""
|
|
community = await Communities.fetch_or_create(twitchid=ctx.channel.id, name=ctx.channel.name)
|
|
cid = community.communityid
|
|
|
|
if name is not None:
|
|
name = name.lower()
|
|
koans = await Koan.fetch_where(
|
|
communityid=cid,
|
|
name=name
|
|
)
|
|
if koans:
|
|
koan = koans[0]
|
|
await ctx.reply(koan.message)
|
|
else:
|
|
await ctx.reply(f"The requested koan '{name}' does not exist! Use '{ctx.prefix}koans' to see all the koans.")
|
|
else:
|
|
koans = await Koan.fetch_where(communityid=cid)
|
|
if koans:
|
|
koan = random.choice(koans)
|
|
await ctx.reply(koan.message)
|
|
else:
|
|
await ctx.reply("This channel doesn't have any koans!")
|