import os import requests import json import datetime from config import settings import logging # Configure Logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("superoffice-client") class ContactNotFoundException(Exception): """Custom exception for 404 errors on Contact/Person lookups.""" pass class SuperOfficeClient: """A client for interacting with the SuperOffice REST API.""" def __init__(self): # Configuration self.client_id = settings.SO_CLIENT_ID self.client_secret = settings.SO_CLIENT_SECRET self.refresh_token = settings.SO_REFRESH_TOKEN self.env = settings.SO_ENVIRONMENT self.cust_id = settings.SO_CONTEXT_IDENTIFIER logger.info(f"DEBUG CONFIG: Env='{self.env}', CustID='{self.cust_id}', ClientID='{self.client_id[:4]}...'") if not all([self.client_id, self.client_secret, self.refresh_token]): # Graceful failure: Log error but allow init (for help/docs/discovery scripts) logger.error("❌ SuperOffice credentials missing in .env file (or environment variables).") self.base_url = None self.access_token = None return self.base_url = f"https://{self.env}.superoffice.com/{self.cust_id}/api/v1" self.access_token = self._refresh_access_token() if not self.access_token: logger.error("❌ Failed to authenticate with SuperOffice.") else: self.headers = { "Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json", "Accept": "application/json" } logger.info("✅ SuperOffice Client initialized and authenticated.") def _refresh_access_token(self): """Refreshes and returns a new access token.""" # OAuth token endpoint is ALWAYS online.superoffice.com for production, # or sod.superoffice.com for sandbox. token_domain = "online.superoffice.com" if "online" in self.env.lower() else "sod.superoffice.com" url = f"https://{token_domain}/login/common/oauth/tokens" logger.debug(f"DEBUG: Refresh URL: '{url}' (Env: '{self.env}')") data = { "grant_type": "refresh_token", "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, "redirect_uri": settings.SO_REDIRECT_URI } try: resp = requests.post(url, data=data) # Catch non-JSON responses early if resp.status_code != 200: logger.error(f"❌ Token Refresh Failed (Status {resp.status_code})") logger.error(f"Response Body: {resp.text[:500]}") return None resp.raise_for_status() return resp.json().get("access_token") except requests.exceptions.JSONDecodeError: logger.error(f"❌ Token Refresh Error: Received non-JSON response from {url}") logger.debug(f"Raw Response: {resp.text[:500]}") return None except Exception as e: logger.error(f"❌ Connection Error during token refresh: {e}") return None def _request_with_retry(self, method, endpoint, payload=None, retry=True): """Helper to handle 401 Unauthorized with auto-refresh.""" if not self.access_token: if not self._refresh_access_token(): return None url = f"{self.base_url}/{endpoint}" try: if method == "GET": resp = requests.get(url, headers=self.headers) elif method == "POST": resp = requests.post(url, headers=self.headers, json=payload) elif method == "PUT": resp = requests.put(url, headers=self.headers, json=payload) elif method == "PATCH": resp = requests.patch(url, headers=self.headers, json=payload) elif method == "DELETE": resp = requests.delete(url, headers=self.headers) # 401 Handling if resp.status_code == 401 and retry: logger.warning(f"⚠️ 401 Unauthorized for {endpoint}. Attempting Token Refresh...") new_token = self._refresh_access_token() if new_token: logger.info("✅ Token refreshed successfully during retry.") self.access_token = new_token self.headers["Authorization"] = f"Bearer {self.access_token}" # Recursive retry with the new token return self._request_with_retry(method, endpoint, payload, retry=False) else: logger.error("❌ Token Refresh failed during retry.") return None if resp.status_code == 204: return True resp.raise_for_status() return resp.json() except requests.exceptions.HTTPError as e: # Explicitly handle 404 Not Found for GET requests if method == "GET" and e.response.status_code == 404: logger.warning(f"🔍 404 Not Found for GET request to {endpoint}.") raise ContactNotFoundException(f"Entity not found at {endpoint}") from e logger.error(f"❌ API {method} Error for {endpoint} (Status: {e.response.status_code}): {e.response.text}") return None except Exception as e: logger.error(f"❌ Connection Error during {method} for {endpoint}: {e}") return None def _get(self, endpoint): return self._request_with_retry("GET", endpoint) def _put(self, endpoint, payload): return self._request_with_retry("PUT", endpoint, payload) def _patch(self, endpoint, payload): return self._request_with_retry("PATCH", endpoint, payload) def _post(self, endpoint, payload): return self._request_with_retry("POST", endpoint, payload) def _delete(self, endpoint): return self._request_with_retry("DELETE", endpoint) # --- Convenience Wrappers --- def get_person(self, person_id, select: list = None): endpoint = f"Person/{person_id}" if select: endpoint += f"?$select={','.join(select)}" return self._get(endpoint) def get_contact(self, contact_id, select: list = None): endpoint = f"Contact/{contact_id}" if select: endpoint += f"?$select={','.join(select)}" return self._get(endpoint) def search(self, query_string: str): """ Performs a search using OData syntax and handles pagination. Example: "Person?$select=personId&$filter=lastname eq 'Godelmann'" """ if not self.access_token: return None all_results = [] next_page_url = f"{self.base_url}/{query_string}" while next_page_url: try: resp = requests.get(next_page_url, headers=self.headers) resp.raise_for_status() data = resp.json() all_results.extend(data.get('value', [])) # Robust Pagination: Check both OData standard and legacy property next_page_url = data.get('odata.nextLink') or data.get('next_page_url') except requests.exceptions.HTTPError as e: logger.error(f"❌ API Search Error for {query_string}: {e.response.text}") return None return all_results def find_contact_by_criteria(self, name=None, email=None): """Searches for a contact by name or email.""" if name: query = f"Contact?$filter=name eq '{name}'" elif email: # Note: This depends on the specific SO API version/setup for email filtering query = f"Contact?$filter=email/address eq '{email}'" else: return None results = self.search(query) # Handle OData 'value' wrap if search doesn't do it return results[0] if results else None def create_contact(self, name: str, url: str = None, org_nr: str = None): """Creates a new contact.""" payload = { "Name": name, "UrlAddress": url, "OrgNr": org_nr } return self._post("Contact", payload) def create_person(self, first_name: str, last_name: str, contact_id: int, email: str = None): """Creates a new person linked to a contact.""" payload = { "Firstname": first_name, "Lastname": last_name, "ContactId": contact_id } if email: payload["Emails"] = [{"Value": email, "Description": "Primary", "IsPrimary": True}] return self._post("Person", payload) def create_sale(self, title: str, contact_id: int, person_id: int = None, amount: float = 0.0): """Creates a new sale.""" payload = { "Heading": title, "ContactId": contact_id, "Amount": amount, "Saledate": datetime.datetime.utcnow().isoformat() + "Z", "Probablity": 50, "Status": "Open" } if person_id: payload["PersonId"] = person_id return self._post("Sale", payload) def create_project(self, name: str, contact_id: int, person_id: int = None): """Creates a new project linked to a contact, and optionally adds a person.""" payload = { "Name": name, "Contact": {"ContactId": contact_id} } if person_id: # Adding a person to a project requires a ProjectMember object payload["ProjectMembers"] = [ { "Person": {"PersonId": person_id}, "Role": "Member" # Default role, can be configured if needed } ] print(f"Creating new project: {name}...") return self._post("Project", payload) def create_appointment(self, subject: str, description: str, contact_id: int, person_id: int = None): """Creates a new appointment (to simulate a sent activity).""" now = datetime.datetime.utcnow().isoformat() + "Z" # SuperOffice UI limit: 42 chars. # We put exactly this in the FIRST line of the description. short_title = (subject[:39] + '...') if len(subject) > 42 else subject # SuperOffice often 'steals' the first line of the description for the list view header. # So we give it exactly the subject it wants, then two newlines for the real body. formatted_description = f"{short_title}\n\n{description}" payload = { "Description": formatted_description, "Contact": {"ContactId": contact_id}, "StartDate": now, "EndDate": now, "MainHeader": short_title, "Task": {"Id": 1} } if person_id: payload["Person"] = {"PersonId": person_id} logger.info(f"Creating new appointment: {short_title}...") return self._post("Appointment", payload) def update_entity_udfs(self, entity_id: int, entity_type: str, udf_data: dict): """ Updates UDFs for a given entity (Contact or Person) using PATCH. entity_type: 'Contact' or 'Person' udf_data: {ProgId: Value} """ endpoint = f"{entity_type}/{entity_id}" # Construct PATCH payload payload = { "UserDefinedFields": udf_data } logger.info(f"Patching {entity_type} {entity_id} UDFs: {udf_data}...") # PATCH update result = self._patch(endpoint, payload) return bool(result) def update_person_position(self, person_id: int, position_id: int): """ Updates the standard 'Position' field of a Person using PATCH. """ endpoint = f"Person/{person_id}" # Construct PATCH payload payload = { "Position": {"Id": int(position_id)} } logger.info(f"Patching Person {person_id} Position to ID {position_id}...") # PATCH update result = self._patch(endpoint, payload) return bool(result) def patch_contact(self, contact_id: int, patch_data: dict): """ Generic PATCH for Contact entity. """ endpoint = f"Contact/{contact_id}" logger.info(f"Patching Contact {contact_id} with data keys: {list(patch_data.keys())}...") result = self._patch(endpoint, patch_data) return bool(result)