""" 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 /, stable across body -- edits, so a redrafted post is one row. The image is copied, content-addressed -- at /data/post-images/., 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 /") 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