from typing import Any, Optional import datetime as dt from datetime import datetime, timedelta import aiohttp from data import ORDER, Condition from data.conditions import NULL, condition from utils.lib import utc_now from .lib import LOWER, asexpr, ThreadedWebhook from .data import ( CampaignData, Campaign, EarnedReward, ) from . import logger class RewardCampaign: def __init__(self, row: Campaign, session: aiohttp.ClientSession | None = None): self.row = row self._session = session self._webhook: ThreadedWebhook | None = None self._cached_webhookurl: str | None = None @property def is_active(self): return self.row.started_at is not None and self.row.completed_at is None @property def webhook(self): if self._cached_webhookurl != self.row.logging_webhook_url: url = self._cached_webhookurl = self.row.logging_webhook_url if url is not None: # TODO: I don't know if these needs a client # Might be hard if so given we need to run this from Twitch client as well self._webhook = ThreadedWebhook.from_url(url, session=self._session) else: self._webhook = None return self._webhook 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_from: 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_from=earned_from, event_id=event_id, twitch_user_id=twitch_user_id, twitch_user_name=twitch_user_name, earned_at=earned_at or utc_now(), ) await self.try_to_log_reward(row) return row async def update_reward(self, rewardid: int, **kwargs): # In particular, log or update the logged message if kwargs: reward = await EarnedReward.fetch(rewardid) if reward is None: raise ValueError("Reward doesn't exist") await reward.update(**kwargs) await self.try_to_log_reward(reward) 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 async def try_to_log_reward(self, reward: EarnedReward): import discord if self.webhook: embed = await self._log_format_reward(reward) if reward.log_messageid: # Try and edit message try: await self.webhook.edit_message(reward.log_messageid, embed=embed) except discord.HTTPException: await reward.update(log_messageid=None) if not reward.log_messageid: try: message = await self.webhook.send(embed=embed, wait=True) await reward.update(log_messageid=message.id) except discord.HTTPException: # Couldn't send, give up. logger.warning( f"Failed to log campaign reward {reward!r}", exc_info=True ) async def _log_format_reward(self, reward: EarnedReward): """ Quick and ugly embed format for the webhook. """ import discord embed = discord.Embed( title=f"Reward #{reward.earned_id} in {self.row.campaign_name}", ) # User Field embed.add_field( name="User Information", value=( f"`{reward.twitch_user_name or 'Unknown'}`\n" f"`(ID: {reward.twitch_user_id or 'Unknown'})`.\n" f"Internal ID `{reward.profileid}`" ), inline=True, ) # Earning field embed.add_field( name="Reward Earned", value=( f"{discord.utils.format_dt(reward.earned_at, 'F')}.\n" f"{reward.earned_from}" ), inline=True, ) # Reference field embed.add_field( name="Reference", value=reward.reference or "No Reference information saved.", inline=False, ) # Modnote field embed.add_field( name="Notes", value=reward.modnote or "No notes added", inline=False ) # Fulfilled field fluffed_emoji = "✅" if reward.fulfilled_at else "🔳" if reward.fulfilled_at is not None: fluffed = f"{fluffed_emoji} Fluffed at {discord.utils.format_dt(reward.fulfilled_at, 'F')}" if reward.fulfilled_note: fluffed += "\n" + "Fluff Note: " + reward.fulfilled_note else: fluffed = f"{fluffed_emoji} Not yet fluffed" embed.add_field(name="Fulfilled", value=fluffed) embed.set_footer(text="Last Updated") embed.timestamp = utc_now() return embed class CampaignRegistry: VERSION = CampaignData.VERSION def __init__( self, data: CampaignData, session: aiohttp.ClientSession | None = None ): self.data = data # TODO: Actually pass in a session self._session = aiohttp.ClientSession() 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, session=self._session) 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) & (Campaign.completed_at == NULL) ) if active: condition = condition & active_condition else: condition = condition & ~active_condition rows = await Campaign.fetch_where( condition, ) campaigns = [RewardCampaign(row, session=self._session) 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), LOWER(asexpr(name)) ), communityid=cid, ) if results: row = results[0] camp = RewardCampaign(row, session=self._session) else: camp = None return camp async def create_campaign( self, cid: int, campaign_name: str, **kwargs ) -> RewardCampaign: """ Create a new campaign. The name must be unique (ignoring case) to facilitate easy lookup. """ row = await Campaign.create( communityid=cid, campaign_name=campaign_name, **kwargs ) return RewardCampaign(row, session=self._session) async def update_campaign(self, campaign_id: int, **kwargs) -> RewardCampaign: """ Update a campaign with the given data args. Registry handles this for caching and dispatch reasons. """ campaign = await Campaign.fetch(campaign_id) await campaign.update(**kwargs) return RewardCampaign(campaign, session=self._session)