diff --git a/src/mail/actions/calendar_invite.py b/src/mail/actions/calendar_invite.py new file mode 100644 index 0000000..bf11324 --- /dev/null +++ b/src/mail/actions/calendar_invite.py @@ -0,0 +1,237 @@ +"""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.call_from_thread(app.notify, f"Searching for calendar event: {subject[:40]}...") + + event = await find_event_by_subject(subject, organizer_email) + + if not event: + app.call_from_thread( + app.notify, + f"Could not find calendar event matching: {subject[:40]}", + severity="warning", + ) + return + + event_id = event.get("id") + if not event_id: + app.call_from_thread( + 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.call_from_thread( + app.notify, "Already accepted this invite", severity="information" + ) + return + elif current_response == "declined" and response == "decline": + app.call_from_thread( + 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.call_from_thread(app.notify, message, severity=severity) diff --git a/src/mail/app.py b/src/mail/app.py index 017cf64..cadee47 100644 --- a/src/mail/app.py +++ b/src/mail/app.py @@ -8,6 +8,11 @@ from .screens.SearchPanel import SearchPanel from .actions.task import action_create_task from .actions.open import action_open from .actions.delete import delete_current +from .actions.calendar_invite import ( + action_accept_invite, + action_decline_invite, + action_tentative_invite, +) from src.services.taskwarrior import client as taskwarrior_client from src.services.himalaya import client as himalaya_client from src.utils.shared_config import get_theme_name @@ -134,6 +139,9 @@ class EmailViewerApp(App): Binding("escape", "clear_selection", "Clear selection"), Binding("/", "search", "Search"), Binding("u", "toggle_read", "Toggle read/unread"), + Binding("A", "accept_invite", "Accept invite"), + Binding("D", "decline_invite", "Decline invite"), + Binding("T", "tentative_invite", "Tentative"), ] ) @@ -854,6 +862,18 @@ class EmailViewerApp(App): def action_create_task(self) -> None: action_create_task(self) + def action_accept_invite(self) -> None: + """Accept the calendar invite from the current email.""" + action_accept_invite(self) + + def action_decline_invite(self) -> None: + """Decline the calendar invite from the current email.""" + action_decline_invite(self) + + def action_tentative_invite(self) -> None: + """Tentatively accept the calendar invite from the current email.""" + action_tentative_invite(self) + def action_open_links(self) -> None: """Open the link panel showing links from the current message.""" content_container = self.query_one(ContentContainer)