639 lines
27 KiB
Python
639 lines
27 KiB
Python
import asyncio
|
|
import copy
|
|
import datetime
|
|
import logging
|
|
import os
|
|
from typing import Type, Dict, List
|
|
|
|
import aiohttp
|
|
from interactions import (
|
|
Extension,
|
|
SlashContext,
|
|
component_callback,
|
|
ComponentContext,
|
|
StringSelectMenu,
|
|
Message,
|
|
Embed,
|
|
EmbedField,
|
|
is_owner,
|
|
check,
|
|
StringSelectOption, integration_types,
|
|
Modal,
|
|
ShortText,
|
|
modal_callback,
|
|
ModalContext,
|
|
)
|
|
from interactions import Task, IntervalTrigger
|
|
from interactions import slash_command, listen
|
|
from interactions.api.events import Component
|
|
from interactions.api.events import Startup
|
|
from interactions.client.errors import Forbidden
|
|
|
|
from token_bot.controller.alerts import AlertsController
|
|
from token_bot.controller.users import UsersController
|
|
from token_bot.history_manager.history_manager import HistoryManager, History
|
|
from token_bot.persistant_database.alert_category import AlertCategory
|
|
from token_bot.persistant_database.alert_schema import Alert
|
|
from token_bot.persistant_database.alert_type import AlertType
|
|
from token_bot.persistant_database.user_schema import User
|
|
from token_bot.token_database import database as tdb
|
|
from token_bot.token_database.flavor import Flavor
|
|
from token_bot.token_database.region import Region
|
|
from token_bot.ui.action_row.tracker import ALERT_TYPE_ROW
|
|
from token_bot.ui.select_menus.alert_menu import HIGH_ALERT_MENU, LOW_ALERT_MENU
|
|
from token_bot.ui.select_menus.flavor_menu import FLAVOR_MENU
|
|
from token_bot.ui.select_menus.region_menu import REGION_MENU
|
|
|
|
|
|
#### Static Helper Functions
|
|
|
|
|
|
async def gather_alerts_by_flavor(alerts: List[Alert]) -> Dict[Flavor, List[Alert]]:
|
|
alerts_by_flavor = {}
|
|
for alert in alerts:
|
|
if alert.flavor not in alerts_by_flavor:
|
|
alerts_by_flavor[alert.flavor] = [alert]
|
|
else:
|
|
alerts_by_flavor[alert.flavor].append(alert)
|
|
return alerts_by_flavor
|
|
|
|
|
|
class Tracker(Extension):
|
|
def __init__(self, bot):
|
|
self._users: UsersController | None = None
|
|
self._alerts: AlertsController | None = None
|
|
self._tdb: tdb.Database | None = None
|
|
self._history_manager: HistoryManager | None = None
|
|
self._session: aiohttp.ClientSession | None = None
|
|
|
|
###################################
|
|
# Task Functions #
|
|
###################################
|
|
|
|
@Task.create(IntervalTrigger(minutes=1))
|
|
async def update_data(self):
|
|
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Updating Price")
|
|
users_alerts: Dict[User, List[Alert]] = {}
|
|
for flavor in Flavor:
|
|
for region in Region:
|
|
alerts = await self._history_manager.update_data(flavor, Region(region))
|
|
for alert in alerts:
|
|
users = await self._alerts.get_users(alert)
|
|
for user in users:
|
|
if user not in users_alerts:
|
|
users_alerts[user] = [alert]
|
|
else:
|
|
users_alerts[user].append(alert)
|
|
if users_alerts:
|
|
self.bot.logger.log(
|
|
logging.INFO, "TokenBot Tracker: Processing User Alerts"
|
|
)
|
|
for user in users_alerts:
|
|
discord_user = await self.bot.fetch_user(user.user_id)
|
|
alerts_by_flavor = await gather_alerts_by_flavor(users_alerts[user])
|
|
alert_tally = 0
|
|
for flavor in alerts_by_flavor:
|
|
for _ in alerts_by_flavor[flavor]:
|
|
alert_tally += 1
|
|
alert_word = "alert" if alert_tally == 1 else "alerts"
|
|
embeds = [
|
|
Embed(
|
|
title="GoblinBot Tracker Alert Triggered",
|
|
color=0xB10000,
|
|
description=f"You requested to be alerted on the WoW token price. You have {alert_tally} {alert_word}\n\n",
|
|
)
|
|
]
|
|
for flavor in alerts_by_flavor:
|
|
embeds.append(
|
|
await self._render_alert_flavor(
|
|
alerts_by_flavor[flavor], user=user
|
|
)
|
|
)
|
|
embeds.append(
|
|
Embed(
|
|
title="",
|
|
color=0xB10000,
|
|
description=f"You can remove an alert via ```/remove-alert```\n"
|
|
f"or you can remove all alerts and user data via ```/remove-registration```\n",
|
|
)
|
|
)
|
|
try:
|
|
await discord_user.send(embeds=embeds)
|
|
except Forbidden:
|
|
self.bot.logger.log(
|
|
logging.ERROR, f"User: {discord_user.id} has no permissions to send alerts, skipping")
|
|
|
|
self.bot.logger.log(
|
|
logging.INFO, "TokenBot Tracker: Done Processing User Alerts"
|
|
)
|
|
|
|
###################################
|
|
# Slash Commands #
|
|
###################################
|
|
|
|
@listen(Startup)
|
|
async def on_start(self):
|
|
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Initializing")
|
|
# Create a single shared ClientSession for all components
|
|
self._session = aiohttp.ClientSession()
|
|
self._users = UsersController(self._session)
|
|
self._alerts = AlertsController(self._session)
|
|
self._tdb = tdb.Database(self._session)
|
|
self._history_manager = HistoryManager(self._tdb)
|
|
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Initialized")
|
|
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Loading Historical Data")
|
|
await self._history_manager.load_data()
|
|
self.bot.logger.log(
|
|
logging.INFO, "TokenBot Tracker: Loading Historical Data Finished"
|
|
)
|
|
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Loading Custom Price Alerts")
|
|
# Load all SPECIFIC_PRICE alerts from database (AlertType.SPECIFIC_PRICE = 11)
|
|
custom_alerts = await self._alerts.get_all_by_type(AlertType.SPECIFIC_PRICE.value)
|
|
await self._history_manager.load_custom_alerts(custom_alerts)
|
|
self.bot.logger.log(
|
|
logging.INFO, f"TokenBot Tracker: Loaded {len(custom_alerts)} Custom Price Alerts"
|
|
)
|
|
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Started")
|
|
self.update_data.start()
|
|
|
|
def extension_unload(self):
|
|
"""Clean up resources when the extension is unloaded"""
|
|
if self._session and not self._session.closed:
|
|
asyncio.create_task(self._session.close())
|
|
self.bot.logger.log(logging.INFO, "TokenBot Tracker: ClientSession closed")
|
|
|
|
@slash_command(
|
|
name="register",
|
|
description="Register with a new GoblinBot Region for alerts on token price changes.",
|
|
)
|
|
@integration_types(guild=True, user=True)
|
|
async def register(self, ctx: SlashContext):
|
|
text = (
|
|
"## Select a region to register with \n\n"
|
|
"Please note: \n"
|
|
"* You can only be registered with one region at a time \n"
|
|
"* Changing your region will remove all previous alerts you have signed up for \n"
|
|
"* You can remove all alerts and user data using ```/remove-registration```"
|
|
)
|
|
menu = copy.deepcopy(REGION_MENU)
|
|
await ctx.send(text, components=menu, ephemeral=True)
|
|
|
|
@slash_command(
|
|
name="remove-registration",
|
|
description="Remove all alerts and registration from GoblinBot",
|
|
)
|
|
@integration_types(guild=True, user=True)
|
|
async def remove_registration(self, ctx: SlashContext):
|
|
if await self._users.exists(ctx.user.id):
|
|
user = await self._users.get(ctx.user.id)
|
|
for alert in user.subscribed_alerts:
|
|
await self._alerts.remove_user(alert, user)
|
|
await self._users.delete(ctx.user.id)
|
|
|
|
await ctx.send("All alert subscriptions and user data deleted", ephemeral=True)
|
|
|
|
@slash_command(description="The current retail token cost")
|
|
@integration_types(guild=True, user=True)
|
|
async def current(self, ctx: SlashContext):
|
|
current_str = await self.get_current_token(ctx, tdb.Flavor.RETAIL)
|
|
await ctx.send(current_str, ephemeral=True)
|
|
|
|
@slash_command(description="The current classic token cost")
|
|
@integration_types(guild=True, user=True)
|
|
async def current_classic(self, ctx: SlashContext):
|
|
current_str = await self.get_current_token(ctx, tdb.Flavor.CLASSIC)
|
|
await ctx.send(current_str, ephemeral=True)
|
|
|
|
@slash_command(name="add-alert", description="Add an alert listener")
|
|
@integration_types(guild=True, user=True)
|
|
async def add_alert(self, ctx: SlashContext):
|
|
if not await self._users.exists(ctx.user.id):
|
|
try:
|
|
await self.region_select_menu(ctx)
|
|
except TimeoutError:
|
|
return
|
|
|
|
user = await self._users.get(ctx.user.id)
|
|
|
|
try:
|
|
flavor = await self.flavor_select_menu(ctx)
|
|
alert_category, price = await self.alert_category_select_menu(ctx)
|
|
match alert_category:
|
|
case AlertCategory.LOW:
|
|
alert_type = await self.low_alert_select_menu(ctx)
|
|
case AlertCategory.HIGH:
|
|
alert_type = await self.high_alert_select_menu(ctx)
|
|
case AlertCategory.CUSTOM:
|
|
alert_type = AlertType.SPECIFIC_PRICE
|
|
|
|
except TimeoutError:
|
|
return
|
|
|
|
else:
|
|
alert = Alert(alert_type, flavor, user.region, price)
|
|
if not await self._users.is_subscribed(user, alert):
|
|
await asyncio.gather(
|
|
self._users.add_alert(user, alert),
|
|
self._alerts.add_user(alert, user),
|
|
)
|
|
await ctx.send("Successfully added alert", ephemeral=True)
|
|
|
|
else:
|
|
await ctx.send(
|
|
"You are already subscribed to this alert", ephemeral=True
|
|
)
|
|
|
|
@slash_command(
|
|
name="remove-alert", description="Remove an alert you have signed up for"
|
|
)
|
|
@integration_types(guild=True, user=True)
|
|
async def remove_alert(self, ctx: SlashContext):
|
|
if not await self._user_is_registered(ctx):
|
|
return
|
|
user = await self._users.get(ctx.user.id)
|
|
alerts = await self._users.list_alerts(user)
|
|
if len(alerts) == 0:
|
|
await ctx.send("You do not have any alerts registered", ephemeral=True)
|
|
return
|
|
|
|
try:
|
|
alert = await self.remove_alert_select_menu(ctx, user)
|
|
except TimeoutError:
|
|
return
|
|
else:
|
|
await asyncio.gather(
|
|
self._users.remove_alert(user, alert),
|
|
self._alerts.remove_user(alert, user),
|
|
)
|
|
await ctx.send("Successfully removed alert", ephemeral=True)
|
|
|
|
@slash_command(
|
|
name="list-alerts", description="List all alerts you have signed up for"
|
|
)
|
|
@integration_types(guild=True, user=True)
|
|
async def list_alerts(self, ctx: SlashContext):
|
|
if not await self._user_is_registered(ctx):
|
|
return
|
|
user = await self._users.get(ctx.user.id)
|
|
alerts = await self._users.list_alerts(user)
|
|
if len(alerts) == 0:
|
|
await ctx.send("You do not have any alerts registered", ephemeral=True)
|
|
return
|
|
alerts_str = f"You have {len(alerts)} out of 25 maximum alerts registered"
|
|
embeds = [
|
|
Embed(
|
|
title="List of GoblinBot Tracker Alerts",
|
|
color=0x0000B1,
|
|
description=alerts_str,
|
|
)
|
|
]
|
|
alerts_by_flavor = await gather_alerts_by_flavor(alerts)
|
|
for flavor in alerts_by_flavor:
|
|
embeds.append(
|
|
await self._render_alert_flavor(alerts_by_flavor[flavor], user=user)
|
|
)
|
|
await ctx.send(embeds=embeds, ephemeral=True)
|
|
|
|
###################################
|
|
# Callbacks Commands #
|
|
###################################
|
|
|
|
# Note: Callbacks for flavor_menu, high_alert_menu, low_alert_menu, and alert buttons
|
|
# are disabled because they interfere with wait_for_component manual handling
|
|
|
|
@component_callback("region_menu")
|
|
async def region_menu_cb(self, ctx: ComponentContext):
|
|
discord_user = await self.bot.fetch_user(ctx.user.id)
|
|
await discord_user.send(
|
|
"You have successfully registered your region with GoblinBot!\n"
|
|
"Most interactions will happen in direct messages with GoblinBot here.\n"
|
|
"You can remove your user data and alerts at any time using ```/remove-registration```\n"
|
|
)
|
|
await ctx.defer(edit_origin=True, suppress_error=True)
|
|
|
|
###################################
|
|
# Helper Functions #
|
|
###################################
|
|
|
|
async def get_current_token(self, ctx: SlashContext, flavor: Flavor) -> str:
|
|
user: User = await self._users.get(ctx.user.id)
|
|
if user.region is None:
|
|
return (
|
|
f"Please register with a region before attempting to list alerts using\n"
|
|
"```/register```"
|
|
)
|
|
region = user.region.name
|
|
region_history = self._history_manager.get_history(flavor, user.region)
|
|
price_movement_str = format(region_history.last_price_movement, ",")
|
|
if region_history.last_price_movement > 0:
|
|
price_movement_str = f"+{price_movement_str}"
|
|
|
|
return (
|
|
f"Last Price Value for {region}: {format(region_history.last_price_datum[1], ",")}\n"
|
|
f"Last Update Time: <t:{int(region_history.last_price_datum[0].timestamp())}:F> local time\n"
|
|
f"Last Price Movement: {price_movement_str}"
|
|
)
|
|
|
|
async def remove_alert_select_menu(self, ctx: SlashContext, user: User):
|
|
alerts_by_flavor = await gather_alerts_by_flavor(user.subscribed_alerts)
|
|
select_options: List[StringSelectOption] = []
|
|
for flavor in alerts_by_flavor:
|
|
for alert in alerts_by_flavor[flavor]:
|
|
select_options.append(
|
|
StringSelectOption(
|
|
label=f"{alert.flavor.name.lower().title()} {alert.to_human_string()}",
|
|
value=f"{alert.flavor.name.lower()} {alert.to_human_string()}",
|
|
)
|
|
)
|
|
menu = StringSelectMenu(
|
|
select_options,
|
|
placeholder="Select an alert to remove",
|
|
custom_id="remove_alert_menu",
|
|
)
|
|
message = await ctx.send(
|
|
"Select an alert to remove", components=menu, ephemeral=True
|
|
)
|
|
|
|
try:
|
|
alert_component: Component = await self.bot.wait_for_component(
|
|
messages=message, components=menu, timeout=30
|
|
)
|
|
except TimeoutError:
|
|
menu.disabled = True
|
|
await message.edit(context=ctx, components=menu, content="Timed out")
|
|
raise TimeoutError
|
|
else:
|
|
menu.disabled = True
|
|
await message.edit(context=ctx, components=menu)
|
|
selection_split = alert_component.ctx.values[0].split(" ")
|
|
flavor = Flavor[selection_split[0].upper()]
|
|
alert_type_str = " ".join(selection_split[1:])
|
|
alert_type = AlertType.from_str(alert_type_str)
|
|
|
|
# Parse price for custom alerts
|
|
price = 0
|
|
if alert_type == AlertType.SPECIFIC_PRICE:
|
|
# Extract price from "Custom Price: 250,000g"
|
|
price_part = alert_type_str.split(": ")[1].rstrip("g").replace(",", "")
|
|
price_gold = int(price_part)
|
|
# Convert gold to copper
|
|
price = price_gold
|
|
|
|
return Alert(alert_type, flavor, user.region, price)
|
|
|
|
async def region_select_menu(self, ctx: SlashContext, user: User | None = None):
|
|
region_menu = copy.deepcopy(REGION_MENU)
|
|
region_menu_str = str()
|
|
|
|
if user is None:
|
|
region_menu_str += "You are not currently registered with a region, please select a region to register with.\n"
|
|
region_menu_str += (
|
|
"* You can only be registered with one region at a time.\n"
|
|
"* Registering for a new region will remove your old region's registration.\n"
|
|
)
|
|
region_message = await ctx.send(
|
|
region_menu_str,
|
|
components=region_menu,
|
|
ephemeral=True,
|
|
)
|
|
try:
|
|
region_component = await self.bot.wait_for_component(
|
|
messages=region_message, components=region_menu, timeout=30
|
|
)
|
|
except TimeoutError:
|
|
region_menu.disabled = True
|
|
await region_message.edit(
|
|
context=ctx, components=region_menu, content="Timed out"
|
|
)
|
|
raise TimeoutError
|
|
else:
|
|
# Acknowledge the component interaction to avoid 404 Unknown Interaction
|
|
await region_component.ctx.defer(edit_origin=True, suppress_error=True)
|
|
region_menu.disabled = True
|
|
region = Region(region_component.ctx.values[0].lower())
|
|
user = User(ctx.user.id, region, subscribed_alerts=[])
|
|
await asyncio.gather(
|
|
self._users.add(user),
|
|
region_message.edit(context=ctx, components=region_menu),
|
|
)
|
|
return region
|
|
|
|
async def flavor_select_menu(self, ctx: SlashContext) -> Type[Flavor]:
|
|
flavor_menu = copy.deepcopy(FLAVOR_MENU)
|
|
|
|
flavor_message = await ctx.send(
|
|
"Select a flavor to add alerts for", components=flavor_menu, ephemeral=True
|
|
)
|
|
try:
|
|
flavor_component: Component = await self.bot.wait_for_component(
|
|
messages=flavor_message, components=flavor_menu, timeout=30
|
|
)
|
|
except TimeoutError:
|
|
flavor_menu.disabled = True
|
|
await flavor_message.edit(
|
|
context=ctx, components=flavor_menu, content="Timed out"
|
|
)
|
|
raise TimeoutError
|
|
else:
|
|
# Acknowledge the component interaction to avoid 404 Unknown Interaction
|
|
await flavor_component.ctx.defer(edit_origin=True, suppress_error=True)
|
|
flavor = Flavor[flavor_component.ctx.values[0].upper()]
|
|
flavor_menu.disabled = True
|
|
await flavor_message.edit(context=ctx, components=flavor_menu)
|
|
return flavor
|
|
|
|
async def alert_category_select_menu(self, ctx: SlashContext) -> tuple[AlertCategory, int]:
|
|
alert_type_button = copy.deepcopy(ALERT_TYPE_ROW)
|
|
alert_type_message = await ctx.send(
|
|
"Select an alert type to add", components=alert_type_button, ephemeral=True
|
|
)
|
|
try:
|
|
alert_type_component: Component = await self.bot.wait_for_component(
|
|
messages=alert_type_message, components=alert_type_button, timeout=30
|
|
)
|
|
except TimeoutError:
|
|
for button in alert_type_button[0].components:
|
|
button.disabled = True
|
|
await alert_type_message.edit(
|
|
context=ctx, components=alert_type_button, content="Timed out"
|
|
)
|
|
raise TimeoutError
|
|
else:
|
|
alert_type = AlertCategory.from_str(alert_type_component.ctx.custom_id)
|
|
|
|
# If custom alert, send modal as response to button press
|
|
if alert_type == AlertCategory.CUSTOM:
|
|
price = await self.custom_price_modal(alert_type_component.ctx)
|
|
else:
|
|
# Acknowledge the component interaction to avoid 404 Unknown Interaction
|
|
await alert_type_component.ctx.defer(edit_origin=True, suppress_error=True)
|
|
price = 0
|
|
|
|
for button in alert_type_button[0].components:
|
|
button.disabled = True
|
|
await alert_type_message.edit(context=ctx, components=alert_type_button)
|
|
return alert_type, price
|
|
|
|
async def _alert_select_menu_handler(
|
|
self, ctx: SlashContext, menu: StringSelectMenu, message: Message
|
|
) -> AlertType:
|
|
try:
|
|
component: Component = await self.bot.wait_for_component(
|
|
messages=message, components=menu, timeout=30
|
|
)
|
|
except TimeoutError:
|
|
menu.disabled = True
|
|
await message.edit(context=ctx, components=menu, content="Timed out")
|
|
raise TimeoutError
|
|
else:
|
|
menu.disabled = True
|
|
await component.ctx.defer(edit_origin=True, suppress_error=True)
|
|
await message.edit(context=ctx, components=menu)
|
|
return AlertType.from_str(component.ctx.values[0])
|
|
|
|
async def high_alert_select_menu(self, ctx: SlashContext) -> AlertType:
|
|
high_menu = copy.deepcopy(HIGH_ALERT_MENU)
|
|
high_message = await ctx.send(
|
|
"Select a time range to add a High Alert for",
|
|
components=high_menu,
|
|
ephemeral=True,
|
|
)
|
|
return await self._alert_select_menu_handler(ctx, high_menu, high_message)
|
|
|
|
async def low_alert_select_menu(self, ctx: SlashContext) -> AlertType:
|
|
low_menu = copy.deepcopy(LOW_ALERT_MENU)
|
|
low_message = await ctx.send(
|
|
"Select a time range to add a Low Alert for",
|
|
components=low_menu,
|
|
ephemeral=True,
|
|
)
|
|
return await self._alert_select_menu_handler(ctx, low_menu, low_message)
|
|
|
|
async def custom_price_modal(self, ctx: ComponentContext) -> int:
|
|
modal = Modal(
|
|
ShortText(
|
|
label="Price (in gold)",
|
|
custom_id="price_input",
|
|
placeholder="e.g., 250000 for 250k gold",
|
|
required=True,
|
|
),
|
|
title="Custom Price Alert",
|
|
custom_id="custom_price_modal",
|
|
)
|
|
await ctx.send_modal(modal)
|
|
|
|
try:
|
|
modal_ctx: ModalContext = await self.bot.wait_for_modal(modal, timeout=300)
|
|
except TimeoutError:
|
|
await ctx.send("Modal timed out", ephemeral=True)
|
|
raise TimeoutError
|
|
else:
|
|
price_str = modal_ctx.responses["price_input"]
|
|
try:
|
|
price_gold = int(price_str.replace(",", "").replace(" ", "").replace("g", ""))
|
|
if price_gold <= 0:
|
|
await modal_ctx.send("Price must be greater than 0", ephemeral=True)
|
|
# Convert gold to copper (1 gold = 10000 copper)
|
|
price_copper = price_gold
|
|
await modal_ctx.send(f"Custom price alert set for {format(price_gold, ',')}g", ephemeral=True)
|
|
return price_copper
|
|
except ValueError:
|
|
await modal_ctx.send("Invalid price. Please enter a valid number.", ephemeral=True)
|
|
|
|
async def _user_is_registered(self, ctx: SlashContext) -> bool:
|
|
if not await self._users.exists(ctx.user.id):
|
|
await ctx.send(
|
|
"You are not registered with any region\n"
|
|
"Please add an alert to get started ```/add-alert```",
|
|
ephemeral=True,
|
|
)
|
|
return False
|
|
return True
|
|
|
|
async def _render_alert_flavor(
|
|
self, alerts: List[Alert], user: User | None = None
|
|
) -> Embed:
|
|
region = alerts[0].region
|
|
flavor = alerts[0].flavor
|
|
fields: List[EmbedField] = []
|
|
for alert in alerts:
|
|
history = self._history_manager.get_history(alert.flavor, alert.region)
|
|
trigger = await history.find_update_trigger_from_alert(alert)
|
|
if trigger.last_trigger is not None:
|
|
alert_str = (
|
|
f"Last Alerting Price Value: {format(trigger.last_alerting[1], ",")}\n"
|
|
f"Last Alerting Time: <t:{int(trigger.last_alerting[0].timestamp())}:F> local time\n"
|
|
f"[Link to this Chart]({self._render_token_url(alert)})\n"
|
|
)
|
|
if os.getenv("ENV") == "DEBUG":
|
|
alert_str += (
|
|
f"\nShowing you some internals since this is a DEBUG build:\n"
|
|
f"```history.last_price_datum:\n"
|
|
f"\t{history.last_price_datum[0].strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
|
|
f"\t{history.last_price_datum[1]}\n"
|
|
f"trigger.last_alerting:\n"
|
|
f"\t{trigger.last_alerting[0].strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
|
|
f"\t{trigger.last_alerting[1]}\n"
|
|
f"trigger.last_trigger:\n"
|
|
f"\t{trigger.last_trigger[0].strftime('%Y-%m-%d %H:%M:%S UTC')}\n"
|
|
f"\t{trigger.last_trigger[1]}\n"
|
|
f"trigger.squelched:\n\t{trigger.squelched}```"
|
|
)
|
|
else:
|
|
# For custom price alerts, show current status vs threshold
|
|
if alert.alert_type == AlertType.SPECIFIC_PRICE:
|
|
current_price = history.last_price_datum[1]
|
|
target_price_gold = alert.price
|
|
current_price_gold = current_price
|
|
|
|
alert_str = (
|
|
f"Threshold has never been crossed\n"
|
|
f"Current Price: {format(current_price_gold, ',')}g\n"
|
|
f"Target Price: {format(target_price_gold, ',')}g\n"
|
|
f"[Link to this Chart]({self._render_token_url(alert, time_range='72h')})\n"
|
|
)
|
|
else:
|
|
alert_str = "You should only be seeing this if the bot has not finished importing history at startup."
|
|
fields.append(
|
|
EmbedField(
|
|
name=f"{alert.to_human_string()} Alert",
|
|
value=alert_str,
|
|
inline=False,
|
|
)
|
|
)
|
|
embed = Embed(
|
|
title=f"Alerts for {region.name} {flavor.name.lower().title()}",
|
|
color=0xB10000,
|
|
fields=fields,
|
|
)
|
|
return embed
|
|
|
|
def _render_token_url(self, alert: Alert, time_range: str | None = None) -> str:
|
|
match alert.flavor:
|
|
case Flavor.CLASSIC:
|
|
url = "https://classic.wowtoken.app/?"
|
|
case Flavor.RETAIL:
|
|
url = "https://wowtoken.app/?"
|
|
case _:
|
|
raise NotImplementedError
|
|
url += f"region={alert.region.value}&"
|
|
|
|
# If time_range is explicitly provided, use it
|
|
if time_range:
|
|
url += f"time={time_range}&"
|
|
else:
|
|
# Otherwise, determine time range based on alert type
|
|
match alert.alert_type:
|
|
case AlertType.WEEKLY_LOW | AlertType.WEEKLY_HIGH:
|
|
url += "time=168h&"
|
|
case AlertType.MONTHLY_LOW | AlertType.MONTHLY_HIGH:
|
|
url += "time=720h&"
|
|
case AlertType.YEARLY_LOW | AlertType.YEARLY_HIGH:
|
|
url += "time=1y&"
|
|
case AlertType.ALL_TIME_LOW | AlertType.ALL_TIME_HIGH:
|
|
url += "time=all&"
|
|
case _:
|
|
url += "time=72h&"
|
|
|
|
return url
|