from datetime import datetime, timedelta from collections import defaultdict, Counter from llama_index.llms.openai import OpenAI from composio_llamaindex import ComposioToolSet, App, Action import gradio as gr import os import json from dotenv import load_dotenv # Load environment variables load_dotenv() llm = OpenAI(model='gpt-4', api_key=os.getenv('OPENAI_API_KEY')) class CalendarService: def __init__(self): self.toolset = ComposioToolSet(api_key=os.getenv('COMPOSIO_API_KEY')) self.connection_request = None def analyze_calendar_events(self, response_data): """ Analyze calendar events and return statistics about meetings. """ current_year = datetime.now().year meetings = [] participants = [] meeting_times = [] total_duration = timedelta() monthly_meetings = defaultdict(int) daily_meetings = defaultdict(int) events = response_data.get('data', {}).get('event_data', {}).get('event_data', []) for event in events: start_data = event.get('start', {}) end_data = event.get('end', {}) try: start = datetime.fromisoformat(start_data.get('dateTime').replace('Z', '+00:00')) end = datetime.fromisoformat(end_data.get('dateTime').replace('Z', '+00:00')) if start.year == current_year: duration = end - start total_duration += duration monthly_meetings[start.strftime('%B')] += 1 daily_meetings[start.strftime('%A')] += 1 meeting_times.append(start.strftime('%H:%M')) if 'attendees' in event: for attendee in event['attendees']: if attendee.get('responseStatus') != 'declined': participants.append(attendee.get('email')) organizer_email = event.get('organizer', {}).get('email') if organizer_email: participants.append(organizer_email) meetings.append({ 'start': start, 'duration': duration, 'summary': event.get('summary', 'No Title') }) except (ValueError, TypeError, AttributeError) as e: print(f"Error processing event: {e}") continue total_meetings = len(meetings) stats = { "total_meetings_this_year": total_meetings } if total_meetings > 0: stats.update({ "total_time_spent": str(total_duration), "busiest_month": max(monthly_meetings.items(), key=lambda x: x[1])[0] if monthly_meetings else "N/A", "busiest_day": max(daily_meetings.items(), key=lambda x: x[1])[0] if daily_meetings else "N/A", "most_frequent_participant": Counter(participants).most_common(1)[0][0] if participants else "N/A", "average_meeting_duration": str(total_duration / total_meetings), "most_common_meeting_time": Counter(meeting_times).most_common(1)[0][0] if meeting_times else "N/A", "monthly_breakdown": dict(monthly_meetings), "daily_breakdown": dict(daily_meetings) }) else: stats.update({ "total_time_spent": "0:00:00", "busiest_month": "N/A", "busiest_day": "N/A", "most_frequent_participant": "N/A", "average_meeting_duration": "0:00:00", "most_common_meeting_time": "N/A", "monthly_breakdown": {}, "daily_breakdown": {} }) return stats def initiate_connection(self, entity_id: str, redirect_url: str = "https://calendar-wrapped-eight.vercel.app/") -> dict: try: self.connection_request = self.toolset.initiate_connection( entity_id=entity_id, app=App.GOOGLECALENDAR, ) return { 'success': True, 'data': { 'redirect_url': self.connection_request.redirectUrl, 'message': "Please authenticate using the provided link." } } except Exception as e: return { 'success': False, 'error': str(e) } def check_connection_status(self, entity_id: str) -> dict: try: # if not self.connection_request: # return { # 'success': False, # 'error': 'No active connection request found' # } entity_id = self.toolset.get_entity(id=entity_id) connection = entity_id.get_connection(app=App.GOOGLECALENDAR) status = connection.status #status = self.connection_request.connectionStatus return { 'success': True, 'data': { 'status': status, 'message': f"Connection status: {status}" } } except Exception as e: return { 'success': False, 'error': str(e) } def generate_wrapped(self, entity_id: str) -> dict: try: current_year = datetime.now().year request_params = { "calendar_id": "primary", "timeMin": f"{current_year},1,1,0,0,0", "timeMax": f"{current_year},12,31,23,59,59", "single_events": True, "max_results": 2500, "order_by": "startTime" } events_response = self.toolset.execute_action( action=Action.GOOGLECALENDAR_FIND_EVENT, params=request_params, entity_id=entity_id ) if events_response["successfull"]: stats = self.analyze_calendar_events(events_response) # Generate prompts for LLM analysis billionaire_prompt = f"""Based on these calendar stats, which tech billionaire's schedule does this most resemble and why? Stats: - {stats['total_meetings_this_year']} total meetings - {stats['total_time_spent']} total time in meetings - Most active on {stats['busiest_day']}s - Busiest month is {stats['busiest_month']} - Average meeting duration: {stats['average_meeting_duration']} Suggest a different billionaire each time, dont say elon. Return as JSON with format: {{"name": "billionaire name", "reason": "explanation"}} """ stats_prompt = f"""Analyze these calendar stats and write a brief, insightful one-sentence comment for each metric: - Total meetings: {stats['total_meetings_this_year']} - Total time in meetings: {stats['total_time_spent']} - Busiest month: {stats['busiest_month']} - Busiest day: {stats['busiest_day']} - Average meeting duration: {stats['average_meeting_duration']} - Most common meeting time: {stats['most_common_meeting_time']} - Most frequent participant: {stats['most_frequent_participant']} Return as JSON with format: {{"total_meetings_comment": "", "time_spent_comment": "", "busiest_times_comment": "", "collaborator_comment": "", "habits_comment": ""}} """ try: billionaire_response = json.loads(llm.complete(billionaire_prompt).text) stats_comments = json.loads(llm.complete(stats_prompt).text) stats["schedule_analysis"] = billionaire_response stats["metric_insights"] = stats_comments except Exception as e: print(f"Error processing LLM responses: {e}") stats["schedule_analysis"] = {"name": "Unknown", "reason": "Analysis unavailable"} stats["metric_insights"] = { "total_meetings_comment": "", "time_spent_comment": "", "busiest_times_comment": "", "collaborator_comment": "", "habits_comment": "" } return { 'success': True, 'data': stats } else: return { 'success': False, 'error': events_response.get("error", "Failed to fetch calendar events") } except Exception as e: return { 'success': False, 'error': str(e) } def create_gradio_interface(): service = CalendarService() def handle_connection(entity_id: str, redirect_url: str = None) -> str: return json.dumps(service.initiate_connection(entity_id, redirect_url)) def check_status(entity_id: str) -> str: return json.dumps(service.check_connection_status(entity_id)) def generate_wrapped(entity_id: str) -> str: return json.dumps(service.generate_wrapped(entity_id)) # Create Gradio interface with gr.Blocks(title="Calendar Wrapped API") as interface: gr.Markdown("# Calendar Wrapped API") with gr.Tab("Connect"): entity_input = gr.Textbox(label="Entity ID") redirect_input = gr.Textbox( label="Redirect URL", placeholder="https://yourwebsite.com/connection/success", value="https://calendar-wrapped-eight.vercel.app/" ) connect_btn = gr.Button("Initialize Connection") connect_output = gr.JSON() connect_btn.click( fn=handle_connection, inputs=[entity_input, redirect_input], outputs=connect_output ) with gr.Tab("Check Status"): status_input = gr.Textbox(label="Entity ID") status_btn = gr.Button("Check Status") status_output = gr.JSON() status_btn.click( fn=check_status, inputs=status_input, outputs=status_output ) with gr.Tab("Generate Wrapped"): wrapped_input = gr.Textbox(label="Entity ID") wrapped_btn = gr.Button("Generate Wrapped") wrapped_output = gr.JSON() wrapped_btn.click( fn=generate_wrapped, inputs=wrapped_input, outputs=wrapped_output ) return interface if __name__ == "__main__": interface = create_gradio_interface() interface.launch(server_name="0.0.0.0", server_port=7860)