feat: Add dashboard support and enhance Discord webhook functionality
- Updated docker-compose.yml to expose dashboard on port 8080. - Enhanced main.py with timezone database caching and improved state management. - Introduced a minimal dashboard using Flask to display webhook status and notifications. - Added templates.json for customizable embed messages in Discord notifications. - Created templates.py for loading and saving notification templates. - Implemented tests for dashboard rendering and main functionality. - Added requirements for Flask and tzdata to support new features. - Included test cases for timezone handling and template management.
This commit is contained in:
+143
@@ -0,0 +1,143 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime
|
||||
|
||||
from flask import Flask, request
|
||||
|
||||
from templates import load_templates, parse_color, save_templates
|
||||
|
||||
|
||||
def _fmt_dt(dt: datetime | None) -> str:
|
||||
if dt is None:
|
||||
return "—"
|
||||
try:
|
||||
return dt.isoformat(timespec="seconds")
|
||||
except Exception:
|
||||
return str(dt)
|
||||
|
||||
|
||||
def create_app(
|
||||
*,
|
||||
get_state: Callable[[], dict],
|
||||
get_next_event: Callable[[], dict],
|
||||
) -> Flask:
|
||||
app = Flask(__name__)
|
||||
|
||||
@app.get("/")
|
||||
def index() -> str:
|
||||
state = get_state() or {}
|
||||
next_event = get_next_event() or {}
|
||||
|
||||
locations = state.get("last_locations") or []
|
||||
locations_html = "".join(f"<li>{loc}</li>" for loc in locations)
|
||||
|
||||
return (
|
||||
"<!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>"
|
||||
"<h2>Status</h2>"
|
||||
"<ul>"
|
||||
f"<li>running: {state.get('running', True)}</li>"
|
||||
f"<li>started_at: {_fmt_dt(state.get('started_at'))}</li>"
|
||||
f"<li>last_type: {state.get('last_type') or '—'}</li>"
|
||||
f"<li>last_attempt_at: {_fmt_dt(state.get('last_attempt_at'))}</li>"
|
||||
f"<li>last_success_at: {_fmt_dt(state.get('last_success_at'))}</li>"
|
||||
f"<li>last_status_code: {state.get('last_status_code') or '—'}</li>"
|
||||
f"<li>last_error: {state.get('last_error') or '—'}</li>"
|
||||
"</ul>"
|
||||
"<h2>Next scheduled</h2>"
|
||||
"<ul>"
|
||||
f"<li>type: {next_event.get('type') or '—'}</li>"
|
||||
f"<li>at: {_fmt_dt(next_event.get('at'))}</li>"
|
||||
"</ul>"
|
||||
"<h2>Locations (latest)</h2>"
|
||||
f"<p>Total: {len(locations)}</p>"
|
||||
f"<ul>{locations_html}</ul>"
|
||||
"</body></html>"
|
||||
)
|
||||
|
||||
@app.get("/admin")
|
||||
def admin_get() -> str:
|
||||
templates_path = app.config.get("TEMPLATES_PATH", "templates.json")
|
||||
templates = load_templates(templates_path)
|
||||
|
||||
blocks = []
|
||||
for key, tpl in templates.items():
|
||||
text = (tpl.get("text") or "").replace("'", "'")
|
||||
color = tpl.get("color")
|
||||
image_url = tpl.get("image_url") or ""
|
||||
blocks.append(
|
||||
"<fieldset style='margin-bottom:16px;'>"
|
||||
f"<legend><strong>{key}</strong></legend>"
|
||||
f"<label>Text<br><textarea name='{key}__text' rows='3' style='width:100%'>{text}</textarea></label><br>"
|
||||
f"<label>Color<br><input name='{key}__color' value='{color}' style='width:200px'></label><br>"
|
||||
f"<label>Image URL (optional)<br><input name='{key}__image_url' value='{image_url}' style='width:100%'></label>"
|
||||
"</fieldset>"
|
||||
)
|
||||
|
||||
blocks_html = "".join(blocks)
|
||||
return (
|
||||
"<!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>"
|
||||
"<form method='post'>"
|
||||
f"{blocks_html}"
|
||||
"<button type='submit'>Save</button>"
|
||||
"</form>"
|
||||
"</body></html>"
|
||||
)
|
||||
|
||||
@app.post("/admin")
|
||||
def admin_post() -> tuple[str, int]:
|
||||
templates_path = app.config.get("TEMPLATES_PATH", "templates.json")
|
||||
current = load_templates(templates_path)
|
||||
|
||||
errors: list[str] = []
|
||||
updated: dict[str, dict] = {}
|
||||
for key in current.keys():
|
||||
text = request.form.get(f"{key}__text", "").strip()
|
||||
color_raw = request.form.get(f"{key}__color", "").strip()
|
||||
image_url = request.form.get(f"{key}__image_url", "").strip()
|
||||
|
||||
if not text:
|
||||
errors.append(f"{key}: text is required")
|
||||
continue
|
||||
try:
|
||||
color = parse_color(color_raw)
|
||||
except Exception as e:
|
||||
errors.append(f"{key}: invalid color ({e})")
|
||||
continue
|
||||
|
||||
tpl: dict = {"text": text, "color": color}
|
||||
if image_url:
|
||||
tpl["image_url"] = image_url
|
||||
updated[key] = tpl
|
||||
|
||||
if errors:
|
||||
msg = "<br>".join(errors)
|
||||
return (
|
||||
"<!doctype html><html><body>"
|
||||
"<h1>Admin: templates</h1>"
|
||||
f"<p style='color:#b00;'>Errors:<br>{msg}</p>"
|
||||
"<p><a href='/admin'>Back</a></p>"
|
||||
"</body></html>",
|
||||
400,
|
||||
)
|
||||
|
||||
save_templates(templates_path, updated)
|
||||
return (
|
||||
"<!doctype html><html><body>"
|
||||
"<h1>Admin: templates</h1>"
|
||||
"<p>Saved.</p>"
|
||||
"<p><a href='/admin'>Back</a></p>"
|
||||
"</body></html>",
|
||||
200,
|
||||
)
|
||||
|
||||
return app
|
||||
Reference in New Issue
Block a user