zwitschi/thc-webhook
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add HTML template rendering and test notification functionality
Build and Deploy Docker Container / test (push) Successful in 32s
Details
Build and Deploy Docker Container / build-and-deploy (push) Failing after 58s
Details

Browse Source
This commit is contained in:
zwitschi
2026-05-10 12:01:25 +02:00
parent 1dcd80a8bb
commit 6bae1e2f66
4 changed files with 85 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dashboard.py
+21 -22
View File
@@ -17,6 +17,21 @@ def _fmt_dt(dt: datetime | None) -> str:
return str(dt) return str(dt)
HTML_TEMPLATE = (
"<!doctype html><html><head><meta charset='utf-8'><title>thc-webhook admin</title>"
"<style>body{{font-family:sans-serif;max-width:900px;margin:24px;}}textarea,input{{font-family:inherit;}}</style>"
"</head><body>"
"<h1>Admin: templates</h1>"
"<p>Edits are saved to the templates JSON file on disk.</p>"
"{content}"
"</body></html>"
)
def get_html_template(content) -> str:
return HTML_TEMPLATE.format(content=content)
def create_app( def create_app(
*, *,
get_state: Callable[[], dict], get_state: Callable[[], dict],
@@ -32,11 +47,7 @@ def create_app(
locations = state.get("last_locations") or [] locations = state.get("last_locations") or []
locations_html = "".join(f"<li>{loc}</li>" for loc in locations) locations_html = "".join(f"<li>{loc}</li>" for loc in locations)
return ( return get_html_template(
"<!doctype html>"
"<html><head><meta charset='utf-8'><title>thc-webhook</title>"
"<style>body{font-family:sans-serif;max-width:900px;margin:24px;}</style>"
"</head><body>"
"<h1>thc-webhook</h1>" "<h1>thc-webhook</h1>"
"<h2>Status</h2>" "<h2>Status</h2>"
"<ul>" "<ul>"
@@ -56,7 +67,6 @@ def create_app(
"<h2>Locations (latest)</h2>" "<h2>Locations (latest)</h2>"
f"<p>Total: {len(locations)}</p>" f"<p>Total: {len(locations)}</p>"
f"<ul>{locations_html}</ul>" f"<ul>{locations_html}</ul>"
"</body></html>"
) )
@app.get("/admin") @app.get("/admin")
@@ -79,18 +89,13 @@ def create_app(
) )
blocks_html = "".join(blocks) blocks_html = "".join(blocks)
return ( return get_html_template(
"<!doctype html>"
"<html><head><meta charset='utf-8'><title>thc-webhook admin</title>"
"<style>body{font-family:sans-serif;max-width:900px;margin:24px;}textarea,input{font-family:inherit;}</style>"
"</head><body>"
"<h1>Admin: templates</h1>" "<h1>Admin: templates</h1>"
"<p>Edits are saved to the templates JSON file on disk.</p>" "<p>Edits are saved to the templates JSON file on disk.</p>"
"<form method='post'>" "<form method='post'>"
f"{blocks_html}" f"{blocks_html}"
"<button type='submit'>Save</button>" "<button type='submit'>Save</button>"
"</form>" "</form>"
"</body></html>"
) )
@app.post("/admin") @app.post("/admin")
@@ -121,23 +126,17 @@ def create_app(
if errors: if errors:
msg = "<br>".join(errors) msg = "<br>".join(errors)
return ( return get_html_template(
"<!doctype html><html><body>"
"<h1>Admin: templates</h1>" "<h1>Admin: templates</h1>"
f"<p style='color:#b00;'>Errors:<br>{msg}</p>" f"<p style='color:#b00;'>Errors:<br>{msg}</p>"
"<p><a href='/admin'>Back</a></p>" "<p><a href='/admin'>Back</a></p>"
"</body></html>", ), 400
400,
)
save_templates(templates_path, updated) save_templates(templates_path, updated)
return ( return get_html_template(
"<!doctype html><html><body>"
"<h1>Admin: templates</h1>" "<h1>Admin: templates</h1>"
"<p>Saved.</p>" "<p>Saved.</p>"
"<p><a href='/admin'>Back</a></p>" "<p><a href='/admin'>Back</a></p>"
"</body></html>", ), 200
200,
)
return app return app
main.py
+46 -16
View File
@@ -10,6 +10,14 @@ import schedule
from dotenv import load_dotenv from dotenv import load_dotenv
from templates import load_templates 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 TZDB_CACHE: dict | None = None
@@ -42,6 +50,15 @@ DISCORD_CHANNEL_ID = os.getenv('DISCORD_CHANNEL_ID')
GUILD_ID = os.getenv('DISCORD_GUILD_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: def init_tzdb_cache() -> dict:
"""Initialize a cached lookup structure for tzdb data. """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: if now is None:
now = datetime.now() now = datetime.now()
schedule_map = [
(15, "reminder"),
(20, "420"),
(45, "reminder_halftime"),
(50, "halftime"),
]
candidates: list[tuple[datetime, str]] = [] 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) candidate = now.replace(minute=minute, second=0, microsecond=0)
if candidate > now: if candidate > now:
candidates.append((candidate, msg_type)) candidates.append((candidate, msg_type))
@@ -155,8 +165,8 @@ def get_next_scheduled_event(now: datetime | None = None) -> dict:
if not candidates: if not candidates:
base = (now + timedelta(hours=1)).replace(minute=0, base = (now + timedelta(hours=1)).replace(minute=0,
second=0, microsecond=0) second=0, microsecond=0)
next_dt = base.replace(minute=schedule_map[0][0]) next_dt = base.replace(minute=SCHEDULED_NOTIFICATIONS[0][0])
return {"at": next_dt, "type": schedule_map[0][1]} return {"at": next_dt, "type": SCHEDULED_NOTIFICATIONS[0][1]}
next_dt, next_type = min(candidates, key=lambda x: x[0]) next_dt, next_type = min(candidates, key=lambda x: x[0])
return {"at": next_dt, "type": next_type} return {"at": next_dt, "type": next_type}
@@ -706,15 +716,33 @@ def where_is_it_420(
return results 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: def main() -> None:
""" """
Main function to run the scheduler. Main function to run the scheduler.
""" """
# Schedule notifications # Start the dashboard in a separate thread
schedule.every().hour.at(":15").do(send_notification, "reminder") dashboard_thread = threading.Thread(target=start_dashboard, daemon=True)
schedule.every().hour.at(":20").do(send_notification, "420") dashboard_thread.start()
# schedule.every().hour.at(":45").do(send_notification, "reminder_halftime")
# schedule.every().hour.at(":50").do(send_notification, "halftime") # 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 deletion of old messages every 5 minutes
schedule.every(5).minutes.do(delete_old_messages, 6) schedule.every(5).minutes.do(delete_old_messages, 6)
@@ -722,7 +750,9 @@ def main() -> None:
logging.info("Scheduler started.") logging.info("Scheduler started.")
# Test the notification on startup # 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 on startup to clean up any previous notifications
# delete_old_messages(6) # delete_old_messages(6)
templates.json
+4
View File
@@ -16,5 +16,9 @@
"reminder_halftime": { "reminder_halftime": {
"color": 15105570, "color": 15105570,
"text": "Half-time in 5 minutes!" "text": "Half-time in 5 minutes!"
},
"test": {
"color": 3447003,
"text": "This is a test notification."
} }
} }
templates.py
+14 -9
View File
@@ -6,12 +6,8 @@ from pathlib import Path
DEFAULT_TEMPLATES: dict[str, dict] = { DEFAULT_TEMPLATES: dict[str, dict] = {
"reminder_halftime": { "420": {
"text": "Half-time in 5 minutes!", "text": "Blaze it!",
"color": 0xE67E22,
},
"halftime": {
"text": "Half-time!",
"color": 0x2ECC71, "color": 0x2ECC71,
"image_url": "https://copyparty.allucanget.biz/img/weed.png", "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!", "text": "This is your 5 minute reminder to 420!",
"color": 0xE67E22, "color": 0xE67E22,
}, },
"420": { "halftime": {
"text": "Blaze it!", "text": "Half-time!",
"color": 0x2ECC71, "color": 0x2ECC71,
"image_url": "https://copyparty.allucanget.biz/img/weed.png", "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) p.parent.mkdir(parents=True, exist_ok=True)
tmp = p.with_suffix(p.suffix + ".tmp") 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) tmp.replace(p)