HoloTech/twitch-subathon-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add lb and adjust commands.

Browse Source
This commit is contained in:
Interitio
2025-09-02 21:31:19 +10:00
parent a3a3b36230
commit b10098d28f
3 changed files with 197 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

269
subathon/component.py
View File

@@ -6,7 +6,8 @@ import twitchio
from twitchio import PartialUser, Scopes, eventsub
from twitchio.ext import commands as cmds
from meta import Bot
from data.queries import ORDER
from meta import Bot, Context
from meta.sockets import Channel, register_channel
from utils.lib import utc_now, strfdelta
@@ -42,6 +43,11 @@ class TimerChannel(Channel):
websocket=websocket,
)
async def del_connection(self, websocket):
for wss in self.communities.values():
wss.discard(websocket)
await super().del_connection(websocket)
async def send_updates(self, communityid: int):
args = await self.get_args_for(communityid)
for ws in self.communities[communityid]:
@@ -396,85 +402,197 @@ class SubathonComponent(cmds.Component):
await ctx.reply("No active subathon running!")
# subathon start
@group_subathon.command(name='setup')
@group_subathon.command(name='setup', alias='start')
@cmds.is_broadcaster()
async def cmd_setup(self, ctx: cmds.Context, name: str, initial_hours: float, sub1: float, sub2: float, sub3: float, bit: float, timescore: int, timecap: Optional[int]=None):
if ctx.broadcaster:
# TODO: Usage. Maybe implement ? commands?
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
await ctx.reply("There is already an active subathon running! Use !subathon stop to stop it!")
return
initial_time = initial_hours * 60 * 60
active = await Subathon.create(
communityid=cid,
name=name,
initial_time=initial_time,
sub1_score=sub1,
sub2_score=sub2,
sub3_score=sub3,
bit_score=bit,
score_time=timescore,
timecap=timecap
)
timer_link = f"https://izashi.thewisewolf.dev/tracker/timer?channelid={ctx.channel.id}"
await ctx.reply(f"Setup your {name}! Use !subathon resume to get the timer running. Your timer link: {timer_link}")
await self.channel.send_updates(cid)
"""
Creates a new subathon.
USAGE: {prefix}subathon setup <initial_hours> <sub1_points> <sub2_points> <sub3_points> <bit_points> <timepoints> [timecap]
Arguments:
initial_hours = number of hours to start the timer with
sub1_points = points per T1 sub
sub2_points = points per T2 sub
sub3_points = points per T3 sub
bit_points = points per bit
timepoints = seconds to be added to the timer per point
timecap (optional) = number of seconds to cap the timer at.
"""
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
await ctx.reply("There is already an active subathon running! Use !subathon stop to stop it!")
return
initial_time = initial_hours * 60 * 60
active = await Subathon.create(
communityid=cid,
name=name,
initial_time=initial_time,
sub1_score=sub1,
sub2_score=sub2,
sub3_score=sub3,
bit_score=bit,
score_time=timescore,
timecap=timecap
)
# TODO: Add this to config not hardcode
timer_link = f"https://izashi.thewisewolf.dev/tracker/timer?channelid={ctx.channel.id}"
await ctx.reply(f"Setup your {name}! Use !subathon resume to get the timer running. Your timer link: {timer_link}")
await self.channel.send_updates(cid)
# subathon stop
@group_subathon.command(name='stop')
@cmds.is_broadcaster()
async def cmd_stop(self, ctx: cmds.Context):
if ctx.broadcaster:
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
if active.running:
await active.pause()
await self.channel.send_updates(cid)
await active.subathondata.update(ended_at=utc_now())
total = await active.get_score()
dursecs = active.get_duration()
dur = strfdelta(timedelta(seconds=dursecs))
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
if active.running:
await active.pause()
await self.channel.send_updates(cid)
await active.subathondata.update(ended_at=utc_now())
total = await active.get_score()
dursecs = active.get_duration()
dur = strfdelta(timedelta(seconds=dursecs))
await ctx.reply(
f"{active.name} complete after {dur} with a total of {total} points, congratulations!"
)
else:
await ctx.reply("No active subathon to stop.")
await ctx.reply(
f"{active.name} complete after {dur} with a total of {total} points, congratulations!"
)
else:
await ctx.reply("No active subathon to stop.")
# subathon pause
@group_subathon.command(name='pause')
@cmds.is_moderator()
async def cmd_pause(self, ctx: cmds.Context):
if ctx.broadcaster or ctx.author.moderator:
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
if active.running:
await active.pause()
await ctx.reply(f"{active.name} timer paused!")
await self.channel.send_updates(cid)
else:
await ctx.reply(f"{active.name} timer already paused!")
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
if active.running:
await active.pause()
await ctx.reply(f"{active.name} timer paused!")
await self.channel.send_updates(cid)
else:
await ctx.reply("No active subathon to pause")
await ctx.reply(f"{active.name} timer already paused!")
else:
await ctx.reply("No active subathon to pause")
# subathon resume
@group_subathon.command(name='resume')
@cmds.is_moderator()
async def cmd_resume(self, ctx: cmds.Context):
if ctx.broadcaster or ctx.author.moderator:
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
if not active.running:
await active.resume()
await ctx.reply(f"{active.name} timer resumed!")
await self.channel.send_updates(cid)
else:
await ctx.reply(f"{active.name} timer already running!")
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
if not active.running:
await active.resume()
await ctx.reply(f"{active.name} timer resumed!")
await self.channel.send_updates(cid)
else:
await ctx.reply("No active subathon to resume")
await ctx.reply(f"{active.name} timer already running!")
else:
await ctx.reply("No active subathon to resume")
# subathon adjust
@group_subathon.command(name='adjust')
@cmds.is_moderator()
async def cmd_subathon_adjust(self, ctx: Context, amount: int, *, user: Optional[twitchio.User] = None):
"""
Directly add or remove points from the subathon.
If a user is provided, will adjust their contributed amount.
USAGE:
{prefix}subathon adjust <amount> [@user]
Arguments:
'amount' is an integer number of points to adjust by.
It may be negative to remove points.
'@user' is an optional user name or mention to adjust for.
Examples:
'{prefix}subathon adjust 10'
'{prefix}subathon adjust -10 @machinestalkerwolfie'
"""
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
if user is not None:
profile = await self.bot.profiles.fetch_profile(user)
name = user.display_name or profile.nickname or 'Unknown'
pid = profile.profileid
else:
profile = None
name = None
pid = None
await active.add_contribution(pid, amount, None)
# Build message
if amount > 0:
amountstr = f"Added {amount} point(s) to the timer"
elif amount < 0:
amountstr = f"Removed {amount} point(s) from the timer"
else:
amountstr = "Did nothing to the timer"
namestr = f"on behalf of {name}" if profile else ""
message = f"{amountstr} {namestr}"
await ctx.reply(message)
else:
await ctx.reply("No active subathon to adjust")
@group_subathon.command(name='leaderboard', aliases=('top', 'lb',))
async def cmd_subathon_lb(self, ctx: Context):
"""
Display top contributors by points contributed, up to 10.
Anonymous contributions are gathered as one user.
"""
# TODO: Might want to extend stats to select/show last subathon as well
# IDEA: We could also offer full export as csv via web
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
caller_profile = await self.bot.profiles.fetch_profile(ctx.chatter)
caller_pid = caller_profile.profileid
if (active := await self.get_active_subathon(cid)) is not None:
# Get totals for all contributors
query = self.data.subathon_contributions.select_where(subathonid=active.subathondata.subathon_id)
query.join('user_profiles', using=('profileid',))
query.select('profileid', total="SUM(score)")
query.order_by('total', direction=ORDER.DESC)
query.with_no_adapter()
results = await query
parts = []
caller_idx = None
for i, row in enumerate(results):
if pid := row['profileid']:
profile = await self.bot.profiles.profiles.get_profile(pid)
name = (profile.nickname if profile else None) or 'Unknown'
if pid == caller_pid:
caller_idx = i
else:
name = 'Anonymous'
score = row['total']
part = f"{name}: {score} points"
parts.append(part)
header = ""
leaderboard = ', '.join(parts[:10])
footer = ""
if len(parts) > 10:
header = f"{active.name} top 10 leaderboard: "
leaderboard = ', '.join(parts[:10])
if caller_idx is not None and caller_idx >= 10:
caller_part = parts[caller_idx]
footer = f" ... {caller_part}"
elif parts:
header = f"{active.name} contribution leaderboard: "
else:
header = "No contributions to show yet!"
message = f"{header} {leaderboard} {footer}"
await ctx.reply(message)
else:
await ctx.reply("No active subathon to show leaderboard of!")
# Subathon goals
@cmds.group(name='goals', invoke_fallback=True)
async def group_goals(self, ctx: cmds.Context):
# List the goals
@@ -496,18 +614,17 @@ class SubathonComponent(cmds.Component):
await ctx.reply("No active subathon running!")
@group_goals.command(name='add')
@cmds.is_moderator()
async def cmd_add(self, ctx: cmds.Context, required: int, *, description: str):
if ctx.broadcaster or ctx.author.moderator:
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
await SubathonGoal.create(
subathon_id=active.subathondata.subathon_id,
required_score=required,
description=description,
)
await ctx.reply("Goal added!")
else:
await ctx.reply("No active subathon to add goal to!")
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
cid = community.communityid
if (active := await self.get_active_subathon(cid)) is not None:
await SubathonGoal.create(
subathon_id=active.subathondata.subathon_id,
required_score=required,
description=description,
)
await ctx.reply("Goal added!")
else:
await ctx.reply("No active subathon to add goal to!")
# TODO:

2
subathon/data.py
View File

@@ -1,8 +1,6 @@
from data import Registry, RowModel, Table
from data.columns import String, Timestamp, Integer, Bool
# User contributed {} subs and added {}:{} to the timer! Thank you :love:
# We have reached goal #5: Karaoke !! Thank you everyone for your support <3
class Subathon(RowModel):
_tablename_ = 'subathons'