e30920067f
- 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.
144 lines
5.0 KiB
Python
144 lines
5.0 KiB
Python
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
|