generated from HoloTech/holotech-plugin-template
262 lines
8.3 KiB
Python
262 lines
8.3 KiB
Python
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
|
|
if reward.fulfilled_at is not None:
|
|
fluffed = f"Fluffed at {discord.utils.format_dt(reward.fulfilled_at, 'F')}"
|
|
if reward.fulfilled_note:
|
|
fluffed += "\n" + "Fluff Note: " + reward.fulfilled_note
|
|
else:
|
|
fluffed = "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 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, 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)
|