generated from HoloTech/holotech-plugin-template
559 lines
21 KiB
Python
559 lines
21 KiB
Python
from typing import Optional
|
|
import asyncio
|
|
|
|
from aiohttp import client
|
|
import discord
|
|
from discord.ext import commands as cmds
|
|
from discord import Forbidden, User, app_commands as appcmds
|
|
|
|
from meta import LionBot, LionCog, LionContext
|
|
from meta import logger
|
|
from meta.errors import SafeCancellation, UserInputError
|
|
from meta.logger import log_wrap
|
|
from utils.lib import utc_now
|
|
|
|
from ..data import CampaignData, EarnedReward
|
|
from ..campaign import CampaignRegistry, RewardCampaign
|
|
from ..lib import ThreadedWebhook
|
|
from .ui import RewardList, RewardEditor, CampaignDashboard
|
|
|
|
|
|
class CampaignCog(LionCog):
|
|
def __init__(self, bot: LionBot):
|
|
self.bot = bot
|
|
|
|
self.data = bot.db.load_registry(CampaignData())
|
|
self.campaigns = CampaignRegistry(self.data)
|
|
|
|
async def cog_load(self):
|
|
await self.data.init()
|
|
await self.bot.version_check(*self.data.VERSION)
|
|
await self.campaigns.init()
|
|
|
|
async def resolve_campaign(
|
|
self, cid: int, campaign_name: str | None
|
|
) -> RewardCampaign:
|
|
campaign = None
|
|
if campaign_name is not None:
|
|
campaign = await self.campaigns.fetch_campaign_by_name(cid, campaign_name)
|
|
if campaign is None:
|
|
raise UserInputError(
|
|
f"Sorry, no campaign found named '{campaign_name}'"
|
|
)
|
|
else:
|
|
# Find active campaign
|
|
active = await self.campaigns.fetch_campaigns(cid, active=True)
|
|
if len(active) > 1:
|
|
names = ", ".join(camp.row.campaign_name for camp in active)
|
|
raise UserInputError(f"Multiple active campaigns running: {names}")
|
|
elif not active:
|
|
raise UserInputError("No active campaigns running")
|
|
else:
|
|
campaign = active[0]
|
|
return campaign
|
|
|
|
async def campaign_modcheck(self, campaign: RewardCampaign, member: discord.Member):
|
|
if member.guild_permissions.administrator:
|
|
return True
|
|
if campaign.row.moderator_role_id and campaign.row.moderator_role_id in [
|
|
r.id for r in member.roles
|
|
]:
|
|
return True
|
|
return False
|
|
|
|
async def _campaign_acmpl(
|
|
self, interaction: discord.Interaction, partial: str
|
|
) -> list[appcmds.Choice]:
|
|
"""
|
|
Generate a list of campaigns, with active campaigns listed first.
|
|
|
|
Campaign values are their name.
|
|
"""
|
|
if not interaction.guild:
|
|
return []
|
|
community = await self.bot.profiles.fetch_community(
|
|
interaction.guild, interaction=interaction
|
|
)
|
|
cid = community.communityid
|
|
|
|
campaigns = await self.campaigns.fetch_campaigns(cid, active=None)
|
|
sorted_campaigns = sorted(
|
|
campaigns,
|
|
key=lambda camp: (camp.is_active, camp.row.started_at, camp.row.created_at),
|
|
reverse=True,
|
|
)
|
|
names = [
|
|
camp.row.campaign_name
|
|
for camp in sorted_campaigns
|
|
if partial.lower() in camp.row.campaign_name.lower()
|
|
]
|
|
|
|
choices = [appcmds.Choice(name=name[:100], value=name) for name in names]
|
|
return choices
|
|
|
|
@cmds.hybrid_group(
|
|
name="campaign", description="Command group for administering reward campaigns"
|
|
)
|
|
@appcmds.guild_only()
|
|
@appcmds.default_permissions(manage_guild=True)
|
|
async def campaign_group(self, ctx: LionContext):
|
|
"""
|
|
As a base command group this never gets executed.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
@campaign_group.command(
|
|
name="dashboard",
|
|
description="Show summary dashboard for active or selected campaign.",
|
|
)
|
|
@appcmds.describe(campaign_name="Name of the campaign to display")
|
|
@appcmds.rename(campaign_name="campaign")
|
|
async def campaign_dashboard_cmd(
|
|
self, ctx: LionContext, campaign_name: Optional[str] = None
|
|
):
|
|
if not ctx.guild:
|
|
return
|
|
if not ctx.interaction:
|
|
return
|
|
|
|
campaign = await self.resolve_campaign(ctx.community.communityid, campaign_name)
|
|
if not await self.campaign_modcheck(campaign, ctx.author):
|
|
await ctx.interaction.response.send_message(
|
|
"You need to be an administrator or have the configured modrole to use that!",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
# Open campaign UI
|
|
widget = CampaignDashboard(
|
|
bot=self.bot, campaign=campaign, callerid=ctx.author.id
|
|
)
|
|
await widget.run(ctx.interaction)
|
|
await widget.wait()
|
|
|
|
campaign_dashboard_cmd.autocomplete("campaign_name")(_campaign_acmpl)
|
|
|
|
@campaign_group.command(name="start", description="Setup and start a campaign")
|
|
@appcmds.describe(
|
|
campaign_name="Name of the campaign to create. Must be unique.",
|
|
rewards_cap="Optional maximum number of rewards to give.",
|
|
moderator_role="Optional discord role to allow to edit and moderate the campaign.",
|
|
logging_webhook="Optional discord webhook URL to log earned rewards to.",
|
|
)
|
|
async def campaign_start_cmd(
|
|
self,
|
|
ctx: LionContext,
|
|
campaign_name: str,
|
|
rewards_cap: Optional[int] = None,
|
|
moderator_role: Optional[discord.Role] = None,
|
|
logging_webhook: Optional[str] = None,
|
|
):
|
|
if not ctx.guild:
|
|
return
|
|
if not ctx.interaction:
|
|
return
|
|
|
|
# Permission check
|
|
if not ctx.author.guild_permissions.administrator:
|
|
await ctx.interaction.response.send_message(
|
|
"You need to be an administrator to start a campaign!", ephemeral=True
|
|
)
|
|
return
|
|
|
|
# Check name clash
|
|
existing = await self.campaigns.fetch_campaign_by_name(
|
|
ctx.community.communityid, campaign_name
|
|
)
|
|
if existing:
|
|
await ctx.reply("A campaign with that name already exists!")
|
|
return
|
|
|
|
# Check active campaign
|
|
campaigns = await self.campaigns.fetch_campaigns(
|
|
ctx.community.communityid, 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
|
|
|
|
# If logging webhook is given, check that it works
|
|
if logging_webhook is not None:
|
|
webhook = ThreadedWebhook.from_url(logging_webhook, client=self.bot)
|
|
try:
|
|
await webhook.test_webhook()
|
|
except (discord.HTTPException, discord.Forbidden):
|
|
await ctx.reply("Couldn't post to the logging webhook provided!")
|
|
return
|
|
|
|
# Passed all the checks, now we can create
|
|
campaign = await self.campaigns.create_campaign(
|
|
cid=ctx.community.communityid,
|
|
campaign_name=campaign_name,
|
|
target_rewards=rewards_cap,
|
|
moderator_role_id=moderator_role.id if moderator_role else None,
|
|
logging_webhook_url=logging_webhook,
|
|
)
|
|
await campaign.start()
|
|
|
|
# Ack creation
|
|
# TODO: Can also show dashboard
|
|
await ctx.reply(
|
|
f"Setup and started your reward campaign {campaign.row.campaign_name}! Good luck"
|
|
)
|
|
# Open campaign UI
|
|
widget = CampaignDashboard(
|
|
bot=self.bot, campaign=campaign, callerid=ctx.author.id
|
|
)
|
|
await widget.run(ctx.interaction)
|
|
await widget.wait()
|
|
|
|
@campaign_group.command(name="configure", description="Update campaign settings")
|
|
@appcmds.describe(
|
|
campaign_name="Name of the campaign you want to update.",
|
|
new_name="New name for the campaign. Must be unique.",
|
|
new_cap="New cap for the number of rewards. Use 0 or negative for no cap.",
|
|
moderator_role="Optional discord role to allow to edit and moderate the campaign.",
|
|
logging_webhook="Optional discord webhook URL to log earned rewards to.",
|
|
)
|
|
@appcmds.rename(campaign_name="campaign")
|
|
async def campaign_configure_cmd(
|
|
self,
|
|
ctx: LionContext,
|
|
campaign_name: Optional[str] = None,
|
|
new_name: Optional[str] = None,
|
|
new_cap: Optional[int] = None,
|
|
moderator_role: Optional[discord.Role] = None,
|
|
logging_webhook: Optional[str] = None,
|
|
):
|
|
if not ctx.guild:
|
|
return
|
|
if not ctx.interaction:
|
|
return
|
|
|
|
campaign = await self.resolve_campaign(ctx.community.communityid, campaign_name)
|
|
if not await self.campaign_modcheck(campaign, ctx.author):
|
|
await ctx.interaction.response.send_message(
|
|
"You need to be an administrator or have the configured modrole to use that!",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
# Verify each of the given settings to update
|
|
update_args = {}
|
|
|
|
# Check name clash
|
|
if (
|
|
new_name is not None
|
|
and new_name.lower() != campaign.row.campaign_name.lower()
|
|
):
|
|
existing = await self.campaigns.fetch_campaign_by_name(
|
|
ctx.community.communityid, campaign_name
|
|
)
|
|
if existing:
|
|
await ctx.reply("A campaign with that name already exists!")
|
|
return
|
|
update_args["campaign_name"] = new_name
|
|
|
|
if new_cap is not None:
|
|
update_args["target_rewards"] = new_cap if new_cap > 0 else None
|
|
|
|
# If logging webhook is given, check that it works
|
|
if logging_webhook is not None:
|
|
webhook = ThreadedWebhook.from_url(logging_webhook, client=self.bot)
|
|
try:
|
|
await webhook.test_webhook()
|
|
except (discord.HTTPException, discord.Forbidden):
|
|
await ctx.reply("Couldn't post to the logging webhook provided!")
|
|
return
|
|
update_args["logging_webhook_url"] = logging_webhook
|
|
|
|
if moderator_role is not None:
|
|
# TODO: Currently no way to actually unset this
|
|
update_args["moderator_role_id"] = moderator_role.id
|
|
|
|
# Passed all the checks, now we can update
|
|
|
|
campaign = await self.campaigns.update_campaign(
|
|
campaign.row.campaign_id,
|
|
**update_args,
|
|
)
|
|
|
|
# Ack creation
|
|
await ctx.reply("Updated your campaign, good luck!", ephemeral=True)
|
|
|
|
campaign_configure_cmd.autocomplete("campaign_name")(_campaign_acmpl)
|
|
|
|
@campaign_group.command(
|
|
name="rewards",
|
|
description="View and edit the rewards earned so far this campaign",
|
|
)
|
|
@appcmds.describe(campaign_name="Name of the campaign to display rewards for")
|
|
@appcmds.rename(campaign_name="campaign")
|
|
async def campaign_rewards_cmd(
|
|
self,
|
|
ctx: LionContext,
|
|
campaign_name: Optional[str] = None,
|
|
):
|
|
if not ctx.guild:
|
|
return
|
|
if not ctx.interaction:
|
|
return
|
|
|
|
campaign = await self.resolve_campaign(ctx.community.communityid, campaign_name)
|
|
if not await self.campaign_modcheck(campaign, ctx.author):
|
|
await ctx.interaction.response.send_message(
|
|
"You need to be an administrator or have the configured modrole to use that!",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
# Show rewardlist widget
|
|
widget = RewardList(bot=self.bot, campaign=campaign, callerid=ctx.author.id)
|
|
await widget.run(ctx.interaction)
|
|
await widget.wait()
|
|
|
|
campaign_rewards_cmd.autocomplete("campaign_name")(_campaign_acmpl)
|
|
|
|
@campaign_group.command(
|
|
name="delreward",
|
|
description="Remove a campaign reward",
|
|
)
|
|
@appcmds.describe(
|
|
rewardid="Earned reward to delete",
|
|
)
|
|
@appcmds.rename(rewardid="reward")
|
|
async def campaign_delreward_cmd(
|
|
self,
|
|
ctx: LionContext,
|
|
rewardid: str,
|
|
):
|
|
# For this we'll just open the reward editor
|
|
# Will need to identify the reward with autocomplete. Can use the reward id directly..
|
|
if not rewardid.isdigit():
|
|
raise UserInputError(
|
|
"Please enter the reward number or select a reward from the argument menu"
|
|
)
|
|
|
|
reward = await EarnedReward.fetch(int(rewardid))
|
|
if not reward:
|
|
raise UserInputError(
|
|
"Reward not found. Please enter the reward number or select a reward from the argument menu"
|
|
)
|
|
|
|
# Fetch campaign and then do a perm check
|
|
campaign = await self.campaigns.fetch_campaign(reward.campaign_id)
|
|
if campaign is None:
|
|
raise SafeCancellation("Something went wrong.. please try again soon.")
|
|
if campaign.row.communityid != ctx.community.communityid:
|
|
raise UserInputError("This reward doesn't belong to this community!")
|
|
|
|
if not await self.campaign_modcheck(campaign, ctx.author):
|
|
await ctx.interaction.response.send_message(
|
|
"You need to be an administrator or have the configured modrole to use that!",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
# Okay, they have permission, let's delete the reward.
|
|
await campaign.delete_reward(reward.earned_id)
|
|
|
|
# Ack
|
|
await ctx.reply(content="Campaign reward deleted!", ephemeral=True)
|
|
|
|
|
|
@campaign_group.command(
|
|
name="editreward",
|
|
description="Add or edit details for a given reward in a campaign (see also /campaign rewards)",
|
|
)
|
|
@appcmds.describe(
|
|
rewardid="Earned reward to edit",
|
|
fulfilled="Whether this reward has been completed or not.",
|
|
reference="Reference information for this reward, e.g. image or message URL",
|
|
notes="Any additional notes",
|
|
)
|
|
@appcmds.rename(rewardid="reward")
|
|
async def campaign_editreward_cmd(
|
|
self,
|
|
ctx: LionContext,
|
|
rewardid: str,
|
|
fulfilled: Optional[bool] = None,
|
|
reference: Optional[str] = None,
|
|
notes: Optional[str] = None
|
|
):
|
|
# For this we'll just open the reward editor
|
|
# Will need to identify the reward with autocomplete. Can use the reward id directly..
|
|
if not rewardid.isdigit():
|
|
raise UserInputError(
|
|
"Please enter the reward number or select a reward from the argument menu"
|
|
)
|
|
|
|
reward = await EarnedReward.fetch(int(rewardid))
|
|
if not reward:
|
|
raise UserInputError(
|
|
"Reward not found. Please enter the reward number or select a reward from the argument menu"
|
|
)
|
|
|
|
# Fetch campaign and then do a perm check
|
|
campaign = await self.campaigns.fetch_campaign(reward.campaign_id)
|
|
if campaign is None:
|
|
raise SafeCancellation("Something went wrong.. please try again soon.")
|
|
if campaign.row.communityid != ctx.community.communityid:
|
|
raise UserInputError("This reward doesn't belong to this community!")
|
|
|
|
if not await self.campaign_modcheck(campaign, ctx.author):
|
|
await ctx.interaction.response.send_message(
|
|
"You need to be an administrator or have the configured modrole to use that!",
|
|
ephemeral=True,
|
|
)
|
|
return
|
|
|
|
# Okay, this is our reward, and author has permission to modify it.
|
|
currently_fluffed = reward.fulfilled_at is not None
|
|
|
|
update_args = {}
|
|
if fulfilled is not None:
|
|
if fulfilled and not currently_fluffed:
|
|
# Reward has been fluffed
|
|
update_args["fulfilled_at"] = utc_now()
|
|
elif currently_fluffed and not fulfilled:
|
|
# Reward has been unfluffed
|
|
update_args["fulfilled_at"] = None
|
|
if reference is not None and reference != reward.reference:
|
|
update_args["reference"] = reference
|
|
if notes is not None and notes != reward.modnote:
|
|
update_args["modnote"] = notes
|
|
|
|
if update_args:
|
|
await ctx.interaction.response.defer(thinking=True, ephemeral=True)
|
|
await campaign.update_reward(reward.earned_id, **update_args)
|
|
await ctx.interaction.followup.send(
|
|
content="Reward updated!"
|
|
)
|
|
else:
|
|
modal = RewardEditor.from_reward(reward)
|
|
|
|
@modal.submit_callback()
|
|
async def on_editor_submit(interaction: discord.Interaction):
|
|
update_args = {}
|
|
|
|
if modal.flufbox.component.value and not currently_fluffed:
|
|
# Reward has been fluffed
|
|
update_args["fulfilled_at"] = utc_now()
|
|
elif currently_fluffed and not modal.flufbox.component.value:
|
|
# Reward has been unfluffed
|
|
update_args["fulfilled_at"] = None
|
|
|
|
new_ref_value = modal.reference.component.value or None
|
|
if new_ref_value != reward.reference:
|
|
update_args["reference"] = new_ref_value
|
|
|
|
new_notes_value = modal.notes.component.value or None
|
|
if new_notes_value != reward.modnote:
|
|
update_args["modnote"] = new_notes_value
|
|
|
|
if update_args:
|
|
await interaction.response.defer(thinking=True, ephemeral=True)
|
|
await campaign.update_reward(reward.earned_id, **update_args)
|
|
await interaction.followup.send(
|
|
content="Reward updated!", ephemeral=True
|
|
)
|
|
else:
|
|
await interaction.response.defer(thinking=False)
|
|
|
|
await ctx.interaction.response.send_modal(modal)
|
|
|
|
def _reward_acmpl_format(
|
|
self, campaign: RewardCampaign, reward: EarnedReward
|
|
) -> str:
|
|
"""
|
|
Format an EarnedReward for completion.
|
|
|
|
#100: username in campaign
|
|
"""
|
|
return "#{rewardid}: {username} in {cname}".format(
|
|
rewardid=reward.earned_id,
|
|
username=reward.twitch_user_name or reward.twitch_user_id or "Unknown",
|
|
cname=campaign.row.campaign_name,
|
|
)
|
|
|
|
@campaign_delreward_cmd.autocomplete("rewardid")
|
|
@campaign_editreward_cmd.autocomplete("rewardid")
|
|
async def _reward_acmpl(
|
|
self, interaction: discord.Interaction, partial: str
|
|
) -> list[appcmds.Choice]:
|
|
"""
|
|
Display list of rewards.
|
|
|
|
If possible, use the active campaign.
|
|
If no matching choices, search all campaigns of the community.
|
|
|
|
Values are the earned_id as a string
|
|
"""
|
|
if not interaction.guild:
|
|
return []
|
|
community = await self.bot.profiles.fetch_community(
|
|
interaction.guild, interaction=interaction
|
|
)
|
|
cid = community.communityid
|
|
|
|
choices = []
|
|
|
|
campaigns = await self.campaigns.fetch_campaigns(cid, active=None)
|
|
sorted_campaigns = sorted(
|
|
campaigns,
|
|
key=lambda camp: (camp.is_active, camp.row.started_at, camp.row.created_at),
|
|
reverse=True,
|
|
)
|
|
for campaign in sorted_campaigns:
|
|
rewards = await campaign.get_rewards()
|
|
for reward in reversed(rewards):
|
|
formatted = self._reward_acmpl_format(campaign, reward)
|
|
if partial in formatted:
|
|
choices.append(
|
|
appcmds.Choice(name=formatted, value=str(reward.earned_id))
|
|
)
|
|
if len(choices) >= 20:
|
|
break
|
|
if len(choices) >= 20:
|
|
break
|
|
|
|
return choices[:25]
|
|
|
|
# @campaign_group.command(
|
|
# name="reward",
|
|
# description="Add a reward to a given user"
|
|
# )
|
|
# async def campaign_reward_cmd(self, ctx: LionContext, twitch_username: str, reward_reason: str, campaign_name: Optional[str] = None):
|
|
# # TODO: This does
|
|
# ...
|
|
|
|
# @campaign_group.command(
|
|
# name="start", description="Add or edit details for a given reward in a campaign"
|
|
# )
|
|
# async def campaign_rewardnote_cmd(self, ctx: LionContext): ...
|
|
#
|
|
# @campaign_group.command(name="finish", description="Close a campaign")
|
|
# @appcmds.describe(campaign_name="Name of the campaign to mark as complete")
|
|
# @appcmds.rename(campaign_name='campaign')
|
|
# async def campaign_finish_cmd(
|
|
# self, ctx: LionContext, campaign_name: Optional[str] = None
|
|
# ):
|
|
# if not ctx.guild:
|
|
# return
|
|
# if not ctx.interaction:
|
|
# return
|
|
#
|
|
# campaign = await self.resolve_campaign(ctx.community.communityid, campaign_name)
|
|
# if not await self.campaign_modcheck(campaign, ctx.author):
|
|
# await ctx.interaction.response.send_message(
|
|
# "You need to be an administrator or have the configured modrole to use that!",
|
|
# ephemeral=True
|
|
# )
|
|
# return
|
|
# campaign_finish_cmd.autocomplete('campaign_name')(_campaign_acmpl)
|