path: root/desktop/island/calendar/calctl

#!/usr/bin/python3
#
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# The island calendar's backend: the vdirsyncer calendars as JSON. Reads the
# calendar list from khal's config so there is no second list to keep in
# step; khal and cal-notif read the same files and are not affected.
#
#   calctl calendars                 {default, calendars: [{name, color, path}]}
#   calctl list FROM TO              occurrences in [FROM, TO), local time
#   calctl get UID CAL               the series itself, for the editor
#   calctl new EVENT-JSON            {uid}
#   calctl edit UID CAL EVENT-JSON   {uid}; CAL is where it is now
#   calctl delete UID CAL            {}
#
# Every command prints one object: {"ok": true, "data": ...} or
# {"ok": false, "error": "..."} with exit 1. Environment, for the tests:
# CALCTL_KHAL_CONFIG, CALCTL_OVERRIDES, CALCTL_NOSYNC.

import json
import os
import subprocess
import sys
import tempfile
import tomllib
import uuid
from datetime import date, datetime, time, timedelta, timezone
from pathlib import Path
from zoneinfo import ZoneInfo

import icalendar
from dateutil.rrule import rrulestr

HOME = Path.home()
KHAL_CONFIG = Path(os.environ.get("CALCTL_KHAL_CONFIG", HOME / ".config/khal/config"))
OVERRIDES = Path(os.environ.get("CALCTL_OVERRIDES", HOME / ".config/cal-notif/overrides.toml"))
SYNC_LOG = HOME / ".local/share/vdirsyncer/sync.log"

PRESET_REPEAT = {"daily": "DAILY", "weekly": "WEEKLY", "monthly": "MONTHLY", "yearly": "YEARLY"}
PRESET_ALARM = {"10m": timedelta(minutes=10), "1h": timedelta(hours=1), "1d": timedelta(days=1)}

# khal's colour names as Theme property names; a #rrggbb colour passes
# through and anything else is the accent.
COLORS = {
    "black": "crust", "white": "text", "gray": "subtext", "light gray": "subtext",
    "dark gray": "overlay", "brown": "peach",
    "red": "red", "dark red": "red", "light red": "red",
    "green": "green", "dark green": "green", "light green": "green",
    "yellow": "yellow", "light yellow": "yellow",
    "blue": "blue", "dark blue": "blue", "light blue": "blue",
    "magenta": "mauve", "dark magenta": "mauve", "light magenta": "mauve",
    "cyan": "teal", "dark cyan": "teal", "light cyan": "teal",
}

USAGE = ("usage: calctl calendars | list FROM TO | get UID CAL | new JSON"
         " | edit UID CAL JSON | delete UID CAL")


class Fail(Exception):
    pass


def local_zone():
    tz = os.environ.get("TZ", "").lstrip(":")
    if tz:
        try:
            return ZoneInfo(tz)
        except Exception:
            pass
    real = os.path.realpath("/etc/localtime")
    if "zoneinfo/" in real:
        return ZoneInfo(real.split("zoneinfo/", 1)[1])
    return ZoneInfo("UTC")


LOCAL = local_zone()


def calendars():
    """([{name, path, color}], default name) from khal's [calendars].

    ponytail: khal's `type = discover` globs are not expanded; every
    calendar here is a plain path.
    """
    try:
        lines = KHAL_CONFIG.read_text().splitlines()
    except OSError as e:
        raise Fail(f"cannot read {KHAL_CONFIG}: {e.strerror}")
    cals, section, cur, default = [], None, None, None
    for raw in lines:
        line = raw.strip()
        # Only whole-line comments: a colour value can start with '#'.
        if not line or line.startswith("#"):
            continue
        if line.startswith("[["):
            cur = None
            if section == "calendars":
                cur = {"name": line.strip("[]").strip(), "path": None, "color": "accent"}
                cals.append(cur)
            continue
        if line.startswith("["):
            section, cur = line.strip("[]").strip(), None
            continue
        if "=" not in line:
            continue
        key, value = (s.strip() for s in line.split("=", 1))
        value = value.strip("'\"")
        if cur is not None and key == "path":
            cur["path"] = Path(os.path.expanduser(value))
        elif cur is not None and key == "color":
            cur["color"] = COLORS.get(value, value if value.startswith("#") else "accent")
        elif section == "default" and key == "default_calendar":
            default = value
    cals = [c for c in cals if c["path"]]
    if not cals:
        raise Fail(f"no calendars in {KHAL_CONFIG}")
    names = [c["name"] for c in cals]
    return cals, default if default in names else names[0]


