1000 lines
38 KiB
Python
1000 lines
38 KiB
Python
#!/usr/bin/env python3
|
|
# quickshell-cal.py — calendar + weather backend for the QuickShell bar popup.
|
|
#
|
|
# Runtime architecture:
|
|
# * A home-manager user systemd timer runs `qs-cal-sync sync` every 20 min:
|
|
# - Nextcloud CalDAV -> vdirsyncer -> khal (recurrence-aware) -> events.json
|
|
# - Open-Meteo forecast (keyless) -> weather.json
|
|
# * The QuickShell popup calls `qs-cal-sync read` (instant, merges caches to
|
|
# stdout as a single JSON doc) whenever it opens, and `qs-cal-sync
|
|
# setloc "<city or lat,lon>"` from the in-popup location editor.
|
|
#
|
|
# Nextcloud credentials are NOT compiled into anything: the sync step reads the
|
|
# env file rendered from the SOPS secret (default /run/secrets/hp-laptop/
|
|
# nextcloud-cal-env; override with $QS_CAL_SECRET_ENV), writes a temporary
|
|
# vdirsyncer config with them (mode 0600), syncs, then deletes it again.
|
|
#
|
|
# Every write is atomic (tmp + rename) so the shell never reads a half file.
|
|
|
|
import datetime
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import uuid
|
|
import wave
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
CACHE = os.path.expanduser("~/.cache/quickshell-cal")
|
|
SECRET_ENV = os.environ.get("QS_CAL_SECRET_ENV", "/run/secrets/hp-laptop/nextcloud-cal-env")
|
|
|
|
# Paths are substituted at build time by the Nix home-manager module so the
|
|
# script keeps working regardless of what PATH looks like in a user unit.
|
|
VDIRSYNCER = "@vdirsyncer@/bin/vdirsyncer"
|
|
|
|
VDIR = os.path.expanduser("~/.local/share/vdirsyncer")
|
|
STATUS = os.path.expanduser("~/.local/state/vdirsyncer")
|
|
|
|
EVENTS_FILE = os.path.join(CACHE, "events.json")
|
|
WEATHER_FILE = os.path.join(CACHE, "weather.json")
|
|
LOC_FILE = os.path.join(CACHE, "loc.json")
|
|
# Last-sync health, surfaced as the header badge in the shell popup.
|
|
SYNC_FILE = os.path.join(CACHE, "sync.json")
|
|
TMP_CONF = os.path.join(CACHE, "vdirsyncer.conf")
|
|
|
|
# Reminder plumbing (paths baked in by the Nix module). notify-send talks to
|
|
# the QuickShell NotificationServer (org.freedesktop.Notifications) so toasts
|
|
# appear in our own notification center.
|
|
NOTIFY = "@libnotify@/bin/notify-send"
|
|
PW_PLAY = "@pipewire@/bin/pw-play"
|
|
CHIME_FILE = os.path.join(CACHE, "chime.wav")
|
|
# Remember which (uid, occurrence, trigger) we already fired so the per-minute
|
|
# remind timer never double-fires.
|
|
REMINDER_STATE = os.path.join(CACHE, "reminders-fired.json")
|
|
# "uid|occurrence|trigger" -> ISO time at which a snoozed reminder may fire
|
|
# again. The shell writes it via `qs-cal-sync snooze <key> <minutes>`; remind()
|
|
# skips anything whose deadline is still in the future.
|
|
SNOOZE_FILE = os.path.join(CACHE, "snoozes.json")
|
|
|
|
# First-run bootstrap location (never leaves the machine; change in the popup).
|
|
DEFAULT_LOC = {"name": "Edinburgh", "lat": "55.9533", "lon": "-3.1883"}
|
|
|
|
EVENT_DAYS = 31
|
|
|
|
UA = "Nix-Vibe-quickshell/1.0"
|
|
|
|
|
|
def log(msg):
|
|
sys.stderr.write("[qs-cal] %s\n" % msg)
|
|
sys.stderr.flush()
|
|
|
|
|
|
def rd(blob, fallback):
|
|
"""json.load a file, tolerating absence/corruption."""
|
|
try:
|
|
with open(blob, "r") as f:
|
|
return json.load(f)
|
|
except (OSError, ValueError):
|
|
return fallback
|
|
|
|
|
|
def wr(blob, obj):
|
|
"""Atomic json write."""
|
|
os.makedirs(CACHE, exist_ok=True)
|
|
fd, tmp = tempfile.mkstemp(dir=CACHE, suffix=".tmp")
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(obj, f, ensure_ascii=False)
|
|
os.replace(tmp, blob)
|
|
|
|
|
|
def http_get(url, timeout=10):
|
|
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
return r.read().decode("utf-8", "replace")
|
|
|
|
|
|
def parse_env_file(path):
|
|
"""Read a sops-rendered KEY=VALUE env file into a dict."""
|
|
if not os.path.exists(path):
|
|
return {}
|
|
out = {}
|
|
try:
|
|
with open(path, "r") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
k, _, v = line.partition("=")
|
|
if k:
|
|
out[k.strip()] = v.strip().strip('"').strip("'")
|
|
except OSError as e:
|
|
log("cannot read secret env %s: %s" % (path, e))
|
|
return {}
|
|
return out
|
|
|
|
|
|
def write_tmp_vdirsyncer_conf(env):
|
|
"""Temporary vdirsyncer config that syncs EVERY calendar under the CalDAV
|
|
principal (collections = ["from a", "from b"] -> discovery)."""
|
|
os.makedirs(os.path.dirname(TMP_CONF), exist_ok=True)
|
|
url = env.get("NEXTCLOUD_CALDAV_URL", "")
|
|
user = env.get("NEXTCLOUD_CALDAV_USERNAME", "")
|
|
pw = env.get("NEXTCLOUD_CALDAV_PASSWORD", "")
|
|
# Nextcloud CalDAV requires the username in the URL path. If the user
|
|
# supplies the base path (/remote.php/dav/calendars/), append the username.
|
|
if url and user and url.rstrip("/").endswith("/calendars"):
|
|
url = url.rstrip("/") + "/" + user
|
|
conf = "\n".join([
|
|
"[general]",
|
|
'status_path = "%s"' % STATUS,
|
|
"",
|
|
"[pair nextcloud]",
|
|
'a = "nextcloud_local"',
|
|
'b = "nextcloud_remote"',
|
|
'collections = ["from a", "from b"]',
|
|
"",
|
|
"[storage nextcloud_local]",
|
|
'type = "filesystem"',
|
|
'path = "%s"' % VDIR,
|
|
'fileext = ".ics"',
|
|
"",
|
|
"[storage nextcloud_remote]",
|
|
'type = "caldav"',
|
|
'url = "%s"' % url,
|
|
'username = "%s"' % user,
|
|
'password = "%s"' % pw,
|
|
"",
|
|
])
|
|
with open(TMP_CONF, "w") as f:
|
|
f.write(conf)
|
|
os.chmod(TMP_CONF, 0o600)
|
|
|
|
|
|
def run_sync_cmd(argv, timeout=180, stdin_input=None):
|
|
try:
|
|
p = subprocess.run(
|
|
argv, input=stdin_input, capture_output=True, text=True, timeout=timeout)
|
|
if p.returncode != 0:
|
|
log("failed: %s -> %s" % (" ".join(argv), (p.stderr or p.stdout)[:400]))
|
|
return False
|
|
return True
|
|
except (OSError, subprocess.TimeoutExpired) as e:
|
|
log("error running %s: %s" % (argv[0], e))
|
|
return False
|
|
|
|
|
|
def list_calendars():
|
|
"""Names of the calendars vdirsyncer discovered (one subdir per calendar)."""
|
|
if not os.path.isdir(VDIR):
|
|
return []
|
|
try:
|
|
return sorted(n for n in os.listdir(VDIR)
|
|
if os.path.isdir(os.path.join(VDIR, n)))
|
|
except OSError:
|
|
return []
|
|
|
|
|
|
def ics_escape(s):
|
|
"""RFC 5545 TEXT escaping (backslash, comma, semicolon, newline)."""
|
|
return (str(s).replace("\\", "\\\\")
|
|
.replace(";", "\\;")
|
|
.replace(",", "\\,")
|
|
.replace("\n", "\\n"))
|
|
|
|
|
|
# Repeat choices offered by the shell form -> RFC 5545 FREQ.
|
|
REPEAT_FREQ = {"DAILY": "DAILY", "WEEKLY": "WEEKLY", "MONTHLY": "MONTHLY"}
|
|
|
|
|
|
def reminder_minutes(vevent):
|
|
"""Minutes-before from the first relative TRIGGER VALARM, or 0 when the
|
|
event has none (or only absolute/dated triggers)."""
|
|
for arm in vevent.walk("VALARM"):
|
|
try:
|
|
trig = arm.get("TRIGGER")
|
|
if trig is None:
|
|
continue
|
|
delta = trig.dt
|
|
except Exception:
|
|
continue
|
|
if not isinstance(delta, datetime.timedelta):
|
|
continue
|
|
# -PT5M -> 5 (minutes before start)
|
|
m = re.fullmatch(r"-PT(\d+)M", str(trig))
|
|
if m:
|
|
return int(m.group(1))
|
|
return 0
|
|
|
|
|
|
def vdir_push():
|
|
"""Push local .ics changes up to Nextcloud (vdirsyncer sync). No-op when
|
|
the secret env file is missing."""
|
|
secret = parse_env_file(SECRET_ENV)
|
|
if not secret.get("NEXTCLOUD_CALDAV_URL"):
|
|
return True
|
|
write_tmp_vdirsyncer_conf(secret)
|
|
try:
|
|
y = ("y\n") * 50
|
|
return run_sync_cmd([VDIRSYNCER, "-c", TMP_CONF, "sync"], stdin_input=y)
|
|
finally:
|
|
if os.path.exists(TMP_CONF):
|
|
os.remove(TMP_CONF)
|
|
|
|
|
|
def add_event(params):
|
|
"""Write a VEVENT into the chosen calendar's vdirsyncer local dir, then
|
|
push it to Nextcloud. params is the JSON dict from the shell popup."""
|
|
cal = (params.get("cal") or "").strip()
|
|
title = (params.get("title") or "").strip()
|
|
if not cal or not title:
|
|
return {"ok": False, "error": "calendar and title are required"}
|
|
caldir = os.path.join(VDIR, cal)
|
|
if not os.path.isdir(caldir):
|
|
return {"ok": False, "error": "unknown calendar %r" % cal}
|
|
|
|
date = (params.get("date") or "").strip()
|
|
all_day = bool(params.get("allDay"))
|
|
start = (params.get("start") or "").strip()
|
|
end = (params.get("end") or "").strip()
|
|
loc = (params.get("loc") or "").strip()
|
|
reminder = int(params.get("reminder") or 0)
|
|
repeat = (params.get("repeat") or "NONE").strip().upper()
|
|
|
|
tzlocal = datetime.datetime.now().astimezone().tzinfo
|
|
uid = "%s@nix-vibe" % uuid.uuid4()
|
|
now = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
|
|
# normalise dates/times -> ISO parts
|
|
try:
|
|
dt = datetime.date.fromisoformat(date)
|
|
except ValueError:
|
|
return {"ok": False, "error": "invalid date %r" % date}
|
|
|
|
def dt_line(key, d):
|
|
return d.astimezone(datetime.timezone.utc).strftime(key + ":%Y%m%dT%H%M%SZ")
|
|
|
|
lines = ["BEGIN:VCALENDAR", "VERSION:2.0",
|
|
"PRODID:-//Nix-Vibe//quickshell//EN",
|
|
"BEGIN:VEVENT", "UID:" + uid, "DTSTAMP:" + now]
|
|
|
|
if all_day:
|
|
# RFC 5545 all-day events use VALUE=DATE and exclusive end date.
|
|
lines.append("DTSTART;VALUE=DATE:%s" % dt.strftime("%Y%m%d"))
|
|
endd = dt + datetime.timedelta(days=1)
|
|
lines.append("DTEND;VALUE=DATE:%s" % endd.strftime("%Y%m%d"))
|
|
else:
|
|
start_dt = datetime.datetime.strptime(start or "09:00", "%H:%M").replace(
|
|
year=dt.year, month=dt.month, day=dt.day, tzinfo=tzlocal)
|
|
end_dt = datetime.datetime.strptime(end or "10:00", "%H:%M").replace(
|
|
year=dt.year, month=dt.month, day=dt.day, tzinfo=tzlocal)
|
|
if end_dt <= start_dt:
|
|
end_dt += datetime.timedelta(days=1)
|
|
lines.append(dt_line("DTSTART", start_dt))
|
|
lines.append(dt_line("DTEND", end_dt))
|
|
|
|
if repeat in REPEAT_FREQ:
|
|
lines.append("RRULE:FREQ=%s" % REPEAT_FREQ[repeat])
|
|
lines += ["SUMMARY:" + ics_escape(title)]
|
|
if loc:
|
|
lines.append("LOCATION:" + ics_escape(loc))
|
|
if reminder and reminder > 0:
|
|
lines += _alarm_lines(title, reminder)
|
|
lines += ["END:VEVENT", "END:VCALENDAR", ""]
|
|
vcal = "\r\n".join(lines)
|
|
|
|
path = os.path.join(caldir, uid + ".ics")
|
|
try:
|
|
with open(path, "w") as f:
|
|
f.write(vcal)
|
|
except OSError as e:
|
|
return {"ok": False, "error": str(e)}
|
|
|
|
if not vdir_push():
|
|
log("add: push failed for %s" % title)
|
|
_mark_sync_error("push failed for %r" % title)
|
|
fetch_events()
|
|
return {"ok": True, "uid": uid, "file": os.path.basename(path),
|
|
"cal": cal, "title": title}
|
|
|
|
|
|
def _alarm_lines(title, reminder):
|
|
return [
|
|
"BEGIN:VALARM",
|
|
"ACTION:DISPLAY",
|
|
"TRIGGER:-PT%dM" % reminder,
|
|
"DESCRIPTION:" + ics_escape(title),
|
|
"END:VALARM",
|
|
]
|
|
|
|
|
|
def _event_file(cal, file):
|
|
"""Resolve a (cal, basename/path) pair from the shell back to an existing
|
|
.ics on disk. Returns None when it has vanished (deleted server-side)."""
|
|
if not cal:
|
|
return None
|
|
base = os.path.basename(str(file))
|
|
cand = os.path.join(VDIR, cal, base)
|
|
if base and base.endswith(".ics") and os.path.isfile(cand):
|
|
return cand
|
|
return None
|
|
|
|
|
|
def edit_event(params):
|
|
"""Rewrite a VEVENT that the shell already knows about (matched by uid),
|
|
then push + refetch. The whole series is edited for recurring events."""
|
|
import icalendar
|
|
|
|
uid = (params.get("uid") or "").strip()
|
|
file = _event_file(params.get("cal"), params.get("file"))
|
|
title = (params.get("title") or "").strip()
|
|
if not uid or not file:
|
|
return {"ok": False, "error": "missing event reference"}
|
|
if not title:
|
|
return {"ok": False, "error": "title is required"}
|
|
|
|
date = (params.get("date") or "").strip()
|
|
all_day = bool(params.get("allDay"))
|
|
start = (params.get("start") or "").strip()
|
|
end = (params.get("end") or "").strip()
|
|
loc = (params.get("loc") or "").strip()
|
|
reminder = int(params.get("reminder") or 0)
|
|
repeat = (params.get("repeat") or "NONE").strip().upper()
|
|
|
|
try:
|
|
dt = datetime.date.fromisoformat(date)
|
|
except ValueError:
|
|
return {"ok": False, "error": "invalid date %r" % date}
|
|
|
|
try:
|
|
with open(file, "rb") as f:
|
|
cal = icalendar.Calendar.from_ical(f.read().decode("utf-8", "replace"))
|
|
except (OSError, ValueError) as e:
|
|
log("edit: cannot read %s: %s" % (file, e))
|
|
return {"ok": False, "error": "cannot read event file"}
|
|
|
|
vevent = None
|
|
for v in cal.walk("VEVENT"):
|
|
if str(v.get("UID") or "") == uid:
|
|
vevent = v
|
|
break
|
|
if vevent is None:
|
|
return {"ok": False, "error": "event uid no longer on disk (refetch?)"}
|
|
|
|
vevent["DTSTAMP"] = datetime.datetime.now(datetime.timezone.utc).strftime(
|
|
"%Y%m%dT%H%M%SZ")
|
|
vevent.pop("DTSTART", None)
|
|
vevent.pop("DTEND", None)
|
|
vevent.pop("RRULE", None)
|
|
vevent["SUMMARY"] = ics_escape(title)
|
|
|
|
if all_day:
|
|
# vDDDTypes: a plain date -> DTSTART;VALUE=DATE (exclusive DTEND +1d)
|
|
vevent["DTSTART"] = dt
|
|
vevent["DTEND"] = dt + datetime.timedelta(days=1)
|
|
else:
|
|
tzlocal = datetime.datetime.now().astimezone().tzinfo
|
|
start_dt = datetime.datetime.strptime(start or "09:00", "%H:%M").replace(
|
|
year=dt.year, month=dt.month, day=dt.day, tzinfo=tzlocal)
|
|
end_dt = datetime.datetime.strptime(end or "10:00", "%H:%M").replace(
|
|
year=dt.year, month=dt.month, day=dt.day, tzinfo=tzlocal)
|
|
if end_dt <= start_dt:
|
|
end_dt += datetime.timedelta(days=1)
|
|
def ics_dt(d):
|
|
return d.astimezone(datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
vevent["DTSTART"] = ics_dt(start_dt)
|
|
vevent["DTEND"] = ics_dt(end_dt)
|
|
|
|
if repeat in REPEAT_FREQ:
|
|
vevent["RRULE"] = "FREQ=%s" % REPEAT_FREQ[repeat]
|
|
if loc:
|
|
vevent["LOCATION"] = ics_escape(loc)
|
|
else:
|
|
vevent.pop("LOCATION", None)
|
|
|
|
# replace alarms entirely (single-child vdirsyncer items usually only hold
|
|
# one VEVENT plus optional VTIMEZONE; keep anything non-VALARM intact)
|
|
vevent.subcomponents = [
|
|
c for c in vevent.subcomponents if str(c.name) != "VALARM"]
|
|
if reminder and reminder > 0:
|
|
arm = icalendar.Alarm()
|
|
arm.add("ACTION", "DISPLAY")
|
|
arm.add("TRIGGER", datetime.timedelta(minutes=-reminder))
|
|
arm.add("DESCRIPTION", title)
|
|
vevent.add_component(arm)
|
|
|
|
try:
|
|
with open(file, "wb") as f:
|
|
f.write(cal.to_ical())
|
|
except OSError as e:
|
|
return {"ok": False, "error": str(e)}
|
|
|
|
if not vdir_push():
|
|
log("edit: push failed for %s" % uid)
|
|
_mark_sync_error("push failed for edit (uid %s)" % uid)
|
|
fetch_events()
|
|
return {"ok": True, "uid": uid, "cal": params.get("cal"), "title": title}
|
|
|
|
|
|
def delete_event(params):
|
|
"""Delete the .ics backing an event, push the removal to Nextcloud and
|
|
refetch. Deletes the whole event (every occurrence of a series)."""
|
|
file = _event_file(params.get("cal"), params.get("file"))
|
|
uid = (params.get("uid") or "").strip()
|
|
if not file or not uid:
|
|
return {"ok": False, "error": "missing event reference"}
|
|
try:
|
|
os.remove(file)
|
|
except OSError as e:
|
|
return {"ok": False, "error": str(e)}
|
|
if not vdir_push():
|
|
log("delete: push failed for %s" % uid)
|
|
_mark_sync_error("push failed for delete (uid %s)" % uid)
|
|
fetch_events()
|
|
return {"ok": True, "uid": uid}
|
|
|
|
|
|
def ensure_chime():
|
|
"""Synthesize a short two-tone 'ping' if we don't have one cached."""
|
|
if os.path.exists(CHIME_FILE):
|
|
return CHIME_FILE
|
|
os.makedirs(CACHE, exist_ok=True)
|
|
rate = 44100
|
|
n = (3.0 * 4) // 8 # make it a short pluck
|
|
# 0.55s decayed 880Hz ping
|
|
samples = []
|
|
for i in range(int(rate * 0.45)):
|
|
t = i / rate
|
|
env = math.exp(-t * 6)
|
|
samples.append(int(12000 * env * math.sin(2 * math.pi * 880 * t)))
|
|
for i in range(int(rate * 0.45)):
|
|
t = i / rate
|
|
env = math.exp(-t * 6)
|
|
samples.append(int(12000 * env * math.sin(2 * math.pi * 1320 * t)))
|
|
with wave.open(CHIME_FILE, "wb") as w:
|
|
w.setnchannels(1)
|
|
w.setsampwidth(2)
|
|
w.setframerate(rate)
|
|
w.writeframes(b"".join(int(s).to_bytes(2, "little", signed=True)
|
|
for s in samples))
|
|
return CHIME_FILE
|
|
|
|
|
|
def remind():
|
|
"""Scan every .ics in the vdirsyncer dirs for VALARM triggers, expand
|
|
RRULE recurrences, and fire any whose reminder time has just arrived:
|
|
play the cached chime and send a desktop notification (lands in the
|
|
QuickShell notification center). Tracked by (uid, occurrence, trigger)
|
|
so the per-minute timer never fires the same reminder twice."""
|
|
import icalendar
|
|
import recurring_ical_events
|
|
|
|
fired = rd(REMINDER_STATE, {})
|
|
snoozes = rd(SNOOZE_FILE, {})
|
|
now = datetime.datetime.now().astimezone()
|
|
fired_any = False
|
|
snooze_dirty = False
|
|
|
|
for calname in list_calendars():
|
|
caldir = os.path.join(VDIR, calname)
|
|
if not os.path.isdir(caldir):
|
|
continue
|
|
for fn in os.listdir(caldir):
|
|
if not fn.endswith(".ics"):
|
|
continue
|
|
path = os.path.join(caldir, fn)
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = f.read()
|
|
except OSError as e:
|
|
log("remind: cannot read %s: %s" % (path, e))
|
|
continue
|
|
# fast pre-filter: >90% of files have no VALARM at all, so skip
|
|
# the (expensive) full icalendar + recurrence expansion for them.
|
|
if b"VALARM" not in data:
|
|
continue
|
|
try:
|
|
cal = icalendar.Calendar.from_ical(data.decode("utf-8", "replace"))
|
|
except Exception as e:
|
|
log("remind: cannot parse %s: %s" % (path, e))
|
|
continue
|
|
|
|
for vevent in cal.walk("VEVENT"):
|
|
try:
|
|
uid = str(vevent.get("UID") or str(uuid.uuid4()))
|
|
summary = str(vevent.get("SUMMARY") or "(no title)")
|
|
loc = str(vevent.get("LOCATION") or "")
|
|
alarms = vevent.walk("VALARM")
|
|
if not alarms:
|
|
continue
|
|
dtstart = vevent.get("DTSTART")
|
|
if dtstart is None:
|
|
continue
|
|
# run recurrence expansion over a generous window
|
|
comp = icalendar.Calendar()
|
|
for tz in cal.walk("VTIMEZONE"):
|
|
comp.add_component(tz)
|
|
comp.add_component(vevent)
|
|
occs = recurring_ical_events.of(comp).between(
|
|
now - datetime.timedelta(days=1),
|
|
now + datetime.timedelta(days=31))
|
|
for occ in occs:
|
|
start = occ.get("DTSTART").dt
|
|
if isinstance(start, datetime.date) and not isinstance(start, datetime.datetime):
|
|
start = datetime.datetime.combine(start, datetime.time.min, tzinfo=now.tzinfo)
|
|
if start.tzinfo is None:
|
|
start = start.replace(tzinfo=now.tzinfo)
|
|
for alarm in alarms:
|
|
try:
|
|
trig = alarm.get("TRIGGER")
|
|
delta = trig.dt
|
|
except Exception:
|
|
continue
|
|
if not isinstance(delta, datetime.timedelta):
|
|
continue # absolute triggers skipped (rare)
|
|
# TRIGGER:-PT5M means 5 min BEFORE the start, so
|
|
# a negative delta moves `when` earlier.
|
|
when = start + delta
|
|
key = "%s|%s|%s" % (uid, start.isoformat(), delta)
|
|
snoozed_until = snoozes.get(key)
|
|
refire = False
|
|
if snoozed_until:
|
|
try:
|
|
if datetime.datetime.fromisoformat(snoozed_until) > now:
|
|
continue # still snoozing this one
|
|
except ValueError:
|
|
pass
|
|
# deadline passed: fire now even though we're
|
|
# outside the usual 2-minute lateness window,
|
|
# and let it happen again later.
|
|
snoozes.pop(key, None)
|
|
snooze_dirty = True
|
|
refire = True
|
|
if key in fired:
|
|
continue
|
|
if when <= now and ((now - when) <= datetime.timedelta(minutes=2) or refire):
|
|
fired[key] = now.isoformat()
|
|
fired_any = True
|
|
body = calname
|
|
if loc:
|
|
body += " \u00b7 " + loc
|
|
if start.time() != datetime.time.min or not isinstance(occ.get("DTSTART").dt, datetime.date):
|
|
body += " \u00b7 " + start.strftime("%H:%M")
|
|
try:
|
|
subprocess.run([PW_PLAY, ensure_chime()],
|
|
capture_output=True, timeout=10)
|
|
except Exception as e:
|
|
log("chime failed: %s" % e)
|
|
try:
|
|
# Identifier payloads (Open carries the
|
|
# event date; Snooze carries the exact
|
|
# reminder key) let the shell react without
|
|
# a DBus round trip — see quickshell-shell.qml.
|
|
subprocess.run(
|
|
[NOTIFY, "-a", "Calendar Reminder",
|
|
"-u", "normal",
|
|
"-c", "calendar",
|
|
"-t", "0",
|
|
"--action=cal-open:%s=OPEN" % start.strftime("%Y-%m-%d"),
|
|
"--action=cal-snooze:%s=SNOOZE 10M" % key,
|
|
"Reminder: " + summary,
|
|
body],
|
|
capture_output=True, timeout=10)
|
|
except Exception as e:
|
|
log("notify failed: %s" % e)
|
|
log("reminder fired: %s (%s)" % (summary, calname))
|
|
except Exception as e:
|
|
log("remind: event error: %s" % e)
|
|
|
|
if fired_any:
|
|
# prune old entries (older than 40 days)
|
|
cutoff = (now - datetime.timedelta(days=40)).isoformat()
|
|
fired = {k: v for k, v in fired.items() if v >= cutoff}
|
|
wr(REMINDER_STATE, fired)
|
|
|
|
if snooze_dirty:
|
|
# drop expired snoozes (older than 40 days) and persist the rest
|
|
cutoff = (now - datetime.timedelta(days=40)).isoformat()
|
|
snoozes = {k: v for k, v in snoozes.items() if v >= cutoff}
|
|
wr(SNOOZE_FILE, snoozes)
|
|
|
|
|
|
def _mark_sync_ok():
|
|
"""Record that a sync/event write succeeded (calendar header badge)."""
|
|
now_iso = datetime.datetime.now().isoformat(timespec="seconds")
|
|
st = rd(SYNC_FILE, {})
|
|
st.update({
|
|
"ok": True,
|
|
"lastOk": now_iso,
|
|
"lastEffort": now_iso,
|
|
"error": "",
|
|
})
|
|
wr(SYNC_FILE, st)
|
|
|
|
|
|
def _mark_sync_error(msg):
|
|
"""Record a failed sync/push (calendar header badge)."""
|
|
now_iso = datetime.datetime.now().isoformat(timespec="seconds")
|
|
st = rd(SYNC_FILE, {})
|
|
st.update({
|
|
"ok": False,
|
|
"lastOk": st.get("lastOk", ""),
|
|
"lastEffort": now_iso,
|
|
"error": msg,
|
|
})
|
|
wr(SYNC_FILE, st)
|
|
|
|
|
|
def fetch_events():
|
|
"""vdirsyncer sync -> icalendar self-scan of the local stores -> events.json.
|
|
|
|
Each event carries uid + file so the shell popup can edit/delete a specific
|
|
item. Recurrences are expanded via recurring_ical_events (the same library
|
|
the reminder scanner uses)."""
|
|
secret = parse_env_file(SECRET_ENV)
|
|
url = secret.get("NEXTCLOUD_CALDAV_URL", "")
|
|
user = secret.get("NEXTCLOUD_CALDAV_USERNAME", "")
|
|
pw = secret.get("NEXTCLOUD_CALDAV_PASSWORD", "")
|
|
if not (url and user and pw):
|
|
log("nextcloud secret missing or incomplete; keeping existing events.json")
|
|
_mark_sync_error("Nextcloud cal secret missing/incomplete")
|
|
return
|
|
try:
|
|
write_tmp_vdirsyncer_conf(secret)
|
|
effective_url = url
|
|
if url.rstrip("/").endswith("/calendars"):
|
|
effective_url = url.rstrip("/") + "/" + user
|
|
log("syncing from %s" % effective_url)
|
|
# `collections = ["from a", "from b"]` triggers auto-discovery; feed y
|
|
# answers so the "Should it attempt to create?" prompts never hang
|
|
# in our headless runner.
|
|
y = ("y\n") * 50
|
|
if not run_sync_cmd([VDIRSYNCER, "-c", TMP_CONF, "discover", "nextcloud"],
|
|
stdin_input=y):
|
|
return
|
|
if not run_sync_cmd([VDIRSYNCER, "-c", TMP_CONF, "sync"],
|
|
stdin_input=y):
|
|
return
|
|
if not run_sync_cmd([VDIRSYNCER, "-c", TMP_CONF, "sync"],
|
|
stdin_input=y):
|
|
return
|
|
finally:
|
|
if os.path.exists(TMP_CONF):
|
|
os.remove(TMP_CONF)
|
|
os.makedirs(VDIR, exist_ok=True)
|
|
wkdir = os.path.dirname(STATUS)
|
|
if wkdir:
|
|
os.makedirs(wkdir, exist_ok=True)
|
|
|
|
try:
|
|
events = scan_local_events()
|
|
except Exception as e:
|
|
log("scan_local_events failed: %s" % e)
|
|
_mark_sync_error("scan failed: %s" % e)
|
|
return
|
|
|
|
events.sort(key=lambda e: (e["d"], e["t"] if e["t"] else "00:00"))
|
|
now_iso = datetime.datetime.now().isoformat(timespec="seconds")
|
|
wr(EVENTS_FILE, {"fetched": now_iso, "events": events})
|
|
_mark_sync_ok()
|
|
log("wrote %d events" % len(events))
|
|
|
|
|
|
def scan_local_events():
|
|
"""Expand every .ics in the vdirsyncer stores into flat event rows, expanded
|
|
from recurrence rules. Returns list of dicts; the shell popup and the
|
|
edit/delete commands both rely on the uid/file fields."""
|
|
import icalendar
|
|
import recurring_ical_events
|
|
|
|
now = datetime.datetime.now()
|
|
window_start = now.date()
|
|
window_end = window_start + datetime.timedelta(days=EVENT_DAYS)
|
|
events = []
|
|
|
|
for calname in list_calendars():
|
|
caldir = os.path.join(VDIR, calname)
|
|
try:
|
|
names = sorted(os.listdir(caldir))
|
|
except OSError:
|
|
continue
|
|
for fn in names:
|
|
if not fn.endswith(".ics"):
|
|
continue
|
|
path = os.path.join(caldir, fn)
|
|
try:
|
|
with open(path, "rb") as f:
|
|
data = f.read().decode("utf-8", "replace")
|
|
except OSError as e:
|
|
log("scan: cannot read %s: %s" % (path, e))
|
|
continue
|
|
try:
|
|
cal = icalendar.Calendar.from_ical(data)
|
|
except Exception as e:
|
|
log("scan: cannot parse %s: %s" % (path, e))
|
|
continue
|
|
|
|
for vevent in cal.walk("VEVENT"):
|
|
uid = str(vevent.get("UID") or str(uuid.uuid4()))
|
|
summary = str(vevent.get("SUMMARY") or "(no title)").strip()
|
|
loc = str(vevent.get("LOCATION") or "").strip()
|
|
rrule = str(vevent.get("RRULE") or "")
|
|
if vevent.get("DTSTART") is None:
|
|
continue
|
|
comp = icalendar.Calendar()
|
|
for tz in cal.walk("VTIMEZONE"):
|
|
comp.add_component(tz)
|
|
comp.add_component(vevent)
|
|
try:
|
|
occs = recurring_ical_events.of(comp).between(
|
|
window_start, window_end)
|
|
except Exception as e:
|
|
log("scan: recurrence error in %s: %s" % (fn, e))
|
|
continue
|
|
for occ in occs:
|
|
st_raw = occ.get("DTSTART").dt
|
|
allDay = isinstance(st_raw, datetime.date) and not isinstance(
|
|
st_raw, datetime.datetime)
|
|
st = _to_local_naive(st_raw, allDay)
|
|
datepart = st.strftime("%Y-%m-%d")
|
|
timepart = st.strftime("%H:%M") if not allDay else ""
|
|
|
|
en_raw = occ.get("DTEND")
|
|
if en_raw is not None:
|
|
en_raw = en_raw.dt
|
|
if en_raw is None:
|
|
end_naive = st + (datetime.timedelta(days=1) if allDay
|
|
else datetime.timedelta(hours=1))
|
|
else:
|
|
end_naive = _to_local_naive(en_raw, allDay)
|
|
endpart = end_naive.strftime("%H:%M") if not allDay else ""
|
|
|
|
# drop timed occurrences that already finished
|
|
if not allDay and end_naive < now:
|
|
continue
|
|
|
|
events.append({
|
|
"d": datepart,
|
|
"t": timepart,
|
|
"e": endpart,
|
|
"title": summary,
|
|
"cal": calname,
|
|
"allDay": allDay,
|
|
"rep": bool(rrule),
|
|
"rrule": rrule,
|
|
"loc": loc,
|
|
"rem": reminder_minutes(vevent),
|
|
"uid": uid,
|
|
"file": path,
|
|
})
|
|
return events
|
|
|
|
|
|
def _to_local_naive(dt, allDay):
|
|
"""Normalise an occurrence start/end to a local naive datetime."""
|
|
if allDay and isinstance(dt, datetime.date) and not isinstance(dt, datetime.datetime):
|
|
return datetime.datetime.combine(dt, datetime.time.min)
|
|
if dt.tzinfo is not None:
|
|
return dt.astimezone().replace(tzinfo=None)
|
|
return dt
|
|
|
|
|
|
def refresh_local_cache():
|
|
"""Re-scan the local vdirsyncer stores (no network) and rewrite the event
|
|
cache so a mutation (add/edit/delete) shows up in the popup immediately
|
|
without waiting for the next timed sync."""
|
|
try:
|
|
events = scan_local_events()
|
|
wr(EVENTS_FILE, {
|
|
"events": events,
|
|
"fetched": datetime.datetime.now().isoformat(timespec="seconds"),
|
|
})
|
|
log("rescanned %d events" % len(events))
|
|
except Exception as e:
|
|
log("refresh_local_cache failed: %s" % e)
|
|
|
|
|
|
def fetch_weather():
|
|
"""Open-Meteo forecast for the stored location -> weather.json. Keeps a
|
|
current block (temp/feels/code/humidity/wind), per-day rows (min/max/precip
|
|
chance/max wind, 7 days) and an hourly trace for the next 24h so the shell
|
|
can draw an hourly strip / more days."""
|
|
loc = rd(LOC_FILE, None)
|
|
if not loc:
|
|
loc = dict(DEFAULT_LOC)
|
|
wr(LOC_FILE, loc)
|
|
lat, lon = str(loc.get("lat", "")), str(loc.get("lon", ""))
|
|
if not (lat and lon):
|
|
return
|
|
params = {
|
|
"latitude": lat,
|
|
"longitude": lon,
|
|
"current": "temperature_2m,apparent_temperature,weather_code,relative_humidity_2m,wind_speed_10m",
|
|
"hourly": "temperature_2m,weather_code,precipitation_probability,wind_speed_10m",
|
|
"daily": "weather_code,temperature_2m_max,temperature_2m_min,precipitation_probability_max,wind_speed_10m_max",
|
|
"forecast_days": "7",
|
|
"timezone": "auto",
|
|
}
|
|
url = "https://api.open-meteo.com/v1/forecast?" + urllib.parse.urlencode(params)
|
|
try:
|
|
data = json.loads(http_get(url))
|
|
except (OSError, ValueError) as e:
|
|
log("weather fetch failed: %s" % e)
|
|
return
|
|
|
|
cur = data.get("current") or {}
|
|
daily = data.get("daily") or {}
|
|
hourly = data.get("hourly") or {}
|
|
htime = hourly.get("time") or []
|
|
|
|
def hourly_row(i):
|
|
return {
|
|
"h": (htime[i][11:16] if i < len(htime) else ""),
|
|
"t": (hourly.get("temperature_2m") or [])[i],
|
|
"code": (hourly.get("weather_code") or [])[i],
|
|
"pop": (hourly.get("precipitation_probability") or [])[i],
|
|
"wind": (hourly.get("wind_speed_10m") or [])[i],
|
|
}
|
|
|
|
out = {
|
|
"name": loc.get("name", "?"),
|
|
"lat": lat,
|
|
"lon": lon,
|
|
"fetched": datetime.datetime.now().isoformat(timespec="seconds"),
|
|
"current": {
|
|
"t": cur.get("temperature_2m"),
|
|
"feels": cur.get("apparent_temperature"),
|
|
"code": cur.get("weather_code"),
|
|
"hum": cur.get("relative_humidity_2m"),
|
|
"wind": cur.get("wind_speed_10m"),
|
|
},
|
|
"hourly": [
|
|
hourly_row(i)
|
|
for i in range(min(24, len(htime)))
|
|
],
|
|
"daily": [
|
|
{
|
|
"d": (daily.get("time") or [])[i],
|
|
"code": (daily.get("weather_code") or [])[i],
|
|
"tmin": (daily.get("temperature_2m_min") or [])[i],
|
|
"tmax": (daily.get("temperature_2m_max") or [])[i],
|
|
"pop": (daily.get("precipitation_probability_max") or [])[i],
|
|
"wind": (daily.get("wind_speed_10m_max") or [])[i],
|
|
}
|
|
for i in range(min(7, len(daily.get("time") or [])))
|
|
],
|
|
}
|
|
wr(WEATHER_FILE, out)
|
|
log("wrote weather for %s" % out["name"])
|
|
|
|
|
|
def resolve_location(query):
|
|
"""'lat,lon' passthrough or Open-Meteo geocoding of a town name."""
|
|
q = query.strip()
|
|
if re.fullmatch(r"[-+]?\d+(?:\.\d+)?\s*,\s*[-+]?\d+(?:\.\d+)?", q):
|
|
lat, lon = [p.strip() for p in q.split(",")]
|
|
return {"name": q, "lat": lat, "lon": lon}
|
|
url = ("https://geocoding-api.open-meteo.com/v1/search?name="
|
|
+ urllib.parse.quote(q) + "&count=1&language=en&format=json")
|
|
try:
|
|
data = json.loads(http_get(url))
|
|
except (OSError, ValueError) as e:
|
|
log("geocoding failed: %s" % e)
|
|
return None
|
|
res = (data.get("results") or [None])[0]
|
|
if not res:
|
|
log("geocoding: no match for %r" % q)
|
|
return None
|
|
name = ", ".join(x for x in [
|
|
res.get("name"),
|
|
res.get("admin1"),
|
|
res.get("country_code"),
|
|
] if x)
|
|
return {"name": name, "lat": str(res["latitude"]), "lon": str(res["longitude"])}
|
|
|
|
|
|
def snooze(key, minutes):
|
|
"""Delay a specific reminder (key = uid|occurrence|trigger as shown in the
|
|
notification) by N minutes. The shell calls this when the SNOOZE action is
|
|
picked; remind() skips the key until the deadline passes, then fires it
|
|
again (the fired marker is cleared so the re-fire isn't suppressed)."""
|
|
try:
|
|
minutes = int(minutes)
|
|
except (TypeError, ValueError):
|
|
minutes = 10
|
|
if minutes < 1:
|
|
minutes = 10
|
|
snoozes = rd(SNOOZE_FILE, {})
|
|
snoozes[key] = (datetime.datetime.now().astimezone()
|
|
+ datetime.timedelta(minutes=minutes)).isoformat(timespec="seconds")
|
|
wr(SNOOZE_FILE, snoozes)
|
|
# let the notification appear again once the deadline passes
|
|
fired = rd(REMINDER_STATE, {})
|
|
fired.pop(key, None)
|
|
wr(REMINDER_STATE, fired)
|
|
print(json.dumps({"ok": True, "key": key, "until": snoozes[key]},
|
|
ensure_ascii=False))
|
|
sys.stdout.flush()
|
|
|
|
|
|
def dump_cache():
|
|
"""Merge the cache files into one stdout doc for the shell."""
|
|
ev = rd(EVENTS_FILE, None)
|
|
wx = rd(WEATHER_FILE, None)
|
|
loc = rd(LOC_FILE, None)
|
|
print(json.dumps({
|
|
"events": (ev or {}).get("events", []),
|
|
"eventsFetched": (ev or {}).get("fetched", ""),
|
|
"weather": wx,
|
|
"loc": loc,
|
|
"cals": list_calendars(),
|
|
"sync": rd(SYNC_FILE, None),
|
|
}, ensure_ascii=False))
|
|
sys.stdout.flush()
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
mode = args[0] if args else "read"
|
|
|
|
if mode == "sync":
|
|
fetch_events()
|
|
fetch_weather()
|
|
elif mode == "setloc" and len(args) > 1:
|
|
loc = resolve_location(args[1])
|
|
if loc:
|
|
wr(LOC_FILE, loc)
|
|
fetch_weather()
|
|
f = rd(WEATHER_FILE, None)
|
|
if f:
|
|
f["name"] = loc["name"]
|
|
f["lat"] = loc["lat"]
|
|
f["lon"] = loc["lon"]
|
|
wr(WEATHER_FILE, f)
|
|
dump_cache()
|
|
elif mode == "read":
|
|
dump_cache()
|
|
elif mode == "cals":
|
|
print(json.dumps(list_calendars(), ensure_ascii=False))
|
|
sys.stdout.flush()
|
|
elif mode == "add" and len(args) > 1:
|
|
try:
|
|
params = json.loads(args[1])
|
|
except ValueError:
|
|
params = {}
|
|
print(json.dumps(add_event(params), ensure_ascii=False))
|
|
sys.stdout.flush()
|
|
refresh_local_cache()
|
|
dump_cache()
|
|
elif mode == "edit" and len(args) > 1:
|
|
try:
|
|
params = json.loads(args[1])
|
|
except ValueError:
|
|
params = {}
|
|
print(json.dumps(edit_event(params), ensure_ascii=False))
|
|
sys.stdout.flush()
|
|
refresh_local_cache()
|
|
dump_cache()
|
|
elif mode == "delete" and len(args) > 1:
|
|
try:
|
|
params = json.loads(args[1])
|
|
except ValueError:
|
|
params = {}
|
|
print(json.dumps(delete_event(params), ensure_ascii=False))
|
|
sys.stdout.flush()
|
|
refresh_local_cache()
|
|
dump_cache()
|
|
elif mode == "snooze" and len(args) > 1:
|
|
snooze(args[1], args[2] if len(args) > 2 else "10")
|
|
elif mode == "remind":
|
|
remind()
|
|
else:
|
|
log("unknown mode %r" % mode)
|
|
sys.exit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |