Announcements: a site-wide notice with a start and an end

The one string on this site where a deploy is the wrong latency is
"tonight's meeting is cancelled, the lot is flooded" at 4pm on a
Tuesday. That is a record with a lifecycle, not site copy, so it gets
a table and a write path rather than a commit.

store.py
  announcements table, created by the existing IF NOT EXISTS path so
  there is no migration. ends_at is REQUIRED: an announcement that
  never expires is site copy, and site copy belongs in the repo where
  it has a diff. Nothing is hard-deleted; taking one down early sets
  revoked_at, so what the site said and when survives.

  Ranking is urgent first, then most recent. Recency alone would let a
  routine Wednesday notice bury a Tuesday cancellation still live.

  Two caps, enforced here rather than in the route so the future panel
  inherits them: 200 characters, and 3 live at once. Both reject rather
  than truncate. Clipping a cancellation mid-sentence is worse than
  making someone shorten it, and a 4th live notice is a signal nobody
  is expiring things rather than something to render.

app.py
  announcement_bar() above the sticky nav. Native <details>, no
  JavaScript, which matters on a read-only rootfs with no build step.
  Collapsed clamps to one line with a count; expanded lists all of
  them and caps at 40vh. One notice renders with no chevron and no
  count: the common case must not look like a widget.

  FAILS OPEN. The admin API fails closed because it serves family
  phone numbers. This is the opposite case, and a broken announcement
  must never take down the public homepage.

  Colours are the existing note and rust token pairs from the brand
  standard. No new colour enters the palette.

admin_api.py
  GET/POST/DELETE on /api/admin/announcements. Leads stay read-only;
  announcements are the deliberate exception, because being mutable
  and expiring is the entire feature rather than a guess at a process.
  list returns a computed state per row (live, scheduled, expired,
  revoked, over_cap) so "why is my notice not showing" is answerable
  from the API and not from the homepage.
