| """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: |
| 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: |
| 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, |
| } |
|
|
