CarmiCoven/pluscampaign-plugin
2
0
Fork 0
generated from HoloTech/holotech-plugin-template
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Registry, tw interface draft, sub handler

Browse Source
This commit is contained in:
Interitio
2026-07-25 09:49:42 +03:00
parent 67a04ef82f
commit fc5c3d07e4
4 changed files with 384 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
data/pluscampaign-v1.sql
+5 -3
View File
@@ -13,7 +13,7 @@ INSERT INTO version_history (component, from_version, to_version, author) VALUES
CREATE TABLE campaigns(
campaign_id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
communityid INTEGER NOT NULL REFERENCES communities(communityid) ON DELETE CASCADE ON UPDATE CASCADE,
target_rewards INTEGER NOT NULL,
target_rewards INTEGER,
campaign_name TEXT NOT NULL,
started_at TIMESTAMPTZ,
completed_at TIMESTAMPTZ,
@@ -21,6 +21,8 @@ CREATE TABLE campaigns(
_timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE UNIQUE INDEX campaigns_communities_names ON campaigns(communityid, LOWER(campaign_name));
CREATE TABLE campaign_rewards_earned(
earned_id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
campaign_id INTEGER NOT NULL REFERENCES campaigns ON DELETE CASCADE ON UPDATE CASCADE,
@@ -30,9 +32,9 @@ CREATE TABLE campaign_rewards_earned(
twitch_user_name TEXT,
fulfilled_at TIMESTAMPTZ,
fulfilled_note TEXT,
earned_at TIMESTAMPTZ NOT NULL,
modnote TEXT,
earned_from TEXT NOT NULL,
earned_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
earned_reason TEXT NOT NULL,
_timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
plugin/campaign.py
+131 -1
View File
@@ -1,4 +1,12 @@
from typing import Optional
from typing import Any, Optional
import datetime as dt
from datetime import datetime, timedelta
from data import ORDER, Condition
from data.conditions import NULL, condition
from utils.lib import utc_now
from .lib import LOWER, asexpr
from .data import (
CampaignData,
Campaign,
@@ -6,6 +14,55 @@ from .data import (
)
class RewardCampaign:
def __init__(self, row: Campaign):
self.row = row
@property
def is_active(self):
return self.row.started_at is not None and self.row.completed_at is None
async def start(self):
"""Start the reward campaign."""
if self.row.started_at is not None:
raise ValueError("This campaign has already been started.")
await self.row.update(started_at=utc_now())
async def finish(self):
"""End the reward campaign."""
if self.row.completed_at is not None:
raise ValueError("This campaign is already finished.")
await self.row.update(completed_at=utc_now())
async def add_reward(
self,
profileid: int,
earned_reason: str,
event_id: Optional[int] = None,
twitch_user_id: Optional[str] = None,
twitch_user_name: Optional[str] = None,
earned_at: Optional[datetime] = None,
) -> EarnedReward:
row = await EarnedReward.create(
campaign_id=self.row.campaign_id,
profileid=profileid,
earned_reason=earned_reason,
event_id=event_id,
twitch_user_id=twitch_user_id,
twitch_user_name=twitch_user_name,
earned_at=earned_at or utc_now(),
)
return row
async def get_rewards(self) -> list[EarnedReward]:
rows = await EarnedReward.fetch_where(
campaign_id=self.row.campaign_id
).order_by("earned_at", direction=ORDER.ASC)
return rows
class CampaignRegistry:
VERSION = CampaignData.VERSION
@@ -14,3 +71,76 @@ class CampaignRegistry:
async def init(self):
await self.data.init()
async def fetch_campaign(self, campaign_id: int) -> RewardCampaign:
"""
Fetch a campaign by id. The campaign must exist.
"""
row = await Campaign.fetch(campaign_id)
if row is None:
raise ValueError("Campign %s doesn't exist." % campaign_id)
camp = RewardCampaign(row)
return camp
async def fetch_campaigns(
self,
cid: int,
*,
active: bool | None = True,
) -> list[RewardCampaign]:
"""
Fetch all campaigns in the given community, matching the given conditions.
"""
condition = Campaign.communityid == cid
if active is not None:
active_condition = (
Campaign.started_at != NULL and Campaign.completed_at == NULL
)
if active:
condition = condition and active_condition
else:
condition = condition and ~active_condition
rows = await Campaign.fetch_where(condition)
campaigns = [RewardCampaign(row) for row in rows]
return campaigns
async def fetch_campaign_by_name(
self, cid: int, name: str
) -> Optional[RewardCampaign]:
"""
Fetch a campaign for this community by name.
The name is matched without case.
"""
results = await Campaign.fetch_where(
Condition._expression_equality(
LOWER(Campaign.campaign_name.expr), LOWER(asexpr(name))
),
campaignid=cid,
)
if results:
row = results[0]
camp = RewardCampaign(row)
else:
camp = None
return camp
async def create_campaign(
self,
cid: int,
campaign_name: str,
target_rewards: Optional[int] = None,
) -> RewardCampaign:
"""
Create a new campaign.
The name must be unique (ignoring case) to facilitate easy lookup.
"""
row = await Campaign.create(
communityid=cid,
target_rewards=target_rewards,
campaign_name=campaign_name,
)
return RewardCampaign(row)
plugin/lib.py
+21
View File
@@ -0,0 +1,21 @@
from typing import Any
from psycopg import sql
from data import RawExpr, Expression
def LOWER(expression: Expression) -> RawExpr:
"""
Wrap an Expression in the SQL function LOWER().
"""
expr, values = expression.as_tuple()
final_expr = sql.SQL("LOWER({})").format(expr)
final_values = values
return RawExpr(final_expr, final_values)
def asexpr(value: Any) -> RawExpr:
"""
Turn a value into an expression.
"""
return RawExpr(sql.Placeholder(), (value,))
plugin/twitch/component.py
+227
View File
@@ -30,4 +30,231 @@ class CampaignComponent(cmds.Component):
async def component_teardown(self):
pass
# ------ 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:
# Add a reward to the database with the correct info.
await campaign.add_reward(
profileid=event_row["profileid"],
earned_reason=f"(SUB NOTICE): 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"],
)
# TODO: Webhook logging maybe..
# Or just general logging.
@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 finish your campaign before starting a new one."
)
return
# Now can create
campaign = await self.campaigns.create_campaign(cid, name, 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!"
)
@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)
@group_campaign.command(name="reward")
@cmds.is_moderator()
async def cmd_campaign_reward(
self,
ctx: cmds.Context,
user: twitchio.PartialUser,
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
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,
reward_reason,
twitch_user_id=user.id,
twitch_user_name=user.name,
)
await ctx.reply(
f"Successfully added campaign reward to {user.mention}'s account."
)