Compare commits

...

19 Commits

Author SHA1 Message Date
509a7d8a74 Make tracebacks only in debug builds 2026-01-28 18:22:06 -08:00
28fbdab904 Fix listing of alerts 2026-01-28 18:17:52 -08:00
4cb9d4ab70 Version Bump 2026-01-28 17:50:33 -08:00
35ad75cd7c Bug fix
- Validate and enforce positive price for SPECIFIC_PRICE alerts
- improve error handling and alert squelching logic.
2026-01-28 17:49:28 -08:00
62b20cf4ab Version bump 2025-11-06 02:47:37 -08:00
ed79f4b65c Initial implementation of custom price triggers
- likely to have some bugs, but this is good enough for a preview release.
2025-11-06 02:46:03 -08:00
19eb0a4e24 Version bump 2025-11-06 00:23:05 -08:00
faab0d5f7e Improve alert handling, centralize aiohttp.ClientSession, and fix minor bugs 2025-11-06 00:22:21 -08:00
7e387b6cd7 Version bump 2025-11-04 22:09:17 -08:00
f18b129907 Potential fix for "Unknown integration" 2025-11-04 22:05:53 -08:00
6a4b1f117b Version bump 2025-11-04 21:07:12 -08:00
fc57a7d172 Dependency bumps 2025-11-04 21:06:52 -08:00
0d0fe8e685 - If you unregister and then list your alerts, the bot will be sad 2025-11-04 21:06:40 -08:00
c5bb53c0e7 - Add error handling for Forbidden exceptions when sending alerts
- Improve alert message formatting
2025-11-04 20:34:07 -08:00
2164e98730 General cleanup
- Moved alert removal reminder to bottom of message

- debug output of alerts to discord only on debug environments
2024-12-20 18:58:42 -08:00
f20a8c6476 Version bump 2024-12-16 14:45:49 -08:00
3fffcf86c9 Format timestamps to be in local time zone provided by Discord 2024-12-16 14:45:36 -08:00
84561948da Help code update 2024-12-14 20:58:01 -08:00
621bd16d71 Formatting 2024-12-14 20:57:53 -08:00
13 changed files with 346 additions and 136 deletions

View File

@@ -1,4 +1,4 @@
FROM python:3.12-bookworm AS base
FROM python:3.12-trixie AS base
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

View File

@@ -1,5 +1,5 @@
import token_bot.token_bot as token_bot
if __name__ == '__main__':
if __name__ == "__main__":
bot = token_bot.TokenBot()
bot.run()

View File

@@ -1,27 +1,42 @@
aiodynamo==24.1
aiohttp==3.9.5
aiosignal==1.3.1
anyio==4.4.0
attrs==23.2.0
black==24.10.0
certifi==2024.7.4
croniter==2.0.5
discord-py-interactions==5.12.1
aiodynamo==24.7
aiohappyeyeballs==2.6.1
aiohttp==3.13.2
aiosignal==1.4.0
anyio==4.11.0
attrs==25.4.0
black==25.9.0
certifi==2025.10.5
cfgv==3.4.0
click==8.3.0
croniter==6.0.0
discord-py-interactions==5.15.0
discord-typings==0.9.0
emoji==2.12.1
frozenlist==1.4.1
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
multidict==6.0.5
pre-commit==4.0.1
distlib==0.4.0
emoji==2.15.0
filelock==3.20.0
frozenlist==1.8.0
h11==0.16.0
httpcore==1.0.9
httpx==0.28.1
identify==2.6.15
idna==3.11
multidict==6.7.0
mypy_extensions==1.1.0
nodeenv==1.9.1
packaging==25.0
pathspec==0.12.1
platformdirs==4.5.0
pre_commit==4.3.0
propcache==0.4.1
python-dateutil==2.9.0.post0
python-dotenv==1.0.1
pytz==2024.1
six==1.16.0
python-dotenv==1.2.1
pytokens==0.2.0
pytz==2025.2
PyYAML==6.0.3
six==1.17.0
sniffio==1.3.1
tomli==2.0.1
typing_extensions==4.12.2
uvloop==0.21.0
yarl==1.9.4
tomli==2.3.0
typing_extensions==4.15.0
uvloop==0.22.1
virtualenv==20.35.4
yarl==1.22.0

View File