def master(ics):
    """The series' VEVENT: the first one that is not an occurrence override."""
    for ev in ics.walk("VEVENT"):
        if "RECURRENCE-ID" not in ev:
            return ev
    return None


def events():
    """(calendar, path, VEVENT) for every event file. A file that fails to
    parse is skipped: one bad file must not blank the month."""
    cals, _ = calendars()
    for cal in cals:
        for path in sorted(cal["path"].glob("*.ics")):
            try:
                ev = master(icalendar.Calendar.from_ical(path.read_bytes()))
            except Exception:
                continue
            # ponytail: occurrence overrides (RECURRENCE-ID) and EXDATE are not
            # applied; none exist in the live calendars (spec, Scope).
            if ev is not None:
                yield cal, path, ev


def find(uid, cal_name):
    """An event by UID within one calendar: the same invite synced into two
    calendars has one UID and two files, and each row must act on its own."""
    for cal, path, ev in events():
        if cal["name"] == cal_name and str(ev.get("UID", "")) == uid:
            return cal, path
    raise Fail(f"no event {uid} in {cal_name}")


def as_local(value):
    """A date-time as naive local time; a date stays a date. A floating
    time (no zone) is local by definition."""
    if isinstance(value, datetime) and value.tzinfo is not None:
        return value.astimezone(LOCAL).replace(tzinfo=None)
    return value


def span(ev):
    """(start, end, all_day): naive local datetimes, or dates with an
    exclusive end for an all-day event."""
    start = as_local(ev["DTSTART"].dt)
    all_day = not isinstance(start, datetime)
    if "DTEND" in ev:
        end = as_local(ev["DTEND"].dt)
    elif "DURATION" in ev:
        end = start + ev["DURATION"].dt
    else:
        end = start + timedelta(days=1) if all_day else start
    return start, end, all_day


def repeat_of(ev):
    """(repeat, until) as the form shows them."""
    rule = ev.get("RRULE")
    if rule is None:
        return "none", ""
    names = {v: k for k, v in PRESET_REPEAT.items()}
    freq = rule.get("FREQ", [""])[0]
    # WKST changes nothing for a plain FREQ; the live birthdays all carry it.
    if set(rule) - {"WKST"} <= {"FREQ", "UNTIL"} and freq in names:
        until = ""
        if "UNTIL" in rule:
            u = as_local(rule["UNTIL"][0])
            until = (u.date() if isinstance(u, datetime) else u).isoformat()
        return names[freq], until
    return "custom", ""


def alarm_of(ev):
    alarms = [c for c in ev.subcomponents if c.name == "VALARM"]
    if not alarms:
        return "none"
    if len(alarms) == 1:
        trig = alarms[0].get("TRIGGER")
        if (trig is not None and isinstance(trig.dt, timedelta)
                and trig.params.get("RELATED", "START") == "START"):
            for name, before in PRESET_ALARM.items():
                if trig.dt == -before:
                    return name
    return "custom"


def overrides():
    """UIDs with a per-event alarm in cal-notif, which wins over the VALARM."""
    try:
        return set(tomllib.loads(OVERRIDES.read_text()))
    except (OSError, tomllib.TOMLDecodeError):
        return set()


def stamp(value):
    if isinstance(value, datetime):
        return value.isoformat(timespec="minutes")
    return value.isoformat()


def to_json(cal, ev, start, end, all_day, ov):
    repeat, until = repeat_of(ev)
    uid = str(ev.get("UID", ""))
    return {
        "uid": uid,
        "calendar": cal["name"],
        "color": cal["color"],
        "title": str(ev.get("SUMMARY", "")),
        "allDay": all_day,
        "start": stamp(start),
        # All-day ends go out inclusive: the last day the event covers.
        "end": stamp(max(start, end - timedelta(days=1))) if all_day else stamp(end),
        "location": str(ev.get("LOCATION", "")),
        "notes": str(ev.get("DESCRIPTION", "")),
        "repeat": repeat,
        "until": until,
        "alarm": alarm_of(ev),
        "override": uid in ov,
    }


