"""Create an event in Google Calendar.""" import re from datetime import datetime from typing import Any, Dict, List, Optional, Type, Union from uuid import uuid4 from langchain_core.callbacks import CallbackManagerForToolRun from pydantic import BaseModel, Field from langchain_google_community.calendar.base import CalendarBaseTool from langchain_google_community.calendar.utils import is_all_day_event class CreateEventSchema(BaseModel): """Input for CalendarCreateEvent.""" summary: str = Field(..., description="The title of the event.") start_datetime: str = Field( ..., description=( "The start datetime for the event in 'YYYY-MM-DD HH:MM:SS' format." "If the event is an all-day event, set the time to 'YYYY-MM-DD' format." "If you do not know the current datetime, use the tool to get it." ), ) end_datetime: str = Field( ..., description=( "The end datetime for the event in 'YYYY-MM-DD HH:MM:SS' format. " "If the event is an all-day event, set the time to 'YYYY-MM-DD' format." ), ) timezone: str = Field(..., description="The timezone of the event.") calendar_id: str = Field( default="primary", description="The calendar ID to create the event in." ) recurrence: Optional[Dict[str, Any]] = Field( default=None, description=( "The recurrence of the event. " "Format: {'FREQ': <'DAILY' or 'WEEKLY'>, 'INTERVAL': , " "'COUNT': , 'UNTIL': <'YYYYMMDD' or None>, " "'BYDAY': <'MO', 'TU', 'WE', 'TH', 'FR', 'SA', 'SU' or None>}. " "Use either COUNT or UNTIL, but not both; set the other to None." ), ) location: Optional[str] = Field( default=None, description="The location of the event." ) description: Optional[str] = Field( default=None, description="The description of the event." ) attendees: Optional[List[str]] = Field( default=None, description="A list of attendees' email addresses for the event." ) reminders: Union[None, bool, List[Dict[str, Any]]] = Field( default=None, description=( "Reminders for the event. " "Set to True for default reminders, or provide a list like " "[{'method': 'email', 'minutes': }, ...]. " "Valid methods are 'email' and 'popup'." ), ) conference_data: Optional[bool] = Field( default=None, description="Whether to include conference data." ) color_id: Optional[str] = Field( default=None, description=( "The color ID of the event. None for default. " "'1': Lavender, '2': Sage, '3': Grape, '4': Flamingo, '5': Banana, " "'6': Tangerine, '7': Peacock, '8': Graphite, '9': Blueberry, " "'10': Basil, '11': Tomato." ), ) transparency: Optional[str] = Field( default=None, description=( "User availability for the event." "transparent for available and opaque for busy." ), ) class CalendarCreateEvent(CalendarBaseTool): # type: ignore[override, override] """Tool that creates an event in Google Calendar.""" name: str = "create_calendar_event" description: str = ( "Use this tool to create an event. " "The input must include the summary, start, and end datetime for the event." ) args_schema: Type[CreateEventSchema] = CreateEventSchema def _prepare_event( self, summary: str, start_datetime: str, end_datetime: str, timezone: str, recurrence: Optional[Dict[str, Any]] = None, location: Optional[str] = None, description: Optional[str] = None, attendees: Optional[List[str]] = None, reminders: Union[None, bool, List[Dict[str, Any]]] = None, conference_data: Optional[bool] = None, color_id: Optional[str] = None, transparency: Optional[str] = None, ) -> Dict[str, Any]: """Prepare the event body.""" try: if is_all_day_event(start_datetime, end_datetime): start = {"date": start_datetime} end = {"date": end_datetime} else: datetime_format = "%Y-%m-%d %H:%M:%S" start_dt = datetime.strptime(start_datetime, datetime_format) end_dt = datetime.strptime(end_datetime, datetime_format) start = { "dateTime": start_dt.astimezone().isoformat(), "timeZone": timezone, } end = { "dateTime": end_dt.astimezone().isoformat(), "timeZone": timezone, } except ValueError as error: raise ValueError("The datetime format is incorrect.") from error recurrence_data = None if recurrence: if isinstance(recurrence, dict): recurrence_items = [ f"{k}={v}" for k, v in recurrence.items() if v is not None ] recurrence_data = "RRULE:" + ";".join(recurrence_items) attendees_emails: List[Dict[str, str]] = [] if attendees: email_pattern = r"^[^@]+@[^@]+\.[^@]+$" for email in attendees: if not re.match(email_pattern, email): raise ValueError(f"Invalid email address: {email}") attendees_emails.append({"email": email}) reminders_info: Dict[str, Union[bool, List[Dict[str, Any]]]] = {} if reminders is True: reminders_info.update({"useDefault": True}) elif isinstance(reminders, list): for reminder in reminders: if "method" not in reminder or "minutes" not in reminder: raise ValueError( "Each reminder must have 'method' and 'minutes' keys." ) if reminder["method"] not in ["email", "popup"]: raise ValueError("The reminder method must be 'email' or 'popup") reminders_info.update({"useDefault": False, "overrides": reminders}) else: reminders_info.update({"useDefault": False}) conference_data_info = None if conference_data: conference_data_info = { "createRequest": { "requestId": str(uuid4()), "conferenceSolutionKey": {"type": "hangoutsMeet"}, } } event_body: Dict[str, Any] = {"summary": summary, "start": start, "end": end} if location: event_body["location"] = location if description: event_body["description"] = description if recurrence_data: event_body["recurrence"] = [recurrence_data] if len(attendees_emails) > 0: event_body["attendees"] = attendees_emails if len(reminders_info) > 0: event_body["reminders"] = reminders_info if conference_data_info: event_body["conferenceData"] = conference_data_info if color_id: event_body["colorId"] = color_id if transparency: event_body["transparency"] = transparency return event_body def _run( self, summary: str, start_datetime: str, end_datetime: str, timezone: str, calendar_id: str = "primary", recurrence: Optional[Dict[str, Any]] = None, location: Optional[str] = None, description: Optional[str] = None, attendees: Optional[List[str]] = None, reminders: Union[None, bool, List[Dict[str, Any]]] = None, conference_data: Optional[bool] = None, color_id: Optional[str] = None, transparency: Optional[str] = None, run_manager: Optional[CallbackManagerForToolRun] = None, ) -> str: """Run the tool to create an event in Google Calendar.""" try: body = self._prepare_event( summary=summary, start_datetime=start_datetime, end_datetime=end_datetime, timezone=timezone, recurrence=recurrence, location=location, description=description, attendees=attendees, reminders=reminders, conference_data=conference_data, color_id=color_id, transparency=transparency, ) conference_version = 1 if conference_data else 0 event = ( self.api_resource.events() .insert( calendarId=calendar_id, body=body, conferenceDataVersion=conference_version, ) .execute() ) return f"Event created: {event.get('htmlLink')}" except Exception as error: raise Exception(f"An error occurred: {error}") from error