room_cal/scripts/fetch_busy_ics.py

#!/usr/bin/env python3
"""Fetch a public CalDAV calendar and write a combined ICS to stdout or a file.

Sends a CalDAV REPORT to the given URL, extracts all VEVENT components, and
combines them into a single VCALENDAR document. No authentication required.

Usage:
  python scripts/fetch_busy_ics.py [--url URL] [-o FILE]
"""
import argparse
import sys
import xml.etree.ElementTree as ET

import httpx
from icalendar import Calendar, vText

DEFAULT_URL = (
  'https://dav.kj.zi3.fi/dav/calendars/'
  'blonde.neutral.jazz.legacy.episode.phrase@zi3.xyz/'
  '45911EE3-80B1-4D3E-9D07-20BA775E110F'
)

_REPORT_BODY = b"""\
<?xml version="1.0" encoding="utf-8" ?>
<C:calendar-query xmlns:D="DAV:" xmlns:C="urn:ietf:params:xml:ns:caldav">
  <D:prop>
    <D:getetag/>
    <C:calendar-data/>
  </D:prop>
  <C:filter>
    <C:comp-filter name="VCALENDAR">
      <C:comp-filter name="VEVENT"/>
    </C:comp-filter>
  </C:filter>
</C:calendar-query>"""

_DAV_NS    = 'DAV:'
_CALDAV_NS = 'urn:ietf:params:xml:ns:caldav'

PRODID = '-//room_cal//Busy//EN'


def fetch_calendar_data(url: str) -> list[str]:
  """Send a CalDAV REPORT and return one iCal string per event resource."""
  url = url.rstrip('/') + '/'
  resp = httpx.request(
    'REPORT',
    url,
    content=_REPORT_BODY,
    headers={
      'Content-Type': 'application/xml; charset=utf-8',
      'Depth': '1',
    },
  )
  resp.raise_for_status()

  root = ET.fromstring(resp.content)
  results = []
  for response in root.iter(f'{{{_DAV_NS}}}response'):
    for cal_data in response.iter(f'{{{_CALDAV_NS}}}calendar-data'):
      if cal_data.text:
        results.append(cal_data.text)
  return results


def build_ics(cal_data_list: list[str]) -> bytes:
  """Parse each calendar-data string and merge all VEVENTs into one VCALENDAR."""
  combined = Calendar()
  combined.add('prodid', PRODID)
  combined.add('version', '2.0')
  combined.add('calscale', vText('GREGORIAN'))
  combined.add('method', vText('PUBLISH'))

  for cal_text in cal_data_list:
    cal = Calendar.from_ical(cal_text)
    for component in cal.walk():
      if component.name == 'VEVENT':
        combined.add_component(component)

  return combined.to_ical()


def main() -> None:
  parser = argparse.ArgumentParser(
    description=__doc__,
    formatter_class=argparse.RawDescriptionHelpFormatter,
  )
  parser.add_argument('--url', default=DEFAULT_URL, help='CalDAV calendar URL (default: Busy calendar)')
  parser.add_argument('-o', '--output', default=None, metavar='FILE', help='Output file (default: stdout)')
  args = parser.parse_args()

  cal_data_list = fetch_calendar_data(args.url)
  ics = build_ics(cal_data_list)

  if args.output:
    with open(args.output, 'wb') as f:
      f.write(ics)
    print(f'{len(cal_data_list)} event(s) written to {args.output}', file=sys.stderr)
  else:
    sys.stdout.buffer.write(ics)


if __name__ == '__main__':
  main()