def cmd_list(lo_s, hi_s):
    lo, hi = datetime.fromisoformat(lo_s), datetime.fromisoformat(hi_s)
    ov, out = overrides(), []
    for cal, _, ev in events():
        try:
            start, end, all_day = span(ev)
            s = datetime.combine(start, time()) if all_day else start
            e = datetime.combine(end, time()) if all_day else end
            dur = e - s
            if "RRULE" in ev:
                # Naive local time keeps the wall-clock time across DST.
                # ponytail: right for local and same-rule zones (all the live
                # data); a series in a zone with other DST dates shows an hour
                # off in the weeks they differ. Expand in the event's own zone
                # if such events arrive.
                rule = ev["RRULE"]
                if "UNTIL" in rule:
                    # UNTIL is UTC on disk; read it as local time like the
                    # start, or a 22:30 series loses its last occurrence.
                    rule = icalendar.vRecur(rule)
                    rule["UNTIL"] = [as_local(u) for u in rule["UNTIL"]]
                rule = rrulestr(rule.to_ical().decode(), dtstart=s, ignoretz=True)
                starts = rule.between(lo - dur, hi, inc=True)
            else:
                starts = [s]
        except Exception:
            continue
        for o in starts:
            if not (o < hi and (o + dur > lo or (not dur and o >= lo))):
                continue
            if all_day:
                out.append(to_json(cal, ev, o.date(), (o + dur).date(), True, ov))
            else:
                out.append(to_json(cal, ev, o, o + dur, False, ov))
    out.sort(key=lambda x: (x["start"][:10], not x["allDay"], x["start"]))
    return out


def cmd_get(uid, cal_name):
    cal, path = find(uid, cal_name)
    ev = master(icalendar.Calendar.from_ical(path.read_bytes()))
    start, end, all_day = span(ev)
    return to_json(cal, ev, start, end, all_day, overrides())


def parse_payload(text, cals):
    """The form's JSON, checked. Dates become date/datetime; an all-day end
    becomes exclusive."""
    try:
        p = json.loads(text)
    except json.JSONDecodeError as e:
        raise Fail(f"bad event json: {e}")
    if not isinstance(p, dict):
        raise Fail("bad event json: not an object")
    by_name = {c["name"]: c for c in cals}
    if p.get("calendar") not in by_name:
        raise Fail(f"unknown calendar {p.get('calendar')!r}")
    title = str(p.get("title", "")).strip()
    if not title:
        raise Fail("title is empty")
    all_day = bool(p.get("allDay"))
    try:
        if all_day:
            start = date.fromisoformat(p["start"])
            end = date.fromisoformat(p["end"]) + timedelta(days=1)
        else:
            start = datetime.fromisoformat(p["start"])
            end = datetime.fromisoformat(p["end"])
        until = date.fromisoformat(p["until"]) if p.get("until") else None
    except (KeyError, TypeError, ValueError) as e:
        raise Fail(f"bad date: {e}")
    if end < start:
        raise Fail("end is before start")
    repeat = p.get("repeat", "none")
    if repeat not in PRESET_REPEAT and repeat not in ("none", "custom"):
        raise Fail(f"unknown repeat {repeat!r}")
    alarm = p.get("alarm", "none")
    if alarm not in PRESET_ALARM and alarm not in ("none", "custom"):
        raise Fail(f"unknown alarm {alarm!r}")
    return {"cal": by_name[p["calendar"]], "title": title, "allDay": all_day,
            "start": start, "end": end, "until": until, "repeat": repeat, "alarm": alarm,
            "location": str(p.get("location", "")).strip(), "notes": str(p.get("notes", ""))}


def apply(ev, p):
    """Write the form's fields into a VEVENT, touching only what changed: an
    unchanged time keeps its zone and DURATION, an unchanged alarm keeps its
    ACTION and extras, an unchanged title keeps its parameters. Everything
    else on it is left alone; a `custom` repeat or alarm keeps what is there."""
    if "DTSTART" not in ev or span(ev) != (p["start"], p["end"], p["allDay"]):
        for key in ("DTSTART", "DTEND", "DURATION"):
            ev.pop(key, None)
        if p["allDay"]:
            ev.add("DTSTART", p["start"])
            ev.add("DTEND", p["end"])
        else:
            ev.add("DTSTART", p["start"].replace(tzinfo=LOCAL))
            ev.add("DTEND", p["end"].replace(tzinfo=LOCAL))
    for key, prop in (("title", "SUMMARY"), ("location", "LOCATION"), ("notes", "DESCRIPTION")):
        if str(ev.get(prop, "")) != p[key]:
            ev.pop(prop, None)
            if p[key]:
                ev.add(prop, p[key])
    until = p["until"].isoformat() if p["until"] and p["repeat"] != "none" else ""
    if p["repeat"] != "custom" and repeat_of(ev) != (p["repeat"], until):
        ev.pop("RRULE", None)
        if p["repeat"] != "none":
            rule = {"FREQ": PRESET_REPEAT[p["repeat"]]}
            if p["until"]:
                # RFC 5545: UNTIL is a DATE for a DATE start, UTC otherwise.
                rule["UNTIL"] = (p["until"] if p["allDay"] else
                                 datetime.combine(p["until"], time(23, 59, 59), LOCAL)
                                 .astimezone(timezone.utc))
            ev.add("RRULE", rule)
    if p["alarm"] != "custom" and alarm_of(ev) != p["alarm"]:
        ev.subcomponents = [c for c in ev.subcomponents if c.name != "VALARM"]
        if p["alarm"] != "none":
            alarm = icalendar.Alarm()
            alarm.add("ACTION", "DISPLAY")
            alarm.add("DESCRIPTION", p["title"])
            alarm.add("TRIGGER", -PRESET_ALARM[p["alarm"]])
            ev.add_component(alarm)
    now = datetime.now(timezone.utc).replace(microsecond=0)
    ev.pop("DTSTAMP", None)
    ev.pop("LAST-MODIFIED", None)
    ev.add("DTSTAMP", now)
    ev.add("LAST-MODIFIED", now)


