generated from HoloTech/holotech-plugin-template
157 lines
4.9 KiB
Python
157 lines
4.9 KiB
Python
from typing import Optional, TypeAlias, TypedDict
|
|
import json
|
|
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta
|
|
from dataclasses import dataclass
|
|
|
|
from data.queries import JOINTYPE, ORDER
|
|
from meta.sockets import Channel
|
|
from utils.lib import utc_now
|
|
from modules.profiles.profiles.profiles import ProfilesRegistry
|
|
|
|
from . import logger
|
|
from .data import (
|
|
Campaign,
|
|
CampaignData,
|
|
EarnedReward,
|
|
)
|
|
from .campaign import CampaignRegistry, RewardCampaign
|
|
|
|
|
|
# ISO formatted timestamp
|
|
ISOTimestamp: TypeAlias = str
|
|
|
|
|
|
class CampaignPayload(TypedDict):
|
|
communityid: int
|
|
campaign_id: int
|
|
campaign_name: str
|
|
|
|
started_at: ISOTimestamp | None
|
|
completed_at: ISOTimestamp | None
|
|
|
|
reward_progress: int
|
|
reward_cap: int | None
|
|
|
|
|
|
async def prepare_campaign(campaign: RewardCampaign) -> CampaignPayload:
|
|
reward_progress = len(await campaign.get_rewards())
|
|
|
|
return CampaignPayload(
|
|
communityid=campaign.row.communityid,
|
|
campaign_id=campaign.row.campaign_id,
|
|
campaign_name=campaign.row.campaign_name,
|
|
started_at=campaign.row.started_at.isoformat()
|
|
if campaign.row.started_at
|
|
else None,
|
|
completed_at=campaign.row.completed_at.isoformat()
|
|
if campaign.row.completed_at
|
|
else None,
|
|
reward_progress=reward_progress,
|
|
reward_cap=campaign.row.target_rewards,
|
|
)
|
|
|
|
|
|
class CampaignChannel(Channel):
|
|
name = "CampaignRewards"
|
|
|
|
def __init__(
|
|
self, profiler: ProfilesRegistry, campaigns: CampaignRegistry, **kwargs
|
|
):
|
|
super().__init__(**kwargs)
|
|
|
|
self.profiler: ProfilesRegistry = profiler
|
|
self.campaigns: CampaignRegistry = campaigns
|
|
|
|
# Map of communities to webhooks listening for this community
|
|
self.communities = defaultdict(
|
|
set
|
|
) # Map of communityid -> listening websockets
|
|
|
|
async def on_connection(self, websocket, event):
|
|
if not (cidstr := event.get("community")):
|
|
logger.warning("Campaign connection missing communityid")
|
|
await super().on_connection(websocket, event)
|
|
await self.send_sample(websocket=websocket)
|
|
return
|
|
elif not cidstr.isdigit():
|
|
raise ValueError("Community id provided is not an integer")
|
|
cid = int(cidstr)
|
|
community = await self.profiler.get_community(cid)
|
|
if community is None:
|
|
raise ValueError("Unknown community provided.")
|
|
|
|
await super().on_connection(websocket, event)
|
|
self.communities[cid].add(websocket)
|
|
|
|
# Fetch active campaign for this cid and send if it exists
|
|
active = await self.campaigns.fetch_campaigns(cid)
|
|
if active:
|
|
# TODO: Presumes only one active campaign
|
|
campaign = active[0]
|
|
payload = await prepare_campaign(campaign)
|
|
await self.send_campaign_update(cid, payload, websocket)
|
|
else:
|
|
await self.send_no_campaign(cid, websocket)
|
|
|
|
async def send_sample(self, websocket):
|
|
import random
|
|
|
|
payload = CampaignPayload(
|
|
communityid=1,
|
|
campaign_id=1,
|
|
campaign_name="PartnerPlus2026",
|
|
started_at=utc_now().isoformat(),
|
|
completed_at=None,
|
|
reward_progress=random.randint(0, 105),
|
|
reward_cap=150,
|
|
)
|
|
await self.send_campaign_update(1, payload, websocket=websocket)
|
|
|
|
async def del_connection(self, websocket):
|
|
for wss in self.communities.values():
|
|
wss.discard(websocket)
|
|
await super().del_connection(websocket)
|
|
|
|
async def send_campaign_update(
|
|
self, communityid: int, payload: CampaignPayload, websocket=None
|
|
):
|
|
for ws in (websocket,) if websocket else self.communities[communityid]:
|
|
await self.send_event(
|
|
{
|
|
"type": "DO",
|
|
"method": "setTimer",
|
|
"args": payload,
|
|
},
|
|
websocket=ws,
|
|
)
|
|
|
|
async def send_campaign_ended(
|
|
self, communityid: int, payload: CampaignPayload, websocket=None
|
|
):
|
|
for ws in (websocket,) if websocket else self.communities[communityid]:
|
|
await self.send_event(
|
|
{
|
|
"type": "DO",
|
|
"method": "endTimer",
|
|
"args": payload,
|
|
},
|
|
websocket=ws,
|
|
)
|
|
|
|
async def send_no_campaign(self, communityid: int, websocket=None):
|
|
for ws in (websocket,) if websocket else self.communities[communityid]:
|
|
await self.send_event(
|
|
{
|
|
"type": "DO",
|
|
"method": "noTimer",
|
|
"args": {},
|
|
},
|
|
websocket=ws,
|
|
)
|
|
|
|
async def send_event(self, event, **kwargs):
|
|
logger.info(f"Sending websocket event: {json.dumps(event, indent=1)}")
|
|
await super().send_event(event, **kwargs)
|