Compare commits

...

13 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
12 changed files with 304 additions and 109 deletions

View File

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

View File

@@ -1,27 +1,42 @@
aiodynamo==24.1 aiodynamo==24.7
aiohttp==3.9.5 aiohappyeyeballs==2.6.1
aiosignal==1.3.1 aiohttp==3.13.2
anyio==4.4.0 aiosignal==1.4.0
attrs==23.2.0 anyio==4.11.0
black==24.10.0 attrs==25.4.0
certifi==2024.7.4 black==25.9.0
croniter==2.0.5 certifi==2025.10.5
discord-py-interactions==5.12.1 cfgv==3.4.0
click==8.3.0
croniter==6.0.0
discord-py-interactions==5.15.0
discord-typings==0.9.0 discord-typings==0.9.0
emoji==2.12.1 distlib==0.4.0
frozenlist==1.4.1 emoji==2.15.0
h11==0.14.0 filelock==3.20.0
httpcore==1.0.5 frozenlist==1.8.0
httpx==0.27.0 h11==0.16.0
idna==3.7 httpcore==1.0.9
multidict==6.0.5 httpx==0.28.1
pre-commit==4.0.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-dateutil==2.9.0.post0
python-dotenv==1.0.1 python-dotenv==1.2.1
pytz==2024.1 pytokens==0.2.0
six==1.16.0 pytz==2025.2
PyYAML==6.0.3
six==1.17.0
sniffio==1.3.1 sniffio==1.3.1
tomli==2.0.1 tomli==2.3.0
typing_extensions==4.12.2 typing_extensions==4.15.0
uvloop==0.21.0 uvloop==0.22.1
yarl==1.9.4 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: class AlertsController:
def __init__(self, session: aiohttp.ClientSession): def __init__(self, session: aiohttp.ClientSession):
self._pdb: pdb.Database = pdb.Database(session) 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") os.getenv("ALERTS_TABLE")
) )
@@ -41,3 +41,18 @@ class AlertsController:
alert = self._alert_to_obj(alert) alert = self._alert_to_obj(alert)
user = self._user_to_obj(user) user = self._user_to_obj(user)
await alert.remove_user(self.table, 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 pdb
from token_bot.token_database import database as tdb from token_bot.token_database import database as tdb
VERSION = "0.9.3" VERSION = "0.9.12"
class Core(Extension): class Core(Extension):

View File

@@ -15,10 +15,13 @@ class History:
self._last_price_movement: int = 0 self._last_price_movement: int = 0
self._latest_price_datum: Tuple[datetime.datetime, int] | None = None self._latest_price_datum: Tuple[datetime.datetime, int] | None = None
self._update_triggers: List[UpdateTrigger] = [] self._update_triggers: List[UpdateTrigger] = []
# Create triggers for all non-custom alert types
for alert_type in AlertType: for alert_type in AlertType:
self._update_triggers.append( if alert_type != AlertType.SPECIFIC_PRICE:
UpdateTrigger(Alert(alert_type, flavor, self._region)) self._update_triggers.append(
) UpdateTrigger(Alert(alert_type, flavor, self._region))
)
# SPECIFIC_PRICE triggers are created on-demand as they have unique prices
@property @property
def flavor(self) -> Flavor: def flavor(self) -> Flavor:
@@ -55,8 +58,21 @@ class History:
self._history.append(datum) self._history.append(datum)
return await self._process_update_triggers() 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: for trigger in self._update_triggers:
if trigger.alert == alert: if trigger.alert == alert:
return trigger 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 raise ValueError

View File

@@ -20,7 +20,7 @@ class HistoryManager:
self._history[flavor][Region(region)] = History(flavor, Region(region)) self._history[flavor][Region(region)] = History(flavor, Region(region))
async def _retrieve_data( async def _retrieve_data(
self, flavor: Flavor, region: Region self, flavor: Flavor, region: Region
) -> List[Tuple[datetime.datetime, int]]: ) -> List[Tuple[datetime.datetime, int]]:
high_fidelity_time = datetime.datetime.now( high_fidelity_time = datetime.datetime.now(
tz=datetime.UTC tz=datetime.UTC
@@ -32,7 +32,7 @@ class HistoryManager:
final_response = [] final_response = []
def _convert_to_datetime( def _convert_to_datetime(
data: Tuple[str, int] data: Tuple[str, int]
) -> Tuple[datetime.datetime, int]: ) -> Tuple[datetime.datetime, int]:
return datetime.datetime.fromisoformat(data[0]), data[1] return datetime.datetime.fromisoformat(data[0]), data[1]
@@ -57,6 +57,16 @@ class HistoryManager:
await history.add_price(item) await history.add_price(item)
self._history[flavor][region] = history 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]: async def update_data(self, flavor: Flavor, region: Region) -> List[Alert]:
history = self._history[flavor][region] history = self._history[flavor][region]
current_price_data = await self._tdb.current(flavor) current_price_data = await self._tdb.current(flavor)

View File

@@ -29,25 +29,29 @@ class UpdateTrigger:
def squelched(self): def squelched(self):
return self._squelched return self._squelched
@squelched.setter
def squelched(self, value):
self._squelched = value
def _find_next_trigger( def _find_next_trigger(
self, self,
comparison_operator: Callable, comparison_operator: Callable,
starting_point: datetime.datetime, starting_point: datetime.datetime,
history: List[Tuple[datetime.datetime, int]], history: List[Tuple[datetime.datetime, int]],
): ):
candidate_datum: Tuple[datetime.datetime, int] | None = None candidate_datum: Tuple[datetime.datetime, int] | None = None
for datum in history: for datum in history:
if datum[0] > starting_point and datum != history[-1]: if datum[0] > starting_point and datum != history[-1]:
if candidate_datum is None or comparison_operator( if candidate_datum is None or comparison_operator(
datum[1], candidate_datum[1] datum[1], candidate_datum[1]
): ):
candidate_datum = datum candidate_datum = datum
self._last_trigger = candidate_datum self._last_trigger = candidate_datum
def check_and_update( def check_and_update(
self, self,
new_datum: Tuple[datetime.datetime, int], new_datum: Tuple[datetime.datetime, int],
history: List[Tuple[datetime.datetime, int]], history: List[Tuple[datetime.datetime, int]],
) -> bool: ) -> bool:
match self.alert.flavor: match self.alert.flavor:
case Flavor.RETAIL: case Flavor.RETAIL:
@@ -93,12 +97,49 @@ class UpdateTrigger:
case AlertType.ALL_TIME_HIGH: case AlertType.ALL_TIME_HIGH:
time_range = now - start_time time_range = now - start_time
comparison_operator = operator.gt 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 _: case _:
# TODO: The logic here is certainly wrong for Custom
time_range = datetime.timedelta(days=int(365.25 * 6)) time_range = datetime.timedelta(days=int(365.25 * 6))
comparison_operator = operator.eq 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: if self._last_trigger is None:
self._last_trigger = new_datum self._last_trigger = new_datum
self._last_alerting = new_datum self._last_alerting = new_datum

View File

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

View File

@@ -38,4 +38,7 @@ class AlertType(Enum):
case "All Time Low": case "All Time Low":
return AlertType.ALL_TIME_LOW return AlertType.ALL_TIME_LOW
case _: 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 return AlertType.SPECIFIC_PRICE

View File

@@ -15,7 +15,13 @@ class TokenBot:
) )
log = logging.getLogger("TokenBotLogger") log = logging.getLogger("TokenBotLogger")
log.setLevel(logging.INFO) 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): def run(self):
self.bot.load_extension("token_bot.core") self.bot.load_extension("token_bot.core")

