Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| from datetime import datetime, timedelta | |
| import re | |
| import os | |
| import json | |
| from scheduler import get_calendar_service, find_meetings_with_person, remove_event | |
| st.set_page_config(page_title="Autonomous Calendar Assistant", page_icon="📅") | |
| def login_page(): | |
| st.title("Login to Google Calendar") | |
| st.info("This app needs access to your Google Calendar to function properly.") | |
| if st.button("Login with Google Calendar"): | |
| with st.spinner("Authenticating with Google..."): | |
| try: | |
| # Perform authentication - this will redirect if needed | |
| service = get_calendar_service() | |
| st.session_state['authenticated'] = True | |
| st.success("Authentication successful!") | |
| st.rerun() # Rerun the app with authenticated state | |
| except Exception as e: | |
| st.error(f"Authentication failed: {str(e)}") | |
| def agent_page(): | |
| st.title("Autonomous Calendar Agent") | |
| try: | |
| # Get the service | |
| service = get_calendar_service() | |
| # User Interface | |
| st.write("### Your AI Calendar Assistant") | |
| st.write("I understand natural language. Tell me what you need help with for your calendar.") | |
| # Command input - larger text area for natural language | |
| command = st.text_area( | |
| "What would you like to do with your calendar?", | |
| height=120, | |
| placeholder="Examples:\n- Schedule a meeting with Alex tomorrow at 2pm\n- Find all meetings with Morgan this month\n- Show my calendar for next Monday\n- When am I free tomorrow?\n- Check my availability for Friday" | |
| ) | |
| # Process the command | |
| if st.button("Process", key="process_cmd"): | |
| if not command: | |
| st.warning("Please enter a command first.") | |
| return | |
| with st.spinner("Processing your request..."): | |
| # Convert command to lowercase for easier pattern matching | |
| command_lower = command.lower() | |
| # FINDING MEETINGS WITH PERSON | |
| if re.search(r'find.*(?:meeting|appointment).*with\s+(\w+)', command_lower) or \ | |
| re.search(r'show.*(?:meeting|appointment).*with\s+(\w+)', command_lower): | |
| # Extract person name | |
| person_match = re.search(r'(?:with|and)\s+(\w+(?:\s+\w+)?)', command_lower) | |
| if person_match: | |
| person = person_match.group(1).strip() | |
| # Extract time constraints if any | |
| time_min = datetime.now().isoformat() + 'Z' | |
| time_max = (datetime.now() + timedelta(days=30)).isoformat() + 'Z' | |
| if 'today' in command_lower: | |
| today = datetime.now() | |
| time_min = datetime.combine(today.date(), datetime.min.time()).isoformat() + 'Z' | |
| time_max = datetime.combine(today.date(), datetime.max.time()).isoformat() + 'Z' | |
| elif 'tomorrow' in command_lower: | |
| tomorrow = datetime.now() + timedelta(days=1) | |
| time_min = datetime.combine(tomorrow.date(), datetime.min.time()).isoformat() + 'Z' | |
| time_max = datetime.combine(tomorrow.date(), datetime.max.time()).isoformat() + 'Z' | |
| elif 'this week' in command_lower: | |
| today = datetime.now() | |
| start_of_week = today - timedelta(days=today.weekday()) | |
| end_of_week = start_of_week + timedelta(days=6) | |
| time_min = datetime.combine(start_of_week.date(), datetime.min.time()).isoformat() + 'Z' | |
| time_max = datetime.combine(end_of_week.date(), datetime.max.time()).isoformat() + 'Z' | |
| elif 'this month' in command_lower: | |
| today = datetime.now() | |
| last_day = today.replace(day=28) + timedelta(days=4) # Move to next month | |
| last_day = last_day.replace(day=1) - timedelta(days=1) # Last day of current month | |
| time_min = datetime.combine(today.replace(day=1).date(), datetime.min.time()).isoformat() + 'Z' | |
| time_max = datetime.combine(last_day.date(), datetime.max.time()).isoformat() + 'Z' | |
| # Find meetings with this person | |
| meetings = find_meetings_with_person(service, "primary", person, time_min, time_max) | |
| if meetings: | |
| st.success(f"I found {len(meetings)} meetings with {person}") | |
| for i, meeting in enumerate(meetings): | |
| start_time = meeting['start'] | |
| if 'T' in start_time: | |
| start_dt = datetime.fromisoformat(start_time.replace('Z', '+00:00')) | |
| formatted_time = start_dt.strftime("%A, %b %d, %Y at %I:%M %p") | |
| else: | |
| formatted_time = start_time | |
| with st.expander(f"{i+1}. {meeting['summary']} - {formatted_time}"): | |
| st.write(f"**Time:** {formatted_time}") | |
| st.markdown(f"[View in Calendar]({meeting['htmlLink']})") | |
| # Cancel button | |
| if st.button(f"Cancel this meeting", key=f"cancel_{i}"): | |
| result = remove_event(service, "primary", meeting['id']) | |
| st.success("Meeting cancelled!") | |
| st.rerun() | |
| else: | |
| st.info(f"I couldn't find any meetings with {person} in the specified time range.") | |
| # VIEWING CALENDAR FOR A SPECIFIC DAY | |
| elif re.search(r'(?:show|view|display).*(?:calendar|schedule|meeting).*(?:for|on)\s+(\w+)', command_lower) or \ | |
| re.search(r'what.*(?:calendar|schedule|meeting).*(?:for|on)\s+(\w+)', command_lower): | |
| # Try to extract date | |
| date_value = None | |
| if 'today' in command_lower: | |
| date_value = datetime.now().date() | |
| elif 'tomorrow' in command_lower: | |
| date_value = (datetime.now() + timedelta(days=1)).date() | |
| else: | |
| # Extract day name if present | |
| day_mapping = { | |
| 'monday': 0, 'tuesday': 1, 'wednesday': 2, | |
| 'thursday': 3, 'friday': 4, 'saturday': 5, 'sunday': 6 | |
| } | |
| for day_name, day_number in day_mapping.items(): | |
| if day_name in command_lower: | |
| today = datetime.now() | |
| days_ahead = (day_number - today.weekday()) % 7 | |
| if days_ahead == 0: # Today | |
| if 'next' in command_lower: | |
| days_ahead = 7 # Next week | |
| date_value = (today + timedelta(days=days_ahead)).date() | |
| break | |
| if date_value: | |
| # Get the day's events | |
| start_dt = datetime.combine(date_value, datetime.min.time()) | |
| end_dt = datetime.combine(date_value, datetime.max.time()) | |
| time_min = start_dt.isoformat() + 'Z' | |
| time_max = end_dt.isoformat() + 'Z' | |
| try: | |
| events_result = service.events().list( | |
| calendarId="primary", | |
| timeMin=time_min, | |
| timeMax=time_max, | |
| singleEvents=True, | |
| orderBy='startTime' | |
| ).execute() | |
| events = events_result.get('items', []) | |
| if events: | |
| st.success(f"Here's your schedule for {date_value.strftime('%A, %B %d')}:") | |
| for i, event in enumerate(events): | |
| summary = event.get('summary', 'No Title') | |
| start = event['start'].get('dateTime', event['start'].get('date')) | |
| if 'T' in start: | |
| start_dt = datetime.fromisoformat(start.replace('Z', '+00:00')) | |
| time_str = start_dt.strftime("%I:%M %p") | |
| else: | |
| time_str = "All day" | |
| with st.expander(f"{summary} - {time_str}"): | |
| st.write(f"**Event:** {summary}") | |
| st.write(f"**Time:** {time_str}") | |
| st.markdown(f"[View in Calendar]({event.get('htmlLink', '#')})") | |
| # Cancel button | |
| if st.button("Cancel this event", key=f"cancel_event_{i}"): | |
| result = remove_event(service, "primary", event['id']) | |
| st.success("Event cancelled!") | |
| st.rerun() | |
| else: | |
| st.info(f"You have no events scheduled for {date_value.strftime('%A, %B %d')}.") | |
| except Exception as e: | |
| st.error(f"Error fetching events: {str(e)}") | |
| else: | |
| st.error(f"I couldn't understand the date in your request. Please try again with a clearer date specification.") | |
| # CHECKING AVAILABILITY | |
| elif re.search(r'(?:when|what time).*(?:free|available)', command_lower) or \ | |
| re.search(r'check.*availability', command_lower) or \ | |
| re.search(r'am i free', command_lower): | |
| # Try to extract date | |
| date_value = None | |
| if 'today' in command_lower: | |
| date_value = datetime.now().date() | |
| elif 'tomorrow' in command_lower: | |
| date_value = (datetime.now() + timedelta(days=1)).date() | |
| else: | |
| # Extract day name if present | |
| day_mapping = { | |
| 'monday': 0, 'tuesday': 1, 'wednesday': 2, | |
| 'thursday': 3, 'friday': 4, 'saturday': 5, 'sunday': 6 | |
| } | |
| for day_name, day_number in day_mapping.items(): | |
| if day_name in command_lower: | |
| today = datetime.now() | |
| days_ahead = (day_number - today.weekday()) % 7 | |
| if days_ahead == 0: # Today | |
| if 'next' in command_lower: | |
| days_ahead = 7 # Next week | |
| date_value = (today + timedelta(days=days_ahead)).date() | |
| break | |
| # Default to tomorrow if no date found | |
| if not date_value: | |
| date_value = datetime.now().date() + timedelta(days=1) | |
| # Set default business hours | |
| start_hour = 9 | |
| end_hour = 17 | |
| # Check availability | |
| start_dt = datetime.combine(date_value, datetime.min.time().replace(hour=start_hour)) | |
| end_dt = datetime.combine(date_value, datetime.min.time().replace(hour=end_hour)) | |
| time_min = start_dt.isoformat() + 'Z' | |
| time_max = end_dt.isoformat() + 'Z' | |
| try: | |
| # Query the freebusy API | |
| freebusy_query = { | |
| "timeMin": time_min, | |
| "timeMax": time_max, | |
| "items": [{"id": "primary"}] | |
| } | |
| freebusy_result = service.freebusy().query(body=freebusy_query).execute() | |
| busy_slots = freebusy_result['calendars']['primary']['busy'] | |
| st.success(f"Here's your availability for {date_value.strftime('%A, %B %d')}:") | |
| if busy_slots: | |
| # Create a list of hours and availability | |
| hour_slots = [] | |
| for hour in range(start_hour, end_hour + 1): | |
| slot_start = datetime.combine(date_value, datetime.min.time().replace(hour=hour)) | |
| slot_end = slot_start + timedelta(hours=1) | |
| # Check if this slot overlaps with any busy time | |
| is_busy = False | |
| for busy in busy_slots: | |
| busy_start = datetime.fromisoformat(busy['start'].replace('Z', '+00:00')) | |
| busy_end = datetime.fromisoformat(busy['end'].replace('Z', '+00:00')) | |
| if (slot_start < busy_end and slot_end > busy_start): | |
| is_busy = True | |
| break | |
| hour_slots.append({ | |
| "hour": hour, | |
| "time_str": slot_start.strftime("%I:%M %p"), | |
| "is_busy": is_busy | |
| }) | |
| # Display as a table | |
| col1, col2 = st.columns([1, 2]) | |
| with col1: | |
| st.markdown("**Time**") | |
| for slot in hour_slots: | |
| st.write(slot["time_str"]) | |
| with col2: | |
| st.markdown("**Status**") | |
| for slot in hour_slots: | |
| if slot["is_busy"]: | |
| st.markdown("🔴 Busy") | |
| else: | |
| st.markdown("🟢 Available") | |
| else: | |
| st.info(f"You're completely free on {date_value.strftime('%A, %B %d')} from {start_hour}:00 to {end_hour}:00!") | |
| except Exception as e: | |
| st.error(f"Error checking availability: {str(e)}") | |
| # DEFAULT: Let the agent handle all other commands | |
| else: | |
| result = agent.process_command(command) | |
| if result["status"] == "created": | |
| st.success(result["message"]) | |
| st.markdown(f"[View in Calendar]({result['event_link']})") | |
| # Show alternatives if available | |
| if "alternatives" in result and result["alternatives"]: | |
| st.markdown("**Alternative times that would also work:**") | |
| for i, alt in enumerate(result["alternatives"]): | |
| if isinstance(alt, dict) and 'start' in alt: | |
| start_dt = datetime.fromisoformat(alt['start'].replace('Z', '+00:00')) | |
| st.write(f"{i+1}. {start_dt.strftime('%A, %b %d at %I:%M %p')}") | |
| elif result["status"] == "rescheduled": | |
| st.warning(result["message"]) | |
| st.markdown(f"[View in Calendar]({result['event_link']})") | |
| elif result["status"] == "suggestions": | |
| st.info(result["message"]) | |
| for i, slot in enumerate(result["slots"]): | |
| st.write(f"{i+1}. {slot}") | |
| else: | |
| st.error(result["message"]) | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}") | |
| st.info("Try restarting the application if the issue persists.") | |
| # Main app logic | |
| if 'authenticated' not in st.session_state: | |
| st.session_state['authenticated'] = False | |
| # Add logout button in sidebar | |
| if st.session_state['authenticated']: | |
| if st.sidebar.button("Logout"): | |
| st.session_state['authenticated'] = False | |
| if 'user_token' in st.session_state: | |
| del st.session_state['user_token'] | |
| st.rerun() | |
| # Show appropriate page | |
| if st.session_state['authenticated']: | |
| agent_page() | |
| else: | |
| login_page() | |