#!/usr/bin/env python3 """ Instantly.ai Test Pipeline ========================== Tests the full Instantly.ai pipeline for Kinan's customer journey test. Steps: 1. Verify API key works (list campaigns) 2. List sending accounts 3. Create a NEW test campaign with email sequence 4. Assign sending accounts to campaign 5. Add Kinan as a test lead with custom variables 6. Report results Created: 2026-02-17 """ import json import sys import time import traceback sys.path.insert(0, "/mnt/e/genesis-system") from core.instantly_client import InstantlyClient, InstantlyAPIError API_KEY = "MjBjODUxNGYtNjA5MC00NjY4LWFhY2UtOWZmYTE3NDhhMmQ1OnRTYkpqRU94RHJveg==" EXISTING_CAMPAIGN_ID = "cf21ba92-4c8c-443c-b67e-4b6530024118" # The report URL - will be set after PDF upload REPORT_URL = "https://api.sunaivadigital.com/reports/Trinity_Landscape_Cairns.pdf" def retry_call(func, *args, max_retries=3, delay=2, **kwargs): """Retry API calls on transient failures.""" for attempt in range(max_retries): try: return func(*args, **kwargs) except InstantlyAPIError as e: if e.status_code == 0 and attempt < max_retries - 1: print(f" [RETRY {attempt+1}/{max_retries}] Connection error, waiting {delay}s...") time.sleep(delay) delay *= 2 else: raise return None def main(): print("=" * 70) print("INSTANTLY.AI TEST PIPELINE") print("=" * 70) print() client = InstantlyClient(api_key=API_KEY) results = {} # ========================================================================= # STEP 1: Verify API key - List campaigns # ========================================================================= print("[STEP 1] Verifying API key - listing campaigns...") try: campaigns = retry_call(client.list_campaigns, limit=20) campaign_list = campaigns.get("items", []) print(f" SUCCESS: API key is valid. Found {len(campaign_list)} campaign(s).") for c in campaign_list: print(f" - {c['name']} (ID: {c['id']}, status: {c.get('status', 'unknown')})") results["api_key_works"] = True results["existing_campaigns"] = len(campaign_list) except Exception as e: print(f" FAILED: {e}") results["api_key_works"] = False return results # ========================================================================= # STEP 2: List sending accounts # ========================================================================= print() print("[STEP 2] Listing sending accounts...") try: accounts = retry_call(client.list_accounts, limit=25) account_list = accounts.get("items", []) print(f" SUCCESS: Found {len(account_list)} sending account(s).") # Group by domain domains = {} for a in account_list: email = a.get("email", "unknown") domain = email.split("@")[1] if "@" in email else "unknown" warmup = a.get("stat_warmup_score", 0) status = a.get("status", "unknown") if domain not in domains: domains[domain] = [] domains[domain].append({"email": email, "warmup": warmup, "status": status}) for domain, accs in domains.items(): print(f" Domain: {domain} ({len(accs)} accounts)") for a in accs: print(f" - {a['email']} (warmup: {a['warmup']}%, status: {a['status']})") results["sending_accounts"] = len(account_list) results["account_emails"] = [a["email"] for a in account_list] except Exception as e: print(f" FAILED: {e}") results["sending_accounts"] = 0 # ========================================================================= # STEP 3: Create NEW test campaign with email sequence # ========================================================================= print() print("[STEP 3] Creating new test campaign 'ReceptionistAI Heatmap - Test Run'...") # Build the email sequence with the heatmap email sequences = [ { "steps": [ { "type": "email", "delay": 0, "variants": [ { "subject": "Your website scores {{ai_score}}/100 - here's your free AI audit", "body": ( "Hi {{first_name}},\n\n" "I ran a quick AI audit on {{company_name}}'s website and wanted to share the results.\n\n" "Your Business AI Score: {{ai_score}}/100\n\n" "Here's what stood out:\n" "- Website Speed: {{website_speed}}/100 (53% of mobile visitors leave after 3 seconds)\n" "- Mobile Experience: {{mobile_score}}/100\n" "- SEO Basics: {{seo_score}}/100\n\n" "I've put together a detailed report with recommendations:\n" "{{report_url}}\n\n" "Want to hear what an AI receptionist would sound like for {{company_name}}?\n" "Call this number: {{demo_phone}}\n\n" "It'll greet you as if it's YOUR receptionist. Takes 30 seconds.\n\n" "Cheers,\n" "The ReceptionistAI Team\n" "receptionistai.au" ) } ] } ] } ] # Campaign schedule - AEST business hours, Mon-Fri campaign_schedule = { "schedules": [ { "name": "AEST Business Hours", "timing": { "from": "06:00", "to": "18:00" }, "days": { "0": True, "1": True, "2": True, "3": True, "4": True, "5": True, "6": True }, "timezone": "Australia/Brisbane" } ] } try: campaign = retry_call( client.create_campaign, name="ReceptionistAI Heatmap - Test Run", sequences=sequences, campaign_schedule=campaign_schedule, ) campaign_id = campaign.get("id", "") print(f" SUCCESS: Campaign created!") print(f" Campaign ID: {campaign_id}") print(f" Name: {campaign.get('name', 'unknown')}") results["campaign_created"] = True results["campaign_id"] = campaign_id except InstantlyAPIError as e: print(f" FAILED: {e}") print(f" Response body: {e.response_body}") results["campaign_created"] = False results["campaign_error"] = str(e) # Try to continue with existing campaign as fallback campaign_id = None # If campaign creation failed, let's try to see if it's a limitation # and just add the lead to the existing campaign instead print() print(" [FALLBACK] Will try to use existing campaign or add lead without campaign...") # ========================================================================= # STEP 4: Assign sending accounts to campaign # ========================================================================= if results.get("campaign_created") and campaign_id: print() print("[STEP 4] Assigning sending accounts to campaign...") try: # Get account IDs - we'll use a few from different domains for diversity accounts_data = retry_call(client.list_accounts, limit=25) all_accounts = accounts_data.get("items", []) # Pick 3 accounts from different domains used_domains = set() selected_accounts = [] for a in all_accounts: email = a.get("email", "") domain = email.split("@")[1] if "@" in email else "" if domain not in used_domains and len(selected_accounts) < 3: selected_accounts.append(email) used_domains.add(domain) if selected_accounts: print(f" Selected accounts: {selected_accounts}") try: mapping = retry_call( client.assign_accounts_to_campaign, campaign_id=campaign_id, account_ids=selected_accounts, ) print(f" Account assignment result: {json.dumps(mapping, indent=2)}") results["accounts_assigned"] = True except InstantlyAPIError as e: print(f" Account assignment failed: {e}") print(f" Response: {e.response_body}") results["accounts_assigned"] = False results["account_assign_error"] = str(e) else: print(" No accounts available to assign") results["accounts_assigned"] = False except Exception as e: print(f" FAILED: {e}") results["accounts_assigned"] = False else: print() print("[STEP 4] SKIPPED (no campaign created)") results["accounts_assigned"] = False # ========================================================================= # STEP 5: Add Kinan as a test lead # ========================================================================= print() target_campaign = campaign_id if results.get("campaign_created") else EXISTING_CAMPAIGN_ID print(f"[STEP 5] Adding Kinan as a lead to campaign {target_campaign}...") try: lead = retry_call( client.create_lead, email="kinan@protonmail.com", campaign_id=target_campaign, first_name="Kinan", last_name="Test", company_name="Trinity Landscape Cairns", website="https://www.trinitylandscapecairns.com.au/", custom_variables={ "ai_score": "48", "website_speed": "30", "mobile_score": "30", "seo_score": "85", "report_url": REPORT_URL, "demo_phone": "+61 7 3130 4377", }, ) lead_id = lead.get("id", "") print(f" SUCCESS: Lead created!") print(f" Lead ID: {lead_id}") print(f" Email: {lead.get('email', 'unknown')}") print(f" Full response: {json.dumps(lead, indent=2)}") results["lead_created"] = True results["lead_id"] = lead_id except InstantlyAPIError as e: print(f" FAILED: {e}") print(f" Response body: {e.response_body}") results["lead_created"] = False results["lead_error"] = str(e) # ========================================================================= # STEP 6: Get campaign status (don't activate yet) # ========================================================================= if results.get("campaign_created") and campaign_id: print() print("[STEP 6] Getting campaign details (pre-activation check)...") try: details = retry_call(client.get_campaign, campaign_id) print(f" Campaign status: {details.get('status', 'unknown')}") print(f" Sequences: {json.dumps(details.get('sequences', []), indent=2)[:500]}") results["campaign_status"] = details.get("status") except Exception as e: print(f" FAILED to get details: {e}") # ========================================================================= # SUMMARY # ========================================================================= print() print("=" * 70) print("PIPELINE RESULTS SUMMARY") print("=" * 70) print(f" 1. API Key Works: {results.get('api_key_works', False)}") print(f" 2. Sending Accounts: {results.get('sending_accounts', 0)}") print(f" 3. Campaign Created: {results.get('campaign_created', False)}") if results.get("campaign_id"): print(f" Campaign ID: {results['campaign_id']}") if results.get("campaign_error"): print(f" Campaign Error: {results['campaign_error'][:100]}") print(f" 4. Accounts Assigned: {results.get('accounts_assigned', False)}") print(f" 5. Lead Created: {results.get('lead_created', False)}") if results.get("lead_id"): print(f" Lead ID: {results['lead_id']}") if results.get("lead_error"): print(f" Lead Error: {results['lead_error'][:100]}") print(f" 6. PDF URL: {REPORT_URL}") print() # Activation note if results.get("campaign_created"): print("NOTE: Campaign is created but NOT activated.") print("To activate, run:") print(f" python3 -c \"") print(f" from core.instantly_client import InstantlyClient") print(f" c = InstantlyClient(api_key='{API_KEY}')") print(f" print(c.activate_campaign('{results.get('campaign_id', '')}'))\"") print() return results if __name__ == "__main__": results = main()