""" Reputation Rocket — GHL API v2 Deployment Script ================================================= Programmatically deploys the Review Generator module to a GHL sub-account. Deploys: - 4x Custom Fields (google_review_link, technician_name, job_type, review_rating) - 1x Pipeline (Reputation Tracker) with 3 stages - 2x SMS Templates (Review Request, Follow-Up) Does NOT deploy (must be imported via GHL snapshot): - Workflows (GHL API v2 does not support workflow creation) - Funnels (must be built or cloned via GHL UI) Usage: python ghl_api_deploy.py \ --sub-account-id "abc123locationId" \ --google-review-link "https://g.page/r/XXXXXX/review" \ --business-name "Smith Plumbing" Environment Variables: GHL_API_KEY — Agency API key with write access to the sub-account Dependencies: pip install requests """ import argparse import json import os import sys import time from datetime import datetime import requests # --------------------------------------------------------------------------- # Configuration # --------------------------------------------------------------------------- GHL_BASE_URL = "https://services.leadconnectorhq.com" GHL_API_VERSION = "2021-07-28" HEADERS_TEMPLATE = { "Authorization": "Bearer {api_key}", "Content-Type": "application/json", "Version": GHL_API_VERSION, } # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def get_headers(api_key: str) -> dict: """Return headers with bearer token injected.""" return { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", "Version": GHL_API_VERSION, } def handle_response(response: requests.Response, action: str) -> dict: """ Check HTTP response. Raise a descriptive RuntimeError on failure. Returns parsed JSON on success. """ try: response.raise_for_status() except requests.HTTPError: body = "" try: body = response.json() except Exception: body = response.text raise RuntimeError( f"[{action}] HTTP {response.status_code}: {body}" ) return response.json() def log(message: str, level: str = "INFO") -> None: """Print a timestamped log line.""" ts = datetime.utcnow().strftime("%H:%M:%S") print(f"[{ts}] [{level}] {message}") def pause(seconds: float = 0.5) -> None: """Polite delay between API calls to avoid rate limiting.""" time.sleep(seconds) # --------------------------------------------------------------------------- # Step 1 — Custom Fields # --------------------------------------------------------------------------- CUSTOM_FIELDS = [ { "name": "google_review_link", "dataType": "TEXT", "placeholder": "https://g.page/r/...", "isRequired": False, "fieldKey": "contact.google_review_link", "model": "contact", }, { "name": "technician_name", "dataType": "TEXT", "placeholder": "e.g. Jake", "isRequired": False, "fieldKey": "contact.technician_name", "model": "contact", }, { "name": "job_type", "dataType": "TEXT", "placeholder": "e.g. Hot water replacement", "isRequired": False, "fieldKey": "contact.job_type", "model": "contact", }, { "name": "review_rating", "dataType": "NUMERICAL", "placeholder": "1-5", "isRequired": False, "fieldKey": "contact.review_rating", "model": "contact", }, ] def deploy_custom_fields(location_id: str, api_key: str) -> list[dict]: """ Create all 4 custom fields in the GHL sub-account. Returns a list of {name, id} dicts for the created fields. """ url = f"{GHL_BASE_URL}/locations/{location_id}/customFields" headers = get_headers(api_key) created = [] log(f"Creating {len(CUSTOM_FIELDS)} custom fields...") for field in CUSTOM_FIELDS: payload = { "name": field["name"], "dataType": field["dataType"], "placeholder": field["placeholder"], "isRequired": field["isRequired"], "model": field["model"], } try: response = requests.post(url, headers=headers, json=payload) data = handle_response(response, f"Create custom field: {field['name']}") field_id = data.get("customField", {}).get("id") or data.get("id", "unknown") created.append({"name": field["name"], "id": field_id}) log(f" Created field: {field['name']} (id: {field_id})") except RuntimeError as err: # Field may already exist — treat as non-fatal if "already exists" in str(err).lower() or "duplicate" in str(err).lower(): log(f" Field '{field['name']}' already exists — skipping.", "WARN") created.append({"name": field["name"], "id": "pre-existing"}) else: raise pause(0.3) return created # --------------------------------------------------------------------------- # Step 2 — Pipeline # --------------------------------------------------------------------------- PIPELINE_STAGES = [ {"name": "Review Requested", "probability": 50}, {"name": "Negative Feedback", "probability": 0}, {"name": "Review Posted", "probability": 100}, ] def deploy_pipeline(location_id: str, api_key: str) -> str: """ Create the Reputation Tracker pipeline with 3 stages. Returns the pipeline ID. """ url = f"{GHL_BASE_URL}/opportunities/pipelines" headers = get_headers(api_key) payload = { "name": "Reputation Tracker", "locationId": location_id, "stages": [ {"name": stage["name"], "probability": stage["probability"]} for stage in PIPELINE_STAGES ], } log("Creating pipeline: Reputation Tracker...") try: response = requests.post(url, headers=headers, json=payload) data = handle_response(response, "Create pipeline") pipeline_id = ( data.get("pipeline", {}).get("id") or data.get("id", "unknown") ) log(f" Pipeline created (id: {pipeline_id})") for stage in PIPELINE_STAGES: log(f" Stage: {stage['name']}") return pipeline_id except RuntimeError as err: if "already exists" in str(err).lower() or "duplicate" in str(err).lower(): log(" Pipeline 'Reputation Tracker' already exists — skipping.", "WARN") return "pre-existing" raise # --------------------------------------------------------------------------- # Step 3 — SMS Templates # --------------------------------------------------------------------------- def build_review_request_template(business_name: str, rating_page_placeholder: str) -> dict: """Build the initial review request SMS template payload.""" body = ( f"Hi {{{{contact.first_name}}}}, it's {{{{contact.technician_name}}}} " f"from {business_name}. Thanks for having us today for your " f"{{{{contact.job_type}}}}.\n\n" f"We'd love to know how we did — it takes 10 seconds: {rating_page_placeholder}\n\n" f"Thanks, {{{{contact.technician_name}}}}" ) return { "name": "RR - Review Request", "type": "SMS", "body": body, } def build_followup_template(business_name: str) -> dict: """Build the 48-hour follow-up SMS template payload.""" body = ( f"Hi {{{{contact.first_name}}}}, just a quick note from {business_name}.\n\n" f"Your review means the world to us and helps other families find a " f"tradesperson they can trust. If you have 30 seconds: " f"{{{{contact.google_review_link}}}}\n\n" f"Thank you sincerely." ) return { "name": "RR - Review Follow-Up", "type": "SMS", "body": body, } def deploy_sms_templates( location_id: str, api_key: str, business_name: str, rating_page_url: str = "[RATING_PAGE_URL — UPDATE AFTER FUNNEL IS PUBLISHED]", ) -> list[dict]: """ Create both SMS templates in the GHL sub-account. Returns a list of {name, id} dicts for the created templates. """ url = f"{GHL_BASE_URL}/locations/{location_id}/templates" headers = get_headers(api_key) templates = [ build_review_request_template(business_name, rating_page_url), build_followup_template(business_name), ] created = [] log(f"Creating {len(templates)} SMS templates...") for template in templates: try: response = requests.post(url, headers=headers, json=template) data = handle_response(response, f"Create SMS template: {template['name']}") template_id = ( data.get("template", {}).get("id") or data.get("id", "unknown") ) created.append({"name": template["name"], "id": template_id}) log(f" Created template: {template['name']} (id: {template_id})") except RuntimeError as err: if "already exists" in str(err).lower() or "duplicate" in str(err).lower(): log(f" Template '{template['name']}' already exists — skipping.", "WARN") created.append({"name": template["name"], "id": "pre-existing"}) else: raise pause(0.3) return created # --------------------------------------------------------------------------- # Step 4 — Pre-populate Contact Custom Field Defaults # --------------------------------------------------------------------------- def set_location_defaults( location_id: str, api_key: str, google_review_link: str, business_name: str, ) -> None: """ Patch the location settings to store the Google Review link and business name at the sub-account level. These serve as fallback defaults if individual contact fields are not populated at trigger time. Note: GHL does not have a direct "default custom field value" API endpoint. This function stores the values in the location's custom values (snippets) which can be referenced in workflow actions. """ url = f"{GHL_BASE_URL}/locations/{location_id}/customValues" headers = get_headers(api_key) custom_values = [ { "name": "Default Google Review Link", "value": google_review_link, }, { "name": "Business Name", "value": business_name, }, ] log("Setting sub-account custom values (location defaults)...") for cv in custom_values: try: response = requests.post(url, headers=headers, json=cv) data = handle_response(response, f"Set custom value: {cv['name']}") cv_id = data.get("customValue", {}).get("id") or data.get("id", "unknown") log(f" Set: {cv['name']} = {cv['value'][:50]}... (id: {cv_id})") except RuntimeError as err: # Non-fatal — these are convenience defaults log(f" Could not set custom value '{cv['name']}': {err}", "WARN") pause(0.3) # --------------------------------------------------------------------------- # Deployment Summary # --------------------------------------------------------------------------- def print_summary( location_id: str, business_name: str, google_review_link: str, custom_fields: list[dict], pipeline_id: str, templates: list[dict], ) -> None: """Print a formatted deployment summary.""" separator = "-" * 60 print(f"\n{separator}") print(" REPUTATION ROCKET — DEPLOYMENT SUMMARY") print(separator) print(f" Sub-Account ID : {location_id}") print(f" Business Name : {business_name}") print(f" Google Review : {google_review_link}") print(f" Deployed At : {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC") print(separator) print("\n CUSTOM FIELDS") for field in custom_fields: status = "OK" if field["id"] != "pre-existing" else "EXISTS" print(f" [{status}] {field['name']} (id: {field['id']})") print("\n PIPELINE") status = "OK" if pipeline_id != "pre-existing" else "EXISTS" print(f" [{status}] Reputation Tracker (id: {pipeline_id})") print("\n SMS TEMPLATES") for template in templates: status = "OK" if template["id"] != "pre-existing" else "EXISTS" print(f" [{status}] {template['name']} (id: {template['id']})") print(f"\n{separator}") print(" NEXT STEPS") print(separator) print(" 1. Import the Reputation Rocket GHL snapshot into this sub-account.") print(" (GHL UI: Settings > Snapshots > Import Snapshot)") print("") print(" 2. Build or publish the Rating Landing Page funnel.") print(" (GHL UI: Sites > Funnels > Reputation Rocket — Rating Page)") print("") print(" 3. Update the SMS template 'RR - Review Request' with the") print(" published funnel URL (replace [RATING_PAGE_URL] placeholder).") print("") print(" 4. Set the internal notification email in Workflow 2 (5-Star Filter)") print(" to the client's email address.") print("") print(" 5. Activate all 3 workflows in the sub-account.") print("") print(" 6. Run an end-to-end test (see README.md Step 5).") print(separator) print("") # --------------------------------------------------------------------------- # Main Entry Point # --------------------------------------------------------------------------- def main() -> None: parser = argparse.ArgumentParser( description="Deploy the Reputation Rocket module to a GHL sub-account." ) parser.add_argument( "--sub-account-id", required=True, help="GHL location/sub-account ID.", ) parser.add_argument( "--google-review-link", required=True, help="Client's Google Business Profile review link (https://g.page/r/...).", ) parser.add_argument( "--business-name", required=True, help="Client's trading name as it should appear in SMS messages.", ) parser.add_argument( "--rating-page-url", default="[RATING_PAGE_URL — UPDATE AFTER FUNNEL IS PUBLISHED]", help=( "Published GHL funnel URL for the rating page. " "Can be updated later in the SMS template." ), ) parser.add_argument( "--skip-custom-values", action="store_true", help="Skip setting sub-account custom values (location defaults).", ) args = parser.parse_args() # Retrieve API key from environment api_key = os.environ.get("GHL_API_KEY") if not api_key: print( "[ERROR] Environment variable GHL_API_KEY is not set.\n" " Export it before running: set GHL_API_KEY=your_key_here", file=sys.stderr, ) sys.exit(1) location_id = args.sub_account_id google_review_link = args.google_review_link business_name = args.business_name rating_page_url = args.rating_page_url log("Starting Reputation Rocket deployment...") log(f"Sub-account: {location_id}") log(f"Business: {business_name}") try: # Step 1 — Custom fields custom_fields = deploy_custom_fields(location_id, api_key) # Step 2 — Pipeline pipeline_id = deploy_pipeline(location_id, api_key) # Step 3 — SMS templates templates = deploy_sms_templates( location_id, api_key, business_name, rating_page_url ) # Step 4 — Location defaults (non-fatal) if not args.skip_custom_values: set_location_defaults( location_id, api_key, google_review_link, business_name ) except RuntimeError as err: log(f"Deployment failed: {err}", "ERROR") sys.exit(1) except requests.ConnectionError as err: log(f"Network error — check your internet connection: {err}", "ERROR") sys.exit(1) except requests.Timeout: log("Request timed out — GHL API may be slow. Retry in a few minutes.", "ERROR") sys.exit(1) # Print summary print_summary( location_id=location_id, business_name=business_name, google_review_link=google_review_link, custom_fields=custom_fields, pipeline_id=pipeline_id, templates=templates, ) if __name__ == "__main__": main()