View File

@@ -17,7 +17,11 @@ from interactions import (
EmbedField, EmbedField,
is_owner, is_owner,
check, check,
StringSelectOption, StringSelectOption, integration_types,
Modal,
ShortText,
modal_callback,
ModalContext,
) )
from interactions import Task, IntervalTrigger from interactions import Task, IntervalTrigger
from interactions import slash_command, listen from interactions import slash_command, listen
@@ -60,6 +64,7 @@ class Tracker(Extension):
self._alerts: AlertsController | None = None self._alerts: AlertsController | None = None
self._tdb: tdb.Database | None = None self._tdb: tdb.Database | None = None
self._history_manager: HistoryManager | None = None self._history_manager: HistoryManager | None = None
self._session: aiohttp.ClientSession | None = None
################################### ###################################
# Task Functions # # Task Functions #
@@ -87,12 +92,10 @@ class Tracker(Extension):
discord_user = await self.bot.fetch_user(user.user_id) discord_user = await self.bot.fetch_user(user.user_id)
alerts_by_flavor = await gather_alerts_by_flavor(users_alerts[user]) alerts_by_flavor = await gather_alerts_by_flavor(users_alerts[user])
alert_tally = 0 alert_tally = 0
alert_word = "alert"
if alert_tally > 2:
alert_word += 's'
for flavor in alerts_by_flavor: for flavor in alerts_by_flavor:
for _ in alerts_by_flavor[flavor]: for _ in alerts_by_flavor[flavor]:
alert_tally += 1 alert_tally += 1
alert_word = "alert" if alert_tally == 1 else "alerts"
embeds = [ embeds = [
Embed( Embed(
title="GoblinBot Tracker Alert Triggered", title="GoblinBot Tracker Alert Triggered",
@@ -131,9 +134,11 @@ class Tracker(Extension):
@listen(Startup) @listen(Startup)
async def on_start(self): async def on_start(self):
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Initializing") self.bot.logger.log(logging.INFO, "TokenBot Tracker: Initializing")
self._users = UsersController(aiohttp.ClientSession()) # Create a single shared ClientSession for all components
self._alerts = AlertsController(aiohttp.ClientSession()) self._session = aiohttp.ClientSession()
self._tdb = tdb.Database(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._history_manager = HistoryManager(self._tdb)
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Initialized") self.bot.logger.log(logging.INFO, "TokenBot Tracker: Initialized")
self.bot.logger.log(logging.INFO, "TokenBot Tracker: Loading Historical Data") self.bot.logger.log(logging.INFO, "TokenBot Tracker: Loading Historical Data")
@@ -141,13 +146,27 @@ class Tracker(Extension):
self.bot.logger.log( self.bot.logger.log(
logging.INFO, "TokenBot Tracker: Loading Historical Data Finished" 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.bot.logger.log(logging.INFO, "TokenBot Tracker: Started")
self.update_data.start() 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( @slash_command(
name="register", name="register",
description="Register with a new GoblinBot Region for alerts on token price changes.", 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): async def register(self, ctx: SlashContext):
text = ( text = (
"## Select a region to register with \n\n" "## Select a region to register with \n\n"
@@ -163,6 +182,7 @@ class Tracker(Extension):
name="remove-registration", name="remove-registration",
description="Remove all alerts and registration from GoblinBot", description="Remove all alerts and registration from GoblinBot",
) )
@integration_types(guild=True, user=True)
async def remove_registration(self, ctx: SlashContext): async def remove_registration(self, ctx: SlashContext):
if await self._users.exists(ctx.user.id): if await self._users.exists(ctx.user.id):
user = await self._users.get(ctx.user.id) user = await self._users.get(ctx.user.id)
@@ -173,16 +193,19 @@ class Tracker(Extension):
await ctx.send("All alert subscriptions and user data deleted", ephemeral=True) await ctx.send("All alert subscriptions and user data deleted", ephemeral=True)
@slash_command(description="The current retail token cost") @slash_command(description="The current retail token cost")
@integration_types(guild=True, user=True)
async def current(self, ctx: SlashContext): async def current(self, ctx: SlashContext):
current_str = await self.get_current_token(ctx, tdb.Flavor.RETAIL) current_str = await self.get_current_token(ctx, tdb.Flavor.RETAIL)
await ctx.send(current_str, ephemeral=True) await ctx.send(current_str, ephemeral=True)
@slash_command(description="The current classic token cost") @slash_command(description="The current classic token cost")
@integration_types(guild=True, user=True)
async def current_classic(self, ctx: SlashContext): async def current_classic(self, ctx: SlashContext):
current_str = await self.get_current_token(ctx, tdb.Flavor.CLASSIC) current_str = await self.get_current_token(ctx, tdb.Flavor.CLASSIC)
await ctx.send(current_str, ephemeral=True) await ctx.send(current_str, ephemeral=True)
@slash_command(name="add-alert", description="Add an alert listener") @slash_command(name="add-alert", description="Add an alert listener")
@integration_types(guild=True, user=True)
async def add_alert(self, ctx: SlashContext): async def add_alert(self, ctx: SlashContext):
if not await self._users.exists(ctx.user.id): if not await self._users.exists(ctx.user.id):
try: try:
@@ -194,20 +217,20 @@ class Tracker(Extension):
try: try:
flavor = await self.flavor_select_menu(ctx) 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: match alert_category:
case AlertCategory.LOW: case AlertCategory.LOW:
alert_type = await self.low_alert_select_menu(ctx) alert_type = await self.low_alert_select_menu(ctx)
case AlertCategory.HIGH: case AlertCategory.HIGH:
alert_type = await self.high_alert_select_menu(ctx) alert_type = await self.high_alert_select_menu(ctx)
case _: case AlertCategory.CUSTOM:
raise NotImplementedError alert_type = AlertType.SPECIFIC_PRICE
except TimeoutError: except (TimeoutError, ValueError):
return return
else: 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): if not await self._users.is_subscribed(user, alert):
await asyncio.gather( await asyncio.gather(
self._users.add_alert(user, alert), self._users.add_alert(user, alert),
@@ -223,6 +246,7 @@ class Tracker(Extension):
@slash_command( @slash_command(
name="remove-alert", description="Remove an alert you have signed up for" 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): async def remove_alert(self, ctx: SlashContext):
if not await self._user_is_registered(ctx): if not await self._user_is_registered(ctx):
return return
@@ -246,6 +270,7 @@ class Tracker(Extension):
@slash_command( @slash_command(
name="list-alerts", description="List all alerts you have signed up for" 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): async def list_alerts(self, ctx: SlashContext):
if not await self._user_is_registered(ctx): if not await self._user_is_registered(ctx):
return return
@@ -273,24 +298,8 @@ class Tracker(Extension):
# Callbacks Commands # # Callbacks Commands #
################################### ###################################
@component_callback("flavor_menu") # Note: Callbacks for flavor_menu, high_alert_menu, low_alert_menu, and alert buttons
async def flavor_menu(self, ctx: ComponentContext): # are disabled because they interfere with wait_for_component manual handling
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,
)
@component_callback("region_menu") @component_callback("region_menu")
async def region_menu_cb(self, ctx: ComponentContext): async def region_menu_cb(self, ctx: ComponentContext):
@@ -302,24 +311,17 @@ class Tracker(Extension):
) )
await ctx.defer(edit_origin=True, suppress_error=True) 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 # # Helper Functions #
################################### ###################################
async def get_current_token(self, ctx: SlashContext, flavor: Flavor) -> str: async def get_current_token(self, ctx: SlashContext, flavor: Flavor) -> str:
user: User = await self._users.get(ctx.user.id) 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 = user.region.name
region_history = self._history_manager.get_history(flavor, user.region) region_history = self._history_manager.get_history(flavor, user.region)
price_movement_str = format(region_history.last_price_movement, ",") price_movement_str = format(region_history.last_price_movement, ",")
@@ -365,8 +367,19 @@ class Tracker(Extension):
await message.edit(context=ctx, components=menu) await message.edit(context=ctx, components=menu)
selection_split = alert_component.ctx.values[0].split(" ") selection_split = alert_component.ctx.values[0].split(" ")
flavor = Flavor[selection_split[0].upper()] flavor = Flavor[selection_split[0].upper()]
alert_type = AlertType.from_str(" ".join(selection_split[1:])) alert_type_str = " ".join(selection_split[1:])
return Alert(alert_type, flavor, user.region) 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): async def region_select_menu(self, ctx: SlashContext, user: User | None = None):
region_menu = copy.deepcopy(REGION_MENU) region_menu = copy.deepcopy(REGION_MENU)
@@ -394,6 +407,8 @@ class Tracker(Extension):
) )
raise TimeoutError raise TimeoutError
else: 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_menu.disabled = True
region = Region(region_component.ctx.values[0].lower()) region = Region(region_component.ctx.values[0].lower())
user = User(ctx.user.id, region, subscribed_alerts=[]) user = User(ctx.user.id, region, subscribed_alerts=[])
@@ -420,12 +435,14 @@ class Tracker(Extension):
) )
raise TimeoutError raise TimeoutError
else: 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 = Flavor[flavor_component.ctx.values[0].upper()]
flavor_menu.disabled = True flavor_menu.disabled = True
await flavor_message.edit(context=ctx, components=flavor_menu) await flavor_message.edit(context=ctx, components=flavor_menu)
return flavor 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_button = copy.deepcopy(ALERT_TYPE_ROW)
alert_type_message = await ctx.send( alert_type_message = await ctx.send(
"Select an alert type to add", components=alert_type_button, ephemeral=True "Select an alert type to add", components=alert_type_button, ephemeral=True
@@ -443,10 +460,19 @@ class Tracker(Extension):
raise TimeoutError raise TimeoutError
else: else:
alert_type = AlertCategory.from_str(alert_type_component.ctx.custom_id) 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: for button in alert_type_button[0].components:
button.disabled = True button.disabled = True
await alert_type_message.edit(context=ctx, components=alert_type_button) await alert_type_message.edit(context=ctx, components=alert_type_button)
return alert_type return alert_type, price
async def _alert_select_menu_handler( async def _alert_select_menu_handler(
self, ctx: SlashContext, menu: StringSelectMenu, message: Message self, ctx: SlashContext, menu: StringSelectMenu, message: Message
@@ -483,6 +509,39 @@ class Tracker(Extension):
) )
return await self._alert_select_menu_handler(ctx, low_menu, low_message) 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: async def _user_is_registered(self, ctx: SlashContext) -> bool:
if not await self._users.exists(ctx.user.id): if not await self._users.exists(ctx.user.id):
await ctx.send( await ctx.send(
@@ -502,7 +561,7 @@ class Tracker(Extension):
for alert in alerts: for alert in alerts:
history = self._history_manager.get_history(alert.flavor, alert.region) history = self._history_manager.get_history(alert.flavor, alert.region)
trigger = await history.find_update_trigger_from_alert(alert) 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 = ( alert_str = (
f"Last Alerting Price Value: {format(trigger.last_alerting[1], ",")}\n" 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"Last Alerting Time: <t:{int(trigger.last_alerting[0].timestamp())}:F> local time\n"
@@ -523,7 +582,20 @@ class Tracker(Extension):
f"trigger.squelched:\n\t{trigger.squelched}```" f"trigger.squelched:\n\t{trigger.squelched}```"
) )
else: 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( fields.append(
EmbedField( EmbedField(
name=f"{alert.to_human_string()} Alert", name=f"{alert.to_human_string()} Alert",
@@ -538,7 +610,7 @@ class Tracker(Extension):
) )
return embed 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: match alert.flavor:
case Flavor.CLASSIC: case Flavor.CLASSIC:
url = "https://classic.wowtoken.app/?" url = "https://classic.wowtoken.app/?"
@@ -547,14 +619,22 @@ class Tracker(Extension):
case _: case _:
raise NotImplementedError raise NotImplementedError
url += f"region={alert.region.value}&" url += f"region={alert.region.value}&"
match alert.alert_type:
case AlertType.WEEKLY_LOW | AlertType.WEEKLY_HIGH: # If time_range is explicitly provided, use it
url += "time=168h&" if time_range:
case AlertType.MONTHLY_LOW | AlertType.MONTHLY_HIGH: url += f"time={time_range}&"
url += "time=720h&" else:
case AlertType.YEARLY_LOW | AlertType.YEARLY_HIGH: # Otherwise, determine time range based on alert type
url += "time=1y&" match alert.alert_type:
case AlertType.ALL_TIME_LOW | AlertType.ALL_TIME_HIGH: case AlertType.WEEKLY_LOW | AlertType.WEEKLY_HIGH:
url += "time=all&" 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 return url

View File

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