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.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), ) 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["total_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="editreward", description="Add or edit details for a given reward in a campaign (see also /campaign rewards)", ) @appcmds.describe( rewardid="Earned reward to edit", ) @appcmds.rename(rewardid="reward") async def campaign_editreward_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, this is our reward, and author has permission to modify it. Spin the modal. modal = RewardEditor.from_reward(reward) currently_fluffed = reward.fulfilled_at is not None @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.value or None if new_ref_value != reward.reference: update_args["reference"] = new_ref_value new_notes_value = modal.notes.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) def _reward_acmpl_format( self, campaign: RewardCampaign, reward: EarnedReward ) -> str: """ Format an EarnedReward for completion. #100: username in campaign """ ... @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), ) 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)