Edits are saved to the templates JSON file on disk.
" + "{content}" + "" +) + + +def get_html_template(content) -> str: + return HTML_TEMPLATE.format(content=content) + + def create_app( *, get_state: Callable[[], dict], @@ -32,11 +47,7 @@ def create_app( locations = state.get("last_locations") or [] locations_html = "".join(f"Total: {len(locations)}
" f"Edits are saved to the templates JSON file on disk.
" "Errors:
{msg}
Saved.
" "" - "", - 200, - ) + ), 200 return app diff --git a/main.py b/main.py index 4505a40..744fa6d 100644 --- a/main.py +++ b/main.py @@ -10,6 +10,14 @@ import schedule from dotenv import load_dotenv from templates import load_templates +from dashboard import create_app + +SCHEDULED_NOTIFICATIONS = [ + (15, "reminder"), + (20, "420"), + (45, "reminder_halftime"), + (50, "halftime"), +] TZDB_CACHE: dict | None = None @@ -42,6 +50,15 @@ DISCORD_CHANNEL_ID = os.getenv('DISCORD_CHANNEL_ID') GUILD_ID = os.getenv('DISCORD_GUILD_ID') +def get_state() -> dict: + with STATE_LOCK: + return dict(STATE) + + +def get_next_event() -> dict: + return get_next_scheduled_event() + + def init_tzdb_cache() -> dict: """Initialize a cached lookup structure for tzdb data. @@ -139,15 +156,8 @@ def get_next_scheduled_event(now: datetime | None = None) -> dict: if now is None: now = datetime.now() - schedule_map = [ - (15, "reminder"), - (20, "420"), - (45, "reminder_halftime"), - (50, "halftime"), - ] - candidates: list[tuple[datetime, str]] = [] - for minute, msg_type in schedule_map: + for minute, msg_type in SCHEDULED_NOTIFICATIONS: candidate = now.replace(minute=minute, second=0, microsecond=0) if candidate > now: candidates.append((candidate, msg_type)) @@ -155,8 +165,8 @@ def get_next_scheduled_event(now: datetime | None = None) -> dict: if not candidates: base = (now + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0) - next_dt = base.replace(minute=schedule_map[0][0]) - return {"at": next_dt, "type": schedule_map[0][1]} + next_dt = base.replace(minute=SCHEDULED_NOTIFICATIONS[0][0]) + return {"at": next_dt, "type": SCHEDULED_NOTIFICATIONS[0][1]} next_dt, next_type = min(candidates, key=lambda x: x[0]) return {"at": next_dt, "type": next_type} @@ -706,15 +716,33 @@ def where_is_it_420( return results +def schedule_notification(interval: str, at: str, type: str) -> None: + """Example: schedule.every().hour.at(":15").do(send_notification, "reminder")""" + if interval == "hour": + schedule.every().hour.at(at).do(send_notification, type) + elif interval == "day": + schedule.every().day.at(at).do(send_notification, type) + else: + logging.error(f"Unsupported interval: {interval}") + + +def start_dashboard() -> None: + """Compatibility hook for tests and optional dashboard startup.""" + app = create_app(get_state=get_state, get_next_event=get_next_event) + app.run(host="127.0.0.1", port=8080, debug=False, use_reloader=False) + + def main() -> None: """ Main function to run the scheduler. """ - # Schedule notifications - schedule.every().hour.at(":15").do(send_notification, "reminder") - schedule.every().hour.at(":20").do(send_notification, "420") - # schedule.every().hour.at(":45").do(send_notification, "reminder_halftime") - # schedule.every().hour.at(":50").do(send_notification, "halftime") + # Start the dashboard in a separate thread + dashboard_thread = threading.Thread(target=start_dashboard, daemon=True) + dashboard_thread.start() + + # Schedule notifications based on the defined SCHEDULED_NOTIFICATIONS + for minute, msg_type in SCHEDULED_NOTIFICATIONS: + schedule_notification("hour", f":{minute:02d}", msg_type) # Schedule deletion of old messages every 5 minutes schedule.every(5).minutes.do(delete_old_messages, 6) @@ -722,7 +750,9 @@ def main() -> None: logging.info("Scheduler started.") # Test the notification on startup - # send_notification("420") + send_notification("test") + # delete the test message after a short delay to keep the channel clean + schedule.every(1).minutes.do(delete_old_messages, 1) # delete old messages on startup to clean up any previous notifications # delete_old_messages(6) diff --git a/templates.json b/templates.json index 21b12ac..ecd3cce 100644 --- a/templates.json +++ b/templates.json @@ -16,5 +16,9 @@ "reminder_halftime": { "color": 15105570, "text": "Half-time in 5 minutes!" + }, + "test": { + "color": 3447003, + "text": "This is a test notification." } } diff --git a/templates.py b/templates.py index dfc4a76..69c762e 100644 --- a/templates.py +++ b/templates.py @@ -6,12 +6,8 @@ from pathlib import Path DEFAULT_TEMPLATES: dict[str, dict] = { - "reminder_halftime": { - "text": "Half-time in 5 minutes!", - "color": 0xE67E22, - }, - "halftime": { - "text": "Half-time!", + "420": { + "text": "Blaze it!", "color": 0x2ECC71, "image_url": "https://copyparty.allucanget.biz/img/weed.png", }, @@ -19,11 +15,19 @@ DEFAULT_TEMPLATES: dict[str, dict] = { "text": "This is your 5 minute reminder to 420!", "color": 0xE67E22, }, - "420": { - "text": "Blaze it!", + "halftime": { + "text": "Half-time!", "color": 0x2ECC71, "image_url": "https://copyparty.allucanget.biz/img/weed.png", }, + "reminder_halftime": { + "text": "Half-time in 5 minutes!", + "color": 0xE67E22, + }, + "test": { + "text": "This is a test notification.", + "color": 0x3498DB, + }, } @@ -73,7 +77,8 @@ def save_templates(path: str | Path, templates: dict) -> None: p.parent.mkdir(parents=True, exist_ok=True) tmp = p.with_suffix(p.suffix + ".tmp") - tmp.write_text(json.dumps(normalized, indent=2, sort_keys=True) + "\n", encoding="utf-8") + tmp.write_text(json.dumps(normalized, indent=2, + sort_keys=True) + "\n", encoding="utf-8") tmp.replace(p)