scout-website/app/calendar_write.py

"""
calendar_write.py - the site's write-back to the public calendar (P4).

Radicale collection scouts/site73 is the source of truth for the public
calendar; the scout-calendar feed converts it to the JSON the site reads.
This module is the one place the site WRITES to that collection, over
CalDAV, as the scoped `scoutsite` principal. Two rules, both enforced here
and not left to callers:

  1. The site owns only UIDs ending UID_SUFFIX and touches nothing else.
     band-cal-sync owns @band.us the same way; the seeded events and anything
     Mike adds in a CalDAV client stay outside the site's reach. A wrong
     UID is a 403 before any network call.
  2. Fail closed. With no RADICALE_* configuration every write is a 503,
     never a silent no-op that reads as success.

The VEVENT layout matches projects/scout-calendar/seed.py exactly (all-day
DTEND exclusive, TZID + VTIMEZONE on timed events, CATEGORIES for the unit,
X-SCOUT73-BADGE, 75-octet folding) so the feed reverses it into the same
row shape the site already renders. Stdlib only, like the rest of the app.
"""

import base64
import datetime
import hashlib
import os
import urllib.error
import urllib.parse
import urllib.request
import uuid

UID_SUFFIX = "@site73.greenlanescouts73.org"
# The 2026-08-29 seed stamped its 32 events site73-<sha1>@greenlanescouts73.org
# before this namespace existed. They are the site's own data too; the rule
# exists to keep the site off @band.us and off anything a person adds in a
# calendar client, not off itself.
SEED_PREFIX, SEED_SUFFIX = "site73-", "@greenlanescouts73.org"
TZID = "America/New_York"
UNITS = ("pack", "troop", "both")
DEFAULT_TIMED_MINUTES = 90
TIMEOUT = 8

RADICALE_URL = (os.environ.get("RADICALE_URL") or "").strip()
RADICALE_USER = (os.environ.get("RADICALE_USER") or "").strip()
RADICALE_PASS = os.environ.get("RADICALE_PASS") or ""

VTIMEZONE_NY = """BEGIN:VTIMEZONE
TZID:America/New_York
X-LIC-LOCATION:America/New_York
BEGIN:DAYLIGHT
TZOFFSETFROM:-0500
TZOFFSETTO:-0400
TZNAME:EDT
DTSTART:19700308T020000
RRULE:FREQ=YEARLY;BYMONTH=3;BYDAY=2SU
END:DAYLIGHT
BEGIN:STANDARD
TZOFFSETFROM:-0400
TZOFFSETTO:-0500
TZNAME:EST
DTSTART:19701101T020000
RRULE:FREQ=YEARLY;BYMONTH=11;BYDAY=1SU
END:STANDARD
END:VTIMEZONE""".split("\n")


class CalendarRejected(Exception):
    def __init__(self, status, detail):
        super().__init__(detail)
        self.status = status
        self.detail = detail


def configured():
    return bool(RADICALE_URL and RADICALE_USER and RADICALE_PASS)


def owns(uid):
    u = str(uid or "")
    return bool(u) and (u.endswith(UID_SUFFIX) or (u.startswith(SEED_PREFIX) and u.endswith(SEED_SUFFIX)))


def object_name(uid):
    """The CalDAV object the UID lives in. Seeded events sit at <uid>.ics as
    the seeder wrote them; the site's own at a hashed site-*.ics. Getting
    this wrong would create a second object with the same UID, which Radicale
    refuses, so the two forms are tested separately."""
    u = str(uid)
    if u.endswith(UID_SUFFIX):
        return slug(u)
    return urllib.parse.quote(u + ".ics", safe="")


def new_uid():
    return str(uuid.uuid4()) + UID_SUFFIX


def slug(uid):
    """Stable object name per UID. The site- prefix marks ownership on disk
    the way band- does for the BAND mirror."""
    return "site-" + hashlib.sha1(uid.encode("utf-8")).hexdigest()[:16] + ".ics"


# ---------------------------------------------------------------------------
# Building the object
# ---------------------------------------------------------------------------

def esc(v):
    return (str(v).replace("\\", "\\\\").replace(";", "\\;")
            .replace(",", "\\,").replace("\n", "\\n"))


def fold(line):
    """RFC 5545 folding at 75 OCTETS."""
    enc = line.encode("utf-8")
    if len(enc) <= 75:
        return [line]
    out, cur, size = [], "", 0
    for ch in line:
        w = len(ch.encode("utf-8"))
        if size + w > 75:
            out.append(cur)
            cur, size = " " + ch, 1 + w
        else:
            cur += ch
            size += w
    if cur:
        out.append(cur)
    return out


def _date(v, what):
    try:
        return datetime.date.fromisoformat(str(v).strip())
    except (TypeError, ValueError):
        raise CalendarRejected(422, "%s must be a YYYY-MM-DD date" % what)


def _clock(v, what):
    try:
        t = datetime.datetime.strptime(str(v).strip(), "%H:%M")
    except (TypeError, ValueError):
        raise CalendarRejected(422, '%s must be 24h "HH:MM"' % what)
    return t.hour, t.minute


