improved_cal_booking.py

import os
from datetime import datetime, timedelta
import requests
import json
from typing import Dict, List, Optional, Union
from dotenv import load_dotenv

class ImprovedCalBookingSystem:
    def __init__(self, api_key: Optional[str] = None, username: Optional[str] = None):
        """Initialize the booking system with API credentials"""
        # Load from environment if not provided
        if api_key is None:
            load_dotenv()
            api_key = os.getenv('CAL_API_KEY')
        if username is None:
            load_dotenv()
            username = os.getenv('CAL_USERNAME')
            
        if not api_key or not username:
            raise ValueError("CAL_API_KEY and CAL_USERNAME are required")
            
        self.api_key = api_key
        self.username = username
        self.base_url = "https://api.cal.com/v1"
        self.headers = {"Content-Type": "application/json"}
        self.params = {"apiKey": self.api_key}

    def get_event_types(self) -> List[Dict]:
        """Get all available event types"""
        try:
            response = requests.get(
                f"{self.base_url}/event-types",
                headers=self.headers,
                params=self.params
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"Error fetching event types: {e}")
            if hasattr(e.response, 'text'):
                print(f"Response: {e.response.text}")
            return []

    def get_available_slots(
        self, 
        event_type_id: int, 
        date: Optional[datetime] = None,
        days_ahead: int = 7
    ) -> List[str]:
        """Get available time slots for a specific event type"""
        if date is None:
            # Start from tomorrow to avoid "booking in the past" errors
            date = datetime.now() + timedelta(days=1)
            date = date.replace(hour=0, minute=0, second=0, microsecond=0)
        
        try:
            availability_params = {
                **self.params,
                "username": self.username,
                "eventTypeId": event_type_id,
                "dateFrom": date.strftime("%Y-%m-%d"),
                "dateTo": (date + timedelta(days=days_ahead)).strftime("%Y-%m-%d"),
                "duration": 30
            }
            
            response = requests.get(
                f"{self.base_url}/availability",
                headers=self.headers,
                params=availability_params
            )
            response.raise_for_status()
            availability_data = response.json()
            
            # Generate available slots based on working hours
            available_slots = []
            date_ranges = availability_data.get("dateRanges", [])
            busy_times = availability_data.get("busy", [])
            
            # Convert busy times to datetime objects
            busy_periods = []
            for busy in busy_times:
                start = datetime.fromisoformat(busy["start"].replace("Z", "+00:00"))
                end = datetime.fromisoformat(busy["end"].replace("Z", "+00:00"))
                busy_periods.append((start, end))
            
            # Generate slots for each date range
            for date_range in date_ranges:
                range_start = datetime.fromisoformat(date_range["start"].replace("Z", "+00:00"))
                range_end = datetime.fromisoformat(date_range["end"].replace("Z", "+00:00"))
                
                current_slot = range_start
                while current_slot + timedelta(minutes=30) <= range_end:
                    slot_end = current_slot + timedelta(minutes=30)
                    
                    # Check if slot is available
                    is_available = True
                    for busy_start, busy_end in busy_periods:
                        if (current_slot >= busy_start and current_slot < busy_end) or \
                           (slot_end > busy_start and slot_end <= busy_end):
                            is_available = False
                            break
                    
                    if is_available:
                        available_slots.append(current_slot.isoformat())
                    
                    current_slot = slot_end
            
            return available_slots
            
        except requests.RequestException as e:
            print(f"Error fetching available slots: {e}")
            if hasattr(e.response, 'text'):
                print(f"Response: {e.response.text}")
            return []

    def book_appointment(
        self,
        event_type_id: int,
        customer: Dict[str, str],
        start_time: str,
        notes: str = "",
        location: str = ""
    ) -> Dict[str, Union[bool, Dict]]:
        """Book an appointment with the provided details"""
        try:
            # Format the booking data according to Cal.com's API requirements
            booking_data = {
                "eventTypeId": event_type_id,
                "start": start_time,
                "email": customer["email"],
                "name": customer["name"],
                "timeZone": "UTC",
                "language": "en",
                "guests": [],
                "notes": notes,
                "location": location,
                "smsReminderNumber": customer.get("phone", ""),
                "rescheduleReason": "",
                "customInputs": [],
                "metadata": {}  # Required field
            }

            response = requests.post(
                f"{self.base_url}/bookings",
                headers=self.headers,
                params=self.params,
                json=booking_data
            )
            response.raise_for_status()
            return {"success": True, "booking": response.json()}
            
        except requests.RequestException as e:
            print(f"Error creating booking: {e}")
            if hasattr(e.response, 'text'):
                print(f"Response: {e.response.text}")
            return {"success": False, "message": str(e)}

    def get_booking_details(self, booking_id: str) -> Optional[Dict]:
        """Get details of a specific booking"""
        try:
            response = requests.get(
                f"{self.base_url}/bookings/{booking_id}",
                headers=self.headers,
                params=self.params
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            print(f"Error fetching booking details: {e}")
            return None

    def cancel_booking(self, booking_id: str, reason: str = "") -> bool:
        """Cancel a booking"""
        try:
            response = requests.delete(
                f"{self.base_url}/bookings/{booking_id}",
                headers=self.headers,
                params={**self.params, "reason": reason}
            )
            response.raise_for_status()
            return True
        except requests.RequestException as e:
            print(f"Error canceling booking: {e}")
            return False

    def reschedule_booking(
        self,
        booking_id: str,
        new_start_time: str,
        reason: str = ""
    ) -> Dict[str, Union[bool, Dict]]:
        """Reschedule a booking to a new time"""
        try:
            reschedule_data = {
                "start": new_start_time,
                "reason": reason
            }
            
            response = requests.patch(
                f"{self.base_url}/bookings/{booking_id}",
                headers=self.headers,
                params=self.params,
                json=reschedule_data
            )
            response.raise_for_status()
            return {"success": True, "booking": response.json()}
        except requests.RequestException as e:
            print(f"Error rescheduling booking: {e}")
            return {"success": False, "message": str(e)}