generated from HoloTech/holotech-plugin-template
147 lines
4.1 KiB
Python
147 lines
4.1 KiB
Python
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,
|
|
EarnedReward,
|
|
)
|
|
|
|
|
|
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
|
|
|
|
def __init__(self, data: CampaignData):
|
|
self.data = data
|
|
|
|
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)
|