def clean(fields):
    """Validate and normalise an event payload. Returns the dict the builder
    uses. Blank strings are treated as absent."""
    f = {k: (v.strip() if isinstance(v, str) else v) for k, v in (fields or {}).items()}
    f = {k: (None if v == "" else v) for k, v in f.items()}
    title = f.get("title")
    if not title:
        raise CalendarRejected(422, "title is required")
    if len(title) > 120:
        raise CalendarRejected(422, "title is over 120 characters")
    unit = (f.get("unit") or "pack").lower()
    if unit not in UNITS:
        raise CalendarRejected(422, "unit must be one of %s" % (UNITS,))
    start = _date(f.get("date"), "date")
    end = _date(f["end"], "end") if f.get("end") else None
    if end is not None and end < start:
        raise CalendarRejected(422, "end must be on or after date")
    if end == start:
        end = None
    time_ = f.get("time")
    end_time = f.get("end_time")
    if end_time and not time_:
        raise CalendarRejected(422, "end_time needs a start time")
    if time_:
        h, m = _clock(time_, "time")
        time_ = "%02d:%02d" % (h, m)
        if end_time:
            eh, em = _clock(end_time, "end_time")
            end_time = "%02d:%02d" % (eh, em)
            if end is None and (eh, em) <= (h, m):
                raise CalendarRejected(422, "end_time must be after time on a one-day event")
    for k, cap in (("location", 160), ("badge", 40), ("description", 2000)):
        if f.get(k) and len(str(f[k])) > cap:
            raise CalendarRejected(422, "%s is over %d characters" % (k, cap))
    return {"title": title, "unit": unit, "date": start, "end": end, "time": time_,
            "end_time": end_time, "location": f.get("location"), "badge": f.get("badge"),
            "description": f.get("description")}


def build_ics(uid, ev, stamp=None):
    """The complete text/calendar object for one event. `ev` is clean()'s
    output. Returns bytes with CRLF line ends."""
    stamp = stamp or datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    start, end = ev["date"], ev["end"]
    lines = ["BEGIN:VEVENT", "UID:%s" % uid, "DTSTAMP:%s" % stamp, "SUMMARY:%s" % esc(ev["title"])]
    timed = bool(ev["time"])
    if timed:
        h, m = _clock(ev["time"], "time")
        lines.append("DTSTART;TZID=%s:%s" % (TZID, "%sT%02d%02d00" % (start.strftime("%Y%m%d"), h, m)))
        if end is not None:
            if ev["end_time"]:
                eh, em = _clock(ev["end_time"], "end_time")
                lines.append("DTEND;TZID=%s:%sT%02d%02d00" % (TZID, end.strftime("%Y%m%d"), eh, em))
            else:
                lines.append("DTEND;TZID=%s:%sT120000" % (TZID, end.strftime("%Y%m%d")))
        else:
            if ev["end_time"]:
                eh, em = _clock(ev["end_time"], "end_time")
                dt_end = datetime.datetime.combine(start, datetime.time(eh, em))
            else:
                dt_end = (datetime.datetime.combine(start, datetime.time(h, m))
                          + datetime.timedelta(minutes=DEFAULT_TIMED_MINUTES))
            lines.append("DTEND;TZID=%s:%s" % (TZID, dt_end.strftime("%Y%m%dT%H%M00")))
    else:
        lines.append("DTSTART;VALUE=DATE:%s" % start.strftime("%Y%m%d"))
        last = end or start
        lines.append("DTEND;VALUE=DATE:%s" % (last + datetime.timedelta(days=1)).strftime("%Y%m%d"))
    if ev.get("location"):
        lines.append("LOCATION:%s" % esc(ev["location"]))
    if ev.get("description"):
        lines.append("DESCRIPTION:%s" % esc(ev["description"]))
    lines.append("CATEGORIES:%s" % esc(ev["unit"]))
    if ev.get("badge"):
        lines.append("X-SCOUT73-BADGE:%s" % esc(ev["badge"]))
    lines.append("END:VEVENT")
    body = ["BEGIN:VCALENDAR", "VERSION:2.0", "PRODID:-//greenlanescouts73.org//site-calendar-write//EN"]
    if timed:
        body += VTIMEZONE_NY
    body += lines + ["END:VCALENDAR"]
    out = []
    for ln in body:
        out.extend(fold(ln))
    return ("\r\n".join(out) + "\r\n").encode("utf-8")


# ---------------------------------------------------------------------------
# CalDAV
# ---------------------------------------------------------------------------

def _request(method, uid, data=None):
    if not configured():
        raise CalendarRejected(503, "calendar write-back is not configured (RADICALE_URL/USER/PASS)")
    if not owns(uid):
        raise CalendarRejected(403, "the site only manages its own events (%s or the seeded %s...%s)"
                               % (UID_SUFFIX, SEED_PREFIX, SEED_SUFFIX))
    url = RADICALE_URL.rstrip("/") + "/" + object_name(uid)
    req = urllib.request.Request(url, data=data, method=method)
    req.add_header("Authorization", "Basic " + base64.b64encode(
        ("%s:%s" % (RADICALE_USER, RADICALE_PASS)).encode()).decode())
    if data is not None:
        req.add_header("Content-Type", "text/calendar; charset=utf-8")
    try:
        with _urlopen(req, timeout=TIMEOUT) as resp:
            return resp.status
    except urllib.error.HTTPError as e:
        if method == "DELETE" and e.code == 404:
            return 404
        raise CalendarRejected(502, "calendar store answered %d on %s" % (e.code, method))
    except Exception as e:
        raise CalendarRejected(502, "calendar store unreachable: %s" % e)


# Seam for tests.
_urlopen = urllib.request.urlopen


def put_event(uid, ev):
    return _request("PUT", uid, build_ics(uid, ev))


def delete_event(uid):
    return _request("DELETE", uid)