Update app.py

app.py

@@ -1,117 +1,274 @@
+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
-from composio import ComposioToolSet, App, Action
-from datetime import datetime
+import os
 import json
+from dotenv import load_dotenv
 
-# Initialize the Composio toolset
-toolset = ComposioToolSet()
+# Load environment variables
+load_dotenv()
 
-# In-memory store for user connections
-user_connections = {}
+llm = OpenAI(model='gpt-4', api_key=os.getenv('OPENAI_API_KEY'))
 
-# Endpoint 1: Connect the user
-def connect_user(entity_id):
-    connection_request = toolset.initiate_connection(
-        entity_id=entity_id,
-        app=App.GMAIL,
-    )
+class CalendarService:
+    def __init__(self):
+        self.toolset = ComposioToolSet(api_key=os.getenv('COMPOSIO_API_KEY'))
+        self.connection_request = None
 
-    user_connections[entity_id] = {
-        "connection_status": connection_request.connectionStatus
-    }
-
-    if connection_request.connectionStatus == "INITIATED":
-        user_connections[entity_id]["redirect_url"] = connection_request.redirectUrl
-        return f"Connection initiated. Redirect user to: {connection_request.redirectUrl}"
-
-    elif connection_request.connectionStatus == "ACTIVE":
-        return "Connection is active."
-
-    else:
-        return "Connection failed. Please try again."
-
-# Endpoint 2: Check connection status
-def check_connection_status(entity_id):
-    if entity_id not in user_connections:
-        return "No connection found for this user."
-
-    return user_connections[entity_id]["connection_status"]
-
-# Endpoint 3: Generate wrapped analysis
-def generate_wrapped(entity_id):
-    if entity_id not in user_connections or user_connections[entity_id]["connection_status"] != "ACTIVE":
-        return "User is not connected or connection is not active."
-
-    current_year = datetime.now().year
-    time_min = f"{current_year}-01-01T00:00:00Z"
-    time_max = f"{current_year}-12-31T23:59:59Z"
-
-    request_params = {
-        "calendar_id": "primary",
-        "timeMin": time_min,
-        "timeMax": time_max,
-        "single_events": True,
-        "max_results": 2500,
-        "order_by": "startTime",
-    }
-
-    try:
-        events_response = toolset.execute_action(
-            action=Action.GOOGLECALENDAR_FIND_EVENT,
-            params=request_params,
-            entity_id=entity_id
-        )
-
-        if events_response["successfull"]:
-            stats = analyze_calendar_events(events_response)
-            return json.dumps(stats, indent=2)
+    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:
-            return "Failed to fetch calendar events."
-
-    except Exception as e:
-        return f"Error: {e}"
-
-# Helper function to analyze events (mocked for simplicity)
-def analyze_calendar_events(events_response):
-    # Mocking the analysis part for simplicity
-    return {
-        "total_meetings_this_year": 120,
-        "total_time_spent": "150 hours",
-        "busiest_day": "Tuesday",
-        "busiest_month": "September",
-        "average_meeting_duration": "1 hour",
-        "most_common_meeting_time": "10:00 AM",
-        "most_frequent_participant": "John Doe",
-    }
-
-# Gradio interface
-def connect(entity_id):
-    return connect_user(entity_id)
-
-def status(entity_id):
-    return check_connection_status(entity_id)
-
-def wrapped(entity_id):
-    return generate_wrapped(entity_id)
-
-with gr.Blocks() as demo:
-    gr.Markdown("# Calendar Wrapped API")
-
-    with gr.Row():
-        entity_id_input = gr.Textbox(label="Entity ID")
-
-    with gr.Row():
-        connect_btn = gr.Button("Connect User")
-        status_btn = gr.Button("Check Status")
-        wrapped_btn = gr.Button("Generate Wrapped")
-
-    with gr.Row():
-        connect_output = gr.Textbox(label="Connect Output")
-        status_output = gr.Textbox(label="Status Output")
-        wrapped_output = gr.Textbox(label="Wrapped Output")
+            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,
+                redirect_url=redirect_url
+            )
+            
+            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'
+                }
+            
+            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']}
+                
+                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)
+            }
 
-    connect_btn.click(connect, inputs=entity_id_input, outputs=connect_output)
-    status_btn.click(status, inputs=entity_id_input, outputs=status_output)
-    wrapped_btn.click(wrapped, inputs=entity_id_input, outputs=wrapped_output)
+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
 
-demo.launch()
+if __name__ == "__main__":
+    interface = create_gradio_interface()
+    interface.launch(server_name="0.0.0.0", server_port=7860)