"""Extract ICS calendar data from emails.""" import email import logging from dataclasses import dataclass from email.message import Message import icalendar log = logging.getLogger(__name__) @dataclass class CalendarEvent: method: str # REQUEST, REPLY, CANCEL, PUBLISH vevent: icalendar.cal.Event uid: str def process_email(raw_bytes: bytes) -> list[CalendarEvent]: """Parse a raw email and extract calendar events from ICS content. Looks for: - text/calendar MIME parts (iMIP inline) - application/ics attachments - Attachments with .ics extension """ msg = email.message_from_bytes(raw_bytes) subject = msg.get("Subject", "(no subject)") log.debug("Processing email: %s", subject) events: list[CalendarEvent] = [] for part in msg.walk(): content_type = part.get_content_type() filename = part.get_filename() or "" is_calendar = content_type in ("text/calendar", "application/ics") is_ics_attachment = filename.lower().endswith(".ics") if not (is_calendar or is_ics_attachment): continue payload = part.get_payload(decode=True) if not payload: continue try: cal = icalendar.Calendar.from_ical(payload) except Exception: log.warning("Failed to parse ICS from email: %s", subject, exc_info=True) continue method = str(cal.get("METHOD", "PUBLISH")).upper() for component in cal.walk(): if component.name != "VEVENT": continue uid = str(component.get("UID", "")) if not uid: log.warning("VEVENT without UID in email: %s", subject) continue events.append(CalendarEvent(method=method, vevent=component, uid=uid)) if events: log.info("Found %d calendar event(s) in email: %s", len(events), subject) else: log.debug("No calendar content in email: %s", subject) return events