@@ -11,7 +11,7 @@ from token_bot.persistant_database import database as pdb
class AlertsController:
def __init__(self, session: aiohttp.ClientSession):
self._pdb: pdb.Database = pdb.Database(session)
self.table = aiodynamo.client.Table = self._pdb.client.table(
self.table: aiodynamo.client.Table = self._pdb.client.table(
os.getenv("ALERTS_TABLE")
)
@@ -41,3 +41,18 @@ class AlertsController:
alert = self._alert_to_obj(alert)
user = self._user_to_obj(user)
await alert.remove_user(self.table, user)
async def get_all_by_type(self, alert_type: int) -> List[Alert]:
"""Query all alerts of a specific type from the database."""
from aiodynamo.expressions import F
alerts = []
async for item in self.table.query(
key_condition=F("alert").equals(alert_type)
):
alert = Alert.from_item(
primary_key=item["alert"],
sort_key=item["flavor-region-price"],
users=item.get("users", [])
)
alerts.append(alert)
return alerts

View File

@@ -8,7 +8,7 @@ from interactions.api.events import Startup
from token_bot.token_database import database as pdb
from token_bot.token_database import database as tdb
VERSION = "0.9.2"
VERSION = "0.9.12"
class Core(Extension):
@@ -22,11 +22,10 @@ class Core(Extension):
self.bot.logger.log(logging.INFO, f"This is bot version {VERSION}")
self._tdb = tdb.Database(aiohttp.ClientSession())
@slash_command()
@check(is_owner())
async def version(self, ctx):
await ctx.send(f"This is bot version {VERSION}", ephemeral=True)
@slash_command()
async def help(self, ctx):
await ctx.send(f"This is bot help command", ephemeral=True)
await ctx.send(
f"For help on using GoblinBot, please visit the help page found "
f"[here](https://blog.emily.sh/token-bot/#getting-started)",
ephemeral=True,
)

View File

@@ -15,10 +15,13 @@ class History:
self._last_price_movement: int = 0
self._latest_price_datum: Tuple[datetime.datetime, int] | None = None
self._update_triggers: List[UpdateTrigger] = []
# Create triggers for all non-custom alert types
for alert_type in AlertType:
self._update_triggers.append(
UpdateTrigger(Alert(alert_type, flavor, self._region))
)
if alert_type != AlertType.SPECIFIC_PRICE:
self._update_triggers.append(
UpdateTrigger(Alert(alert_type, flavor, self._region))
)
# SPECIFIC_PRICE triggers are created on-demand as they have unique prices
@property
def flavor(self) -> Flavor:
@@ -55,8 +58,21 @@ class History:
self._history.append(datum)
return await self._process_update_triggers()
async def find_update_trigger_from_alert(self, alert: Alert) -> UpdateTrigger:
async def find_update_trigger_from_alert(self, alert: Alert, initial_import: bool = False) -> UpdateTrigger:
for trigger in self._update_triggers:
if trigger.alert == alert:
return trigger
# If not found and it's a SPECIFIC_PRICE alert, create it on-demand
if alert.alert_type == AlertType.SPECIFIC_PRICE:
new_trigger = UpdateTrigger(alert)
if initial_import:
new_trigger.squelched = True
self._update_triggers.append(new_trigger)
# Initialize the trigger with current history
if self._latest_price_datum is not None:
new_trigger.check_and_update(self._latest_price_datum, self._history)
new_trigger.squelched = False
return new_trigger
raise ValueError

View File

@@ -20,7 +20,7 @@ class HistoryManager:
self._history[flavor][Region(region)] = History(flavor, Region(region))
async def _retrieve_data(
self, flavor: Flavor, region: Region
self, flavor: Flavor, region: Region
) -> List[Tuple[datetime.datetime, int]]:
high_fidelity_time = datetime.datetime.now(
tz=datetime.UTC
@@ -32,7 +32,7 @@ class HistoryManager:
final_response = []
def _convert_to_datetime(
data: Tuple[str, int]
data: Tuple[str, int]
) -> Tuple[datetime.datetime, int]:
return datetime.datetime.fromisoformat(data[0]), data[1]
@@ -57,6 +57,16 @@ class HistoryManager:
await history.add_price(item)
self._history[flavor][region] = history
async def load_custom_alerts(self, custom_alerts: List[Alert]):
"""Load custom price alerts and initialize their triggers with historical data."""
for alert in custom_alerts:
history = self._history[alert.flavor][alert.region]
# This will create the trigger on-demand via find_update_trigger_from_alert
trigger = await history.find_update_trigger_from_alert(alert)
# Process all historical data through this trigger to initialize its state
for datum in history.history:
trigger.check_and_update(datum, history.history)
async def update_data(self, flavor: Flavor, region: Region) -> List[Alert]:
history = self._history[flavor][region]
current_price_data = await self._tdb.current(flavor)

View File

@@ -29,25 +29,29 @@ class UpdateTrigger:
def squelched(self):
return self._squelched
@squelched.setter
def squelched(self, value):
self._squelched = value
def _find_next_trigger(
self,
comparison_operator: Callable,
starting_point: datetime.datetime,
history: List[Tuple[datetime.datetime, int]],
self,
comparison_operator: Callable,
starting_point: datetime.datetime,
history: List[Tuple[datetime.datetime, int]],
):
candidate_datum: Tuple[datetime.datetime, int] | None = None
for datum in history:
if datum[0] > starting_point and datum != history[-1]:
if candidate_datum is None or comparison_operator(
datum[1], candidate_datum[1]
datum[1], candidate_datum[1]
):
candidate_datum = datum
self._last_trigger = candidate_datum
def check_and_update(
self,
new_datum: Tuple[datetime.datetime, int],
history: List[Tuple[datetime.datetime, int]],
self,
new_datum: Tuple[datetime.datetime, int],
history: List[Tuple[datetime.datetime, int]],
) -> bool:
match self.alert.flavor:
case Flavor.RETAIL:
@@ -93,12 +97,49 @@ class UpdateTrigger:
case AlertType.ALL_TIME_HIGH:
time_range = now - start_time
comparison_operator = operator.gt
case AlertType.SPECIFIC_PRICE:
# For custom price alerts, check if the price crosses the threshold
# We alert when price moves from below to above (or vice versa)
target_price = self._alert.price
if self._last_trigger is None:
# First time - initialize tracking
self._last_trigger = new_datum
if new_datum[1] >= target_price:
# Price already at/above target - alert and squelch
self._last_alerting = new_datum
self._squelched = True
return True
return False
else:
# Check if we crossed the threshold
old_price = self._last_trigger[1]
new_price = new_datum[1]
# Alert if we cross the threshold in either direction
crossed_up = old_price < target_price <= new_price
crossed_down = old_price >= target_price > new_price
# Always update last_trigger for tracking
self._last_trigger = new_datum
if crossed_up or crossed_down:
# We're crossing the threshold
if self._squelched:
# Currently squelched - this crossing unsquelches us
# but doesn't alert (prevents rapid-fire alerts on oscillation)
self._squelched = False
return False
else:
# Not squelched - send alert and squelch
self._last_alerting = new_datum
self._squelched = True
return True
return False
case _:
# TODO: The logic here is certainly wrong for Custom
time_range = datetime.timedelta(days=int(365.25 * 6))
comparison_operator = operator.eq
if new_datum[0] > now - time_range:
if new_datum[0] > now - time_range and self._alert.alert_type != AlertType.SPECIFIC_PRICE:
if self._last_trigger is None:
self._last_trigger = new_datum
self._last_alerting = new_datum

View File

@@ -12,8 +12,13 @@ import token_bot.persistant_database as pdb
class Alert:
def __init__(
self, alert: pdb.AlertType, flavor: Flavor, region: Region, price: int = 0
self, alert: pdb.AlertType, flavor: Flavor, region: Region, price: int = 0
) -> None:
# Validate price for SPECIFIC_PRICE alerts
if alert == AlertType.SPECIFIC_PRICE:
if price is None or price <= 0:
raise ValueError("SPECIFIC_PRICE alerts require a positive price value")
# AlertType is the Primary Key
self._alert_type: pdb.AlertType = alert
# Flavor (Retail, Classic) is the Sort Key
@@ -91,14 +96,16 @@ class Alert:
def __eq__(self, other):
return (
self.alert_type == other.alert_type
and self.flavor == other.flavor
and self.price == other.price
self.alert_type == other.alert_type
and self.flavor == other.flavor
and self.region == other.region
and self.price == other.price
)
def to_human_string(self):
if self.alert_type == AlertType.SPECIFIC_PRICE:
raise NotImplementedError
price_gold = self.price
return f"Custom Price: {format(price_gold, ',')}g"
else:
alert_type_str = " ".join(self.alert_type.name.split("_"))
return f"{alert_type_str.title()}"
@@ -140,7 +147,7 @@ class Alert:
return self.users
async def add_user(
self, table: Table, user: pdb.User, consistent: bool = False
self, table: Table, user: pdb.User, consistent: bool = False
) -> None:
await self._lazy_load(table, consistent=consistent)
@@ -148,7 +155,7 @@ class Alert:
await self._append_user(table=table, user=user)
async def remove_user(
self, table: Table, user: pdb.User, consistent: bool = True
self, table: Table, user: pdb.User, consistent: bool = True
) -> None:
await self._lazy_load(table, consistent=consistent)

View File

@@ -38,4 +38,7 @@ class AlertType(Enum):
case "All Time Low":
return AlertType.ALL_TIME_LOW
case _:
# Check if it's a custom price format like "Custom Price: 250,000g"
if category.startswith("Custom Price"):
return AlertType.SPECIFIC_PRICE
return AlertType.SPECIFIC_PRICE

View File

@@ -15,7 +15,13 @@ class TokenBot:
)
log = logging.getLogger("TokenBotLogger")
log.setLevel(logging.INFO)
self.bot = Client(intents=Intents.DEFAULT, asyncio_debug=True, logger=log)
is_debug = os.getenv("ENV") == "DEBUG"
self.bot = Client(
intents=Intents.DEFAULT,
asyncio_debug=is_debug,
send_command_tracebacks=is_debug,
logger=log,
)
def run(self):
self.bot.load_extension("token_bot.core")

View File

@@ -2,6 +2,7 @@ import asyncio
import copy
import datetime
import logging
import os
from typing import Type, Dict, List
import aiohttp
@@ -16,12 +17,17 @@ from interactions import (
EmbedField,
is_owner,
check,
StringSelectOption,
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
@@ -58,6 +64,7 @@ class Tracker(Extension):
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 #
@@ -78,28 +85,47 @@ class Tracker(Extension):
else:
users_alerts[user].append(alert)
if users_alerts:
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Processing User 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"Hello, you requested to be sent an alert when the price of the World of Warcraft "
f"token reaches a certain value.\n\n"
f"As a reminder, you can remove an alert via ```/remove-alert```\n"
f"or you can remove all alerts and user data via ```/remove-registration```\n\n",
description=f"You requested to be alerted on the WoW token price. You have {alert_tally} {alert_word}\n\n",
)
]
alerts_by_flavor = await gather_alerts_by_flavor(users_alerts[user])
for flavor in alerts_by_flavor:
embeds.append(
await self._render_alert_flavor(alerts_by_flavor[flavor], user=user)
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")
await discord_user.send(embeds=embeds)
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Done Processing User Alerts")
self.bot.logger.log(
logging.INFO, "TokenBot Tracker: Done Processing User Alerts"
)
###################################
# Slash Commands #
@@ -108,9 +134,11 @@ class Tracker(Extension):
@listen(Startup)
async def on_start(self):
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Initializing")
self._users = UsersController(aiohttp.ClientSession())
self._alerts = AlertsController(aiohttp.ClientSession())
self._tdb = tdb.Database(aiohttp.ClientSession())
# 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")
@@ -118,13 +146,27 @@ class Tracker(Extension):
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"
@@ -140,6 +182,7 @@ class Tracker(Extension):
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)
@@ -149,24 +192,20 @@ class Tracker(Extension):
await ctx.send("All alert subscriptions and user data deleted", ephemeral=True)
@slash_command(
name="exists", description="Check if you are registered with GoblinBot" ""
)
@check(is_owner())
async def exists(self, ctx: SlashContext):
await ctx.send(str(await self._users.exists(ctx.user.id)), 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:
@@ -178,20 +217,20 @@ class Tracker(Extension):
try:
flavor = await self.flavor_select_menu(ctx)
alert_category = await self.alert_category_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 _:
raise NotImplementedError
case AlertCategory.CUSTOM:
alert_type = AlertType.SPECIFIC_PRICE
except TimeoutError:
except (TimeoutError, ValueError):
return
else:
alert = Alert(alert_type, flavor, user.region)
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),
@@ -207,6 +246,7 @@ class Tracker(Extension):
@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
@@ -230,6 +270,7 @@ class Tracker(Extension):
@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
@@ -257,24 +298,8 @@ class Tracker(Extension):
# Callbacks Commands #
###################################
@component_callback("flavor_menu")
async def flavor_menu(self, ctx: ComponentContext):
await ctx.send(f"Selected Flavor: {ctx.values[0]}", ephemeral=True)
@component_callback("high_alert_menu")
async def alert_menu(self, ctx: ComponentContext):
await ctx.send(f"Selected Alert: {ctx.values[0]}", ephemeral=True)
@component_callback("low_alert_menu")
async def alert_menu(self, ctx: ComponentContext):
await ctx.send(f"Selected Alert: {ctx.values[0]}", ephemeral=True)
@component_callback("remove_alert_menu")
async def remove_alert_menu(self, ctx: ComponentContext):
await ctx.send(
f"You have selected to remove the following alert: {ctx.values[0].title()}",
ephemeral=True,
)
# 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):
@@ -286,24 +311,17 @@ class Tracker(Extension):
)
await ctx.defer(edit_origin=True, suppress_error=True)
@component_callback("high_alert_button")
async def high_alert_button(self, ctx: ComponentContext):
await ctx.send("You selected to add a High Price Alert", ephemeral=True)
@component_callback("low_alert_button")
async def low_alert_button(self, ctx: ComponentContext):
await ctx.send("You selected to add a Low Price Alert", ephemeral=True)
@component_callback("custom_alert_button")
async def custom_alert_button(self, ctx: ComponentContext):
await ctx.send("You selected to add a Custom Price Alert", ephemeral=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, ",")
@@ -312,7 +330,7 @@ class Tracker(Extension):
return (
f"Last Price Value for {region}: {format(region_history.last_price_datum[1], ",")}\n"
f"Last Update Time: {region_history.last_price_datum[0].strftime('%Y-%m-%d %H:%M:%S UTC')}\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}"
)
@@ -349,8 +367,19 @@ class Tracker(Extension):
await message.edit(context=ctx, components=menu)
selection_split = alert_component.ctx.values[0].split(" ")
flavor = Flavor[selection_split[0].upper()]
alert_type = AlertType.from_str(" ".join(selection_split[1:]))
return Alert(alert_type, flavor, user.region)
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)
@@ -378,6 +407,8 @@ class Tracker(Extension):
)
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=[])
@@ -404,12 +435,14 @@ class Tracker(Extension):
)
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) -> AlertCategory:
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
@@ -427,13 +460,22 @@ class Tracker(Extension):
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
return alert_type, price
async def _alert_select_menu_handler(
self, ctx: SlashContext, menu: StringSelectMenu, message: Message
self, ctx: SlashContext, menu: StringSelectMenu, message: Message
) -> AlertType:
try:
component: Component = await self.bot.wait_for_component(
@@ -467,6 +509,39 @@ class Tracker(Extension):
)
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", ""))
except ValueError:
await modal_ctx.send("Invalid price. Please enter a valid number.", ephemeral=True)
raise
if price_gold <= 0:
await modal_ctx.send("Price must be greater than 0", ephemeral=True)
raise ValueError("Price must be greater than 0")
await modal_ctx.send(f"Custom price alert set for {format(price_gold, ',')}g", ephemeral=True)
return price_gold
async def _user_is_registered(self, ctx: SlashContext) -> bool:
if not await self._users.exists(ctx.user.id):
await ctx.send(
@@ -478,7 +553,7 @@ class Tracker(Extension):
return True
async def _render_alert_flavor(
self, alerts: List[Alert], user: User | None = None
self, alerts: List[Alert], user: User | None = None
) -> Embed:
region = alerts[0].region
flavor = alerts[0].flavor
@@ -486,15 +561,15 @@ class Tracker(Extension):
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:
if trigger.last_alerting is not None:
alert_str = (
f"Last Alerting Price Value: {format(trigger.last_alerting[1], ",")}\n"
f"Last Alerting Time: {trigger.last_alerting[0].strftime('%Y-%m-%d %H:%M:%S UTC')}\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 user is not None and user.user_id == 265678699435655169:
if os.getenv("ENV") == "DEBUG":
alert_str += (
f"\nShowing you some internals since you are the bot owner:\n"
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"
@@ -507,7 +582,20 @@ class Tracker(Extension):
f"trigger.squelched:\n\t{trigger.squelched}```"
)
else:
alert_str = "You should only be seeing this if the bot has not finished importing history at startup."
# 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",
@@ -522,7 +610,7 @@ class Tracker(Extension):
)
return embed
def _render_token_url(self, alert: Alert) -> str:
def _render_token_url(self, alert: Alert, time_range: str | None = None) -> str:
match alert.flavor:
case Flavor.CLASSIC:
url = "https://classic.wowtoken.app/?"
@@ -531,14 +619,22 @@ class Tracker(Extension):
case _:
raise NotImplementedError
url += f"region={alert.region.value}&"
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&"
# 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

View File

@@ -3,11 +3,13 @@ from interactions import ActionRow
from token_bot.ui.buttons.tracker.alert_category import (
HIGH_ALERT_BUTTON,
LOW_ALERT_BUTTON,
CUSTOM_ALERT_BUTTON,
)
ALERT_TYPE_ROW: list[ActionRow] = [
ActionRow(
HIGH_ALERT_BUTTON,
LOW_ALERT_BUTTON,
CUSTOM_ALERT_BUTTON,
)
]