app.py

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)