# /mnt/e/genesis-system/core/revenue/crm_sync.py import logging import os import json import time from typing import Dict, Any, Optional import requests # Configure logging (ensure this is consistent with Genesis system practices) logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class GHLClient: """ Client for interacting with the GoHighLevel API. """ def __init__(self, api_key: str, base_url: str = "https://rest.gohighlevel.com/v1/"): """ Initializes the GHLClient with the API key and base URL. Args: api_key: The GoHighLevel API key. base_url: The base URL of the GoHighLevel API. Defaults to the standard GHL URL. """ self.api_key = api_key self.base_url = base_url self.headers = { "Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json" } def _make_request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """ Makes a request to the GoHighLevel API. Args: method: The HTTP method (e.g., "GET", "POST", "PUT"). endpoint: The API endpoint. data: The request body as a dictionary (optional). Returns: The JSON response from the API as a dictionary. Raises: requests.exceptions.RequestException: If the API request fails. ValueError: If the API response is not valid JSON. """ url = self.base_url + endpoint try: response = requests.request(method, url, headers=self.headers, json=data) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) return response.json() except requests.exceptions.RequestException as e: logger.error(f"API request failed: {e}") raise except json.JSONDecodeError as e: logger.error(f"Failed to decode JSON response: {e}") raise ValueError("Invalid JSON response from API") def get_contact(self, contact_id: str) -> Dict[str, Any]: """ Retrieves a contact from GoHighLevel by ID. Args: contact_id: The ID of the contact. Returns: A dictionary representing the contact. Raises: requests.exceptions.RequestException: If the API request fails. ValueError: If the API response is not valid JSON. """ endpoint = f"contacts/{contact_id}" return self._make_request("GET", endpoint) def create_contact(self, data: Dict[str, Any]) -> Dict[str, Any]: """ Creates a new contact in GoHighLevel. Args: data: A dictionary containing the contact data. Returns: A dictionary containing the created contact's data. Raises: requests.exceptions.RequestException: If the API request fails. ValueError: If the API response is not valid JSON. """ endpoint = "contacts" return self._make_request("POST", endpoint, data) def update_contact(self, contact_id: str, data: Dict[str, Any]) -> Dict[str, Any]: """ Updates an existing contact in GoHighLevel. Args: contact_id: The ID of the contact to update. data: A dictionary containing the updated contact data. Returns: A dictionary containing the updated contact's data. Raises: requests.exceptions.RequestException: If the API request fails. ValueError: If the API response is not valid JSON. """ endpoint = f"contacts/{contact_id}" return self._make_request("PUT", endpoint, data) def find_contacts(self, query: Dict[str, Any]) -> Dict[str, Any]: """ Finds contacts in GoHighLevel based on a query. Args: query: A dictionary containing the query parameters. Returns: A dictionary containing the search results. Raises: requests.exceptions.RequestException: If the API request fails. ValueError: If the API response is not valid JSON. """ endpoint = "contacts" return self._make_request("GET", endpoint, data=query) class CRMSync: """ Synchronizes data between the Genesis system and GoHighLevel CRM. """ def __init__(self, ghl_api_key: str): """ Initializes the CRMSync with the GoHighLevel API key. Args: ghl_api_key: The GoHighLevel API key. """ self.ghl_client = GHLClient(ghl_api_key) def sync_contact(self, genesis_contact: Dict[str, Any]) -> None: """ Synchronizes a contact from the Genesis system to GoHighLevel. If a contact with the same email already exists in GHL, it will be updated. Otherwise, a new contact will be created. Args: genesis_contact: A dictionary containing the contact data from the Genesis system. """ email = genesis_contact.get("email") if not email: logger.warning("Contact has no email address; skipping sync.") return try: # Attempt to find the contact by email query = {"email": email} results = self.ghl_client.find_contacts(query) if results and "contacts" in results and results["contacts"]: # Contact exists, update it ghl_contact = results["contacts"][0] ghl_contact_id = ghl_contact["id"] # Prepare data for update. Map Genesis contact fields to GHL fields. update_data = self._map_genesis_to_ghl(genesis_contact) self.ghl_client.update_contact(ghl_contact_id, update_data) logger.info(f"Updated contact {email} in GHL (ID: {ghl_contact_id})") else: # Contact does not exist, create it create_data = self._map_genesis_to_ghl(genesis_contact) new_contact = self.ghl_client.create_contact(create_data) logger.info(f"Created contact {email} in GHL (ID: {new_contact['id']})") except Exception as e: logger.error(f"Error syncing contact {email}: {e}") def _map_genesis_to_ghl(self, genesis_contact: Dict[str, Any]) -> Dict[str, Any]: """ Maps fields from a Genesis contact to a GoHighLevel contact. Args: genesis_contact: A dictionary representing the Genesis contact. Returns: A dictionary representing the GoHighLevel contact. Only includes fields that are mapped. """ # Define the mapping between Genesis and GHL fields mapping = { "firstName": "first_name", "lastName": "last_name", "email": "email", "phone": "phone", "address": "address1", "city": "city", "state": "state", "postalCode": "postal_code", "companyName": "company_name", # Add custom fields if needed - example: # "genesisCustomField": "customField.goHighLevelCustomFieldId" } ghl_data = {} for ghl_field, genesis_field in mapping.items(): if genesis_field in genesis_contact and genesis_contact[genesis_field] is not None: ghl_data[ghl_field] = str(genesis_contact[genesis_field]) # Ensure all values are strings return ghl_data def main(): """ Main function to demonstrate the CRMSync functionality. This should be replaced with integration into the broader Genesis system. """ # Replace with your actual GHL API key ghl_api_key = os.environ.get("GHL_API_KEY") if not ghl_api_key: logger.error("GHL_API_KEY environment variable not set.") return crm_sync = CRMSync(ghl_api_key) # Example Genesis contact data genesis_contact = { "firstName": "John", "lastName": "Doe", "email": "john.doe@example.com", "phone": "555-123-4567", "address": "123 Main St", "city": "Anytown", "state": "CA", "postalCode": "91234", "companyName": "Acme Corp", } crm_sync.sync_contact(genesis_contact) if __name__ == "__main__": main()