322 lines
12 KiB
Python
Executable File
322 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Create bill due-date events in the user's iCloud Bills calendar via CalDAV.
|
|
|
|
Uses only Python stdlib. Secret is read from a protected password file.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import ssl
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from email.utils import parseaddr
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from xml.etree import ElementTree as ET
|
|
|
|
ICLOUD_USER = "g@germainebrown.com"
|
|
PASSWORD_FILE = Path("/root/.config/himalaya/g-germainebrown-icloud-calendar.pass")
|
|
STATE_DIR = Path("/root/.hermes/email_triage")
|
|
CAL_STATE_FILE = STATE_DIR / "calendar_events.json"
|
|
CAL_LOG_FILE = STATE_DIR / "calendar_events.jsonl"
|
|
CALENDAR_NAME = "Bills"
|
|
DISCOVERY_URL = "https://caldav.icloud.com/.well-known/caldav"
|
|
USER_AGENT = "DAVKit/4.0.3 (732); CalendarStore/4.0.3 (991); iCal/4.0.3 (1388); Mac OS X/10.6.4 (10F569)"
|
|
|
|
MONTHS = {
|
|
"jan": 1, "january": 1,
|
|
"feb": 2, "february": 2,
|
|
"mar": 3, "march": 3,
|
|
"apr": 4, "april": 4,
|
|
"may": 5,
|
|
"jun": 6, "june": 6,
|
|
"jul": 7, "july": 7,
|
|
"aug": 8, "august": 8,
|
|
"sep": 9, "sept": 9, "september": 9,
|
|
"oct": 10, "october": 10,
|
|
"nov": 11, "november": 11,
|
|
"dec": 12, "december": 12,
|
|
}
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def load_password() -> str:
|
|
if not PASSWORD_FILE.exists():
|
|
raise SystemExit(f"Missing iCloud password file: {PASSWORD_FILE}")
|
|
mode = PASSWORD_FILE.stat().st_mode & 0o777
|
|
if mode != 0o600:
|
|
raise SystemExit(f"Password file permissions must be 600, got {mode:o}: {PASSWORD_FILE}")
|
|
return PASSWORD_FILE.read_text().strip()
|
|
|
|
|
|
def auth_header() -> str:
|
|
token = base64.b64encode(f"{ICLOUD_USER}:{load_password()}".encode()).decode()
|
|
return f"Basic {token}"
|
|
|
|
|
|
def request(method: str, url: str, body: str | bytes | None = None, depth: str | None = None) -> tuple[int, bytes, dict[str, str]]:
|
|
data = body.encode("utf-8") if isinstance(body, str) else body
|
|
headers = {
|
|
"Authorization": auth_header(),
|
|
"User-Agent": USER_AGENT,
|
|
}
|
|
if body is not None:
|
|
headers["Content-Type"] = "application/xml; charset=utf-8" if method in {"PROPFIND", "PROPPATCH", "MKCOL", "MKCALENDAR"} else "text/calendar; charset=utf-8"
|
|
if depth is not None:
|
|
headers["Depth"] = depth
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
ctx = ssl.create_default_context()
|
|
try:
|
|
with urllib.request.urlopen(req, context=ctx, timeout=45) as resp:
|
|
return resp.status, resp.read(), dict(resp.headers.items())
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read(), dict(e.headers.items())
|
|
|
|
|
|
def propfind(url: str, body: str, depth: str = "0") -> ET.Element:
|
|
code, data, _ = request("PROPFIND", url, body, depth=depth)
|
|
if code != 207:
|
|
raise RuntimeError(f"PROPFIND {url} failed HTTP {code}: {data[:500]!r}")
|
|
return ET.fromstring(data)
|
|
|
|
|
|
def first_text(root: ET.Element, tag: str) -> str | None:
|
|
el = root.find(f".//{{{tag.split('}', 1)[0].strip('{')}}}{tag.split('}', 1)[1]}") if tag.startswith("{") else root.find(f".//{tag}")
|
|
return el.text if el is not None else None
|
|
|
|
|
|
def discover_calendar_home() -> str:
|
|
principal_body = """<?xml version="1.0" encoding="UTF-8"?>
|
|
<d:propfind xmlns:d="DAV:"><d:prop><d:current-user-principal/></d:prop></d:propfind>"""
|
|
root = propfind(DISCOVERY_URL, principal_body, "0")
|
|
principal = root.findtext(".//{DAV:}current-user-principal/{DAV:}href")
|
|
if not principal:
|
|
raise RuntimeError("Could not discover iCloud current-user-principal")
|
|
principal_url = urllib.parse.urljoin(DISCOVERY_URL, principal)
|
|
home_body = """<?xml version="1.0" encoding="UTF-8"?>
|
|
<d:propfind xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav"><d:prop><c:calendar-home-set/></d:prop></d:propfind>"""
|
|
root = propfind(principal_url, home_body, "0")
|
|
home = root.findtext(".//{urn:ietf:params:xml:ns:caldav}calendar-home-set/{DAV:}href")
|
|
if not home:
|
|
raise RuntimeError("Could not discover iCloud calendar-home-set")
|
|
return home
|
|
|
|
|
|
def list_calendars(home_url: str) -> list[dict[str, Any]]:
|
|
body = """<?xml version="1.0" encoding="UTF-8"?>
|
|
<d:propfind xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav"><d:prop><d:displayname/><d:resourcetype/><c:supported-calendar-component-set/></d:prop></d:propfind>"""
|
|
root = propfind(home_url, body, "1")
|
|
parsed = urllib.parse.urlparse(home_url)
|
|
base = f"{parsed.scheme}://{parsed.netloc}"
|
|
out: list[dict[str, Any]] = []
|
|
for resp in root.findall("{DAV:}response"):
|
|
href = resp.findtext("{DAV:}href") or ""
|
|
name = resp.findtext(".//{DAV:}displayname") or ""
|
|
comps = [c.attrib.get("name", "") for c in resp.findall(".//{urn:ietf:params:xml:ns:caldav}comp")]
|
|
if name:
|
|
out.append({"name": name, "href": href, "url": urllib.parse.urljoin(base, href), "components": comps})
|
|
return out
|
|
|
|
|
|
def get_bills_calendar_url() -> str:
|
|
home = discover_calendar_home()
|
|
calendars = list_calendars(home)
|
|
for cal in calendars:
|
|
if cal["name"] == CALENDAR_NAME and "VEVENT" in cal["components"]:
|
|
return cal["url"]
|
|
names = ", ".join(f"{c['name']}({','.join(c['components'])})" for c in calendars)
|
|
raise RuntimeError(f"Calendar {CALENDAR_NAME!r} with VEVENT support not found. Available: {names}")
|
|
|
|
|
|
def parse_due_date(text: str) -> date:
|
|
s = text.strip().replace(",", "")
|
|
# YYYY-MM-DD
|
|
m = re.fullmatch(r"(\d{4})-(\d{1,2})-(\d{1,2})", s)
|
|
if m:
|
|
return date(int(m.group(1)), int(m.group(2)), int(m.group(3)))
|
|
# mm/dd/yy or mm/dd/yyyy
|
|
m = re.fullmatch(r"(\d{1,2})/(\d{1,2})/(\d{2}|\d{4})", s)
|
|
if m:
|
|
year = int(m.group(3))
|
|
if year < 100:
|
|
year += 2000
|
|
return date(year, int(m.group(1)), int(m.group(2)))
|
|
# Month d yyyy
|
|
m = re.fullmatch(r"([A-Za-z]+)\s+(\d{1,2})\s+(\d{4})", s)
|
|
if m and m.group(1).lower() in MONTHS:
|
|
return date(int(m.group(3)), MONTHS[m.group(1).lower()], int(m.group(2)))
|
|
raise SystemExit(f"Could not parse due date: {text!r}; use YYYY-MM-DD, MM/DD/YYYY, or Month D YYYY")
|
|
|
|
|
|
def clean_text(s: str) -> str:
|
|
return re.sub(r"[\r\n]+", " ", s or "").strip()
|
|
|
|
|
|
def ics_escape(s: str) -> str:
|
|
return clean_text(s).replace("\\", "\\\\").replace(";", r"\;").replace(",", r"\,")
|
|
|
|
|
|
def fold_ical_line(line: str) -> str:
|
|
raw = line.encode("utf-8")
|
|
if len(raw) <= 75:
|
|
return line
|
|
parts: list[str] = []
|
|
cur = ""
|
|
for ch in line:
|
|
prefix = "" if not parts else " "
|
|
if len((prefix + cur + ch).encode("utf-8")) > 75:
|
|
parts.append(("" if not parts else " ") + cur)
|
|
cur = ch
|
|
else:
|
|
cur += ch
|
|
if cur:
|
|
parts.append(("" if not parts else " ") + cur)
|
|
return "\r\n".join(parts)
|
|
|
|
|
|
def make_event_uid(message_id: str, vendor: str, due: date, uid: str) -> str:
|
|
key = "|".join([message_id.strip(), vendor.strip().lower(), due.isoformat(), uid.strip()])
|
|
digest = hashlib.sha256(key.encode()).hexdigest()[:24]
|
|
return f"bill-{digest}@hermes-germainebrown"
|
|
|
|
|
|
def load_state() -> dict[str, Any]:
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
if not CAL_STATE_FILE.exists():
|
|
return {"created": {}, "created_at": now_iso()}
|
|
try:
|
|
return json.loads(CAL_STATE_FILE.read_text())
|
|
except Exception:
|
|
return {"created": {}, "created_at": now_iso(), "state_read_error": True}
|
|
|
|
|
|
def save_state(state: dict[str, Any]) -> None:
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
tmp = CAL_STATE_FILE.with_suffix(".tmp")
|
|
tmp.write_text(json.dumps(state, indent=2, sort_keys=True))
|
|
tmp.replace(CAL_STATE_FILE)
|
|
|
|
|
|
def log_event(entry: dict[str, Any]) -> None:
|
|
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
|
with CAL_LOG_FILE.open("a") as f:
|
|
f.write(json.dumps({"timestamp": now_iso(), **entry}, sort_keys=True) + "\n")
|
|
|
|
|
|
def add_event(args: argparse.Namespace) -> None:
|
|
due = parse_due_date(args.due_date)
|
|
vendor = clean_text(args.vendor) or "Unknown vendor"
|
|
amount = clean_text(args.amount)
|
|
title = f"Bill due: {vendor}" + (f" - {amount}" if amount else "")
|
|
event_uid = make_event_uid(args.message_id or "", vendor, due, args.uid or "")
|
|
state = load_state()
|
|
existing = state.get("created", {})
|
|
for _uid, _info in existing.items():
|
|
if _info.get("vendor") == vendor and _info.get("due_date") == due.isoformat():
|
|
print(json.dumps({"status": "duplicate_skipped", "existing_event_uid": _uid, "vendor": vendor, "due_date": due.isoformat(), "new_uid": event_uid}, indent=2))
|
|
log_event({"action": "duplicate_skipped", "existing_event_uid": _uid, "vendor": vendor, "due_date": due.isoformat()})
|
|
return
|
|
if event_uid in existing:
|
|
print(json.dumps({"status": "exists", "event_uid": event_uid, **state["created"][event_uid]}, indent=2))
|
|
return
|
|
cal_url = get_bills_calendar_url().rstrip("/") + "/"
|
|
event_url = urllib.parse.urljoin(cal_url, urllib.parse.quote(event_uid) + ".ics")
|
|
dtstamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
|
|
dtstart = due.strftime("%Y%m%d")
|
|
dtend = (due + timedelta(days=1)).strftime("%Y%m%d")
|
|
description = "\\n".join([
|
|
f"Source email UID: {args.uid or ''}",
|
|
f"Message-ID: {args.message_id or ''}",
|
|
f"Subject: {args.subject or ''}",
|
|
f"Sender: {args.sender or ''}",
|
|
])
|
|
lines = [
|
|
"BEGIN:VCALENDAR",
|
|
"VERSION:2.0",
|
|
"PRODID:-//Hermes Agent//Bill Calendar//EN",
|
|
"CALSCALE:GREGORIAN",
|
|
"BEGIN:VEVENT",
|
|
f"UID:{event_uid}",
|
|
f"DTSTAMP:{dtstamp}",
|
|
f"DTSTART;VALUE=DATE:{dtstart}",
|
|
f"DTEND;VALUE=DATE:{dtend}",
|
|
f"SUMMARY:{ics_escape(title)}",
|
|
f"DESCRIPTION:{ics_escape(description)}",
|
|
"TRANSP:TRANSPARENT",
|
|
"BEGIN:VALARM",
|
|
"ACTION:DISPLAY",
|
|
f"DESCRIPTION:{ics_escape(title)}",
|
|
"TRIGGER:-P7D",
|
|
"END:VALARM",
|
|
"BEGIN:VALARM",
|
|
"ACTION:DISPLAY",
|
|
f"DESCRIPTION:{ics_escape(title)}",
|
|
"TRIGGER:-P1D",
|
|
"END:VALARM",
|
|
"END:VEVENT",
|
|
"END:VCALENDAR",
|
|
]
|
|
ics = "\r\n".join(fold_ical_line(line) for line in lines) + "\r\n"
|
|
code, data, _ = request("PUT", event_url, ics.encode("utf-8"))
|
|
if code not in (200, 201, 204):
|
|
raise RuntimeError(f"PUT event failed HTTP {code}: {data[:500]!r}")
|
|
record = {
|
|
"calendar": CALENDAR_NAME,
|
|
"calendar_url": cal_url,
|
|
"event_url": event_url,
|
|
"title": title,
|
|
"due_date": due.isoformat(),
|
|
"vendor": vendor,
|
|
"amount": amount,
|
|
"source_uid": args.uid,
|
|
"message_id": args.message_id,
|
|
"subject": args.subject,
|
|
"http_code": code,
|
|
}
|
|
state.setdefault("created", {})[event_uid] = record
|
|
save_state(state)
|
|
log_event({"action": "created", "event_uid": event_uid, **record})
|
|
print(json.dumps({"status": "created", "event_uid": event_uid, **record}, indent=2))
|
|
|
|
|
|
def list_bills_calendar() -> None:
|
|
url = get_bills_calendar_url()
|
|
print(json.dumps({"status": "ok", "calendar": CALENDAR_NAME, "calendar_url": url}, indent=2))
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--verify", action="store_true", help="Verify iCloud login and locate Bills calendar")
|
|
ap.add_argument("--add-event", action="store_true", help="Add a bill all-day event")
|
|
ap.add_argument("--uid", default="")
|
|
ap.add_argument("--message-id", default="")
|
|
ap.add_argument("--vendor", default="")
|
|
ap.add_argument("--due-date", default="")
|
|
ap.add_argument("--amount", default="")
|
|
ap.add_argument("--subject", default="")
|
|
ap.add_argument("--sender", default="")
|
|
args = ap.parse_args()
|
|
if args.verify:
|
|
list_bills_calendar()
|
|
return
|
|
if args.add_event:
|
|
if not args.due_date:
|
|
raise SystemExit("--due-date is required for --add-event")
|
|
add_event(args)
|
|
return
|
|
ap.print_help()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|