Update app.py

app.py

@@ -1,316 +1,117 @@
-from dataclasses import dataclass
-from enum import Enum
-from typing import Optional, Dict, Any
-from composio_llamaindex import ComposioToolSet, App, Action
-from datetime import datetime, timedelta
-from collections import defaultdict, Counter
-from llama_index.llms.openai import OpenAI
 import gradio as gr
-import os
+from composio import ComposioToolSet, App, Action
+from datetime import datetime
 import json
-from dotenv import load_dotenv
 
-# Load environment variables
-load_dotenv()
+# Initialize the Composio toolset
+toolset = ComposioToolSet()
 
-llm = OpenAI(model='gpt-4o', api_key=os.getenv('OPENAI_API_KEY'))
+# In-memory store for user connections
+user_connections = {}
 
-class ConnectionStatus(Enum):
-    PENDING = "pending"
-    ACTIVE = "active"
-    FAILED = "failed"
-    NOT_FOUND = "not_found"
+# Endpoint 1: Connect the user
+def connect_user(entity_id):
+    connection_request = toolset.initiate_connection(
+        entity_id=entity_id,
+        app=App.GMAIL,
+    )
+
+    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."
 
-@dataclass
-class APIResponse:
-    success: bool
-    data: Optional[Dict[str, Any]] = None
-    error: Optional[str] = None
-    
-    def to_json(self) -> str:
-        return json.dumps({
-            "success": self.success,
-            "data": self.data,
-            "error": self.error
-        })
+    return user_connections[entity_id]["connection_status"]
 
-class CalendarService:
-    def __init__(self):
-        self.toolset = ComposioToolSet(api_key=os.getenv('COMPOSIO_API_KEY'))
-        self.connections: Dict[str, Dict[str, Any]] = {}
-        self.connectionRequest = 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)
-            })
+# 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)
         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: Optional[str] = None) -> APIResponse:
-        try:
-            if not redirect_url:
-                redirect_url = "https://calendar-wrapped-eight.vercel.app/"
-                
-            connection_request = self.toolset.initiate_connection(
-                entity_id=entity_id,
-                app=App.GOOGLECALENDAR,
-                redirect_url=redirect_url
-            )
-            
-            self.connections[entity_id] = {
-                'status': ConnectionStatus.PENDING.value,
-                'redirect_url': connection_request.redirectUrl,
-                'created_at': datetime.now().isoformat()
-            }
+            return "Failed to fetch calendar events."
 
-            self.connectionRequest = connection_request
-            
-            return APIResponse(
-                success=True,
-                data={
-                    'status': ConnectionStatus.PENDING.value,
-                    'redirect_url': connection_request.redirectUrl,
-                    'wait_time': 60,
-                    'message': "Please authenticate using the provided link."
-                }
-            )
-            
-        except Exception as e:
-            return APIResponse(
-                success=False,
-                error=f"Failed to initiate connection: {str(e)}"
-            )
-    
-    def check_status(self, entity_id: str) -> APIResponse:
-        try:
-            status = self.connectionRequest.connectionStatus
+    except Exception as e:
+        return f"Error: {e}"
 
-            if status == 'ACTIVE':
-                status='active'
-                connection = self.connections[entity_id]
-                
-                return APIResponse(
-                    success=True,
-                    data={
-                        'status': status,
-                        'message': f"Connection status: {connection['status']}"
-                    }
-                )
-                
-        except Exception as e:
-            return APIResponse(
-                success=False,
-                error=f"Failed to check status: {str(e)}"
-            )
-    
-    def generate_wrapped(self, entity_id: str) -> APIResponse:
-        try:
-            # Get current year's start and end dates
-            current_year = datetime.now().year
-            time_min = f"{current_year},1,1,0,0,0"
-            time_max = f"{current_year},12,31,23,59,59"
-            
-            request_params = {
-                "calendar_id": "primary",
-                "timeMin": time_min,
-                "timeMax": time_max,
-                "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)
-                
-                # Get tech billionaire comparison
-                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"}}
-                """
-                
-                # Get comments for each stat
-                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": ""}}
-                """
-                
-                # Make LLM calls
-                try:
-                    billionaire_response = json.loads(llm.complete(billionaire_prompt).text)
-                    stats_comments = json.loads(llm.complete(stats_prompt).text)
-                    
-                    # Add new fields to stats
-                    stats["schedule_analysis"] = billionaire_response
-                    stats["metric_insights"] = stats_comments
-                except (json.JSONDecodeError, Exception) as e:
-                    print(f"Error processing LLM responses: {e}")
-                    # Add empty defaults if LLM processing fails
-                    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 APIResponse(
-                    success=True,
-                    data=stats
-                )
-            else:
-                return APIResponse(
-                    success=False,
-                    error=events_response["error"] or "Failed to fetch calendar events"
-                )
-            
-        except Exception as e:
-            return APIResponse(
-                success=False,
-                error=f"Failed to generate wrapped: {str(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",
+    }
 
-def create_gradio_api():
-    service = CalendarService()
-    
-    def handle_connection(entity_id: str, redirect_url: Optional[str] = None) -> str:
-        response = service.initiate_connection(entity_id, redirect_url)
-        return response.to_json()
-    
-    def check_status(entity_id: str) -> str:
-        response = service.check_status(entity_id)
-        return response.to_json()
-    
-    def generate_wrapped(entity_id: str) -> str:
-        response = service.generate_wrapped(entity_id)
-        return response.to_json()
-    
-    # Create API endpoints
-    connection_api = gr.Interface(
-        fn=handle_connection,
-        inputs=[
-            gr.Textbox(label="Entity ID"),
-            gr.Textbox(label="Redirect URL", placeholder="https://yourwebsite.com/connection/success")
-        ],
-        outputs=gr.JSON(),
-        title="Initialize Calendar Connection",
-        description="Start a new calendar connection for an entity",
-        examples=[["user123", "https://example.com/callback"]]
-    )
-    
-    status_api = gr.Interface(
-        fn=check_status,
-        inputs=gr.Textbox(label="Entity ID"),
-        outputs=gr.JSON(),
-        title="Check Connection Status",
-        description="Check the status of an existing connection",
-        examples=[["user123"]]
-    )
-    
-    wrapped_api = gr.Interface(
-        fn=generate_wrapped,
-        inputs=gr.Textbox(label="Entity ID"),
-        outputs=gr.JSON(),
-        title="Generate Calendar Wrapped",
-        description="Generate a calendar wrapped summary for an entity",
-        examples=[["user123"]]
-    )
-    
-    # Combine all interfaces
-    api = gr.TabbedInterface(
-        [connection_api, status_api, wrapped_api],
-        ["Connect", "Check Status", "Generate Wrapped"],
-        title="Calendar Wrapped API",
-    )
-    
-    return api
+# 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")
+
+    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)
 
-if __name__ == "__main__":
-    api = create_gradio_api()
-    api.launch(server_name="0.0.0.0", server_port=7860)
+demo.launch()