4e3c6acab6
CalDAV servers (Migadu) reject ICS files with METHOD:REQUEST/REPLY as it's an iTIP email concept, not a storage property. Strip it when saving to local calendar, for both send and reply flows.
418 lines
14 KiB
Python
418 lines
14 KiB
Python
#!/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, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from email.mime.base import MIMEBase
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.text import MIMEText
|
|
|
|
from icalendar import Alarm, Calendar, Event, vCalAddress, 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 _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_email(email_str, account=None):
|
|
"""Send a raw MIME email via himalaya message send (stdin)."""
|
|
cmd = ["himalaya"]
|
|
if account:
|
|
cmd += ["--account", account]
|
|
cmd += ["message", "send"]
|
|
subprocess.run(cmd, input=email_str, text=True, check=True)
|
|
|
|
|
|
def _build_calendar_email(from_addr, to_addr, subject, body, ics_bytes, method="REQUEST"):
|
|
"""Build a MIME email with a text/calendar attachment."""
|
|
msg = MIMEMultipart('mixed')
|
|
msg['From'] = from_addr
|
|
msg['To'] = to_addr
|
|
msg['Subject'] = subject
|
|
msg.attach(MIMEText(body, 'plain', 'utf-8'))
|
|
ics_part = MIMEBase('text', 'calendar', method=method, charset='utf-8')
|
|
ics_part.set_payload(ics_bytes.decode('utf-8'))
|
|
ics_part.add_header('Content-Disposition', 'attachment; filename="invite.ics"')
|
|
msg.attach(ics_part)
|
|
return msg.as_string()
|
|
|
|
|
|
def _strip_method(ics_bytes):
|
|
"""Remove METHOD property from ICS for CalDAV storage.
|
|
|
|
CalDAV servers reject METHOD (it's an iTIP/email concept, not a storage one).
|
|
"""
|
|
cal = Calendar.from_ical(ics_bytes)
|
|
if "method" in cal:
|
|
del cal["method"]
|
|
return cal.to_ical()
|
|
|
|
|
|
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.now(timezone.utc))
|
|
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)
|
|
organizer = vCalAddress(f"mailto:{args.sender}")
|
|
organizer.params["CN"] = vText(organizer_name)
|
|
event.add("organizer", organizer)
|
|
|
|
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",
|
|
})
|
|
|
|
# 1-day reminder
|
|
alarm = Alarm()
|
|
alarm.add("action", "DISPLAY")
|
|
alarm.add("description", f"Reminder: {args.summary}")
|
|
alarm.add("trigger", timedelta(days=-1))
|
|
event.add_component(alarm)
|
|
|
|
cal.add_component(event)
|
|
ics_bytes = cal.to_ical()
|
|
|
|
# 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 MIME email
|
|
email_str = _build_calendar_email(args.sender, all_to, args.subject, body, ics_bytes, method="REQUEST")
|
|
|
|
if args.dry_run:
|
|
print("=== ICS Content ===")
|
|
print(ics_bytes.decode())
|
|
print("=== Email Message ===")
|
|
print(email_str)
|
|
return
|
|
|
|
# Send email via himalaya message send (stdin)
|
|
_send_email(email_str, args.account)
|
|
print(f"Calendar invite sent to: {args.to}")
|
|
|
|
# Save to local calendar (without METHOD for CalDAV compatibility)
|
|
if CALENDAR_DIR.is_dir():
|
|
dest = CALENDAR_DIR / f"{uid}.ics"
|
|
dest.write_bytes(_strip_method(ics_bytes))
|
|
print(f"Saved to local calendar: {dest}")
|
|
_sync_calendar()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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.now(timezone.utc))
|
|
|
|
# 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()
|
|
|
|
# Build email
|
|
prefix = SUBJECT_PREFIX[partstat]
|
|
subject = f"{prefix}: {summary}"
|
|
|
|
body = f"{prefix}: {summary}"
|
|
if args.comment:
|
|
body += f"\n\n{args.comment}"
|
|
|
|
email_str = _build_calendar_email(args.sender, organizer_email, subject, body, reply_ics_bytes, method="REPLY")
|
|
|
|
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("=== Email Message ===")
|
|
print(email_str)
|
|
if cleanup_dir:
|
|
for f in cleanup_dir.iterdir():
|
|
f.unlink()
|
|
cleanup_dir.rmdir()
|
|
return
|
|
|
|
# Send reply
|
|
_send_email(email_str, 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"):
|
|
fwd_body = f"{prefix}: {summary}"
|
|
fwd_email = _build_calendar_email(
|
|
args.sender, DEFAULT_OWNER_EMAIL,
|
|
f"{prefix}: {summary}", fwd_body,
|
|
ics_path.read_bytes(), method="REQUEST",
|
|
)
|
|
try:
|
|
_send_email(fwd_email, args.account)
|
|
print(f"Forwarded invite to {DEFAULT_OWNER_EMAIL}")
|
|
except subprocess.CalledProcessError:
|
|
print(f"Warning: Failed to forward invite to {DEFAULT_OWNER_EMAIL}")
|
|
|
|
# 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 (without METHOD for CalDAV)
|
|
dest.write_bytes(_strip_method(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
|
|
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()
|