generated from HoloTech/holotech-plugin-template
314 lines
13 KiB
Python
314 lines
13 KiB
Python
from typing import Optional
|
|
import asyncio
|
|
|
|
import twitchio
|
|
from twitchio.ext import commands as cmds
|
|
|
|
from meta import Bot
|
|
from meta.logger import log_wrap
|
|
from meta.sockets import Channel, register_channel
|
|
from utils.lib import utc_now
|
|
|
|
from . import logger
|
|
|
|
from ..data import CampaignData
|
|
from ..campaign import CampaignRegistry, RewardCampaign
|
|
from ..channel import CampaignPayload, prepare_campaign, CampaignChannel
|
|
|
|
|
|
class CampaignComponent(cmds.Component):
|
|
def __init__(self, bot: Bot):
|
|
self.bot = bot
|
|
|
|
self.data = bot.dbconn.load_registry(CampaignData())
|
|
self.campaigns = CampaignRegistry(self.data)
|
|
self.channel = CampaignChannel(self.bot.profiles.profiles, self.campaigns)
|
|
|
|
register_channel(self.channel.name, self.channel)
|
|
|
|
# ----- API -----
|
|
async def component_load(self):
|
|
await self.data.init()
|
|
await self.bot.version_check(*self.data.VERSION)
|
|
await self.campaigns.init()
|
|
|
|
async def component_teardown(self):
|
|
pass
|
|
|
|
async def dispatch_update(self, campaign: RewardCampaign):
|
|
cid = campaign.row.communityid
|
|
payload = await prepare_campaign(campaign)
|
|
await self.channel.send_campaign_update(cid, payload)
|
|
|
|
# ------ Event Handlers -----
|
|
@cmds.Component.listener()
|
|
async def event_safe_event_chat_notice_sub(self, payload):
|
|
"""
|
|
This is our most important notice for this event.
|
|
|
|
The subscription chat notice *will* fire whenever a user changes their subscription.
|
|
The subscription chat notice includes the number of months of the subscription.
|
|
And the tier.
|
|
|
|
Almost every single user participating in the 'tier upgrade' event will
|
|
trigger a relevant chat notice, and we would be able to satisfy the event with
|
|
only this event, along with manual intervention for any existing T3 subscribers.
|
|
"""
|
|
event_row, detail_row, data = payload
|
|
|
|
# Check the tier and duration of the subscription.
|
|
# Continue if tier = 3000 and duration is at least 3 months
|
|
if (tier := detail_row["tier"]) == 3000 and (
|
|
duration := detail_row["duration_months"]
|
|
) >= 3:
|
|
# Check if there is an ongoing campaign
|
|
campaigns = await self.campaigns.fetch_campaigns(event_row["communityid"])
|
|
|
|
for campaign in campaigns:
|
|
reward_progress = len(await campaign.get_rewards())
|
|
|
|
if (
|
|
campaign.row.target_rewards is None
|
|
or reward_progress < campaign.row.target_rewards
|
|
):
|
|
# Add a reward to the database with the correct info.
|
|
await campaign.add_reward(
|
|
profileid=event_row["profileid"],
|
|
earned_from=f"(SUB) User subscribed for {duration} months at tier {tier}",
|
|
event_id=event_row["event_id"],
|
|
twitch_user_id=data["chatter_user_id"],
|
|
twitch_user_name=data["chatter_user_name"],
|
|
)
|
|
await self.dispatch_update(campaign)
|
|
# TODO: Webhook logging maybe.. (do this in campaign)
|
|
# Or just general logging. (also do this in campaign, but we can do again here)
|
|
|
|
@cmds.Component.listener()
|
|
async def event_message(self, payload):
|
|
if not payload.text.startswith("%reward%"):
|
|
return
|
|
|
|
community = await self.bot.profiles.fetch_community(payload.broadcaster)
|
|
cid = community.communityid
|
|
profile = await self.bot.profiles.fetch_profile(payload.chatter)
|
|
pid = profile.profileid
|
|
campaigns = await self.campaigns.fetch_campaigns(cid)
|
|
|
|
for campaign in campaigns:
|
|
reward_progress = len(await campaign.get_rewards())
|
|
|
|
if (
|
|
campaign.row.target_rewards is None
|
|
or reward_progress < campaign.row.target_rewards
|
|
):
|
|
# Add a reward to the database with the correct info.
|
|
await campaign.add_reward(
|
|
profileid=pid,
|
|
earned_from=f"(MSG) User messaged in chat: {payload.text}",
|
|
event_id=None,
|
|
twitch_user_id=payload.chatter.id,
|
|
twitch_user_name=payload.chatter.name,
|
|
)
|
|
await self.dispatch_update(campaign)
|
|
|
|
@cmds.Component.listener()
|
|
async def event_safe_event_chat_notice_resub(self, payload):
|
|
# Check that the end of the sub is past the threshold by adding duration to the previous sub
|
|
# Threshold being now + 3 months, by calendar date.
|
|
# Or even if the sub ends on at least the month that is past the three month region.
|
|
# Then check that this resub extends the last existing resub by more than three months.
|
|
# Technically there could be a resub for 6 months 3 months ago that was
|
|
# # TODO: This logic should be done for completeness, but will postpone for now
|
|
# This could be done in subscription_message as well, but the
|
|
# notice has slightly more self-contained metadata.
|
|
...
|
|
|
|
@cmds.Component.listener()
|
|
async def event_safe_subscription(self, payload): ...
|
|
|
|
@cmds.Component.listener()
|
|
async def event_safe_subscription_message(self, payload): ...
|
|
|
|
# ------ Commands -----
|
|
@cmds.group(name="campaign", invoke_fallback=True)
|
|
@cmds.is_moderator()
|
|
async def group_campaign(self, ctx: cmds.Context, name: Optional[str] = None):
|
|
"""Status of the current or named campaign."""
|
|
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
|
|
cid = community.communityid
|
|
|
|
# {name} campaign: {n} rewards redeemed out of {m}!
|
|
# {name} campaign: {n} rewards redeemed!
|
|
|
|
# Get the named campaign or the current one
|
|
campaign = None
|
|
if name is not None:
|
|
campaign = await self.campaigns.fetch_campaign_by_name(cid, name)
|
|
if campaign is None:
|
|
await ctx.reply(f"Sorry, no campaign found named '{name}'")
|
|
else:
|
|
# Find active campaign if it exists
|
|
campaigns = await self.campaigns.fetch_campaigns(cid, active=True)
|
|
if len(campaigns) > 1:
|
|
names = ", ".join(camp.row.campaign_name for camp in campaigns)
|
|
await ctx.reply(f"Multiple active campaigns running: {names}")
|
|
elif not campaigns:
|
|
await ctx.reply(
|
|
"No active campaigns! To view a historical campaign please use its name!"
|
|
)
|
|
else:
|
|
campaign = campaigns[0]
|
|
|
|
if campaign is not None:
|
|
all_rewards = await campaign.get_rewards()
|
|
rewards_earned = len(all_rewards)
|
|
reward_cap = campaign.row.target_rewards
|
|
|
|
if reward_cap is not None:
|
|
response = "{name}: {given} rewards earned out of {cap}!"
|
|
else:
|
|
response = "{name}: {given} rewards earned so far!"
|
|
formatted = response.format(
|
|
name=campaign.row.campaign_name,
|
|
given=rewards_earned,
|
|
cap=reward_cap,
|
|
)
|
|
await ctx.reply(formatted)
|
|
|
|
@group_campaign.command(name="setup", aliases=["start"])
|
|
@cmds.is_moderator()
|
|
async def cmd_campaign_setup(
|
|
self, ctx: cmds.Context, name: str, rewards: Optional[int] = None
|
|
):
|
|
"""Create a new campaign."""
|
|
# Check if there is an active campaign going, decline to create one if there is
|
|
# This is the simplest business-logic way of making sure campaigns are never run simul
|
|
# TODO: Need to make a decision on this. Simultaneous reward campaigns probably have their place
|
|
# when you are running different types of rewards over different time-frames
|
|
# or in response to different types of events.
|
|
# "Sorry, we don't support simultaneous active reward campaigns at this time"
|
|
#
|
|
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
|
|
cid = community.communityid
|
|
|
|
# First check if there is a campaign by that name already
|
|
existing = await self.campaigns.fetch_campaign_by_name(cid, name)
|
|
if existing:
|
|
await ctx.reply("A campaign with that name already exists!")
|
|
return
|
|
|
|
# Then check if there is already an active campaign
|
|
campaigns = await self.campaigns.fetch_campaigns(cid, active=True)
|
|
if campaigns:
|
|
existing_name = campaigns[0].row.campaign_name
|
|
await ctx.reply(
|
|
f"Your campaign '{existing_name}' is already active! "
|
|
"Sorry, we don't yet support multiple active campaigns, "
|
|
"please use '!campaign finish' to end your current campaign before starting a new one."
|
|
)
|
|
return
|
|
|
|
# Now can create
|
|
campaign = await self.campaigns.create_campaign(
|
|
cid, name, target_rewards=rewards
|
|
)
|
|
|
|
# Also start the campaign
|
|
await campaign.start()
|
|
|
|
# Ack to user
|
|
# await ctx.reply(
|
|
# f"Success! You campaign '{name}' has been created. "
|
|
# "When you are ready to start, use !campaign start"
|
|
# )
|
|
await ctx.reply(
|
|
f"Success! Your campaign '{name}' has been created and started. "
|
|
"Best of luck!"
|
|
)
|
|
await self.dispatch_update(campaign)
|
|
|
|
@group_campaign.command(name="finish", aliases=["stop", "complete", "end"])
|
|
@cmds.is_moderator()
|
|
async def cmd_campaign_finish(self, ctx: cmds.Context, name: Optional[str] = None):
|
|
"""End a campaign and show a brief summary."""
|
|
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
|
|
cid = community.communityid
|
|
# Get the named campaign or the current one
|
|
campaign = None
|
|
if name is not None:
|
|
campaign = await self.campaigns.fetch_campaign_by_name(cid, name)
|
|
if campaign is None:
|
|
await ctx.reply(f"Sorry, no campaign found named '{name}'")
|
|
elif campaign.row.completed_at is not None:
|
|
await ctx.reply("This campaign has already been completed!")
|
|
campaign = None
|
|
else:
|
|
# Find active campaign if it exists
|
|
campaigns = await self.campaigns.fetch_campaigns(cid, active=True)
|
|
if len(campaigns) > 1:
|
|
names = ", ".join(camp.row.campaign_name for camp in campaigns)
|
|
await ctx.reply(f"Multiple active campaigns running: {names}")
|
|
elif not campaigns:
|
|
await ctx.reply("No active campaigns to finish!")
|
|
else:
|
|
campaign = campaigns[0]
|
|
|
|
if campaign is not None:
|
|
await campaign.finish()
|
|
|
|
all_rewards = await campaign.get_rewards()
|
|
rewards_earned = len(all_rewards)
|
|
reward_cap = campaign.row.target_rewards
|
|
|
|
if reward_cap is not None:
|
|
response = f"Completed {name}: {rewards_earned} rewards earned out of {reward_cap}!"
|
|
else:
|
|
response = f"Completed {name}: {rewards_earned} rewards earned so far!"
|
|
formatted = response.format(
|
|
name=campaign.row.campaign_name,
|
|
given=rewards_earned,
|
|
cap=reward_cap,
|
|
)
|
|
await ctx.reply(formatted)
|
|
await self.dispatch_update(campaign)
|
|
|
|
@group_campaign.command(name="reward")
|
|
@cmds.is_moderator()
|
|
async def cmd_campaign_reward(
|
|
self,
|
|
ctx: cmds.Context,
|
|
user: twitchio.User,
|
|
*,
|
|
reward_reason: str,
|
|
):
|
|
"""Manually reward a target user, adjusting their points."""
|
|
# TODO: Lacks support for multi-campaign
|
|
# Find active campaign if it exists
|
|
community = await self.bot.profiles.fetch_community(ctx.broadcaster)
|
|
cid = community.communityid
|
|
profile = await self.bot.profiles.fetch_profile(user)
|
|
name = user.display_name or profile.nickname or "Unknown"
|
|
pid = profile.profileid
|
|
|
|
reason = f"(Given by {ctx.author.mention}) {reward_reason}"
|
|
|
|
campaigns = await self.campaigns.fetch_campaigns(cid, active=True)
|
|
if len(campaigns) > 1:
|
|
names = ", ".join(camp.row.campaign_name for camp in campaigns)
|
|
await ctx.reply(f"Multiple active campaigns running: {names}")
|
|
elif not campaigns:
|
|
await ctx.reply("No active campaigns to finish!")
|
|
else:
|
|
campaign = campaigns[0]
|
|
|
|
await campaign.add_reward(
|
|
pid,
|
|
reason,
|
|
twitch_user_id=user.id,
|
|
twitch_user_name=user.name,
|
|
)
|
|
await ctx.reply(
|
|
f"Successfully added campaign reward to {user.mention}'s account."
|
|
)
|
|
await self.dispatch_update(campaign)
|