def write(path, ics):
    """Temporary file in the same directory, then rename: a reader (khal,
    cal-notif, vdirsyncer, the panel's watch) never sees half a file. The
    temporary name does not end in .ics, so nothing lists it."""
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".calctl-", suffix=".tmp")
    try:
        # mkstemp makes 0600; vdirsyncer's own files are 0644.
        os.fchmod(fd, 0o644)
        with os.fdopen(fd, "wb") as f:
            f.write(ics.to_ical())
        os.replace(tmp, path)
    except BaseException:
        if os.path.exists(tmp):
            os.unlink(tmp)
        raise


def sync():
    """Push the change now rather than at the next cron run. Detached and
    not waited for; a failure only means the cron run carries it."""
    if os.environ.get("CALCTL_NOSYNC"):
        return
    try:
        SYNC_LOG.parent.mkdir(parents=True, exist_ok=True)
        with open(SYNC_LOG, "ab") as log:
            subprocess.Popen(["vdirsyncer", "-v", "ERROR", "sync"], stdin=subprocess.DEVNULL,
                             stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
    except OSError:
        pass


def cmd_new(text):
    cals, _ = calendars()
    p = parse_payload(text, cals)
    uid = str(uuid.uuid4())
    ics = icalendar.Calendar()
    ics.add("PRODID", "-//quickshell//calctl//EN")
    ics.add("VERSION", "2.0")
    ev = icalendar.Event()
    ev.add("UID", uid)
    apply(ev, p)
    ics.add_component(ev)
    ics.add_missing_timezones()
    write(p["cal"]["path"] / f"{uid}.ics", ics)
    sync()
    return {"uid": uid}


def cmd_edit(uid, cal_name, text):
    cals, _ = calendars()
    p = parse_payload(text, cals)
    _, path = find(uid, cal_name)
    ics = icalendar.Calendar.from_ical(path.read_bytes())
    ev = master(ics)
    if p["repeat"] == "custom":
        # A rule like BYDAY=MO is tied to its day; moving the start alone
        # would leave the series on the old day.
        day = lambda v: v.date() if isinstance(v, datetime) else v
        if day(span(ev)[0]) != day(p["start"]):
            raise Fail("this event repeats by a rule the panel cannot edit, so its date cannot move here")
    apply(ev, p)
    sequence = int(ev.pop("SEQUENCE", 0))
    ev.add("SEQUENCE", sequence + 1)
    ics.add_missing_timezones()
    target = p["cal"]["path"] / path.name
    if target != path and target.exists():
        raise Fail(f"{target.name} already exists in {p['cal']['name']}")
    write(target, ics)
    if target != path:
        path.unlink()
    sync()
    return {"uid": uid}


def cmd_delete(uid, cal_name):
    _, path = find(uid, cal_name)
    path.unlink()
    sync()
    return {}


def main(argv):
    cmd, args = (argv[1], argv[2:]) if len(argv) > 1 else ("", [])
    try:
        if cmd == "calendars" and not args:
            cals, default = calendars()
            data = {"default": default,
                    "calendars": [{"name": c["name"], "color": c["color"], "path": str(c["path"])}
                                  for c in cals]}
        elif cmd == "list" and len(args) == 2:
            data = cmd_list(*args)
        elif cmd == "get" and len(args) == 2:
            data = cmd_get(*args)
        elif cmd == "new" and len(args) == 1:
            data = cmd_new(*args)
        elif cmd == "edit" and len(args) == 3:
            data = cmd_edit(*args)
        elif cmd == "delete" and len(args) == 2:
            data = cmd_delete(*args)
        else:
            raise Fail(USAGE)
    except Exception as e:
        msg = str(e) if isinstance(e, Fail) else f"{type(e).__name__}: {e}"
        print(json.dumps({"ok": False, "error": msg}))
        print(f"calctl: {msg}", file=sys.stderr)
        return 1
    print(json.dumps({"ok": True, "data": data}, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))