""" Calendar operations - Sans-I/O business logic for Calendar objects. This module contains pure functions for Calendar operations like component class detection, sync token generation, and result processing. Both sync and async clients use these same functions. """ from __future__ import annotations import hashlib from dataclasses import dataclass from typing import Any from urllib.parse import quote # Component type to class name mapping COMPONENT_CLASS_MAP = { "BEGIN:VEVENT": "Event", "BEGIN:VTODO": "Todo", "BEGIN:VJOURNAL": "Journal", "BEGIN:VFREEBUSY": "FreeBusy", } @dataclass class CalendarObjectInfo: """Information about a calendar object extracted from server response.""" url: str data: str | None etag: str | None component_type: str | None # "Event", "Todo", "Journal", "FreeBusy" extra_props: dict def _detect_component_type_from_string(data: str) -> str | None: """ Detect the component type (Event, Todo, etc.) from iCalendar string data. Args: data: iCalendar data as string Returns: Component type name ("Event", "Todo", "Journal", "FreeBusy") or None """ for line in data.split("\n"): line = line.strip() if line in COMPONENT_CLASS_MAP: return COMPONENT_CLASS_MAP[line] return None def _detect_component_type_from_icalendar(ical_obj: Any) -> str | None: """ Detect the component type from an icalendar object. Args: ical_obj: icalendar.Calendar or similar object with subcomponents Returns: Component type name ("Event", "Todo", "Journal", "FreeBusy") or None """ import icalendar ical2name = { icalendar.Event: "Event", icalendar.Todo: "Todo", icalendar.Journal: "Journal", icalendar.FreeBusy: "FreeBusy", } if not hasattr(ical_obj, "subcomponents"): return None if not len(ical_obj.subcomponents): return None for sc in ical_obj.subcomponents: if sc.__class__ in ical2name: return ical2name[sc.__class__] return None def _detect_component_type(data: Any) -> str | None: """ Detect the component type from iCalendar data (string or object). Args: data: iCalendar data as string, bytes, or icalendar object Returns: Component type name ("Event", "Todo", "Journal", "FreeBusy") or None """ if data is None: return None # Try string detection first if hasattr(data, "split"): return _detect_component_type_from_string(data) # Try icalendar object detection if hasattr(data, "subcomponents"): return _detect_component_type_from_icalendar(data) return None def _generate_fake_sync_token(etags_and_urls: list[tuple[str | None, str]]) -> str: """ Generate a fake sync token for servers without sync support. Uses a hash of all ETags/URLs to detect changes. This allows clients to use the sync token API even when the server doesn't support it. Args: etags_and_urls: List of (etag, url) tuples. ETag may be None. Returns: A fake sync token string prefixed with "fake-" """ parts = [] for etag, url in etags_and_urls: if etag: parts.append(str(etag)) else: # Use URL as fallback identifier parts.append(str(url)) parts.sort() # Consistent ordering combined = "|".join(parts) hash_value = hashlib.md5(combined.encode(), usedforsecurity=False).hexdigest() return f"fake-{hash_value}" def _is_fake_sync_token(token: str | None) -> bool: """ Check if a sync token is a fake one generated by the client. Args: token: Sync token string Returns: True if this is a fake sync token """ return token is not None and isinstance(token, str) and token.startswith("fake-") def _normalize_result_url(result_url: str, parent_url: str) -> str: """ Normalize a URL from search/report results. Handles quoting for relative URLs and ensures proper joining with parent. Args: result_url: URL from server response (may be relative or absolute) parent_url: Parent calendar URL Returns: Normalized URL string ready for joining with parent """ # If it's a full URL, return as-is if "://" in result_url: return result_url # Quote relative paths return quote(result_url) def _should_skip_calendar_self_reference(result_url: str, calendar_url: str) -> bool: """ Check if a result URL should be skipped because it's the calendar itself. iCloud and some other servers return the calendar URL along with calendar item URLs. This function helps filter those out. Args: result_url: URL from server response calendar_url: The calendar's URL Returns: True if this URL should be skipped (it's the calendar itself) """ # Normalize both URLs for comparison result_normalized = result_url.rstrip("/") calendar_normalized = calendar_url.rstrip("/") # Check if they're the same return result_normalized == calendar_normalized def _process_report_results( results: dict, calendar_url: str, calendar_data_tag: str = "{urn:ietf:params:xml:ns:caldav}calendar-data", etag_tag: str = "{DAV:}getetag", ) -> list[CalendarObjectInfo]: """ Process REPORT response results into CalendarObjectInfo objects. Args: results: Dict mapping href -> properties dict calendar_url: URL of the calendar (to filter out self-references) calendar_data_tag: XML tag for calendar data property etag_tag: XML tag for etag property Returns: List of CalendarObjectInfo objects """ objects = [] calendar_url_normalized = calendar_url.rstrip("/") for href, props in results.items(): # Skip calendar self-reference if _should_skip_calendar_self_reference(href, calendar_url_normalized): continue # Extract calendar data data = props.pop(calendar_data_tag, None) # Extract etag etag = props.get(etag_tag) # Detect component type component_type = _detect_component_type(data) # Normalize URL normalized_url = _normalize_result_url(href, calendar_url) objects.append( CalendarObjectInfo( url=normalized_url, data=data, etag=etag, component_type=component_type, extra_props=props, ) ) return objects def _build_calendar_object_url( calendar_url: str, object_id: str, ) -> str: """ Build a URL for a calendar object from calendar URL and object ID. Args: calendar_url: URL of the parent calendar object_id: ID of the calendar object (typically UID.ics) Returns: Full URL for the calendar object """ calendar_url = str(calendar_url).rstrip("/") object_id = quote(str(object_id)) if not object_id.endswith(".ics"): object_id += ".ics" return f"{calendar_url}/{object_id}"