"""Calendar invite actions for mail app. Allows responding to calendar invites directly from email. """ import asyncio import logging import re from typing import Optional, Tuple logger = logging.getLogger(__name__) def detect_calendar_invite(message_content: str, headers: dict) -> Optional[str]: """Detect if a message is a calendar invite and extract event ID if possible. Calendar invites from Microsoft/Outlook typically have: - Content-Type: text/calendar or multipart with text/calendar part - Meeting ID patterns in the content - Teams/Outlook meeting links Args: message_content: The message body content headers: Message headers Returns: Event identifier hint if detected, None otherwise """ # Check for calendar-related content patterns calendar_patterns = [ r"Microsoft Teams meeting", r"Join the meeting", r"Meeting ID:", r"teams\.microsoft\.com/l/meetup-join", r"Accept\s+Tentative\s+Decline", r"VEVENT", r"BEGIN:VCALENDAR", ] content_lower = message_content.lower() if message_content else "" for pattern in calendar_patterns: if re.search(pattern, message_content or "", re.IGNORECASE): return "calendar_invite_detected" return None async def find_event_by_subject( subject: str, organizer_email: Optional[str] = None ) -> Optional[dict]: """Find a calendar event by subject and optionally organizer. Args: subject: Event subject to search for organizer_email: Optional organizer email to filter by Returns: Event dict if found, None otherwise """ try: from src.services.microsoft_graph.auth import get_access_token from src.services.microsoft_graph.client import fetch_with_aiohttp from datetime import datetime, timedelta scopes = ["https://graph.microsoft.com/Calendars.Read"] _, headers = get_access_token(scopes) # Search for events in the next 60 days with matching subject start_date = datetime.now() end_date = start_date + timedelta(days=60) start_str = start_date.strftime("%Y-%m-%dT00:00:00Z") end_str = end_date.strftime("%Y-%m-%dT23:59:59Z") # URL encode the subject for the filter subject_escaped = subject.replace("'", "''") url = ( f"https://graph.microsoft.com/v1.0/me/calendarView?" f"startDateTime={start_str}&endDateTime={end_str}&" f"$filter=contains(subject,'{subject_escaped}')&" f"$select=id,subject,organizer,start,end,responseStatus&" f"$top=10" ) response = await fetch_with_aiohttp(url, headers) if not response: return None events = response.get("value", []) if events: # If organizer email provided, try to match if organizer_email: for event in events: org_email = ( event.get("organizer", {}) .get("emailAddress", {}) .get("address", "") ) if organizer_email.lower() in org_email.lower(): return event # Return first match return events[0] return None except Exception as e: logger.error(f"Error finding event by subject: {e}") return None async def respond_to_calendar_invite(event_id: str, response: str) -> Tuple[bool, str]: """Respond to a calendar invite. Args: event_id: Microsoft Graph event ID response: Response type - 'accept', 'tentativelyAccept', or 'decline' Returns: Tuple of (success, message) """ try: from src.services.microsoft_graph.auth import get_access_token from src.services.microsoft_graph.calendar import respond_to_invite scopes = ["https://graph.microsoft.com/Calendars.ReadWrite"] _, headers = get_access_token(scopes) success = await respond_to_invite(headers, event_id, response) if success: response_text = { "accept": "accepted", "tentativelyAccept": "tentatively accepted", "decline": "declined", }.get(response, response) return True, f"Successfully {response_text} the meeting" else: return False, "Failed to respond to the meeting invite" except Exception as e: logger.error(f"Error responding to invite: {e}") return False, f"Error: {str(e)}" def action_accept_invite(app): """Accept the current calendar invite.""" _respond_to_current_invite(app, "accept") def action_decline_invite(app): """Decline the current calendar invite.""" _respond_to_current_invite(app, "decline") def action_tentative_invite(app): """Tentatively accept the current calendar invite.""" _respond_to_current_invite(app, "tentativelyAccept") def _respond_to_current_invite(app, response: str): """Helper to respond to the current message's calendar invite.""" current_message_id = app.current_message_id if not current_message_id: app.notify("No message selected", severity="warning") return # Get message metadata metadata = app.message_store.get_metadata(current_message_id) if not metadata: app.notify("Could not load message metadata", severity="error") return subject = metadata.get("subject", "") from_addr = metadata.get("from", {}).get("addr", "") if not subject: app.notify( "No subject found - cannot match to calendar event", severity="warning" ) return # Run the async response in a worker app.run_worker( _async_respond_to_invite(app, subject, from_addr, response), exclusive=True, name="respond_invite", ) async def _async_respond_to_invite( app, subject: str, organizer_email: str, response: str ): """Async worker to find and respond to calendar invite.""" # First, find the event app.notify(f"Searching for calendar event: {subject[:40]}...") event = await find_event_by_subject(subject, organizer_email) if not event: app.notify( f"Could not find calendar event matching: {subject[:40]}", severity="warning", ) return event_id = event.get("id") if not event_id: app.notify( "Could not get event ID from calendar", severity="error", ) return current_response = event.get("responseStatus", {}).get("response", "") # Check if already responded if current_response == "accepted" and response == "accept": app.notify("Already accepted this invite", severity="information") return elif current_response == "declined" and response == "decline": app.notify("Already declined this invite", severity="information") return # Respond to the invite success, message = await respond_to_calendar_invite(event_id, response) severity = "information" if success else "error" app.notify(message, severity=severity)