This commit is contained in:
claude
2026-09-01 19:22:04 -04:00
parent 985253422d
commit e02125392a
3 changed files with 344 additions and 6 deletions
+70 -5
View File
@@ -6,10 +6,18 @@ to these endpoints; it never opens the SQLite file directly. That keeps the
panel deployable anywhere (separate container, separate host) and keeps this
app the only writer to its own database.
Read-only by design, for now. The old PATCH route set status/assigned_to/notes
on a lead - columns that were removed because they were a guess at an outreach
process nobody has designed. When that process exists it gets its own table and
its own write routes; until then there is nothing on a lead to mutate.
Leads are read-only, and that is deliberate. The old PATCH route set
status/assigned_to/notes on a lead - columns that were removed because they
were a guess at an outreach process nobody has designed. When that process
exists it gets its own table and its own write routes; until then there is
nothing on a lead to mutate.
Announcements ARE writable, and are the deliberate exception to that. The
read-only rule exists because a lead has nothing mutable on it. An
announcement is defined by being mutable and expiring - it is posted, it
shows, it comes down. Writing it is the entire feature. The caps that keep
the banner from becoming a mess live in store.py, not here, so the panel and
any future client inherit them rather than reimplementing them.
Auth: every route requires the X-Admin-Token header to match ADMIN_TOKEN.
If ADMIN_TOKEN is unset the whole router returns 503 - it FAILS CLOSED. These
@@ -20,7 +28,7 @@ so an unconfigured deployment must not serve them.
import hmac
import os
from fastapi import APIRouter, Header, HTTPException, Query
from fastapi import APIRouter, Body, Header, HTTPException, Query
import store
@@ -83,3 +91,60 @@ def retry(target: str = "google_sheet", x_admin_token: str = Header(None)):
store.set_mirror(rec["id"], target, False, e)
failed_ids.append(rec["id"])
return {"retried_ok": done, "still_failing": failed_ids}
# ----------------------------------------------------------------------------
# Announcements - the one writable object here. See the module docstring.
# ----------------------------------------------------------------------------
def _reject(e):
"""Turn a store guardrail into its HTTP answer, carrying the detail so the
caller is told what to do rather than just refused."""
payload = {"error": e.detail}
payload.update(e.extra)
return HTTPException(e.status, payload)
@router.get("/announcements")
def list_announcements(include_expired: bool = False,
limit: int = Query(100, ge=1, le=500),
x_admin_token: str = Header(None)):
"""Every announcement with its computed state: live, scheduled, expired,
revoked, or over_cap. State is returned rather than left to be inferred
from what the homepage happens to render."""
_auth(x_admin_token)
return {"announcements": store.list_announcements(
include_expired=include_expired, limit=limit)}
@router.post("/announcements", status_code=201)
def create_announcement(payload: dict = Body(...), x_admin_token: str = Header(None)):
"""Post a notice. ends_at is required.
422 if the message is over the character cap or the window is invalid.
409 if the live cap is already reached, listing what is up so you can
decide what to revoke."""
_auth(x_admin_token)
try:
return store.create_announcement(
message=payload.get("message"),
ends_at=payload.get("ends_at"),
starts_at=payload.get("starts_at"),
level=payload.get("level", "info"),
link_url=payload.get("link_url"),
link_text=payload.get("link_text"),
created_by=payload.get("created_by"),
)
except store.AnnouncementRejected as e:
raise _reject(e)
@router.delete("/announcements/{announcement_id}")
def revoke_announcement(announcement_id: str, x_admin_token: str = Header(None)):
"""Take one down early. Sets revoked_at; never deletes the row."""
_auth(x_admin_token)
if not store.get_announcement(announcement_id):
raise HTTPException(404, "no such announcement")
if not store.revoke_announcement(announcement_id):
raise HTTPException(409, "already revoked")
return store.get_announcement(announcement_id)
+98 -1
View File
@@ -1,4 +1,4 @@
import json, os, re, time, datetime, urllib.request, urllib.parse
import html, json, os, re, time, datetime, urllib.request, urllib.parse
from pathlib import Path
from fastapi import FastAPI, Form, Request
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
@@ -268,8 +268,104 @@ footer a.fb{display:inline-flex;align-items:center;gap:8px}
.docd{display:block;margin-top:3px;color:#6A7280;font-size:14px}
.docmeta{font-size:12.5px;font-weight:600;color:#8B93A3;white-space:nowrap}
@media(max-width:560px){.docrow{flex-wrap:wrap}.docmeta{width:100%;padding-left:74px}}
.annc{border-bottom:1px solid;font-family:'Public Sans',system-ui,sans-serif}
.annc-info{background:#F3EBD8;border-color:#E8DCBB;color:#6B5510}
.annc-urgent{background:#F6E3DA;border-color:#DFB9A4;color:#8A3B1D}
.anncrow{display:flex;align-items:center;gap:9px;padding:11px 0}
.annc summary{cursor:pointer;list-style:none;-webkit-tap-highlight-color:transparent}
.annc summary::-webkit-details-marker{display:none}
.annc summary:focus-visible{outline:2px solid currentColor;outline-offset:-3px}
.anncico,.anncchev{flex:none;display:flex;align-items:center}
.anncchev{transition:transform .15s ease}
.annc[open] .anncchev{transform:rotate(180deg)}
.anncline{flex:1;min-width:0;font-size:14px;line-height:1.35;overflow:hidden;text-overflow:ellipsis;white-space:nowrap}
.anncopen{display:none}
.annc[open] .ancctop{display:none}
.annc[open] .anncopen{display:block;font-weight:600}
.annccount{flex:none;font-size:11px;font-weight:700;padding:1px 7px;border-radius:999px;border:1px solid currentColor}
.annc[open] .annccount{display:none}
.anncbody{max-height:40vh;overflow-y:auto;padding-bottom:12px}
.anncitem{padding:10px 0 0;font-size:14px;line-height:1.5;overflow-wrap:anywhere}
.anncitem+.anncitem{border-top:1px solid rgba(0,0,0,.09);padding-top:10px;margin-top:10px}
.anncitem a{color:inherit;text-decoration:underline;font-size:13px}
.ancclink{flex:none;color:inherit;text-decoration:underline;font-size:13px;white-space:nowrap}
"""
ICO_INFO = ('<svg viewBox="0 0 24 24" width="17" height="17" fill="none" stroke="currentColor"'
' stroke-width="2" stroke-linecap="round" aria-hidden="true">'
'<circle cx="12" cy="12" r="9"/><path d="M12 11v5M12 7.6v.01"/></svg>')
ICO_URGENT = ('<svg viewBox="0 0 24 24" width="17" height="17" fill="none" stroke="currentColor"'
' stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true">'
'<path d="M10.3 4.3 2.6 17.6a2 2 0 0 0 1.7 3h15.4a2 2 0 0 0 1.7-3L13.7 4.3a2 2 0 0 0-3.4 0Z"/>'
'<path d="M12 9.5v4M12 17.2v.01"/></svg>')
ICO_CHEV = ('<svg viewBox="0 0 24 24" width="16" height="16" fill="none" stroke="currentColor"'
' stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true">'
'<path d="m6 9 6 6 6-6"/></svg>')
def _annc_item(a):
"""One announcement in the expanded list. Text is escaped; the link is
rendered only from a url store.create_announcement already validated."""
out = '<div class="anncitem">%s' % html.escape(a["message"])
if a.get("link_url"):
out += '<br><a href="%s">%s</a>' % (
html.escape(a["link_url"], quote=True),
html.escape(a.get("link_text") or "More details"))
return out + "</div>"
def announcement_bar():
"""Render the live announcements, or nothing at all.
FAILS OPEN, deliberately. The admin API fails closed because it serves
family phone numbers; this is the opposite case. A broken announcement
must never take down the public homepage, so any error here renders an
empty string and logs.
Collapsed shows the top-ranked notice on one clamped line plus a count.
Expanded lists all of them. Native <details>, so there is no JavaScript
and nothing to load - which matters on a read-only rootfs with no build
step. The single-notice case, which is nearly always the case, renders
with no chevron and no count: it must not look like a widget.
"""
try:
items = store.active_announcements()
except Exception as e:
print("announcement_bar: %s" % e, flush=True)
return ""
if not items:
return ""
top = items[0]
level = "urgent" if any(i["level"] == "urgent" for i in items) else "info"
ico = ICO_URGENT if level == "urgent" else ICO_INFO
label = "Urgent notice" if level == "urgent" else "Notice"
if len(items) == 1:
link = ""
if top.get("link_url"):
link = ('<a class="ancclink" href="%s">%s</a>' % (
html.escape(top["link_url"], quote=True),
html.escape(top.get("link_text") or "More details")))
return ('<div class="annc annc-%s" role="region" aria-label="%s"><div class="wrap anncrow">'
'<span class="anncico">%s</span>'
'<span class="anncline">%s</span>%s'
'</div></div>' % (level, label, ico, html.escape(top["message"]), link))
body = "".join(_annc_item(i) for i in items)
return ('<details class="annc annc-%s"><summary><div class="wrap anncrow">'
'<span class="anncico">%s</span>'
'<span class="anncline ancctop">%s</span>'
'<span class="anncline anncopen">%d notices</span>'
'<span class="annccount">+%d</span>'
'<span class="anncchev">%s</span>'
'</div></summary><div class="wrap anncbody">%s</div></details>'
% (level, ico, html.escape(top["message"]), len(items),
len(items) - 1, ICO_CHEV, body))
def page(title, body, active=""):
def on(k): return ' class="on"' if active == k else ""
return f"""<!doctype html>
@@ -290,6 +386,7 @@ def page(title, body, active=""):
<link href="https://fonts.googleapis.com/css2?family=Archivo:wght@500;600;700;800;900&family=Public+Sans:wght@400;600;700&display=swap" rel="stylesheet">
<style>{CSS}</style>
</head><body style="display:flex;flex-direction:column;min-height:100vh">
{announcement_bar()}
<header class="nav"><div class="wrap navrow">
<a class="brand" href="/"><img src="/static/img/patch-73.png" alt="Pack and Troop 73 patch"><span><span class="b1">PACK &amp; TROOP 73</span><span class="b2">Zieglerville, PA · Cradle of Liberty Council</span></span></a>
<nav class="links">
+176
View File
@@ -39,6 +39,13 @@ DB_PATH = Path(os.environ.get("STORE_DB", "/data/scout73.db"))
# Mirror targets - external destinations a row is copied out to.
TARGETS = ("google_sheet", "ntfy")
# Announcement guardrails. Enforced at write time so a bad notice never
# reaches a render. Both are rejections, never silent truncation: clipping
# someone's cancellation mid-sentence is worse than making them shorten it.
ANNOUNCEMENT_MAX_CHARS = 200
ANNOUNCEMENT_MAX_LIVE = 3
ANNOUNCEMENT_LEVELS = ("info", "urgent")
SCHEMA = """
PRAGMA journal_mode=WAL;
@@ -74,6 +81,21 @@ CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS announcements (
id TEXT PRIMARY KEY,
created_at TEXT NOT NULL,
message TEXT NOT NULL,
level TEXT NOT NULL DEFAULT 'info',
starts_at TEXT NOT NULL,
ends_at TEXT NOT NULL,
link_url TEXT,
link_text TEXT,
created_by TEXT,
revoked_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_announcements_window
ON announcements(revoked_at, starts_at, ends_at);
"""
@@ -336,3 +358,157 @@ def _backfill(con, path):
("%s rows at %s" % (n, _now()),))
con.commit()
print("store: backfilled %s rows from %s" % (n, path), flush=True)
# ----------------------------------------------------------------------------
# Announcements
# ----------------------------------------------------------------------------
#
# A site-wide notice with a start and an end. It exists because "tonight's
# meeting is cancelled, the lot is flooded" at 4pm on a Tuesday is the one
# string on this site where a deploy is the wrong latency.
#
# ends_at is REQUIRED. That is the whole point: nothing has to be remembered
# and taken down. An announcement with no end is site copy, and site copy
# lives in git where it has a diff.
#
# Nothing is ever hard-deleted. Taking one down early sets revoked_at, so the
# record of what the site said, and when, survives.
class AnnouncementRejected(Exception):
"""Raised when a write breaks a guardrail. Carries the HTTP status the
admin API should return, so the caps live here rather than in the route."""
def __init__(self, status, detail, extra=None):
super().__init__(detail)
self.status = status
self.detail = detail
self.extra = extra or {}
def _live_at(con, when):
return con.execute(
"SELECT * FROM announcements"
" WHERE revoked_at IS NULL AND starts_at <= ? AND ends_at > ?"
" ORDER BY CASE level WHEN 'urgent' THEN 0 ELSE 1 END, starts_at DESC",
(when, when),
).fetchall()
def active_announcements(now=None):
"""The notices that should render, best first.
Urgent outranks info, then most recent. Recency alone would let a routine
Wednesday notice bury a Tuesday cancellation that is still live.
Capped at ANNOUNCEMENT_MAX_LIVE as a floor under the render even if rows
got in past the write check - the banner is never allowed to be unbounded.
"""
when = now or _now()
con = connect()
try:
return [dict(r) for r in _live_at(con, when)[:ANNOUNCEMENT_MAX_LIVE]]
finally:
con.close()
def list_announcements(include_expired=False, limit=100):
"""Every announcement with a computed live/expired/revoked state.
The state is returned rather than inferred, so 'why is my notice not
showing' is answerable from the API instead of from the homepage.
"""
now = _now()
con = connect()
try:
sql = "SELECT * FROM announcements"
if not include_expired:
sql += " WHERE revoked_at IS NULL AND ends_at > '%s'" % now
sql += " ORDER BY starts_at DESC LIMIT ?"
rows = [dict(r) for r in con.execute(sql, (limit,)).fetchall()]
finally:
con.close()
live_ids = {r["id"] for r in active_announcements(now)}
for r in rows:
if r["revoked_at"]:
r["state"] = "revoked"
elif r["ends_at"] <= now:
r["state"] = "expired"
elif r["starts_at"] > now:
r["state"] = "scheduled"
elif r["id"] in live_ids:
r["state"] = "live"
else:
r["state"] = "over_cap"
return rows
def create_announcement(message, ends_at, starts_at=None, level="info",
link_url=None, link_text=None, created_by=None):
message = (message or "").strip()
if not message:
raise AnnouncementRejected(422, "message is required")
if len(message) > ANNOUNCEMENT_MAX_CHARS:
raise AnnouncementRejected(422, (
"message is %d characters and the cap is %d. Put the long version on a "
"documents page and link to it with link_url."
% (len(message), ANNOUNCEMENT_MAX_CHARS)))
if level not in ANNOUNCEMENT_LEVELS:
raise AnnouncementRejected(422, "level must be one of %s" % (ANNOUNCEMENT_LEVELS,))
if not ends_at:
raise AnnouncementRejected(422, (
"ends_at is required. An announcement that never expires is site copy, "
"and site copy belongs in the repo where it has a diff."))
if link_text and not link_url:
raise AnnouncementRejected(422, "link_text without link_url has nothing to point at")
if link_url and not str(link_url).startswith(("/", "https://")):
raise AnnouncementRejected(422, "link_url must be site-relative or https")
starts = _norm(starts_at) if starts_at else _now()
ends = _norm(ends_at)
if ends <= starts:
raise AnnouncementRejected(422, "ends_at must be after starts_at")
aid = str(uuid.uuid4())
con = connect()
try:
live = _live_at(con, starts)
if len(live) >= ANNOUNCEMENT_MAX_LIVE:
raise AnnouncementRejected(409, (
"%d announcements are already live at that start time and the cap is %d. "
"Revoke one first." % (len(live), ANNOUNCEMENT_MAX_LIVE)),
{"live": [{k: r[k] for k in ("id", "message", "level", "ends_at")}
for r in live]})
con.execute(
"INSERT INTO announcements (id, created_at, message, level, starts_at,"
" ends_at, link_url, link_text, created_by, revoked_at)"
" VALUES (?,?,?,?,?,?,?,?,?,NULL)",
(aid, _now(), message, level, starts, ends,
link_url or None, link_text or None, created_by or None))
con.commit()
finally:
con.close()
return get_announcement(aid)
def get_announcement(aid):
con = connect()
try:
row = con.execute("SELECT * FROM announcements WHERE id=?", (aid,)).fetchone()
return dict(row) if row else None
finally:
con.close()
def revoke_announcement(aid):
"""Take one down early. Never deletes - the site's history is the point."""
con = connect()
try:
cur = con.execute(
"UPDATE announcements SET revoked_at=? WHERE id=? AND revoked_at IS NULL",
(_now(), aid))
con.commit()
return cur.rowcount > 0
finally:
con.close()