import re import logging import requests from zoneinfo import ZoneInfo from datetime import datetime, timedelta from django.utils import timezone from requests.auth import HTTPBasicAuth from typing import Dict, Any, List from django.conf import settings logger = logging.getLogger(__name__) PACIFIC = ZoneInfo("America/Los_Angeles") class XTimeClient: """ Handles XTime API communication. """ TOKEN_URL = "https://auth.coxautoinc.com/token" API_BASE_URL = settings.COXAUTO_API_BASE_URL def __init__(self, client_id: str, client_secret: str, api_key: str, dealer_id: str, scopes: str): self.client_id = client_id self.client_secret = client_secret self.api_key = api_key self.dealer_id = dealer_id self.scopes = scopes self.access_token = None def get_access_token(self) -> str: """ Fetch OAuth2 access token. Caches token in instance for reuse. """ if self.access_token: return self.access_token headers = {"Content-Type": "application/x-www-form-urlencoded"} data = {"grant_type": "client_credentials", "scope": self.scopes} try: resp = requests.post( self.TOKEN_URL, headers=headers, data=data, auth=HTTPBasicAuth(self.client_id, self.client_secret), timeout=30 ) resp.raise_for_status() token_data = resp.json() self.access_token = token_data["access_token"] return self.access_token except requests.RequestException as exc: logger.exception(f"Failed to fetch XTime token {exc}") raise def get_services_for_vehicle(self, year: int, make: str, model: str) -> Dict[str, Any]: """ Fetch available services for a vehicle. """ url = f"{self.API_BASE_URL}/services/dealer/{self.dealer_id}" headers = { "Authorization": f"Bearer {self.get_access_token()}", "x-api-key": self.api_key, "Content-Type": "application/json", } payload = {"vehicle": {"year": year, "make": make, "model": model}} try: resp = requests.post(url, headers=headers, json=payload, timeout=30) resp.raise_for_status() return resp.json() except requests.RequestException as exc: logger.warning(f"Failed to fetch services: {exc}") return {"error": str(exc)} def book_appointment( self, appointment_datetime: str, first_name: str, last_name: str, email: str, phone: str, year: int, make: str, model: str, opcode: str, ) -> Dict[str, Any]: """ Book an appointment in XTime. """ url = f"{self.API_BASE_URL}/appointments-bookings/dealer/{self.dealer_id}" headers = { "Authorization": f"Bearer {self.get_access_token()}", "x-api-key": self.api_key, "dealerCode": self.dealer_id } payload = { "appointmentDateTimeLocal": appointment_datetime, "firstName": first_name, "lastName": last_name, "emailAddress": email, "phoneNumber": phone, "vehicle": {"year": year, "make": make, "model": model}, "services": [{"opcode": opcode}], } try: resp = requests.post(url, headers=headers, json=payload, timeout=30) resp.raise_for_status() return resp.json() except requests.RequestException as exc: logger.warning(f"Failed to book XTime appointment: {exc}") return { "success": False, "message": str(exc), "error": f"Unable to get Services {exc}" } xtime_client = XTimeClient( client_id=settings.XTIME_CLIENT_ID, client_secret=settings.XTIME_CLIENT_SECRET, api_key=settings.XTIME_API_KEY, dealer_id=settings.XTIME_DEALER_ID, scopes=settings.XTIME_SCOPES ) def x_time_native_dt(dt: str) -> str: """ Convert "2024-06-14 10:30:00" -> "2024-06-14T10:30-10:45" Adds 15 minutes to the start time. """ # Parse input string to datetime start_dt = datetime.strptime(dt, "%Y-%m-%d %H:%M:%S") # Calculate end time (start + 15 minutes) end_dt = start_dt + timedelta(minutes=15) # Build string in desired format return f"{start_dt.strftime('%Y-%m-%dT%H:%M')}-{end_dt.strftime('%H:%M')}" def x_time_native_range(start_dt: datetime, end_dt: datetime) -> str: """ Convert: start_dt = 2024-06-14 10:30:00-07:00 end_dt = 2024-06-14 10:45:00-07:00 -> "2024-06-14T10:30-10:45" Expects timezone-aware datetime objects. """ if timezone.is_naive(start_dt) or timezone.is_naive(end_dt): raise ValueError("start_dt and end_dt must be timezone-aware date times") return ( f"{start_dt.strftime('%Y-%m-%dT%H:%M')}" f"-{end_dt.strftime('%H:%M')}" ) def normalize(text: str) -> str: """ Normalize service names for reliable matching. """ text = text.lower() text = re.sub(r'[^a-z0-9\s]', '', text) text = re.sub(r'\s+', ' ', text).strip() return text def find_best_service_opcode( xtime_services_response: Dict[str, Any], booking_type: str ): """ Given XTime services API response and a local booking type string (e.g. "oil change"), return the best matching opcode. Uses scoring instead of naive contains() for better reliability. """ services: List[Dict[str, Any]] = xtime_services_response.get("services", []) if not services: return None booking_norm = normalize(booking_type) best_score = 0 best_opcode = None best_name = None for service in services: service_name = service.get("serviceName", "") service_norm = normalize(service_name) # Split words booking_words = set(booking_norm.split()) service_words = set(service_norm.split()) # Score based on word intersection common_words = booking_words.intersection(service_words) score = len(common_words) # Bonus if whole phrase matches partially if booking_norm in service_norm: score += 3 if score > best_score: best_score = score best_opcode = service.get("opcode") best_name = service.get("serviceName") return best_opcode, best_name def book_appointment_x_time( appointment_start_dt: datetime, appointment_end_dt: datetime, first_name: str, last_name: str, email: str, phone: str, year: int, make: str, model: str, service_name: str, ): xtime_client.get_access_token() services_response = xtime_client.get_services_for_vehicle( year=year, make=make, model=model, ) dt = x_time_native_range( appointment_start_dt, appointment_end_dt ) if services_response.get('error'): return services_response, dt, None, service_name opcode, service_name = find_best_service_opcode( services_response, service_name, ) return xtime_client.book_appointment( appointment_datetime=dt, first_name=first_name, last_name=last_name, email=email, phone=phone, year=year, make=make, model=model, opcode=opcode ), dt, opcode, service_name n=""" from apps.appointments.services import full_test2 full_test2() """ def full_test2(): return book_appointment_x_time( appointment_start_dt=datetime( 2026, 1, 27, 10, 30, tzinfo=PACIFIC ), appointment_end_dt=datetime( 2026, 1, 27, 10, 45, tzinfo=PACIFIC ), first_name="shamas", last_name="haq", email="shamas.haq@flutenai.com", phone="1234323243", year=2018, make="Nissan", model="Armada", service_name="oil change", )