calendar invite
This commit is contained in:
Yanxin Lu
439
skills/calendar-invite/scripts/calendar_invite.py
Normal file
439
skills/calendar-invite/scripts/calendar_invite.py
Normal file
@@ -0,0 +1,439 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Calendar Invite — Send, accept, and decline calendar invites via himalaya.
|
||||
|
||||
Uses the icalendar library for proper RFC 5545 ICS generation and parsing.
|
||||
Uses himalaya CLI for email delivery. Syncs to local CalDAV via vdirsyncer.
|
||||
|
||||
Subcommands:
|
||||
python calendar_invite.py send [options] # create and send an invite
|
||||
python calendar_invite.py reply [options] # accept/decline/tentative
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import subprocess
|
||||
import sys
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from icalendar import Calendar, Event, vText
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
DEFAULT_TIMEZONE = "America/Los_Angeles"
|
||||
DEFAULT_FROM = "youlu@luyanxin.com"
|
||||
DEFAULT_OWNER_EMAIL = "mail@luyx.org" # Always added as attendee
|
||||
CALENDAR_DIR = Path.home() / ".openclaw" / "workspace" / "calendars" / "home"
|
||||
PRODID = "-//OpenClaw//CalendarInvite//EN"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _himalaya(*args):
|
||||
"""Run a himalaya command and return stdout."""
|
||||
result = subprocess.run(
|
||||
["himalaya", *args],
|
||||
capture_output=True, text=True, check=True,
|
||||
)
|
||||
return result.stdout
|
||||
|
||||
|
||||
def _himalaya_with_account(account, *args):
|
||||
"""Run a himalaya command with optional account flag."""
|
||||
cmd = ["himalaya"]
|
||||
if account:
|
||||
cmd += ["--account", account]
|
||||
cmd += list(args)
|
||||
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
||||
return result.stdout
|
||||
|
||||
|
||||
def _sync_calendar():
|
||||
"""Sync local calendar to CalDAV server via vdirsyncer."""
|
||||
try:
|
||||
subprocess.run(
|
||||
["vdirsyncer", "sync"],
|
||||
capture_output=True, text=True, check=True,
|
||||
)
|
||||
print("Synced to CalDAV server")
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
print("Warning: CalDAV sync failed (will retry on next heartbeat)")
|
||||
|
||||
|
||||
def _send_mml(mml, account=None):
|
||||
"""Send an MML message via himalaya template send."""
|
||||
cmd = ["himalaya"]
|
||||
if account:
|
||||
cmd += ["--account", account]
|
||||
cmd += ["template", "send"]
|
||||
subprocess.run(cmd, input=mml, text=True, check=True)
|
||||
|
||||
|
||||
def _parse_iso_datetime(dt_str):
|
||||
"""Parse ISO 8601 datetime string to a datetime object."""
|
||||
# Handle both 2026-03-20T14:00:00 and 2026-03-20T14:00
|
||||
for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M"):
|
||||
try:
|
||||
return datetime.strptime(dt_str, fmt)
|
||||
except ValueError:
|
||||
continue
|
||||
raise ValueError(f"Cannot parse datetime: {dt_str}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Send invite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def cmd_send(args):
|
||||
"""Create and send a calendar invite."""
|
||||
start = _parse_iso_datetime(args.start)
|
||||
end = _parse_iso_datetime(args.end)
|
||||
uid = args.uid or f"{uuid.uuid4()}@openclaw"
|
||||
organizer_name = args.organizer or args.sender
|
||||
|
||||
# Build ICS
|
||||
cal = Calendar()
|
||||
cal.add("prodid", PRODID)
|
||||
cal.add("version", "2.0")
|
||||
cal.add("calscale", "GREGORIAN")
|
||||
cal.add("method", "REQUEST")
|
||||
|
||||
event = Event()
|
||||
event.add("uid", uid)
|
||||
event.add("dtstamp", datetime.utcnow())
|
||||
event.add("dtstart", start, parameters={"TZID": args.timezone})
|
||||
event.add("dtend", end, parameters={"TZID": args.timezone})
|
||||
event.add("summary", args.summary)
|
||||
event.add("status", "CONFIRMED")
|
||||
event.add("sequence", 0)
|
||||
event["organizer"] = f"mailto:{args.sender}"
|
||||
event["organizer"].params["CN"] = vText(organizer_name)
|
||||
|
||||
if args.location:
|
||||
event.add("location", args.location)
|
||||
if args.description:
|
||||
event.add("description", args.description)
|
||||
|
||||
recipients = [addr.strip() for addr in args.to.split(",")]
|
||||
|
||||
# Always include owner as attendee
|
||||
all_attendees = list(recipients)
|
||||
if DEFAULT_OWNER_EMAIL not in all_attendees:
|
||||
all_attendees.append(DEFAULT_OWNER_EMAIL)
|
||||
|
||||
for addr in all_attendees:
|
||||
event.add("attendee", f"mailto:{addr}", parameters={
|
||||
"ROLE": "REQ-PARTICIPANT",
|
||||
"RSVP": "TRUE",
|
||||
})
|
||||
|
||||
cal.add_component(event)
|
||||
ics_bytes = cal.to_ical()
|
||||
|
||||
# Write ICS to temp file
|
||||
tmp_ics = Path(f"/tmp/openclaw-invite-{int(datetime.now().timestamp())}.ics")
|
||||
tmp_ics.write_bytes(ics_bytes)
|
||||
|
||||
# Build plain text body
|
||||
body = f"You're invited to: {args.summary}\n\nWhen: {args.start} - {args.end} ({args.timezone})"
|
||||
if args.location:
|
||||
body += f"\nWhere: {args.location}"
|
||||
if args.description:
|
||||
body += f"\n\n{args.description}"
|
||||
|
||||
# Email goes to all attendees (including owner)
|
||||
all_to = ", ".join(all_attendees)
|
||||
|
||||
# Build MML message
|
||||
mml = (
|
||||
f"From: {args.sender}\n"
|
||||
f"To: {all_to}\n"
|
||||
f"Subject: {args.subject}\n"
|
||||
f"\n"
|
||||
f"<#multipart type=mixed>\n"
|
||||
f"<#part type=text/plain>\n"
|
||||
f"{body}\n"
|
||||
f"<#part type=text/calendar method=REQUEST filename={tmp_ics} name=invite.ics><#/part>\n"
|
||||
f"<#/multipart>"
|
||||
)
|
||||
|
||||
if args.dry_run:
|
||||
print("=== ICS Content ===")
|
||||
print(ics_bytes.decode())
|
||||
print("=== MML Message ===")
|
||||
print(mml)
|
||||
tmp_ics.unlink(missing_ok=True)
|
||||
return
|
||||
|
||||
# Send email
|
||||
_send_mml(mml, args.account)
|
||||
print(f"Calendar invite sent to: {args.to}")
|
||||
|
||||
# Save to local calendar
|
||||
if CALENDAR_DIR.is_dir():
|
||||
dest = CALENDAR_DIR / f"{uid}.ics"
|
||||
dest.write_bytes(ics_bytes)
|
||||
print(f"Saved to local calendar: {dest}")
|
||||
_sync_calendar()
|
||||
|
||||
tmp_ics.unlink(missing_ok=True)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Reply to invite
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
PARTSTAT_MAP = {
|
||||
"accept": "ACCEPTED",
|
||||
"accepted": "ACCEPTED",
|
||||
"decline": "DECLINED",
|
||||
"declined": "DECLINED",
|
||||
"tentative": "TENTATIVE",
|
||||
}
|
||||
|
||||
SUBJECT_PREFIX = {
|
||||
"ACCEPTED": "Accepted",
|
||||
"DECLINED": "Declined",
|
||||
"TENTATIVE": "Tentative",
|
||||
}
|
||||
|
||||
|
||||
def _extract_ics_from_email(envelope_id, folder, account):
|
||||
"""Download attachments from an email and find the .ics file."""
|
||||
download_dir = Path(f"/tmp/openclaw-ics-extract-{envelope_id}")
|
||||
download_dir.mkdir(exist_ok=True)
|
||||
|
||||
cmd = ["himalaya"]
|
||||
if account:
|
||||
cmd += ["--account", account]
|
||||
cmd += ["attachment", "download", "--folder", folder, str(envelope_id), "--dir", str(download_dir)]
|
||||
|
||||
try:
|
||||
subprocess.run(cmd, capture_output=True, text=True, check=True)
|
||||
except subprocess.CalledProcessError:
|
||||
pass # some emails have no attachments
|
||||
|
||||
ics_files = list(download_dir.glob("*.ics"))
|
||||
if not ics_files:
|
||||
print(f"Error: No .ics attachment found in envelope {envelope_id}", file=sys.stderr)
|
||||
# Cleanup
|
||||
for f in download_dir.iterdir():
|
||||
f.unlink()
|
||||
download_dir.rmdir()
|
||||
sys.exit(1)
|
||||
|
||||
return ics_files[0], download_dir
|
||||
|
||||
|
||||
def cmd_reply(args):
|
||||
"""Accept, decline, or tentatively accept a calendar invite."""
|
||||
partstat = PARTSTAT_MAP.get(args.action.lower())
|
||||
if not partstat:
|
||||
print(f"Error: --action must be accept, decline, or tentative", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Get the ICS file
|
||||
cleanup_dir = None
|
||||
if args.envelope_id:
|
||||
ics_path, cleanup_dir = _extract_ics_from_email(args.envelope_id, args.folder, args.account)
|
||||
elif args.ics_file:
|
||||
ics_path = Path(args.ics_file)
|
||||
if not ics_path.is_file():
|
||||
print(f"Error: ICS file not found: {ics_path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
else:
|
||||
print("Error: --envelope-id or --ics-file is required", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Parse original ICS
|
||||
original_cal = Calendar.from_ical(ics_path.read_bytes())
|
||||
|
||||
# Find the VEVENT
|
||||
original_event = None
|
||||
for component in original_cal.walk():
|
||||
if component.name == "VEVENT":
|
||||
original_event = component
|
||||
break
|
||||
|
||||
if not original_event:
|
||||
print("Error: No VEVENT found in ICS file", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# Extract fields from original
|
||||
uid = str(original_event.get("uid", ""))
|
||||
summary = str(original_event.get("summary", ""))
|
||||
organizer = original_event.get("organizer")
|
||||
|
||||
if not organizer:
|
||||
print("Error: No ORGANIZER found in ICS", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
organizer_email = str(organizer).replace("mailto:", "").replace("MAILTO:", "")
|
||||
|
||||
# Build reply calendar
|
||||
reply_cal = Calendar()
|
||||
reply_cal.add("prodid", PRODID)
|
||||
reply_cal.add("version", "2.0")
|
||||
reply_cal.add("calscale", "GREGORIAN")
|
||||
reply_cal.add("method", "REPLY")
|
||||
|
||||
reply_event = Event()
|
||||
reply_event.add("uid", uid)
|
||||
reply_event.add("dtstamp", datetime.utcnow())
|
||||
|
||||
# Copy timing from original
|
||||
if original_event.get("dtstart"):
|
||||
reply_event["dtstart"] = original_event["dtstart"]
|
||||
if original_event.get("dtend"):
|
||||
reply_event["dtend"] = original_event["dtend"]
|
||||
|
||||
reply_event.add("summary", summary)
|
||||
reply_event["organizer"] = original_event["organizer"]
|
||||
reply_event.add("attendee", f"mailto:{args.sender}", parameters={
|
||||
"PARTSTAT": partstat,
|
||||
"RSVP": "FALSE",
|
||||
})
|
||||
|
||||
if original_event.get("sequence"):
|
||||
reply_event.add("sequence", original_event.get("sequence"))
|
||||
|
||||
reply_cal.add_component(reply_event)
|
||||
reply_ics_bytes = reply_cal.to_ical()
|
||||
|
||||
# Write reply ICS to temp file
|
||||
tmp_reply = Path(f"/tmp/openclaw-reply-{int(datetime.now().timestamp())}.ics")
|
||||
tmp_reply.write_bytes(reply_ics_bytes)
|
||||
|
||||
# Build email
|
||||
prefix = SUBJECT_PREFIX[partstat]
|
||||
subject = f"{prefix}: {summary}"
|
||||
|
||||
body = f"{prefix}: {summary}"
|
||||
if args.comment:
|
||||
body += f"\n\n{args.comment}"
|
||||
|
||||
mml = (
|
||||
f"From: {args.sender}\n"
|
||||
f"To: {organizer_email}\n"
|
||||
f"Subject: {subject}\n"
|
||||
f"\n"
|
||||
f"<#multipart type=mixed>\n"
|
||||
f"<#part type=text/plain>\n"
|
||||
f"{body}\n"
|
||||
f"<#part type=text/calendar method=REPLY filename={tmp_reply} name=invite.ics><#/part>\n"
|
||||
f"<#/multipart>"
|
||||
)
|
||||
|
||||
if args.dry_run:
|
||||
print("=== Original Event ===")
|
||||
print(f"Summary: {summary}")
|
||||
print(f"Organizer: {organizer_email}")
|
||||
print(f"Action: {partstat}")
|
||||
print()
|
||||
print("=== Reply ICS ===")
|
||||
print(reply_ics_bytes.decode())
|
||||
print("=== MML Message ===")
|
||||
print(mml)
|
||||
tmp_reply.unlink(missing_ok=True)
|
||||
if cleanup_dir:
|
||||
for f in cleanup_dir.iterdir():
|
||||
f.unlink()
|
||||
cleanup_dir.rmdir()
|
||||
return
|
||||
|
||||
# Send reply
|
||||
_send_mml(mml, args.account)
|
||||
print(f"Calendar invite {partstat.lower()}: {summary} (replied to {organizer_email})")
|
||||
|
||||
# Forward invite to owner on accept/tentative
|
||||
if partstat in ("ACCEPTED", "TENTATIVE"):
|
||||
tmp_fwd = Path(f"/tmp/openclaw-fwd-{int(datetime.now().timestamp())}.ics")
|
||||
tmp_fwd.write_bytes(ics_path.read_bytes())
|
||||
fwd_mml = (
|
||||
f"From: {args.sender}\n"
|
||||
f"To: {DEFAULT_OWNER_EMAIL}\n"
|
||||
f"Subject: {prefix}: {summary}\n"
|
||||
f"\n"
|
||||
f"<#multipart type=mixed>\n"
|
||||
f"<#part type=text/plain>\n"
|
||||
f"{prefix}: {summary}\n"
|
||||
f"<#part type=text/calendar method=REQUEST filename={tmp_fwd} name=invite.ics><#/part>\n"
|
||||
f"<#/multipart>"
|
||||
)
|
||||
try:
|
||||
_send_mml(fwd_mml, args.account)
|
||||
print(f"Forwarded invite to {DEFAULT_OWNER_EMAIL}")
|
||||
except subprocess.CalledProcessError:
|
||||
print(f"Warning: Failed to forward invite to {DEFAULT_OWNER_EMAIL}")
|
||||
tmp_fwd.unlink(missing_ok=True)
|
||||
|
||||
# Save to / remove from local calendar
|
||||
if CALENDAR_DIR.is_dir():
|
||||
dest = CALENDAR_DIR / f"{uid}.ics"
|
||||
if partstat in ("ACCEPTED", "TENTATIVE"):
|
||||
# Save the original event to local calendar
|
||||
dest.write_bytes(ics_path.read_bytes())
|
||||
print(f"Saved to local calendar: {dest}")
|
||||
elif partstat == "DECLINED" and dest.is_file():
|
||||
dest.unlink()
|
||||
print("Removed from local calendar")
|
||||
_sync_calendar()
|
||||
|
||||
# Cleanup
|
||||
tmp_reply.unlink(missing_ok=True)
|
||||
if cleanup_dir:
|
||||
for f in cleanup_dir.iterdir():
|
||||
f.unlink()
|
||||
cleanup_dir.rmdir()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Calendar invite tool")
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
# --- send ---
|
||||
send_p = subparsers.add_parser("send", help="Send a calendar invite")
|
||||
send_p.add_argument("--from", dest="sender", default=DEFAULT_FROM, help="Sender email")
|
||||
send_p.add_argument("--to", required=True, help="Recipient(s), comma-separated")
|
||||
send_p.add_argument("--subject", required=True, help="Email subject")
|
||||
send_p.add_argument("--summary", required=True, help="Event title")
|
||||
send_p.add_argument("--start", required=True, help="Start time (ISO 8601)")
|
||||
send_p.add_argument("--end", required=True, help="End time (ISO 8601)")
|
||||
send_p.add_argument("--timezone", default=DEFAULT_TIMEZONE, help="IANA timezone")
|
||||
send_p.add_argument("--location", default="", help="Event location")
|
||||
send_p.add_argument("--description", default="", help="Event description")
|
||||
send_p.add_argument("--organizer", default="", help="Organizer display name")
|
||||
send_p.add_argument("--uid", default="", help="Custom event UID")
|
||||
send_p.add_argument("--account", default="", help="Himalaya account")
|
||||
send_p.add_argument("--dry-run", action="store_true", help="Preview without sending")
|
||||
|
||||
# --- reply ---
|
||||
reply_p = subparsers.add_parser("reply", help="Reply to a calendar invite")
|
||||
reply_p.add_argument("--from", dest="sender", default=DEFAULT_FROM, help="Your email")
|
||||
reply_p.add_argument("--action", required=True, help="accept, decline, or tentative")
|
||||
reply_p.add_argument("--envelope-id", default="", help="Himalaya envelope ID")
|
||||
reply_p.add_argument("--ics-file", default="", help="Path to .ics file")
|
||||
reply_p.add_argument("--account", default="", help="Himalaya account")
|
||||
reply_p.add_argument("--folder", default="INBOX", help="Himalaya folder")
|
||||
reply_p.add_argument("--comment", default="", help="Message to include in reply")
|
||||
reply_p.add_argument("--dry-run", action="store_true", help="Preview without sending")
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.command == "send":
|
||||
cmd_send(args)
|
||||
elif args.command == "reply":
|
||||
cmd_reply(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user