# The Setmore class is a Python class that provides methods for interacting with the Setmore API, # including retrieving appointments and services, updating appointments, and setting appointment # labels. import json import time from datetime import datetime from http import HTTPStatus import requests from src.config import ANTI_RATELIMIT_WAIT_TIME from src.exceptions import MaxRetriesException, UnauthorisedRequestException, BadResponseException from src.models.model_abstract import HTTPModelAbstract from src.utils.loggers import setmore_log API_OUTPUT_DATE_FORMAT = "%Y-%m-%dT%H:%MZ" API_INPUT_DATE_FORMAT = "%d-%m-%Y" ZAPIER_DATE_FORMAT = "%Y-%m-%dT%H:%M:%SZ" # class Setmore(HTTPModelAbstract): """ A class representing the Setmore API. Attributes: CREDENTIALS_KEY (str): The key used to store Setmore credentials. base_url (str): The base URL of the Setmore API. COOKIE (str): The cookie used for authentication. Methods: __init__(): Initializes a Setmore instance. auth_headers(): Returns the authentication headers. _request(): Sends an HTTP request to the Setmore API. refresh_access_token(): Refreshes the access token for Setmore Authentication. get_headers(): Returns the headers for API requests. get_appointments(end_date: str, start_date: str): Retrieves Setmore appointments within a specified date range. get_services(): Retrieves services offered by Wwave from Setmore. set_appointment_label(appointment_key: str, label: str): Sets the appointment label in the Setmore database. update_appointment(appointment: dict): Updates a Setmore appointment via a PUT request to the Setmore API. test_request(url, method="GET", headers={}, data={}, _retries=0, auth=None): Sends a test request to the Setmore API. setmore_datetime(date_time: str): Converts a Setmore date/time string to a datetime object. """ # Rest of the code... class Setmore(HTTPModelAbstract): CREDENTIALS_KEY = 'Setmore' base_url = "https://developer.setmore.com/api/v1/" COOKIE = 'JSESSIONID=rUyweg9ovL6anXTT0EitvA' def __init__(self): if self.requires_token_refresh(): self.refresh_access_token() @property def auth_headers(self): return { 'Cookie': self.COOKIE } def _request(self, url, method="GET", headers=None, data=None, params=None, _retries=0, auth=None, **kwargs): log_request = self.format_request_for_logging(method=method, url=url, headers=headers, data=data, auth=auth, _retries=_retries) # Hard limit of 5 retries per _request if _retries >= self.MAX_RETRIES_PER_REQUEST: raise MaxRetriesException(request=log_request) response = requests.request(method=method, url=url, headers=headers, data=data, auth=auth) log_response = self.format_response_for_logging(response) # If the status is 401, refresh the access token if response.status_code == HTTPStatus.UNAUTHORIZED.value: raise UnauthorisedRequestException(request=log_request, response=log_response) # If the status is 504, resend the _request with _retries incremented by 1 if response.status_code == HTTPStatus.GATEWAY_TIMEOUT.value: _return = self._request(url, method, headers, data, _retries + 1, auth) # If the status is OK, return the response elif response.status_code in [HTTPStatus.OK.value, HTTPStatus.CREATED.value]: _return = response else: # If the response body contains an 'error' attribute with the value 'too_many_requests', wait the value of # ANTI_RATELIMIT_WAIT_TIME (in seconds) before resending the request if json.loads(response.text).get('error') == 'too_many_requests': time.sleep(ANTI_RATELIMIT_WAIT_TIME) _return = self._request(url, method, headers, data, _retries + 1, auth) else: raise BadResponseException(response.status_code, request=log_request, response=log_response, code=response.status_code) return _return def refresh_access_token(self): # Indentation: 4 spaces (class level) """Refresh the access token for Setmore Authentication.""" # Indentation: 8 spaces try: # Indentation: 8 spaces credentials = self._get_credentials() # Indentation: 12 spaces setmore_log.info(f"Retrieved credentials: {credentials}") # Indentation: 12 spaces refresh_token = credentials.get("refresh_token") # Indentation: 12 spaces if not refresh_token: # Indentation: 12 spaces from src.exceptions import AuthFailureException # Indentation: 16 spaces raise AuthFailureException("Missing Setmore Refresh Token. Speak to a Wwave representative about checking " # Indentation: 16 spaces "legacy Postman requests for Setmore for old refresh tokens.", code=401) # Indentation: 16 spaces (continued line) setmore_log.info(f"Using refresh token: {refresh_token}") # Indentation: 12 spaces setmore_log.info(f"Base URL: {self.base_url}") # Indentation: 12 spaces url = f"{self.base_url}o/oauth2/token?refreshToken={refresh_token}" # Indentation: 12 spaces setmore_log.info(f"Token refresh URL: {url}") # Indentation: 12 spaces response = self._request(method="GET", url=url, headers=self.auth_headers) # Indentation: 12 spaces response_dict = json.loads(response.text) # Indentation: 12 spaces if not response_dict.get("response"): # Indentation: 12 spaces setmore_log.error(f"Token refresh failed: {response_dict.get('msg')}") # Indentation: 16 spaces from src.exceptions import AuthFailureException # Indentation: 16 spaces raise AuthFailureException(f"Token refresh failed: {response_dict.get('msg')}", code=400) # Indentation: 16 spaces data = response_dict.get("data") # Indentation: 12 spaces credentials["access_token"] = data["token"]["access_token"] # Indentation: 12 spaces credentials["refresh_token"] = refresh_token # Indentation: 12 spaces credentials["last_refresh"] = datetime.utcnow().__str__() # Indentation: 12 spaces self._update_credentials(credentials) # Indentation: 12 spaces setmore_log.info("Successfully refreshed access token") # Indentation: 12 spaces except Exception as e: # Indentation: 8 spaces setmore_log.error(f"Failed to refresh access token: {str(e)}") # Indentation: 12 spaces raise # Indentation: 12 spaces def get_headers(self): access_token = self._get_credentials()["access_token"] return { 'Content-Type': 'application/json', 'Authorization': f"Bearer {access_token}", 'Cookie': self.COOKIE } def get_appointments(self, end_date: str, start_date: str): """Get Setmore appointments from start_date to end_date. Date format: DD-MM-YYYY""" url = f"{self.base_url}bookingapi/appointments?customerDetails=true&endDate={end_date}&startDate={start_date}" response = self._request(method="GET", url=url, headers=self.get_headers()) response_dict = json.loads(response.text) data = response_dict.get("data") if data: return data.get("appointments") def get_services(self): """Get services offered by Wwave from Setmore.""" url = f"{self.base_url}bookingapi/services" response = self._request(method="GET", url=url, headers=self.get_headers()) r_content = json.loads(response.text) data = r_content.get("data") if data: return data.get("services") def set_appointment_label(self, appointment_key: str, label: str): """Sets the appointment label in the Setmore database.""" # Validation of inputs assert type(appointment_key) is str, self.format_type_assertion_str(appointment_key, 'appointment_key', 'str') assert appointment_key != "", "Appointment key cannot be empty string." assert type(label) is str, self.format_type_assertion_str(label, 'label', 'str') # Construct the URL url = f"{self.base_url}bookingapi/appointments/{appointment_key}/label?label='{label}'" # Log the request details setmore_log.info(f"Attempting to set label for appointment set_appointment_label in setmore.setmore.py {appointment_key} to '{label}' using URL: {url}") try: # Perform the PUT request response = self._request(method="PUT", url=url, headers=self.get_headers()) # Log the response setmore_log.info(f"Received response for setting label for appointment set_appointment_label in setmore.setmore.py: {response.status_code}, {response.text}") return response except Exception as e: # Log the error if any setmore_log.error(f"Error while setting label for appointment for appointment set_appointment_label in setmore.setmore.py {appointment_key}: {e}") raise def update_label_manual(self, invoice, amount: float, payment_date: str, payment_method: str): """ Appends a payment label to an existing Setmore appointment label, preserving all data and allowing duplicates. Logs every step to verify data collection and update. Args: invoice (Invoice): A valid MYOB invoice object. amount (float): The payment amount. payment_date (str): Payment date in 'YYYY-MM-DD' format. payment_method (str): Method of payment, e.g., 'EFTPOS', 'Cash'. Returns: dict: API response if successful. Raises: ValueError: If CustomerPurchaseOrderNumber is missing or invalid. Exception: If fetching or updating the appointment fails. """ from src.models.Setmore.appointment import Appointment from datetime import datetime setmore_log.info(f"Starting update_label_manual: invoice_number={invoice.Number if invoice else 'None'}, " f"amount={amount}, payment_date={payment_date}, payment_method={payment_method}") if not invoice or not invoice.CustomerPurchaseOrderNumber: setmore_log.error("Missing CustomerPurchaseOrderNumber on invoice") raise ValueError("Missing CustomerPurchaseOrderNumber on invoice.") # Extract the appointment key raw_po = invoice.CustomerPurchaseOrderNumber.strip() if len(raw_po) >= 36 and raw_po[-36:].count('-') == 4: appointment_key = raw_po[-36:] else: appointment_key = raw_po.replace("C ", "") setmore_log.info(f"Extracted appointment key: {appointment_key} from raw PO: {raw_po}") # Fetch the existing appointment to get the current label current_label = "" try: # Use invoice.datetime if available, else fallback to payment_date appointment_date = invoice.datetime.strftime('%d-%m-%Y') if hasattr(invoice, 'datetime') else payment_date try: appointment_date = datetime.strptime(appointment_date, '%Y-%m-%d').strftime('%d-%m-%Y') except ValueError: pass # Already in DD-MM-YYYY format setmore_log.info(f"Fetching appointment with key {appointment_key} on date {appointment_date}") appointment = Appointment.get_appointment(appointment_key, appointment_date) if not appointment: setmore_log.error(f"No appointment found for key {appointment_key} on date {appointment_date}") raise ValueError(f"No appointment found for key {appointment_key} on date {appointment_date}") current_label = appointment.label.strip() setmore_log.info(f"Existing appointment data: {appointment.__dict__}") setmore_log.info(f"Current label: {current_label}") except Exception as e: setmore_log.error(f"Failed to fetch appointment for key {appointment_key}: {str(e)}") raise # Format the new payment label try: payment_date_short = datetime.strptime(payment_date, '%Y-%m-%d').strftime('%d/%m') except ValueError as e: setmore_log.error(f"Invalid payment_date format: {payment_date}. Expected 'YYYY-MM-DD'. Error: {str(e)}") raise ValueError(f"Invalid payment_date format: {payment_date}. Expected 'YYYY-MM-DD'.") new_label_part = f" | {payment_date_short}: {payment_method.upper()} ${float(amount):.2f}" setmore_log.info(f"New payment label part: {new_label_part}") """ # Check for duplicates to avoid redundant additions if new_label_part in current_label: setmore_log.warning(f"Payment label part '{new_label_part}' already exists in current label: {current_label}") return {"status": "skipped", "message": "Payment label already present"} """ # Append the new payment label final_label = current_label + new_label_part if current_label else new_label_part.lstrip(" |") setmore_log.info(f"Final label after append: {final_label}") # Update the label try: response = self.update_label(appointment_key, final_label) setmore_log.info(f"Label update response for {appointment_key}: {response}") return response except Exception as e: setmore_log.error(f"Failed to update label for {appointment_key}: {str(e)}") raise def update_comment_manual(self, invoice, amount: float, payment_date: str, payment_method: str): """ Appends a payment comment to an existing Setmore appointment comment, preserving all data. Logs every step to verify data collection and update. Args: invoice (Invoice): A valid MYOB invoice object. amount (float): The payment amount. payment_date (str): Payment date in 'YYYY-MM-DD' format. payment_method (str): Method of payment, e.g., 'EFTPOS', 'Cash'. Returns: dict: API response from appointment update. Raises: ValueError: If CustomerPurchaseOrderNumber is missing or invalid. Exception: If fetching or updating the appointment fails. """ from src.models.Setmore.appointment import Appointment from datetime import datetime from src.utils.loggers import setmore_log setmore_log.info(f"Starting update_comment_manual: invoice_number={invoice.Number if invoice else 'None'}, " f"amount={amount}, payment_date={payment_date}, payment_method={payment_method}") if not invoice or not invoice.CustomerPurchaseOrderNumber: setmore_log.error("Missing CustomerPurchaseOrderNumber on invoice") raise ValueError("Missing CustomerPurchaseOrderNumber on invoice.") # Clean booking ID raw_po = invoice.CustomerPurchaseOrderNumber.strip() appointment_key = raw_po[-36:] if len(raw_po) >= 36 and raw_po[-36:].count('-') == 4 else raw_po.replace("C ", "") setmore_log.info(f"Extracted appointment key: {appointment_key} from raw PO: {raw_po}") # Format the new comment try: date_str = datetime.strptime(payment_date, '%Y-%m-%d').strftime('%d/%m') except ValueError as e: setmore_log.error(f"Invalid payment_date format: {payment_date}. Expected 'YYYY-MM-DD'. Error: {str(e)}") raise ValueError(f"Invalid payment_date format: {payment_date}. Expected 'YYYY-MM-DD'.") new_comment = f"{date_str}: Manual payment of ${float(amount):.2f} via {payment_method.upper()}." setmore_log.info(f"New comment part: {new_comment}") # Retrieve current appointment try: # Use invoice.datetime if available, else fallback to payment_date appointment_date = invoice.datetime.strftime('%d-%m-%Y') if hasattr(invoice, 'datetime') else payment_date try: appointment_date = datetime.strptime(appointment_date, '%Y-%m-%d').strftime('%d-%m-%Y') except ValueError: pass # Already in DD-MM-YYYY format setmore_log.info(f"Fetching appointment with key {appointment_key} on date {appointment_date}") appointment = Appointment.get_appointment(appointment_key, appointment_date) if not appointment: setmore_log.error(f"No appointment found for key {appointment_key} on date {appointment_date}") raise ValueError(f"No appointment found for key {appointment_key} on date {appointment_date}") setmore_log.info(f"Existing appointment data: {appointment.__dict__}") setmore_log.info(f"Current comment: {appointment.comment}") except Exception as e: setmore_log.error(f"Failed to fetch appointment for key {appointment_key}: {str(e)}") raise # Append the new comment if appointment.comment: appointment.comment += f"\n\n{new_comment}" else: appointment.comment = new_comment setmore_log.info(f"Updated comment: {appointment.comment}") # Update the appointment try: response = appointment.update() setmore_log.info(f"Comment update response for {appointment_key}: {response}") return response except Exception as e: setmore_log.error(f"Failed to update comment for {appointment_key}: {str(e)}") raise def process_appointment_update(self): """ DISABLED: Temporarily prevents any data being sent back to Setmore. Only logs what's happening — no Setmore updates will be made. """ setmore_log.warning(f"[DISABLED] Skipping Setmore update for appointment {self._appointment.key}") # Leave this bit if you still want MYOB to update if not self._invoice: self._invoice = Invoice.get_studio_invoice(self._appointment.key) if self._invoice: try: self._invoice.Date = self._appointment.myob_date self.update_invoice_extras() self._invoice.update() setmore_log.info(f"[MYOB] Invoice updated for {self._appointment.key}") except Exception as e: setmore_log.error(f"[MYOB] Invoice update failed for {self._appointment.key}: {str(e)}") else: setmore_log.warning(f"[MYOB] No invoice found for appointment {self._appointment.key}") def update_appointment(self, appointment: dict): """Update the given Setmore appointment via a PUT request to the Setmore API.""" # Asserting appointment format required_attribute_names = ["key", "staff_key", "service_key", "customer_key", "start_time", "end_time"] for attr in required_attribute_names: assert appointment.get(attr) is not None, f"Request requires the appointment to contain the {attr} attribute." # Hardcode the v2 URL for the update url = f"https://developer.setmore.com/api/v2/bookingapi/appointments/{appointment['key']}" # Fetch the appointment first to confirm it exists (using v1, since that's what works for fetching) start_date = datetime.strptime(appointment["start_time"], API_OUTPUT_DATE_FORMAT).strftime(API_INPUT_DATE_FORMAT) setmore_log.info(f"Fetching appointment {appointment['key']} for date {start_date} using v1") try: # Temporarily set base_url to v1 for fetching (since v1 works for fetching) original_base_url = self.base_url self.base_url = "https://developer.setmore.com/api/v1/" appointments = self.get_appointments(start_date, start_date) self.base_url = original_base_url # Restore the original base_url if not any(appt["key"] == appointment["key"] for appt in appointments): setmore_log.error(f"Appointment {appointment['key']} not found in v1") raise Exception(f"Appointment {appointment['key']} not found in v1") except Exception as e: setmore_log.error(f"Error fetching appointment {appointment['key']} in v1: {e}") raise # Proceed with the update using the hardcoded v2 URL setmore_log.info(f"Sending update request to {url} using v2") response = self._request(method="PUT", url=url, headers=self.get_headers(), data=json.dumps(appointment, indent=4)) setmore_log.debug(f"Update response from v2: {response.text}") return response def update_label(self, appointment_key: str, label: str): """ Updates the label of an appointment using the specific API endpoint. Args: appointment_key (str): The unique key of the appointment. label (str): The label to set for the appointment. Returns: dict: The response data from the Setmore API. Raises: Exception: If the API call fails or the response indicates failure. """ assert isinstance(appointment_key, str) and appointment_key, "Appointment key must be a non-empty string." assert isinstance(label, str) and label, "Label must be a non-empty string." # Directly specify the full endpoint URL url = f"https://developer.setmore.com/api/v1/bookingapi/appointments/{appointment_key}/label?label={requests.utils.quote(label)}" headers = self.get_headers() # Log the final URL and headers setmore_log.info(f"Sending PUT request to using update_label in setmore.setmore.py: {url}") setmore_log.debug(f"Request headers using update_label in setmore.setmore.py: {headers}") try: # Send the PUT request response = requests.put(url, headers=headers) response_data = response.json() # Log the full response setmore_log.debug(f"Response status code using update_label in setmore.setmore.py: {response.status_code}") setmore_log.debug(f"Response body using update_label in setmore.setmore.py: {response_data}") # Check for success in response if response.status_code == 200 and response_data.get("response", False): setmore_log.info(f"Successfully updated label for appointment using update_label in setmore.setmore.py {appointment_key} to '{label}'.") return response_data else: error_message = response_data.get("msg", "Unknown error") setmore_log.error(f"Failed to update label using update_label in setmore.setmore.py: {error_message}") raise Exception(f"Setmore API returned error using update_label in setmore.setmore.py: {error_message}") except Exception as e: setmore_log.error(f"Error updating label for appointment using update_label in setmore.setmore.py {appointment_key}: {e}") raise def test_request(self, url, method="GET", headers={}, data={}, _retries=0, auth=None): """Public method that calls _request. This method should only be used for testing _request.""" log_request = self.format_request_for_logging(method=method, url=url, headers=headers, data=data, auth=auth, _retries=_retries) # Hard limit of 5 retries per _request if _retries >= self.MAX_RETRIES_PER_REQUEST: raise MaxRetriesException(request=log_request) response = requests.request(method=method, url=url, headers=headers, data=data, auth=auth) log_response = self.format_response_for_logging(response) # If the status is 401, refresh the access token if response.status_code == HTTPStatus.UNAUTHORIZED.value: raise UnauthorisedRequestException(request=log_request, response=log_response) # If the status is 504, resend the _request with _retries incremented by 1 if response.status_code == HTTPStatus.GATEWAY_TIMEOUT.value: _return = self._request(url, method, headers, data, _retries + 1, auth) # If the status is OK, return the response elif response.status_code in [HTTPStatus.OK.value, HTTPStatus.CREATED.value]: _return = response else: # If the response body contains an 'error' attribute with the value 'too_many_requests', wait the value of # ANTI_RATELIMIT_WAIT_TIME (in seconds) before resending the request if json.loads(response.text)['data'].get('error') == 'too_many_requests': time.sleep(ANTI_RATELIMIT_WAIT_TIME) _return = self._request(url, method, headers, data, _retries + 1, auth) else: raise BadResponseException(str(response.status_code), request=log_request, response=log_response, code=response.status_code) return _return @staticmethod def setmore_datetime(date_time: str): from datetime import datetime from pytz import timezone from src.models.Setmore import API_INPUT_DATE_FORMAT, API_OUTPUT_DATE_FORMAT, ZAPIER_DATE_FORMAT # Attempt to strip time in each setmore format try: _return = datetime.strptime(date_time, API_INPUT_DATE_FORMAT) except ValueError: try: _return = datetime.strptime(date_time, API_OUTPUT_DATE_FORMAT) except ValueError: date = datetime.strptime(date_time, ZAPIER_DATE_FORMAT) # Zapier returns "start" in UTC , so localise this datetime to UTC date = timezone("UTC").localize(date) _return = date.astimezone(timezone('Australia/Melbourne')) return _return