File size: 4,926 Bytes
582520a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """Cal.com appointment integration with demo fallback."""
from __future__ import annotations
import logging
from datetime import datetime, timedelta, timezone
from typing import Any, Optional
from uuid import uuid4
import httpx
from app.config import get_settings
logger = logging.getLogger(__name__)
async def get_available_slots(
count: int = 3,
) -> list[dict[str, str]]:
"""Return upcoming available slots."""
settings = get_settings()
if settings.has_calcom:
try:
slots = await _fetch_calcom_slots(count)
if slots:
return slots
except Exception as exc: # noqa: BLE001
logger.exception("Cal.com slots failed: %s", exc)
return _demo_slots(count)
async def book_slot(
*,
slot_start: str,
slot_end: str,
attendee_name: Optional[str],
attendee_email: Optional[str],
) -> dict[str, Any]:
"""Create a booking via Cal.com or demo store."""
settings = get_settings()
if settings.has_calcom:
try:
return await _book_calcom(
slot_start=slot_start,
slot_end=slot_end,
attendee_name=attendee_name or "Guest",
attendee_email=attendee_email
or "guest@example.com",
)
except Exception as exc: # noqa: BLE001
logger.exception("Cal.com booking failed: %s", exc)
return {
"external_id": f"demo-{uuid4().hex[:10]}",
"slot_start": slot_start,
"slot_end": slot_end,
"status": "confirmed",
"demo": True,
}
def _demo_slots(count: int) -> list[dict[str, str]]:
now = datetime.now(timezone.utc).replace(
minute=0, second=0, microsecond=0
)
slots: list[dict[str, str]] = []
cursor = now + timedelta(days=1)
while len(slots) < count:
if cursor.weekday() < 5 and cursor.hour in (10, 14, 16):
end = cursor + timedelta(minutes=30)
slots.append(
{
"start": cursor.isoformat(),
"end": end.isoformat(),
"label": cursor.strftime(
"%a %d %b %Y · %H:%M UTC"
),
}
)
cursor += timedelta(hours=2)
return slots
async def _fetch_calcom_slots(count: int) -> list[dict[str, str]]:
settings = get_settings()
start = datetime.now(timezone.utc)
end = start + timedelta(days=14)
url = "https://api.cal.com/v1/slots"
params = {
"apiKey": settings.calcom_api_key,
"eventTypeId": settings.calcom_event_type_id,
"startTime": start.isoformat(),
"endTime": end.isoformat(),
}
async with httpx.AsyncClient(timeout=20) as client:
response = await client.get(url, params=params)
response.raise_for_status()
data = response.json()
slots: list[dict[str, str]] = []
raw_slots = data.get("slots") or data
if isinstance(raw_slots, dict):
for day_slots in raw_slots.values():
if not isinstance(day_slots, list):
continue
for item in day_slots:
start_s = item.get("time") or item.get("start")
if not start_s:
continue
start_dt = datetime.fromisoformat(
start_s.replace("Z", "+00:00")
)
end_dt = start_dt + timedelta(minutes=30)
slots.append(
{
"start": start_dt.isoformat(),
"end": end_dt.isoformat(),
"label": start_dt.strftime(
"%a %d %b %Y · %H:%M UTC"
),
}
)
if len(slots) >= count:
return slots
return slots[:count]
async def _book_calcom(
*,
slot_start: str,
slot_end: str,
attendee_name: str,
attendee_email: str,
) -> dict[str, Any]:
settings = get_settings()
url = "https://api.cal.com/v1/bookings"
payload = {
"eventTypeId": int(settings.calcom_event_type_id),
"start": slot_start,
"responses": {
"name": attendee_name,
"email": attendee_email,
},
"timeZone": settings.demo_timezone,
"language": "en",
"metadata": {},
}
params = {"apiKey": settings.calcom_api_key}
async with httpx.AsyncClient(timeout=20) as client:
response = await client.post(
url, params=params, json=payload
)
response.raise_for_status()
data = response.json()
return {
"external_id": str(
data.get("id") or data.get("uid") or uuid4()
),
"slot_start": slot_start,
"slot_end": slot_end,
"status": "confirmed",
"demo": False,
}
|