thethreemagi/scout-website
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
scout-website/app/store.py
T
thethreemagi b84fc7f064 fb_posts: edit and repost through the publisher's signed reschedule 
POST /fbposts/{id}/reschedule {message, scheduled_for}. The site derives
the per-post signature from the stored cancel link and asks the publisher
to delete the post on Facebook and redraft the queue file; the row goes
to drafted with no Facebook id and no cancel link until the publisher's
next cycle reports the new one. Times must be at least 15 minutes out,
with an offset. An explicit null now clears fb_post_id on upsert; an
absent one still keeps it. tests/smoke_admin.py 160 -> 162.
2026-09-04 19:57:40 -04:00

1252 lines
49 KiB
Python
Raw Permalink Blame History

"""
store.py - internal record store for greenlanescouts73.org
What this holds
---------------
One table per thing the site collects. Today that is `join_leads`: the /join
interest form, one row per submission, one column per form field.
A /join submission lands here FIRST. The Google Sheet and the ntfy push are
mirrors of a row that already exists locally, and each mirror's outcome is
recorded per row so a failure is replayable instead of just shouted.
Why join_leads and not a generic table
-------------------------------------
This started as a generic `records` table keyed by `kind`, carrying
status/assigned_to/notes for a future admin panel. That was wrong twice over.
The PII columns (display_name/email/phone) only make sense for a lead, and the
workflow columns describe an outreach process that has not been designed yet -
one that is plainly one-to-many, since a family gets contacted more than once.
Guessing at it in three columns would have locked in the wrong shape.
So: a table per form, columns for the form's own fields, and outreach gets its
own table when the process is actually known. Anything the form adds later that
is not worth a column still survives in `payload`, which holds the submission
verbatim.
Stdlib only - sqlite3 ships with Python, so this adds no image dependencies.
"""
import json
import hashlib
import os
import sqlite3
import uuid
import datetime
from pathlib import Path
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")
# nearby_units guardrails. pack/troop/ship/club get a recognisable badge on
# /find-a-unit; crew and post are real unit types the district may list and
# render with the generic card. Anything else is a typo, and a typo'd type
# would silently render a unit as generic "Scouting" rather than fail, so it
# is rejected at write time instead.
NEARBY_UNIT_TYPES = ("pack", "troop", "crew", "ship", "post", "club")
NEARBY_SERVES = ("family", "boys", "girls", "coed")
SCHEMA = """
PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS join_leads (
id TEXT PRIMARY KEY,
submitted_at TEXT NOT NULL,
recorded_at TEXT NOT NULL,
parent_name TEXT NOT NULL,
email TEXT NOT NULL,
phone TEXT,
interested_in TEXT,
children TEXT,
heard_from TEXT,
heard_from_detail TEXT,
message TEXT,
payload TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_join_leads_submitted ON join_leads(submitted_at DESC);
CREATE INDEX IF NOT EXISTS idx_join_leads_email ON join_leads(email);
-- Lead claims (outreach v1, decided by Mike 2026-09-04): who picked a lead
-- up and when, so nothing sits waiting unnoticed. No outcomes, no end
-- states - the lead stays the record of what the family said, this is only
-- who has it. One row per claim; the current claim is the latest with no
-- released_at. Nothing here touches join_leads.
CREATE TABLE IF NOT EXISTS lead_claims (
id TEXT PRIMARY KEY,
lead_id TEXT NOT NULL REFERENCES join_leads(id),
person_id TEXT NOT NULL,
email TEXT NOT NULL,
claimed_at TEXT NOT NULL,
released_at TEXT,
released_by TEXT
);
CREATE INDEX IF NOT EXISTS idx_lead_claims_lead ON lead_claims(lead_id, released_at);
-- Roster (decided by Mike 2026-09-04). my.scouting holds the record of
-- truth for registration; this holds what a den leader needs on a Tuesday:
-- who the families are, which scouts are in which den, and a per-year
-- checklist of things collected (dues paid, health form handed in). The
-- checklist records THAT a thing was collected, by whom and when - never
-- the thing. Health forms are never stored here; that is a policy, not a
-- gap. Scouts carry a first name, a last name, a den and an optional BSA
-- member ID (Mike: worth it, it is the recharter join key) and nothing
-- else: no date of birth, no address, nothing medical.
CREATE TABLE IF NOT EXISTS households (
id TEXT PRIMARY KEY,
parent_name TEXT NOT NULL,
email TEXT,
phone TEXT,
second_parent TEXT,
notes TEXT,
source_lead_id TEXT,
active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL,
created_by TEXT,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS scouts (
id TEXT PRIMARY KEY,
household_id TEXT NOT NULL REFERENCES households(id),
first_name TEXT NOT NULL,
last_name TEXT,
unit_id TEXT NOT NULL,
den TEXT,
bsa_member_id TEXT,
active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_scouts_unit ON scouts(unit_id, active, den);
-- Which accounts belong to which family. A household can have two parents
-- with accounts; a person can, rarely, be on two households. This is what
-- lets a signed-in parent see their own scouts and nobody else's.
CREATE TABLE IF NOT EXISTS household_people (
household_id TEXT NOT NULL REFERENCES households(id),
person_id TEXT NOT NULL,
PRIMARY KEY (household_id, person_id)
);
CREATE INDEX IF NOT EXISTS idx_household_people_person ON household_people(person_id);
CREATE TABLE IF NOT EXISTS roster_checks (
scout_id TEXT NOT NULL REFERENCES scouts(id),
year TEXT NOT NULL,
item TEXT NOT NULL,
done_at TEXT NOT NULL,
done_by TEXT,
note TEXT,
PRIMARY KEY (scout_id, year, item)
);
-- Facebook posts (ops/open-items.md section 12, built 2026-09-04). scout-publisher
-- reports every post it drafts, schedules, holds or cancels by POSTing here; it
-- never opens this file. id is <unit>/<queue-file-stem>, stable across body
-- edits, so a redrafted post is one row. The image is copied, content-addressed
-- at /data/post-images/<sha256>.<ext>, hash verified on arrival; image_ref is
-- provenance only (the NAS path it came from). cancel_url is the publisher's
-- own signed per-post link, so the panel gets Cancel with no new secret.
CREATE TABLE IF NOT EXISTS fb_posts (
id TEXT PRIMARY KEY,
unit TEXT NOT NULL,
page_id TEXT,
fb_post_id TEXT,
status TEXT NOT NULL,
message TEXT,
link TEXT,
scheduled_for TEXT,
queue_file TEXT,
cancel_url TEXT,
image_sha256 TEXT,
image_ref TEXT,
image_mime TEXT,
image_bytes INTEGER,
image_width INTEGER,
image_height INTEGER,
reported_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
cancelled_at TEXT,
cancelled_by TEXT
);
CREATE UNIQUE INDEX IF NOT EXISTS idx_fb_posts_fbid ON fb_posts(fb_post_id) WHERE fb_post_id IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_fb_posts_sched ON fb_posts(status, scheduled_for);
CREATE TABLE IF NOT EXISTS mirrors (
record_id TEXT NOT NULL,
target TEXT NOT NULL,
state TEXT NOT NULL,
attempts INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
last_attempt_at TEXT,
PRIMARY KEY (record_id, target)
);
CREATE INDEX IF NOT EXISTS idx_mirrors_state ON mirrors(target, state);
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);
-- Other Scouting units in the Continental District, published as a courtesy
-- directory. This is somebody else's data: it comes from a hand-typed district
-- document and goes stale without telling us, so `verified_at` is shown to the
-- reader rather than kept as bookkeeping. Pack 73 and Troop 73 are deliberately
-- NOT rows here - the rest of this site is us.
CREATE TABLE IF NOT EXISTS nearby_units (
id TEXT PRIMARY KEY,
unit_type TEXT NOT NULL,
unit_number TEXT NOT NULL,
serves TEXT,
chartered_org TEXT,
street TEXT,
town TEXT,
area TEXT,
meets TEXT,
notes TEXT,
link_url TEXT,
contact TEXT,
sort_order INTEGER NOT NULL DEFAULT 100,
active INTEGER NOT NULL DEFAULT 1,
source TEXT,
verified_at TEXT,
updated_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_nearby_units_listing
ON nearby_units(active, sort_order, unit_number);
"""
def _now():
return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")
def _norm(ts):
"""Normalise any incoming timestamp to UTC ISO-8601 with an offset.
app.py builds its submission timestamp with a naive datetime.now(), i.e.
local Eastern time. Storing that next to UTC rows would silently break both
sorting and `since=` filters, so everything that lands in the DB is
converted here.
"""
if not ts:
return _now()
try:
dt = datetime.datetime.fromisoformat(ts)
except Exception:
return _now()
if dt.tzinfo is None:
dt = dt.astimezone() # interpret as this host's local time
return dt.astimezone(datetime.timezone.utc).isoformat(timespec="seconds")
def connect():
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
con = sqlite3.connect(DB_PATH, timeout=10)
con.row_factory = sqlite3.Row
con.execute("PRAGMA foreign_keys=ON")
return con
def init(backfill_jsonl=None):
"""Create the schema. Safe to call on every boot."""
con = connect()
try:
con.executescript(SCHEMA)
con.commit()
if backfill_jsonl:
_backfill(con, Path(backfill_jsonl))
finally:
con.close()
# ----------------------------------------------------------------------------
# Writes
# ----------------------------------------------------------------------------
def insert_lead(parent_name, email, phone=None, interested_in=None, children=None,
heard_from=None, heard_from_detail=None, message=None,
payload=None, submitted_at=None, record_id=None):
"""Insert one /join submission and return its id. Raises on failure -
callers decide what a failed local write means."""
rid = record_id or str(uuid.uuid4())
submitted = _norm(submitted_at)
con = connect()
try:
con.execute(
"INSERT INTO join_leads (id, submitted_at, recorded_at, parent_name, email,"
" phone, interested_in, children, heard_from, heard_from_detail, message, payload)"
" VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
(rid, submitted, _now(), parent_name, email, phone or None,
interested_in or None, children or None, heard_from or None,
heard_from_detail or None, message or None,
json.dumps(payload or {}, ensure_ascii=False)),
)
con.commit()
finally:
con.close()
return rid
def set_mirror_pending(record_id, target):
"""Register a mirror as owed, before it is attempted.
Without this a mirror that never ran leaves no row at all, which reads
identically to a mirror that was never owed. That is how two leads went
missing from the Sheet on 2026-08-26 with a clean `failed_mirrors: {}`.
Never downgrades an existing row - a mirror already ok or failed has been
attempted, and its outcome stands.
"""
con = connect()
try:
con.execute(
"INSERT INTO mirrors (record_id, target, state, attempts, last_attempt_at)"
" VALUES (?,?,'pending',0,NULL)"
" ON CONFLICT(record_id, target) DO NOTHING",
(record_id, target),
)
con.commit()
finally:
con.close()
def set_mirror(record_id, target, ok, error=None):
"""Record the outcome of an attempt to copy a row to an external target."""
ts = _now()
state = "ok" if ok else "failed"
con = connect()
try:
con.execute(
"INSERT INTO mirrors (record_id, target, state, attempts, last_error, last_attempt_at)"
" VALUES (?,?,?,1,?,?)"
" ON CONFLICT(record_id, target) DO UPDATE SET"
" state=excluded.state,"
" attempts=mirrors.attempts+1,"
" last_error=excluded.last_error,"
" last_attempt_at=excluded.last_attempt_at",
(record_id, target, state, (str(error)[:500] if error else None), ts),
)
con.commit()
finally:
con.close()
# ----------------------------------------------------------------------------
# Reads
# ----------------------------------------------------------------------------
def _row(r):
d = dict(r)
if d.get("payload"):
try:
d["payload"] = json.loads(d["payload"])
except Exception:
pass
return d
def get_lead(record_id):
con = connect()
try:
r = con.execute("SELECT * FROM join_leads WHERE id=?", (record_id,)).fetchone()
if not r:
return None
out = _row(r)
out["mirrors"] = [dict(m) for m in con.execute(
"SELECT target, state, attempts, last_error, last_attempt_at"
" FROM mirrors WHERE record_id=?", (record_id,))]
cur = current_claim(con, record_id)
out["claim"] = {"email": cur["email"], "claimed_at": cur["claimed_at"]} if cur else None
out["claim_history"] = [dict(c) for c in con.execute(
"SELECT email, claimed_at, released_at, released_by FROM lead_claims WHERE lead_id=?"
" ORDER BY claimed_at DESC", (record_id,))]
return out
finally:
con.close()
def list_leads(since=None, q=None, limit=100, offset=0):
where, vals = [], []
if since:
where.append("submitted_at>=?"); vals.append(since)
if q:
where.append("(parent_name LIKE ? OR email LIKE ? OR children LIKE ?"
" OR heard_from LIKE ? OR message LIKE ?)")
vals += ["%%%s%%" % q] * 5
sql = "SELECT * FROM join_leads"
if where:
sql += " WHERE " + " AND ".join(where)
sql += " ORDER BY submitted_at DESC LIMIT ? OFFSET ?"
vals += [max(1, min(int(limit), 500)), max(0, int(offset))]
con = connect()
try:
rows = [_row(r) for r in con.execute(sql, vals)]
ids = [r["id"] for r in rows]
if ids:
marks = ",".join("?" * len(ids))
mir = {}
for m in con.execute(
"SELECT record_id, target, state FROM mirrors WHERE record_id IN (%s)" % marks, ids):
mir.setdefault(m["record_id"], {})[m["target"]] = m["state"]
for r in rows:
r["mirrors"] = mir.get(r["id"], {})
claims = {}
for c in con.execute(
"SELECT lead_id, email, claimed_at FROM lead_claims WHERE released_at IS NULL"
" AND lead_id IN (%s)" % marks, ids):
claims[c["lead_id"]] = {"email": c["email"], "claimed_at": c["claimed_at"]}
for r in rows:
r["claim"] = claims.get(r["id"])
return rows
finally:
con.close()
def summary():
"""Counts for a dashboard: totals, interest split, referral sources, failed mirrors."""
con = connect()
try:
return {
"total": con.execute("SELECT COUNT(*) c FROM join_leads").fetchone()["c"],
"unclaimed": con.execute("SELECT count(*) FROM join_leads l WHERE NOT EXISTS"
" (SELECT 1 FROM lead_claims c WHERE c.lead_id=l.id AND c.released_at IS NULL)").fetchone()[0],
"by_interest": {r["interested_in"]: r["c"] for r in con.execute(
"SELECT interested_in, COUNT(*) c FROM join_leads GROUP BY interested_in")},
"by_heard_from": {r["heard_from"]: r["c"] for r in con.execute(
"SELECT heard_from, COUNT(*) c FROM join_leads GROUP BY heard_from"
" ORDER BY c DESC")},
"failed_mirrors": {r["target"]: r["c"] for r in con.execute(
"SELECT target, COUNT(*) c FROM mirrors WHERE state='failed' GROUP BY target")},
"pending_mirrors": {r["target"]: r["c"] for r in con.execute(
"SELECT target, COUNT(*) c FROM mirrors WHERE state='pending' GROUP BY target")},
}
finally:
con.close()
def failed_mirror_records(target, limit=50):
"""Rows owing a copy to `target`: attempted and failed, or never attempted.
`pending` is included deliberately. A mirror that never ran needs replaying
just as much as one that ran and failed, and it is the case that hid two
leads on 2026-08-26.
"""
con = connect()
try:
return [_row(r) for r in con.execute(
"SELECT l.* FROM join_leads l JOIN mirrors m ON m.record_id=l.id"
" WHERE m.target=? AND m.state IN ('failed','pending')"
" ORDER BY l.submitted_at LIMIT ?",
(target, limit))]
finally:
con.close()
# ----------------------------------------------------------------------------
# One-time backfill of the pre-existing raw log
# ----------------------------------------------------------------------------
def split_heard_from(source):
"""app.py used to fold source_place / source_other into one composed string
('Other: Chocolate booth'), destroying the raw answer. Recover both halves."""
s = (source or "").strip()
if s.startswith("School/Daycare: "):
return "Other school or daycare", s[len("School/Daycare: "):].strip()
if s.startswith("Other: "):
return "Other", s[len("Other: "):].strip()
return s or None, None
def _backfill(con, path):
"""Import /data/leads.jsonl once, so history is not stranded outside the DB."""
done = con.execute("SELECT value FROM meta WHERE key='backfill_leads_jsonl'").fetchone()
if done or not path.exists():
return
n = 0
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
rec = json.loads(line)
except Exception:
continue
heard, detail = split_heard_from(rec.get("source"))
ts = _norm(rec.get("ts"))
con.execute(
"INSERT INTO join_leads (id, submitted_at, recorded_at, parent_name, email,"
" phone, interested_in, children, heard_from, heard_from_detail, message, payload)"
" VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
(str(uuid.uuid4()), ts, _now(),
rec.get("parent_name") or "(unknown)", rec.get("email") or "",
rec.get("phone") or None, rec.get("interested_in") or None,
rec.get("children") or None, heard, detail,
rec.get("message") or None,
json.dumps(rec, ensure_ascii=False)),
)
n += 1
con.execute("INSERT INTO meta (key, value) VALUES ('backfill_leads_jsonl', ?)",
("%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 Rejected(Exception):
"""Raised when a write breaks a guardrail. Carries the HTTP status the
admin API should return, so the rules 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 {}
class AnnouncementRejected(Rejected):
pass
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()
# ----------------------------------------------------------------------------
# Nearby units - writes behind the /find-a-unit courtesy directory.
#
# This is somebody else's data, hand-copied from a district document, and two
# rules follow from that.
#
# verified_at is bumped to today on every row write unless the caller passes
# one explicitly. Saving a row IS the claim that a person just checked it
# against the district list - verified_at is the date the honesty line on the
# public page shows a family, not bookkeeping.
#
# Rows are deactivated, never deleted. A unit that folds or moves keeps its
# row with active=0, so "why did that pack disappear from the page" stays
# answerable. Reactivation is an update setting active back to 1.
# ---------------------------------------------------------------------------
# Lead claims - outreach v1. See the schema comment.
# ---------------------------------------------------------------------------
class ClaimRejected(Rejected):
pass
def current_claim(con, lead_id):
r = con.execute("SELECT * FROM lead_claims WHERE lead_id=? AND released_at IS NULL"
" ORDER BY claimed_at DESC LIMIT 1", (lead_id,)).fetchone()
return dict(r) if r else None
def claim_lead(lead_id, person_id, email):
"""Take a lead. If someone else holds it, 409 with who - taking it over
is a deliberate release-then-claim, never a silent overwrite."""
con = connect()
try:
if not con.execute("SELECT 1 FROM join_leads WHERE id=?", (lead_id,)).fetchone():
return None
cur = current_claim(con, lead_id)
if cur:
if cur["person_id"] == person_id:
raise ClaimRejected(409, "you already have this lead")
raise ClaimRejected(409, "%s has this lead; release it first" % cur["email"], {"claimed_by": cur["email"]})
con.execute("INSERT INTO lead_claims (id, lead_id, person_id, email, claimed_at) VALUES (?,?,?,?,?)",
(str(uuid.uuid4()), lead_id, person_id, email, _now()))
con.commit()
return current_claim(con, lead_id)
finally:
con.close()
def release_lead(lead_id, person_id, email, can_release_any=False):
"""Let a lead go. The holder may; so may someone with people:manage
(an owner reassigning), which is what can_release_any means."""
con = connect()
try:
if not con.execute("SELECT 1 FROM join_leads WHERE id=?", (lead_id,)).fetchone():
return None
cur = current_claim(con, lead_id)
if not cur:
raise ClaimRejected(409, "nobody has this lead")
if cur["person_id"] != person_id and not can_release_any:
raise ClaimRejected(403, "%s has this lead; only they or an owner can release it" % cur["email"])
con.execute("UPDATE lead_claims SET released_at=?, released_by=? WHERE id=?", (_now(), email, cur["id"]))
con.commit()
return cur
finally:
con.close()
def unclaimed_count():
con = connect()
try:
return con.execute("SELECT count(*) FROM join_leads l WHERE NOT EXISTS"
" (SELECT 1 FROM lead_claims c WHERE c.lead_id=l.id AND c.released_at IS NULL)").fetchone()[0]
finally:
con.close()
# ----------------------------------------------------------------------------
class NearbyRejected(Rejected):
pass
# Everything a caller may set. id and updated_at are the store's own.
NEARBY_FIELDS = ("unit_type", "unit_number", "serves", "chartered_org", "street",
"town", "area", "meets", "notes", "link_url", "contact",
"sort_order", "active", "source", "verified_at")
def _clean_nearby(fields, creating):
"""Validate and normalise a payload. Unknown keys are rejected rather than
dropped - a silently ignored typo ("unit_typo": "pack") would read as a
successful save that changed nothing."""
unknown = sorted(set(fields) - set(NEARBY_FIELDS))
if unknown:
raise NearbyRejected(422, "unknown fields: %s. Editable fields are %s"
% (", ".join(unknown), ", ".join(NEARBY_FIELDS)))
out = {}
for k, v in fields.items():
if isinstance(v, str):
v = v.strip() or None
out[k] = v
if creating or "unit_type" in out:
if out.get("unit_type") not in NEARBY_UNIT_TYPES:
raise NearbyRejected(422, "unit_type must be one of %s" % (NEARBY_UNIT_TYPES,))
if creating or "unit_number" in out:
if not out.get("unit_number"):
raise NearbyRejected(422, "unit_number is required")
out["unit_number"] = str(out["unit_number"])
if out.get("serves") is not None and out["serves"] not in NEARBY_SERVES:
raise NearbyRejected(422, "serves must be one of %s, or null" % (NEARBY_SERVES,))
# Both of these land inside href="..." attributes on the public page, so a
# quote or bracket is an attribute breakout, not a formatting nit. Same
# character set identity._setting_https_url blocks, for the same reason.
if out.get("link_url") is not None:
v = str(out["link_url"])
if not v.startswith("https://") or any(c in v for c in " \"'<>"):
raise NearbyRejected(422, "link_url must be a plain https:// URL")
if out.get("contact") is not None:
v = str(out["contact"])
if "@" not in v or any(c in v for c in " \"'<>"):
raise NearbyRejected(422, "contact is rendered as a mailto: link and must be an email address")
if "sort_order" in out and out["sort_order"] is not None:
try:
out["sort_order"] = int(out["sort_order"])
except (TypeError, ValueError):
raise NearbyRejected(422, "sort_order must be an integer")
if "active" in out:
if out["active"] not in (0, 1, True, False):
raise NearbyRejected(422, "active must be 0 or 1")
out["active"] = int(out["active"])
if "verified_at" in out and out["verified_at"] is not None:
try:
datetime.date.fromisoformat(str(out["verified_at"]))
except ValueError:
raise NearbyRejected(422, "verified_at must be a plain YYYY-MM-DD date")
out["verified_at"] = str(out["verified_at"])
return out
def _today():
return datetime.datetime.now(datetime.timezone.utc).date().isoformat()
def list_nearby(include_inactive=False):
con = connect()
try:
sql = "SELECT * FROM nearby_units"
if not include_inactive:
sql += " WHERE active = 1"
sql += " ORDER BY sort_order, unit_number"
return [dict(r) for r in con.execute(sql).fetchall()]
finally:
con.close()
def get_nearby(nid):
con = connect()
try:
row = con.execute("SELECT * FROM nearby_units WHERE id=?", (nid,)).fetchone()
return dict(row) if row else None
finally:
con.close()
def create_nearby(fields):
out = _clean_nearby(fields or {}, creating=True)
if not out.get("verified_at"):
out["verified_at"] = _today()
out.setdefault("active", 1)
out.setdefault("sort_order", 100)
nid = str(uuid.uuid4())
cols = ["id"] + list(out) + ["updated_at"]
vals = [nid] + [out[k] for k in out] + [_now()]
con = connect()
try:
con.execute("INSERT INTO nearby_units (%s) VALUES (%s)"
% (", ".join(cols), ", ".join("?" * len(cols))), vals)
con.commit()
finally:
con.close()
return get_nearby(nid)
def update_nearby(nid, fields):
if not fields:
raise NearbyRejected(422, "nothing to update")
out = _clean_nearby(fields, creating=False)
if not out.get("verified_at"):
out["verified_at"] = _today()
out["updated_at"] = _now()
con = connect()
try:
cur = con.execute("UPDATE nearby_units SET %s WHERE id=?"
% ", ".join("%s=?" % k for k in out),
list(out.values()) + [nid])
con.commit()
if cur.rowcount == 0:
return None
finally:
con.close()
return get_nearby(nid)
def deactivate_nearby(nid):
"""Take a unit off the page. Sets active=0; the row and its history stay."""
con = connect()
try:
cur = con.execute(
"UPDATE nearby_units SET active=0, updated_at=? WHERE id=? AND active=1",
(_now(), nid))
con.commit()
return cur.rowcount > 0
finally:
con.close()
# ---------------------------------------------------------------------------
# Roster. See the schema comment. Every write here is a person's Tuesday
# night bookkeeping; the API logs it, the store keeps it simple.
# ---------------------------------------------------------------------------
class RosterRejected(Rejected):
pass
ROSTER_ITEMS = ("dues", "health_form")
ROSTER_ITEM_WORDS = {"dues": "Dues paid", "health_form": "Health form collected (kept on paper, never here)"}
HOUSEHOLD_FIELDS = ("parent_name", "email", "phone", "second_parent", "notes")
SCOUT_FIELDS = ("first_name", "last_name", "unit_id", "den", "bsa_member_id")
def _clean_text(d, fields, required=()):
out = {}
for k in fields:
if k in d:
v = d[k]
v = (v.strip() if isinstance(v, str) else v) or None
out[k] = v
for k in required:
if not out.get(k):
raise RosterRejected(422, "%s is required" % k)
if out.get("email") and "@" not in out["email"]:
raise RosterRejected(422, "email must contain @")
if out.get("bsa_member_id") and not str(out["bsa_member_id"]).isdigit():
raise RosterRejected(422, "bsa_member_id is digits only")
return out
def list_roster(year, unit_id=None, include_inactive=False):
"""Households with their scouts and this year's checks, ordered by
parent name. With unit_id, only households that have a scout in it."""
con = connect()
try:
hh = {r["id"]: dict(r, scouts=[]) for r in con.execute(
"SELECT * FROM households" + ("" if include_inactive else " WHERE active=1") + " ORDER BY parent_name")}
sql = "SELECT s.*, u.slug AS unit_slug, u.short_name AS unit_name FROM scouts s JOIN units u ON u.id=s.unit_id"
sql += "" if include_inactive else " WHERE s.active=1"
sql += " ORDER BY u.sort_order, s.den, s.first_name"
checks = {}
for c in con.execute("SELECT * FROM roster_checks WHERE year=?", (year,)):
checks.setdefault(c["scout_id"], {})[c["item"]] = {"done_at": c["done_at"], "done_by": c["done_by"], "note": c["note"]}
for r in con.execute(sql):
s = dict(r); s["checks"] = checks.get(s["id"], {})
if s["household_id"] in hh:
hh[s["household_id"]]["scouts"].append(s)
rows = list(hh.values())
if unit_id:
rows = [h for h in rows if any(s["unit_id"] == unit_id for s in h["scouts"])]
return rows
finally:
con.close()
def get_household(hid, year):
con = connect()
try:
r = con.execute("SELECT * FROM households WHERE id=?", (hid,)).fetchone()
if not r:
return None
h = dict(r, scouts=[])
for s in con.execute("SELECT s.*, u.slug AS unit_slug, u.short_name AS unit_name FROM scouts s"
" JOIN units u ON u.id=s.unit_id WHERE household_id=? ORDER BY first_name", (hid,)):
sd = dict(s); sd["checks"] = {c["item"]: {"done_at": c["done_at"], "done_by": c["done_by"], "note": c["note"]}
for c in con.execute("SELECT * FROM roster_checks WHERE scout_id=? AND year=?", (s["id"], year))}
h["scouts"].append(sd)
return h
finally:
con.close()
def create_household(fields, created_by=None, source_lead_id=None):
f = _clean_text(fields, HOUSEHOLD_FIELDS, required=("parent_name",))
hid = str(uuid.uuid4())
con = connect()
try:
if source_lead_id and con.execute("SELECT 1 FROM households WHERE source_lead_id=?", (source_lead_id,)).fetchone():
raise RosterRejected(409, "that lead was already imported")
con.execute("INSERT INTO households (id, parent_name, email, phone, second_parent, notes, source_lead_id,"
" active, created_at, created_by, updated_at) VALUES (?,?,?,?,?,?,?,1,?,?,?)",
(hid, f.get("parent_name"), f.get("email"), f.get("phone"), f.get("second_parent"), f.get("notes"),
source_lead_id, _now(), created_by, _now()))
con.commit()
finally:
con.close()
return hid
def import_lead(lead_id, created_by=None):
"""A lead becomes a household: the parent's name and contact copied,
the lead untouched and linked. Children come as a note to sort out by
hand - the /join form's children field is free text."""
lead = get_lead(lead_id)
if not lead:
return None
hid = create_household({"parent_name": lead.get("parent_name") or lead.get("email") or "Unknown",
"email": lead.get("email"), "phone": lead.get("phone"),
"notes": ("From the join form: %s" % lead["children"]) if lead.get("children") else None},
created_by=created_by, source_lead_id=lead_id)
return hid
def update_household(hid, fields):
f = _clean_text(fields, HOUSEHOLD_FIELDS + ("active",))
if not f:
raise RosterRejected(422, "nothing to update")
con = connect()
try:
if not con.execute("SELECT 1 FROM households WHERE id=?", (hid,)).fetchone():
return None
if "active" in f:
f["active"] = 1 if f["active"] in (1, True, "1", "true") else 0
sets = ", ".join("%s=?" % k for k in f)
con.execute("UPDATE households SET %s, updated_at=? WHERE id=?" % sets, (*f.values(), _now(), hid))
con.commit()
return dict(con.execute("SELECT * FROM households WHERE id=?", (hid,)).fetchone())
finally:
con.close()
def add_scout(hid, fields):
f = _clean_text(fields, SCOUT_FIELDS, required=("first_name", "unit_id"))
con = connect()
try:
if not con.execute("SELECT 1 FROM households WHERE id=?", (hid,)).fetchone():
return None
if not con.execute("SELECT 1 FROM units WHERE id=?", (f["unit_id"],)).fetchone():
raise RosterRejected(422, "unknown unit")
sid = str(uuid.uuid4())
con.execute("INSERT INTO scouts (id, household_id, first_name, last_name, unit_id, den, bsa_member_id,"
" active, created_at, updated_at) VALUES (?,?,?,?,?,?,?,1,?,?)",
(sid, hid, f["first_name"], f.get("last_name"), f["unit_id"], f.get("den"), f.get("bsa_member_id"),
_now(), _now()))
con.commit()
return dict(con.execute("SELECT * FROM scouts WHERE id=?", (sid,)).fetchone())
finally:
con.close()
def update_scout(sid, fields):
f = _clean_text(fields, SCOUT_FIELDS + ("active",))
if not f:
raise RosterRejected(422, "nothing to update")
con = connect()
try:
before = con.execute("SELECT * FROM scouts WHERE id=?", (sid,)).fetchone()
if not before:
return None
if "unit_id" in f and not con.execute("SELECT 1 FROM units WHERE id=?", (f["unit_id"],)).fetchone():
raise RosterRejected(422, "unknown unit")
if "active" in f:
f["active"] = 1 if f["active"] in (1, True, "1", "true") else 0
sets = ", ".join("%s=?" % k for k in f)
con.execute("UPDATE scouts SET %s, updated_at=? WHERE id=?" % sets, (*f.values(), _now(), sid))
con.commit()
return dict(before), dict(con.execute("SELECT * FROM scouts WHERE id=?", (sid,)).fetchone())
finally:
con.close()
def set_check(sid, year, item, done, done_by=None, note=None):
"""Mark a checklist item collected (or not) for a scout and year."""
if item not in ROSTER_ITEMS:
raise RosterRejected(422, "item must be one of %s" % (ROSTER_ITEMS,))
con = connect()
try:
if not con.execute("SELECT 1 FROM scouts WHERE id=?", (sid,)).fetchone():
return None
if done:
con.execute("INSERT OR REPLACE INTO roster_checks (scout_id, year, item, done_at, done_by, note)"
" VALUES (?,?,?,?,?,?)", (sid, year, item, _now(), done_by, (note or "").strip() or None))
else:
con.execute("DELETE FROM roster_checks WHERE scout_id=? AND year=? AND item=?", (sid, year, item))
con.commit()
r = con.execute("SELECT * FROM roster_checks WHERE scout_id=? AND year=? AND item=?", (sid, year, item)).fetchone()
return dict(r) if r else {"scout_id": sid, "year": year, "item": item, "done_at": None}
finally:
con.close()
def household_people(hid):
con = connect()
try:
return [r["person_id"] for r in con.execute("SELECT person_id FROM household_people WHERE household_id=?", (hid,))]
finally:
con.close()
def set_household_people(hid, person_ids):
"""Replace the accounts linked to a family."""
con = connect()
try:
if not con.execute("SELECT 1 FROM households WHERE id=?", (hid,)).fetchone():
return None
con.execute("DELETE FROM household_people WHERE household_id=?", (hid,))
for pid in sorted(set(p for p in person_ids or [] if p)):
con.execute("INSERT INTO household_people (household_id, person_id) VALUES (?,?)", (hid, pid))
con.commit()
return sorted(set(p for p in person_ids or [] if p))
finally:
con.close()
def households_for_person(person_id, year):
con = connect()
try:
ids = [r["household_id"] for r in con.execute("SELECT household_id FROM household_people WHERE person_id=?", (person_id,))]
finally:
con.close()
return [h for h in (get_household(i, year) for i in ids) if h and h.get("active")]
# ---------------------------------------------------------------------------
# Facebook posts. See the schema comment.
# ---------------------------------------------------------------------------
class FbRejected(Rejected):
pass
FB_STATUSES = ("drafted", "scheduled", "handed_off", "cancelled", "published", "failed")
IMAGE_DIR = DB_PATH.parent / "post-images"
IMAGE_MIMES = {"image/png": "png", "image/jpeg": "jpg", "image/webp": "webp"}
def upsert_fb_post(rec, image=None):
"""Record what the publisher reports. `image` is (bytes, mime, claimed_sha256)
or None; the hash is recomputed here and a mismatch is refused, so the row
never claims a version of the image nobody sent."""
pid = (rec.get("id") or "").strip()
if not pid or "/" not in pid:
raise FbRejected(422, "id must be <unit>/<queue-file-stem>")
status = (rec.get("status") or "").strip()
if status not in FB_STATUSES:
raise FbRejected(422, "status must be one of %s" % (FB_STATUSES,))
unit = (rec.get("unit") or pid.split("/", 1)[0]).strip()
img = {}
if image is not None:
data, mime, claimed = image
if mime not in IMAGE_MIMES:
raise FbRejected(422, "image mime must be one of %s" % sorted(IMAGE_MIMES))
if len(data) > 15 * 1024 * 1024:
raise FbRejected(413, "image over 15 MB")
actual = hashlib.sha256(data).hexdigest()
if claimed and claimed.lower() != actual:
raise FbRejected(422, "image sha256 mismatch: sent %s, is %s" % (claimed, actual))
IMAGE_DIR.mkdir(parents=True, exist_ok=True)
dest = IMAGE_DIR / ("%s.%s" % (actual, IMAGE_MIMES[mime]))
if not dest.exists():
tmp = dest.with_suffix(".tmp")
tmp.write_bytes(data); tmp.replace(dest)
w, h = _image_size(data, mime)
img = {"image_sha256": actual, "image_mime": mime, "image_bytes": len(data), "image_width": w, "image_height": h}
con = connect()
try:
old = con.execute("SELECT * FROM fb_posts WHERE id=?", (pid,)).fetchone()
fields = {"unit": unit, "page_id": rec.get("page_id"), "fb_post_id": (rec.get("fb_post_id") or None),
"status": status, "message": rec.get("message"), "link": rec.get("link"),
"scheduled_for": rec.get("scheduled_for"), "queue_file": rec.get("queue_file"),
"cancel_url": rec.get("cancel_url"), "image_ref": rec.get("image_ref"), "updated_at": _now()}
fields.update(img)
if status == "cancelled" and not (old and old["cancelled_at"]):
fields["cancelled_at"] = rec.get("cancelled_at") or _now()
fields["cancelled_by"] = rec.get("cancelled_by") or "publisher"
if old:
# A missing fb_post_id keeps the old one; an explicit null clears it
# (a redraft has no Facebook id until the next cycle).
if fields.get("fb_post_id") is None and "fb_post_id" not in rec:
fields.pop("fb_post_id")
sets = ", ".join("%s=?" % k for k in fields)
con.execute("UPDATE fb_posts SET %s WHERE id=?" % sets, (*fields.values(), pid))
else:
cols = ["id", "reported_at"] + list(fields)
con.execute("INSERT INTO fb_posts (%s) VALUES (%s)" % (", ".join(cols), ",".join("?" * len(cols))),
(pid, _now(), *fields.values()))
con.commit()
return dict(con.execute("SELECT * FROM fb_posts WHERE id=?", (pid,)).fetchone())
finally:
con.close()
def _image_size(data, mime):
"""Width and height from the header, no image library. None if unsure."""
try:
if mime == "image/png" and data[:8] == b"\x89PNG\r\n\x1a\n":
return int.from_bytes(data[16:20], "big"), int.from_bytes(data[20:24], "big")
if mime == "image/jpeg":
i = 2
while i < len(data) - 9:
if data[i] != 0xFF:
i += 1; continue
marker = data[i + 1]
if marker in (0xC0, 0xC1, 0xC2):
return int.from_bytes(data[i + 7:i + 9], "big"), int.from_bytes(data[i + 5:i + 7], "big")
i += 2 + int.from_bytes(data[i + 2:i + 4], "big")
except Exception:
pass
return None, None
def fb_state(rec, now=None):
"""The effective state. A scheduled post whose publish time has passed is
published unless something said otherwise: Facebook does not call back,
and a row that stays 'scheduled' forever is a lie of omission. Decided
by Mike 2026-09-04. `status` stays what was last reported; `state` is
what a reader should act on."""
st = rec.get("status")
if st == "scheduled" and rec.get("scheduled_for") and rec["scheduled_for"] <= (now or _now()):
return "published"
return st
def list_fb_posts(limit=100, include_done=True):
con = connect()
try:
rows = [dict(r) for r in con.execute(
"SELECT * FROM fb_posts ORDER BY COALESCE(scheduled_for, updated_at) DESC LIMIT ?",
(max(1, min(int(limit), 500)),))]
finally:
con.close()
now = _now()
for r in rows:
r["state"] = fb_state(r, now)
if not include_done:
rows = [r for r in rows if r["state"] in ("drafted", "scheduled", "handed_off")]
return rows
def get_fb_post(pid):
con = connect()
try:
r = con.execute("SELECT * FROM fb_posts WHERE id=?", (pid,)).fetchone()
finally:
con.close()
if not r:
return None
d = dict(r); d["state"] = fb_state(d)
return d
def mark_fb_cancelled(pid, by):
con = connect()
try:
con.execute("UPDATE fb_posts SET status='cancelled', cancelled_at=?, cancelled_by=?, updated_at=? WHERE id=?",
(_now(), by, _now(), pid))
con.commit()
return dict(con.execute("SELECT * FROM fb_posts WHERE id=?", (pid,)).fetchone())
finally:
con.close()
def image_path(rec):
if not rec or not rec.get("image_sha256"):
return None
p = IMAGE_DIR / ("%s.%s" % (rec["image_sha256"], IMAGE_MIMES.get(rec.get("image_mime"), "bin")))
return p if p.exists() else None
Reference in New Issue View Git Blame Copy Permalink