generated from HoloTech/holotech-plugin-template
conatum
08fb0084fa
Fix several typos in event handler Rewrite sub and resub event handlers to expect resub events Add ppp alias to campaign command. Add ppp test command for testing last sub calculator
477 lines
19 KiB
Python
477 lines
19 KiB
Python
from typing import Optional
|
|
import asyncio
|
|
import datetime as dt
|
|
from datetime import datetime, timedelta
|
|
|
|
import twitchio
|
|
from twitchio.ext import commands as cmds
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
from data.queries import JOINTYPE, ORDER
|
|
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_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 detail_row["tier"] == 3000 and detail_row["duration_months"] >= 3:
|
|
# Check if there is an ongoing campaign
|
|
campaigns = await self.campaigns.fetch_campaigns(event_row["communityid"])
|
|
pid = event_row["profileid"]
|
|
|
|
for campaign in campaigns:
|
|
rewards = await campaign.get_rewards()
|
|
reward_progress = len(rewards)
|
|
|
|
existing = any(reward.profileid == pid for reward in rewards)
|
|
if existing:
|
|
continue
|
|
|
|
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 NOTICE) {data.system_message}",
|
|
event_id=event_row["event_id"],
|
|
twitch_user_id=data.chatter.id,
|
|
twitch_user_name=data.chatter.name,
|
|
)
|
|
await self.dispatch_update(campaign)
|
|
else:
|
|
logger.info(f"Campaigns ignoring sub notice event: {event_row}")
|
|
|
|
# @cmds.Component.listener()
|
|
async def event_custom_redemption_add(self, payload):
|
|
if payload.reward.title not in (
|
|
"hi!",
|
|
"hydrate",
|
|
"stretch",
|
|
"save file",
|
|
"pet lilac",
|
|
):
|
|
return
|
|
|
|
community = await self.bot.profiles.fetch_community(payload.broadcaster)
|
|
cid = community.communityid
|
|
profile = await self.bot.profiles.fetch_profile(payload.user)
|
|
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"(REDEEM) User redeemed {payload.reward.title}",
|
|
event_id=None,
|
|
twitch_user_id=payload.user.id,
|
|
twitch_user_name=payload.user.name,
|
|
reference=f"Redeem text: {payload.user_input}",
|
|
)
|
|
await self.dispatch_update(campaign)
|
|
|
|
# @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_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.
|
|
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 detail_row["tier"] == 3000 and detail_row["duration_months"] >= 3:
|
|
# Check if there is an ongoing campaign
|
|
campaigns = await self.campaigns.fetch_campaigns(event_row["communityid"])
|
|
pid = event_row["profileid"]
|
|
|
|
# Get user's last sub date
|
|
last_sub_date = await self.get_sub_start(
|
|
channelid=event_row["channel_id"],
|
|
userid=event_row["user_id"],
|
|
tier=3000,
|
|
)
|
|
if last_sub_date is None:
|
|
logger.error(f"T3 rsub with no history: {event_row!r}")
|
|
return
|
|
forecast_end = last_sub_date + relativedelta(
|
|
months=detail_row["duration_months"]
|
|
)
|
|
if forecast_end < dt.datetime(2026, 11, 1, tzinfo=dt.UTC):
|
|
logger.warning(f"T3 sub with forecast end too short: {event_row!r}")
|
|
return
|
|
|
|
for campaign in campaigns:
|
|
rewards = await campaign.get_rewards()
|
|
reward_progress = len(rewards)
|
|
|
|
existing = any(reward.profileid == pid for reward in rewards)
|
|
if existing:
|
|
continue
|
|
|
|
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"(RESUB NOTICE) {data.system_message}",
|
|
event_id=event_row["event_id"],
|
|
twitch_user_id=data.chatter.id,
|
|
twitch_user_name=data.chatter.name,
|
|
)
|
|
await self.dispatch_update(campaign)
|
|
else:
|
|
logger.info(f"Campaigns ignoring sub notice event: {event_row}")
|
|
|
|
async def get_sub_start(self, channelid, userid, tier=3000):
|
|
event_tracker = self.bot.get_component("TrackerComponent")
|
|
query = event_tracker.data.events.select_where(
|
|
channel_id=channelid,
|
|
user_id=userid,
|
|
tier=tier,
|
|
)
|
|
query.join("subscribe_events", using=("event_id",), join_type=JOINTYPE.INNER)
|
|
query.order_by("created_at", direction=ORDER.DESC)
|
|
query.select("created_at")
|
|
query.limit(1)
|
|
query.with_no_adapter()
|
|
rows = await query
|
|
if rows:
|
|
return rows[0]["created_at"]
|
|
else:
|
|
return None
|
|
|
|
@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", aliases=["ppp"], invoke_fallback=True)
|
|
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
|
|
|
|
response = (
|
|
"event details can be found here: {event_link} "
|
|
"there are {remaining} sketch slots remaining! ♡♡ "
|
|
)
|
|
|
|
# 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,
|
|
event_link="https://lilac.thewisewolf.dev/provides/partner_plus_poster2.jpg",
|
|
remaining=reward_cap - rewards_earned,
|
|
given=rewards_earned,
|
|
cap=reward_cap,
|
|
)
|
|
await ctx.reply(formatted)
|
|
await self.dispatch_update(campaign)
|
|
|
|
@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 = "Completed {name}: {given} rewards earned out of {cap}!"
|
|
else:
|
|
response = "Completed {name}: {given} 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)
|
|
|
|
@group_campaign.command(name="test")
|
|
@cmds.is_moderator()
|
|
async def cmd_campaign_test(
|
|
self,
|
|
ctx: cmds.Context,
|
|
user: twitchio.User,
|
|
channel: twitchio.User,
|
|
tier: int = 3000,
|
|
):
|
|
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
|
|
campaigns = await self.campaigns.fetch_campaigns(cid)
|
|
campaign = campaigns[0]
|
|
|
|
# Get user's last sub date
|
|
last_sub_date = await self.get_sub_start(
|
|
channelid=channel.id,
|
|
userid=user.id,
|
|
tier=tier,
|
|
)
|
|
lines = []
|
|
if last_sub_date is None:
|
|
subline = f"Last {tier} sub not found"
|
|
lines.append(subline)
|
|
else:
|
|
subline = f"Last {tier} sub at {last_sub_date}"
|
|
lines.append(subline)
|
|
forecast_end = last_sub_date + relativedelta(months=3)
|
|
lines.append(f"Three months after: {forecast_end}")
|
|
|
|
if forecast_end > dt.datetime(2026, 11, 1, tzinfo=dt.UTC):
|
|
lines.append("Which is after the end of the event")
|
|
else:
|
|
lines.append("Which is not after the end of the event")
|
|
|
|
rewards = await campaign.get_rewards()
|
|
|
|
existing = any(reward.profileid == pid for reward in rewards)
|
|
if existing:
|
|
lines.append(f"This user has already been rewarded.")
|
|
else:
|
|
lines.append(f"This user might yet earn a reward")
|
|
|
|
await ctx.reply("; ".join(lines))
|