LeadGenius → Salesforce Import Guide

How to export leads from LeadGenius and import them into a Salesforce org, with custom fields for all AI insights and enrichment data.

Table of Contents

0. Overview
1. Prerequisites
2. Creating the Salesforce Connected App
3. Exporting from LeadGenius
4. LeadGenius CSV Field Reference
5. Creating Custom Fields in Salesforce
6. Deploying Field-Level Security
7. Importing Leads
8. Populating Custom Fields (Update Pass)
9. Creating the List View
10. Verification
11. Complete Automated Script
12. Troubleshooting

0. Overview

What This Does

LeadGenius CSV  →  Salesforce Leads
 120 columns        with custom AI fields
                    + List View
                    + AI Insights visible

Architecture

LeadGenius exports a CSV with up to 120 columns per lead. Most Salesforce orgs have a handful of standard Lead fields. This guide bridges the gap by:

1. Creating 17 custom fields on the Lead object to hold AI and enrichment data
2. Granting field-level security so the API can read/write them
3. Importing leads with standard field mapping (name, email, company, etc.)
4. Updating leads with AI data in dedicated custom fields
5. Creating a List View that surfaces AI Score, Qualification, and Seniority

API Strategy Summary

1. Prerequisites

Software

pip3 install requests python-dotenv

Salesforce Org

• An active Salesforce org (Developer, sandbox, or production)
• Admin-level user account
• A Connected App configured (see Section 2)

LeadGenius

• An active LeadGenius account (https://last.leadgenius.app)
• Leads exported to CSV from the LeadGenius UI

Environment File

Create .env in your working directory:

# Salesforce Connected App credentials
SALESFORCE_CONSUMER_KEY=your_consumer_key_here
SALESFORCE_CONSUMER_SECRET=your_consumer_secret_here

# Salesforce user credentials
SALESFORCE_USERNAME=your-user@salesforce.com
SALESFORCE_PASSWORD=YourPassword123
SALESFORCE_SECURITY_TOKEN=YourSecurityToken

# Salesforce org URL
SALESFORCE_INSTANCE_URL=https://your-org.develop.my.salesforce.com

2. Creating the Salesforce Connected App

A Connected App provides the Consumer Key / Consumer Secret needed for API authentication. This is a one-time setup per Salesforce org.

2.1 Navigate to App Manager

1. Log in to Salesforce
2. Click ⚙️ gear (top right) → Setup
3. Quick Find → type App Manager
4. Click App Manager

🇫🇷 French UI: Gear opens “Configuration”. Search “Gestionnaire d’applications”.

2.2 Create a New Connected App

1. Click New Connected App (top right)
2. Fill in Basic Information:

2.3 Configure OAuth Settings

1. Check ✅ Enable OAuth Settings
2. Set Callback URL: https://login.salesforce.com/services/oauth2/callback
3. Add these OAuth Scopes:
   • Full access (full)
   • Perform requests at any time (refresh_token, offline_access)
   • Access and manage your data (api)

🇫🇷 French: “Accès complet (full)”, “Effectuer des requêtes à tout moment”, “Accéder à vos données et les gérer”

4. Click Save → Click Continue (wait message)

2.4 Get Consumer Key and Consumer Secret

1. On the Connected App detail page, click Manage Consumer Details
2. You may need to verify your identity (email code)
3. Copy and save:
   • Consumer Key → SALESFORCE_CONSUMER_KEY
   • Consumer Secret → SALESFORCE_CONSUMER_SECRET

⚠️ Save these immediately — Consumer Secret is only shown once.

2.5 Get Your Security Token

1. Click your avatar (top right) → Settings
2. Left sidebar → My Personal Information → Reset My Security Token
3. Click Reset Security Token
4. Check your email → copy the token → SALESFORCE_SECURITY_TOKEN

🇫🇷 French: “Paramètres” → “Mes informations personnelles” → “Réinitialiser mon jeton de sécurité”

⚠️ Every time you change your Salesforce password, the security token is also invalidated.

2.6 Enable Password Grant (If Needed)

Some orgs block the OAuth Password Grant by default:

1. Setup → Quick Find: OAuth and OpenID Connect Settings
2. Enable “Allow OAuth Username-Password Flows”

🇫🇷 “Paramètres OAuth et OpenID Connect” → “Autoriser les flux nom d’utilisateur-mot de passe OAuth”

If disabled, you’ll get: {"error": "unsupported_grant_type"}

2.7 Relax IP Restrictions (If Needed)

If you get INVALID_LOGIN with correct credentials:

1. Setup → App Manager → Find leadgenius → ▼ → Manage
2. Click Edit Policies
3. Under IP Relaxation: select “Relax IP restrictions”
4. Save

2.8 Verify Connection

import requests, os
from dotenv import load_dotenv
load_dotenv()

data = {
    'grant_type': 'password',
    'client_id': os.getenv('SALESFORCE_CONSUMER_KEY'),
    'client_secret': os.getenv('SALESFORCE_CONSUMER_SECRET'),
    'username': os.getenv('SALESFORCE_USERNAME'),
    'password': os.getenv('SALESFORCE_PASSWORD') + os.getenv('SALESFORCE_SECURITY_TOKEN'),
}
r = requests.post('https://login.salesforce.com/services/oauth2/token', data=data)
result = r.json()

if 'access_token' in result:
    print(f"✅ Connected to {result['instance_url']}")
else:
    print(f"❌ Auth failed: {result}")

Troubleshooting

3. Exporting from LeadGenius

From the LeadGenius UI

1. Go to https://last.leadgenius.app
2. Navigate to your Client workspace
3. Select the leads you want to export (or select all)
4. Click Export → CSV
5. Save the CSV file to your working directory

From the LeadGenius API

# Authenticate
python3 scripts/auth.py --email your@email.com

# Export all leads for a client
python3 scripts/lgp.py leads list --client-id your-client-slug --limit 5000 --format csv > leads.csv

CSV Filename Convention

Name your file descriptively:

client-name-MMYYYY.csv
# Example: prophix-demo-02026.csv

4. LeadGenius CSV Field Reference

The LeadGenius CSV contains up to 120 columns. Here are the ones we map to Salesforce:

Standard Lead Fields (→ Salesforce Standard Fields)

⚠️ Fields to SKIP

AI Insight Fields (→ Salesforce Custom Fields)

Enrichment Fields (→ Salesforce Custom Fields)

Fields We Put in Description (Too Large for Individual Fields)

Some LeadGenius fields contain structured multi-paragraph content. We aggregate these into the standard Description field:

5. Creating Custom Fields in Salesforce

The 17 Custom Fields

All field API names are prefixed with LG_ (LeadGenius) to avoid conflicts with other integrations.

TOOLING = f'{instance_url}/services/data/v62.0/tooling'

LEADGENIUS_FIELDS = [
    # AI Scores
    {"FullName": "Lead.LG_AI_Score__c", "Metadata": {
        "label": "LG AI Score", "type": "Number", "precision": 5, "scale": 0}},
    {"FullName": "Lead.LG_Lead_Score__c", "Metadata": {
        "label": "LG Lead Score", "type": "Number", "precision": 5, "scale": 0}},
    
    # AI Text Analysis
    {"FullName": "Lead.LG_Qualification__c", "Metadata": {
        "label": "LG Qualification", "type": "Text", "length": 255}},
    {"FullName": "Lead.LG_Justification__c", "Metadata": {
        "label": "LG Justification", "type": "LongTextArea", "length": 32000, "visibleLines": 5}},
    {"FullName": "Lead.LG_Next_Action__c", "Metadata": {
        "label": "LG Next Action", "type": "LongTextArea", "length": 32000, "visibleLines": 3}},
    {"FullName": "Lead.LG_Cold_Email__c", "Metadata": {
        "label": "LG Cold Email", "type": "LongTextArea", "length": 32000, "visibleLines": 5}},
    {"FullName": "Lead.LG_LinkedIn_Connect__c", "Metadata": {
        "label": "LG LinkedIn Connect", "type": "LongTextArea", "length": 32000, "visibleLines": 3}},
    {"FullName": "Lead.LG_Decision_Role__c", "Metadata": {
        "label": "LG Decision Maker Role", "type": "Text", "length": 255}},
    {"FullName": "Lead.LG_Likely_Engage__c", "Metadata": {
        "label": "LG Likely to Engage", "type": "Text", "length": 10}},
    
    # LinkedIn / Enrichment
    {"FullName": "Lead.LG_LinkedIn_URL__c", "Metadata": {
        "label": "LG LinkedIn URL", "type": "Url"}},
    {"FullName": "Lead.LG_Company_LinkedIn__c", "Metadata": {
        "label": "LG Company LinkedIn", "type": "Url"}},
    {"FullName": "Lead.LG_Seniority__c", "Metadata": {
        "label": "LG Seniority", "type": "Text", "length": 100}},
    {"FullName": "Lead.LG_Departments__c", "Metadata": {
        "label": "LG Departments", "type": "Text", "length": 255}},
    {"FullName": "Lead.LG_Engagement_Rate__c", "Metadata": {
        "label": "LG Engagement Rate", "type": "Number", "precision": 10, "scale": 1}},
    {"FullName": "Lead.LG_Total_Engagements__c", "Metadata": {
        "label": "LG Total Engagements", "type": "Number", "precision": 10, "scale": 0}},
    
    # Tracking
    {"FullName": "Lead.LG_Lead_ID__c", "Metadata": {
        "label": "LG Lead ID", "type": "Text", "length": 255,
        "unique": True, "externalId": True}},
    {"FullName": "Lead.LG_Enrichment_Source__c", "Metadata": {
        "label": "LG Enrichment Source", "type": "Text", "length": 255}},
]

How to Create Them

import requests, os, time
from dotenv import load_dotenv
load_dotenv()

# Authenticate (see Section 2.8)
token, url = authenticate()
headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}
TOOLING = f'{url}/services/data/v62.0/tooling'

created = 0
skipped = 0

for fd in LEADGENIUS_FIELDS:
    fname = fd['FullName'].split('.')[1]
    r = requests.post(f'{TOOLING}/sobjects/CustomField', headers=headers, json=fd)
    
    if r.status_code == 201:
        created += 1
        print(f"  ✅ {fname} created")
    elif r.status_code == 400 and 'DUPLICATE' in r.text:
        skipped += 1
        print(f"  ⏭️  {fname} already exists")
    else:
        print(f"  ❌ {fname}: {r.text[:100]}")
    
    time.sleep(0.5)  # Avoid rate limits

print(f"\n✅ Created: {created} | ⏭️ Skipped: {skipped}")

⚠️ Critical: Fields Are NOT Usable Yet

After creation via Tooling API, the fields exist in Salesforce metadata but are invisible to the REST Data API. You MUST deploy Field-Level Security (Section 6) before importing data.

How to verify:

r = requests.get(f'{url}/services/data/v62.0/sobjects/Lead/describe', headers=headers)
visible = [f['name'] for f in r.json()['fields'] if f['name'].startswith('LG_')]
print(f"Visible LG fields: {len(visible)}")
# If 0 → FLS not deployed yet (expected at this stage)

6. Deploying Field-Level Security

This step grants the Admin profile read/write access to all LeadGenius custom fields.

Why This Is Needed

The Tooling API creates fields in metadata, but they’re gated by Field-Level Security (FLS). Without FLS, the REST API returns No such column when you try to use them.

Deploy FLS via Metadata SOAP API

import requests, os, base64, zipfile, io, time
import xml.etree.ElementTree as ET
from dotenv import load_dotenv
load_dotenv()

token, url = authenticate()
metadata_url = f'{url}/services/Soap/m/62.0'

# All LG_ field names
lg_fields = [
    'LG_AI_Score__c', 'LG_Lead_Score__c', 'LG_Qualification__c',
    'LG_Justification__c', 'LG_Next_Action__c', 'LG_Cold_Email__c',
    'LG_LinkedIn_Connect__c', 'LG_Decision_Role__c', 'LG_Likely_Engage__c',
    'LG_LinkedIn_URL__c', 'LG_Company_LinkedIn__c', 'LG_Seniority__c',
    'LG_Departments__c', 'LG_Engagement_Rate__c', 'LG_Total_Engagements__c',
    'LG_Lead_ID__c', 'LG_Enrichment_Source__c',
]

# Build Profile XML
perms = ''.join([f'''
    <fieldPermissions>
        <field>Lead.{f}</field>
        <editable>true</editable>
        <readable>true</readable>
    </fieldPermissions>''' for f in lg_fields])

profile_xml = f'''<?xml version="1.0" encoding="UTF-8"?>
<Profile xmlns="http://soap.sforce.com/2006/04/metadata">{perms}
</Profile>'''

package_xml = '''<?xml version="1.0" encoding="UTF-8"?>
<Package xmlns="http://soap.sforce.com/2006/04/metadata">
    <types><members>Admin</members><name>Profile</name></types>
    <version>62.0</version>
</Package>'''

# ZIP it
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
    zf.writestr('package.xml', package_xml)
    zf.writestr('profiles/Admin.profile', profile_xml)
zip_data = base64.b64encode(buf.getvalue()).decode()

# Deploy via SOAP
soap = f'''<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
               xmlns:met="http://soap.sforce.com/2006/04/metadata">
    <soap:Header>
        <met:SessionHeader>
            <met:sessionId>{token}</met:sessionId>
        </met:SessionHeader>
    </soap:Header>
    <soap:Body>
        <met:deploy>
            <met:ZipFile>{zip_data}</met:ZipFile>
            <met:DeployOptions>
                <met:checkOnly>false</met:checkOnly>
                <met:ignoreWarnings>true</met:ignoreWarnings>
                <met:rollbackOnError>false</met:rollbackOnError>
                <met:singlePackage>true</met:singlePackage>
            </met:DeployOptions>
        </met:deploy>
    </soap:Body>
</soap:Envelope>'''

r = requests.post(metadata_url, data=soap.encode(),
    headers={'Content-Type': 'text/xml; charset=utf-8', 'SOAPAction': 'deploy'})

# Extract deploy ID and poll (see helper function below)
deploy_id = extract_deploy_id(r.text)
success = poll_deploy_status(deploy_id, token, metadata_url)

Helper Functions

def extract_deploy_id(response_text):
    """Extract the 18-char deploy ID from SOAP response"""
    for elem in ET.fromstring(response_text).iter():
        if 'id' in elem.tag.lower() and elem.text and len(elem.text) == 18:
            return elem.text
    return None

def poll_deploy_status(deploy_id, token, metadata_url, max_attempts=20):
    """Poll until deploy completes. Returns True on success."""
    for attempt in range(max_attempts):
        time.sleep(3)
        check = f'''<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
               xmlns:met="http://soap.sforce.com/2006/04/metadata">
    <soap:Header><met:SessionHeader>
        <met:sessionId>{token}</met:sessionId>
    </met:SessionHeader></soap:Header>
    <soap:Body><met:checkDeployStatus>
        <met:asyncProcessId>{deploy_id}</met:asyncProcessId>
        <met:includeDetails>true</met:includeDetails>
    </met:checkDeployStatus></soap:Body>
</soap:Envelope>'''
        r = requests.post(metadata_url, data=check.encode(),
            headers={'Content-Type': 'text/xml; charset=utf-8',
                     'SOAPAction': 'checkDeployStatus'})
        done = ok = False
        for elem in ET.fromstring(r.text).iter():
            tag = elem.tag.split('}')[-1] if '}' in elem.tag else elem.tag
            if tag == 'done' and elem.text == 'true': done = True
            if tag == 'success' and elem.text == 'true': ok = True
        if done:
            print(f"  {'✅ FLS deployed!' if ok else '❌ FLS deploy failed'}")
            return ok
        print(f"  ⏳ Attempt {attempt+1}...")
    return False

Verify FLS Is Working

After deployment, wait 3-5 seconds, then verify:

time.sleep(5)
r = requests.get(f'{url}/services/data/v62.0/sobjects/Lead/describe', headers=headers)
visible = [f['name'] for f in r.json()['fields'] if f['name'].startswith('LG_')]
print(f"✅ Visible LG fields: {len(visible)}/17")
for f in sorted(visible):
    print(f"  ✅ {f}")

Expected: 17/17 fields visible. If 0/17, the deploy failed or hasn’t propagated yet — wait and retry.

7. Importing Leads

Field Mapping

import csv

def safe_str(val, mx=255):
    """Clean string for Salesforce"""
    if not val or str(val).strip() in ('N/A', '(empty)', 'None', ''):
        return None
    return str(val).strip()[:mx] or None

def safe_long(val, mx=32000):
    """Clean long text for Salesforce"""
    if not val or str(val).strip() in ('N/A', '(empty)', 'None', ''):
        return None
    return str(val).strip()[:mx] or None

def safe_num(val):
    """Clean number for Salesforce"""
    if not val or str(val).strip() in ('N/A', '(empty)', 'None', ''):
        return None
    try:
        return float(str(val).strip())
    except:
        return None

def build_lead(row, campaign_name, available_fields):
    """Build a Salesforce Lead record from a LeadGenius CSV row"""
    
    lead = {
        # Standard fields
        'FirstName': safe_str(row.get('First Name'), 40),
        'LastName': safe_str(row.get('Last Name'), 80) or 'Unknown',
        'Title': safe_str(row.get('Title')),
        'Email': safe_str(row.get('Email')),
        'Phone': safe_str(row.get('Phone Number')),
        'Company': safe_str(row.get('Company Name')) or 'Unknown',
        'Industry': safe_str(row.get('Industry')),
        'NumberOfEmployees': int(safe_num(row.get('Estimated Num Employees')) or 0) or None,
        'Country': safe_str(row.get('Country')),
        'City': safe_str(row.get('City')),
        'LeadSource': campaign_name,
        'Status': 'Open - Not Contacted',
        # ⚠️ NO State field — French regions cause validation errors
    }
    
    # Custom LG_ fields (only if FLS is deployed)
    lg_mapping = {
        'LG_AI_Score__c': safe_num(row.get('Ai Score Value')),
        'LG_Lead_Score__c': safe_num(row.get('Ai Lead Score Score')),
        'LG_Qualification__c': safe_str(row.get('Ai Qualification')),
        'LG_Justification__c': safe_long(row.get('Ai Score Justification')),
        'LG_Next_Action__c': safe_long(row.get('Ai Next Action')),
        'LG_Cold_Email__c': safe_long(row.get('Ai Cold Email')),
        'LG_LinkedIn_Connect__c': safe_long(row.get('Ai Linkedin Connect')),
        'LG_Decision_Role__c': safe_str(row.get('Ai Decision Maker Role')),
        'LG_Likely_Engage__c': safe_str(row.get('Is Likely To Engage'), 10),
        'LG_LinkedIn_URL__c': safe_str(row.get('Linkedin Url')),
        'LG_Company_LinkedIn__c': safe_str(row.get('Company Linkedin Url')),
        'LG_Seniority__c': safe_str(row.get('Seniority'), 100),
        'LG_Departments__c': safe_str(row.get('Departments')),
        'LG_Engagement_Rate__c': safe_num(row.get('Enrichment5 Engagement Rate')),
        'LG_Total_Engagements__c': safe_num(row.get('Enrichment5 Total Engagements')),
        'LG_Lead_ID__c': safe_str(row.get('Lead Id')),
        'LG_Enrichment_Source__c': safe_str(row.get('Enrichment Source')),
    }
    
    for fname, val in lg_mapping.items():
        if val is not None and fname in available_fields:
            lead[fname] = val
    
    # Build Description from AI analysis (always works, no custom fields needed)
    desc_parts = []
    justif = safe_long(row.get('Ai Score Justification'))
    if justif:
        desc_parts.append(f"=== AI ANALYSIS ===\n{justif}")
    next_act = safe_long(row.get('Ai Next Action'))
    if next_act:
        desc_parts.append(f"=== NEXT ACTION ===\n{next_act}")
    cold_email = safe_long(row.get('Ai Cold Email'))
    if cold_email:
        desc_parts.append(f"=== COLD EMAIL DRAFT ===\n{cold_email}")
    company_analysis = safe_long(row.get('Company Analysis Result'))
    if company_analysis:
        desc_parts.append(f"=== COMPANY ANALYSIS ===\n{company_analysis}")
    value_prop = safe_long(row.get('Value Proposition'))
    if value_prop:
        desc_parts.append(f"=== VALUE PROPOSITION ===\n{value_prop}")
    if desc_parts:
        lead['Description'] = '\n\n'.join(desc_parts)[:32000]
    
    # Remove None values
    return {k: v for k, v in lead.items() if v is not None}

Import Loop

# Get available fields (respects FLS)
r = requests.get(f'{API}/sobjects/Lead/describe', headers=headers)
available_fields = [f['name'] for f in r.json()['fields']]

# Read CSV
with open('your-leads.csv', 'r', encoding='utf-8-sig') as f:
    rows = list(csv.DictReader(f))

campaign_name = "LeadGenius Import"  # Your campaign identifier

success = 0
errors = 0

for i, row in enumerate(rows):
    lead = build_lead(row, campaign_name, available_fields)
    r = requests.post(f'{API}/sobjects/Lead', headers=headers, json=lead)
    name = f"{row.get('First Name','')} {row.get('Last Name','')}".strip()
    
    if r.status_code in [200, 201]:
        success += 1
        if success % 25 == 0 or success <= 2:
            score = row.get('Ai Score Value', '-')
            print(f"  ✅ {success:3d}/{len(rows)} | {name:30s} | Score: {score}")
    else:
        errors += 1
        msg = r.json()[0]['message'][:80] if isinstance(r.json(), list) else r.text[:80]
        if errors <= 5:
            print(f"  ❌ {name}: {msg}")

print(f"\n✅ Imported: {success} | ❌ Errors: {errors}")

8. Populating Custom Fields (Update Pass)

If the custom fields weren’t ready during initial import (FLS deployed after first import), run a second pass to populate them:

# Get all SF leads by email
r = requests.get(f'{API}/query',
    params={'q': 'SELECT Id, Email FROM Lead'}, headers=headers)
sf_leads = {}
for rec in r.json().get('records', []):
    if rec.get('Email'):
        sf_leads[rec['Email'].lower()] = rec['Id']

# Read CSV and update
updated = 0
for row in rows:
    email = safe_str(row.get('Email'))
    if not email or email.lower() not in sf_leads:
        continue
    
    lead_id = sf_leads[email.lower()]
    update = {}
    
    lg_mapping = {
        'LG_AI_Score__c': safe_num(row.get('Ai Score Value')),
        'LG_Lead_Score__c': safe_num(row.get('Ai Lead Score Score')),
        'LG_Qualification__c': safe_str(row.get('Ai Qualification')),
        'LG_Justification__c': safe_long(row.get('Ai Score Justification')),
        'LG_Next_Action__c': safe_long(row.get('Ai Next Action')),
        'LG_Cold_Email__c': safe_long(row.get('Ai Cold Email')),
        # ... all LG_ fields ...
    }
    
    for fname, val in lg_mapping.items():
        if val is not None and fname in available_fields:
            update[fname] = val
    
    if update:
        r = requests.patch(f'{API}/sobjects/Lead/{lead_id}', headers=headers, json=update)
        if r.status_code == 204:
            updated += 1

print(f"✅ Updated: {updated} leads with AI data")

9. Creating the List View

❌ API Limitation

List Views cannot be created programmatically in some Salesforce orgs (OrgFarm, scratch orgs, some sandboxes). The Metadata API reports success but the view is never created.

✅ Manual Steps (2 Minutes)

1. Go to the Leads tab in Salesforce
2. Click the list view dropdown → New
3. Name: LeadGenius Leads
4. Visibility: All users can see this list view
5. Click Save
6. Add Filter: Lead Source equals LeadGenius Import (or your campaign name)
7. Click the ⚙️ gear → Select Fields to Display
8. Add these columns:

9. Click Save
10. Optionally sort by LG AI Score (descending) to see top leads first

10. Verification

Quick Checks

# Total leads
r = requests.get(f'{API}/query',
    params={'q': 'SELECT COUNT() FROM Lead'}, headers=headers)
print(f"Total leads: {r.json()['totalSize']}")

# Leads with AI Score
r = requests.get(f'{API}/query',
    params={'q': 'SELECT COUNT() FROM Lead WHERE LG_AI_Score__c > 0'}, headers=headers)
print(f"Leads with AI Score: {r.json()['totalSize']}")

# Top 10 by AI Score
r = requests.get(f'{API}/query', params={
    'q': '''SELECT FirstName, LastName, Company, 
            LG_AI_Score__c, LG_Seniority__c, LG_Likely_Engage__c
            FROM Lead 
            WHERE LG_AI_Score__c > 0 
            ORDER BY LG_AI_Score__c DESC 
            LIMIT 10'''
}, headers=headers)
print("\n🏆 Top 10:")
for rec in r.json()['records']:
    print(f"  {rec.get('LG_AI_Score__c',0):5.0f} | "
          f"{rec.get('FirstName','')} {rec.get('LastName','')} | "
          f"{rec.get('Company','')} | "
          f"{rec.get('LG_Seniority__c','')}")

# Verify custom fields are visible
r = requests.get(f'{API}/sobjects/Lead/describe', headers=headers)
lg = [f['name'] for f in r.json()['fields'] if f['name'].startswith('LG_')]
print(f"\n📐 LG custom fields visible: {len(lg)}/17")

11. Complete Automated Script

Here’s a ready-to-run script that does everything in one go:

"""
LeadGenius → Salesforce Import
Usage: python3 lg_to_sf.py --csv leads.csv --campaign "My Campaign"
"""
import requests, os, csv, time, base64, zipfile, io, argparse
import xml.etree.ElementTree as ET
from dotenv import load_dotenv
load_dotenv()

# ============= CONFIG =============
parser = argparse.ArgumentParser()
parser.add_argument('--csv', required=True, help='Path to LeadGenius CSV')
parser.add_argument('--campaign', default='LeadGenius Import', help='Lead Source / campaign name')
parser.add_argument('--clear', action='store_true', help='Delete existing leads first')
args = parser.parse_args()

# ============= AUTH =============
def authenticate():
    data = {
        'grant_type': 'password',
        'client_id': os.getenv('SALESFORCE_CONSUMER_KEY'),
        'client_secret': os.getenv('SALESFORCE_CONSUMER_SECRET'),
        'username': os.getenv('SALESFORCE_USERNAME'),
        'password': os.getenv('SALESFORCE_PASSWORD') + os.getenv('SALESFORCE_SECURITY_TOKEN'),
    }
    r = requests.post('https://login.salesforce.com/services/oauth2/token', data=data)
    result = r.json()
    if 'access_token' not in result:
        raise Exception(f"Auth failed: {result}")
    return result['access_token'], result['instance_url']

token, url = authenticate()
h = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}
API = f'{url}/services/data/v62.0'
TOOLING = f'{url}/services/data/v62.0/tooling'
META = f'{url}/services/Soap/m/62.0'
print(f"✅ Connected to {url}\n")

# ============= HELPERS =============
def s(val, mx=255):
    if not val or str(val).strip() in ('N/A','(empty)','None',''): return None
    return str(val).strip()[:mx] or None
def sl(val, mx=32000):
    if not val or str(val).strip() in ('N/A','(empty)','None',''): return None
    return str(val).strip()[:mx] or None
def n(val):
    if not val or str(val).strip() in ('N/A','(empty)','None',''): return None
    try: return float(str(val).strip())
    except: return None

def extract_id(text):
    for elem in ET.fromstring(text).iter():
        if 'id' in elem.tag.lower() and elem.text and len(elem.text) == 18:
            return elem.text
    return None

def poll(deploy_id):
    for att in range(20):
        time.sleep(3)
        soap = f'''<?xml version="1.0" encoding="utf-8"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/" xmlns:met="http://soap.sforce.com/2006/04/metadata">
<soap:Header><met:SessionHeader><met:sessionId>{token}</met:sessionId></met:SessionHeader></soap:Header>
<soap:Body><met:checkDeployStatus><met:asyncProcessId>{deploy_id}</met:asyncProcessId><met:includeDetails>true</met:includeDetails></met:checkDeployStatus></soap:Body>
</soap:Envelope>'''
        r = requests.post(META, data=soap.encode(), headers={'Content-Type':'text/xml; charset=utf-8','SOAPAction':'checkDeployStatus'})
        done = ok = False
        for elem in ET.fromstring(r.text).iter():
            tag = elem.tag.split('}')[-1] if '}' in elem.tag else elem.tag
            if tag == 'done' and elem.text == 'true': done = True
            if tag == 'success' and elem.text == 'true': ok = True
        if done: return ok
    return False

# ============= STEP 1: OPTIONAL CLEAR =============
if args.clear:
    print("STEP 1: CLEARING LEADS")
    while True:
        r = requests.get(f'{API}/query', params={'q': 'SELECT Id FROM Lead LIMIT 200'}, headers=h)
        recs = r.json().get('records', [])
        if not recs: break
        ids = ','.join([x['Id'] for x in recs])
        requests.delete(f'{API}/composite/sobjects?ids={ids}&allOrNone=false', headers=h)
    print("  ✅ Done\n")

# ============= STEP 2: CREATE FIELDS =============
print("STEP 2: CREATING CUSTOM FIELDS")
FIELDS = [
    {"FullName":"Lead.LG_AI_Score__c","Metadata":{"label":"LG AI Score","type":"Number","precision":5,"scale":0}},
    {"FullName":"Lead.LG_Lead_Score__c","Metadata":{"label":"LG Lead Score","type":"Number","precision":5,"scale":0}},
    {"FullName":"Lead.LG_Qualification__c","Metadata":{"label":"LG Qualification","type":"Text","length":255}},
    {"FullName":"Lead.LG_Justification__c","Metadata":{"label":"LG Justification","type":"LongTextArea","length":32000,"visibleLines":5}},
    {"FullName":"Lead.LG_Next_Action__c","Metadata":{"label":"LG Next Action","type":"LongTextArea","length":32000,"visibleLines":3}},
    {"FullName":"Lead.LG_Cold_Email__c","Metadata":{"label":"LG Cold Email","type":"LongTextArea","length":32000,"visibleLines":5}},
    {"FullName":"Lead.LG_LinkedIn_Connect__c","Metadata":{"label":"LG LinkedIn Connect","type":"LongTextArea","length":32000,"visibleLines":3}},
    {"FullName":"Lead.LG_Decision_Role__c","Metadata":{"label":"LG Decision Maker Role","type":"Text","length":255}},
    {"FullName":"Lead.LG_Likely_Engage__c","Metadata":{"label":"LG Likely to Engage","type":"Text","length":10}},
    {"FullName":"Lead.LG_LinkedIn_URL__c","Metadata":{"label":"LG LinkedIn URL","type":"Url"}},
    {"FullName":"Lead.LG_Company_LinkedIn__c","Metadata":{"label":"LG Company LinkedIn","type":"Url"}},
    {"FullName":"Lead.LG_Seniority__c","Metadata":{"label":"LG Seniority","type":"Text","length":100}},
    {"FullName":"Lead.LG_Departments__c","Metadata":{"label":"LG Departments","type":"Text","length":255}},
    {"FullName":"Lead.LG_Engagement_Rate__c","Metadata":{"label":"LG Engagement Rate","type":"Number","precision":10,"scale":1}},
    {"FullName":"Lead.LG_Total_Engagements__c","Metadata":{"label":"LG Total Engagements","type":"Number","precision":10,"scale":0}},
    {"FullName":"Lead.LG_Lead_ID__c","Metadata":{"label":"LG Lead ID","type":"Text","length":255,"unique":True,"externalId":True}},
    {"FullName":"Lead.LG_Enrichment_Source__c","Metadata":{"label":"LG Enrichment Source","type":"Text","length":255}},
]
for fd in FIELDS:
    fname = fd['FullName'].split('.')[1]
    r = requests.post(f'{TOOLING}/sobjects/CustomField', headers=h, json=fd)
    if r.status_code == 201: print(f"  ✅ {fname}")
    elif 'DUPLICATE' in r.text: print(f"  ⏭️  {fname} (exists)")
    else: print(f"  ❌ {fname}: {r.text[:80]}")
    time.sleep(0.5)

# ============= STEP 3: DEPLOY FLS =============
print("\nSTEP 3: DEPLOYING FIELD-LEVEL SECURITY")
lg_names = [fd['FullName'].split('.')[1] for fd in FIELDS]
perms = ''.join([f'''
    <fieldPermissions><field>Lead.{f}</field><editable>true</editable><readable>true</readable></fieldPermissions>''' for f in lg_names])
prof = f'<?xml version="1.0" encoding="UTF-8"?><Profile xmlns="http://soap.sforce.com/2006/04/metadata">{perms}</Profile>'
pkg = '<?xml version="1.0" encoding="UTF-8"?><Package xmlns="http://soap.sforce.com/2006/04/metadata"><types><members>Admin</members><name>Profile</name></types><version>62.0</version></Package>'
buf = io.BytesIO()
with zipfile.ZipFile(buf,'w',zipfile.ZIP_DEFLATED) as zf:
    zf.writestr('package.xml', pkg); zf.writestr('profiles/Admin.profile', prof)
zd = base64.b64encode(buf.getvalue()).decode()
soap = f'''<?xml version="1.0" encoding="utf-8"?><soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/" xmlns:met="http://soap.sforce.com/2006/04/metadata"><soap:Header><met:SessionHeader><met:sessionId>{token}</met:sessionId></met:SessionHeader></soap:Header><soap:Body><met:deploy><met:ZipFile>{zd}</met:ZipFile><met:DeployOptions><met:checkOnly>false</met:checkOnly><met:ignoreWarnings>true</met:ignoreWarnings><met:rollbackOnError>false</met:rollbackOnError><met:singlePackage>true</met:singlePackage></met:DeployOptions></met:deploy></soap:Body></soap:Envelope>'''
r = requests.post(META, data=soap.encode(), headers={'Content-Type':'text/xml; charset=utf-8','SOAPAction':'deploy'})
did = extract_id(r.text)
ok = poll(did)
print(f"  {'✅ FLS deployed' if ok else '❌ FLS failed'}")

# ============= STEP 4: VERIFY FIELDS =============
print("\nSTEP 4: VERIFYING FIELDS")
time.sleep(5)
r = requests.get(f'{API}/sobjects/Lead/describe', headers=h)
avail = [f['name'] for f in r.json()['fields']]
lg_visible = [f for f in avail if f.startswith('LG_')]
print(f"  LG fields visible: {len(lg_visible)}/17")

# ============= STEP 5: IMPORT LEADS =============
print(f"\nSTEP 5: IMPORTING LEADS FROM {args.csv}")
with open(args.csv, 'r', encoding='utf-8-sig') as f:
    rows = list(csv.DictReader(f))
print(f"  {len(rows)} leads to import\n")

ok_count = err_count = 0
for i, row in enumerate(rows):
    lead = {
        'FirstName': s(row.get('First Name'),40),
        'LastName': s(row.get('Last Name'),80) or 'Unknown',
        'Title': s(row.get('Title')),
        'Email': s(row.get('Email')),
        'Phone': s(row.get('Phone Number')),
        'Company': s(row.get('Company Name')) or 'Unknown',
        'Industry': s(row.get('Industry')),
        'NumberOfEmployees': int(n(row.get('Estimated Num Employees')) or 0) or None,
        'Country': s(row.get('Country')),
        'City': s(row.get('City')),
        'LeadSource': args.campaign,
        'Status': 'Open - Not Contacted',
    }
    # Custom fields
    lg_map = {
        'LG_AI_Score__c': n(row.get('Ai Score Value')),
        'LG_Lead_Score__c': n(row.get('Ai Lead Score Score')),
        'LG_Qualification__c': s(row.get('Ai Qualification')),
        'LG_Justification__c': sl(row.get('Ai Score Justification')),
        'LG_Next_Action__c': sl(row.get('Ai Next Action')),
        'LG_Cold_Email__c': sl(row.get('Ai Cold Email')),
        'LG_LinkedIn_Connect__c': sl(row.get('Ai Linkedin Connect')),
        'LG_Decision_Role__c': s(row.get('Ai Decision Maker Role')),
        'LG_Likely_Engage__c': s(row.get('Is Likely To Engage'),10),
        'LG_LinkedIn_URL__c': s(row.get('Linkedin Url')),
        'LG_Company_LinkedIn__c': s(row.get('Company Linkedin Url')),
        'LG_Seniority__c': s(row.get('Seniority'),100),
        'LG_Departments__c': s(row.get('Departments')),
        'LG_Engagement_Rate__c': n(row.get('Enrichment5 Engagement Rate')),
        'LG_Total_Engagements__c': n(row.get('Enrichment5 Total Engagements')),
        'LG_Lead_ID__c': s(row.get('Lead Id')),
        'LG_Enrichment_Source__c': s(row.get('Enrichment Source')),
    }
    for fname, val in lg_map.items():
        if val is not None and fname in avail:
            lead[fname] = val
    # Description
    parts = []
    j = sl(row.get('Ai Score Justification'))
    if j: parts.append(f"=== AI ANALYSIS ===\n{j}")
    na = sl(row.get('Ai Next Action'))
    if na: parts.append(f"=== NEXT ACTION ===\n{na}")
    ce = sl(row.get('Ai Cold Email'))
    if ce: parts.append(f"=== COLD EMAIL ===\n{ce}")
    if parts: lead['Description'] = '\n\n'.join(parts)[:32000]
    
    lead = {k:v for k,v in lead.items() if v is not None}
    r = requests.post(f'{API}/sobjects/Lead', headers=h, json=lead)
    nm = f"{row.get('First Name','')} {row.get('Last Name','')}".strip()
    if r.status_code in [200,201]:
        ok_count += 1
        if ok_count % 25 == 0 or ok_count <= 2:
            sc = row.get('Ai Score Value','-')
            print(f"  ✅ {ok_count:3d}/{len(rows)} | {nm:30s} | Score: {sc}")
    else:
        err_count += 1
        if err_count <= 5:
            msg = r.json()[0]['message'][:80] if isinstance(r.json(),list) else r.text[:80]
            print(f"  ❌ {nm}: {msg}")

print(f"\n{'='*60}")
print(f"  ✅ Imported: {ok_count}")
print(f"  ❌ Errors:   {err_count}")

# ============= STEP 6: VERIFY =============
print(f"\nSTEP 6: VERIFICATION")
r = requests.get(f'{API}/query', params={'q':'SELECT COUNT() FROM Lead'}, headers=h)
print(f"  Total leads: {r.json()['totalSize']}")
r = requests.get(f'{API}/query', params={'q':'SELECT COUNT() FROM Lead WHERE LG_AI_Score__c > 0'}, headers=h)
print(f"  With AI Score: {r.json()['totalSize']}")
r = requests.get(f'{API}/query', params={'q':'SELECT FirstName, LastName, Company, LG_AI_Score__c, LG_Seniority__c FROM Lead WHERE LG_AI_Score__c > 0 ORDER BY LG_AI_Score__c DESC LIMIT 5'}, headers=h)
print(f"\n🏆 Top 5:")
for rec in r.json()['records']:
    print(f"  {rec.get('LG_AI_Score__c',0):5.0f} | {rec.get('FirstName','')} {rec.get('LastName','')} | {rec.get('Company','')}")
print(f"\n🔗 {url}/lightning/o/Lead/list")
print(f"   Create a list view with filter: Lead Source = '{args.campaign}'")

Usage

# Full import with field creation
python3 lg_to_sf.py --csv prophix-demo-02026.csv --campaign "Prophix Demo"

# Clear existing leads first
python3 lg_to_sf.py --csv leads.csv --campaign "Q1 Campaign" --clear

12. Troubleshooting

Common Errors

Verify Each Step

# 1. Auth works?
python3 -c "from dotenv import load_dotenv; load_dotenv(); exec(open('lg_to_sf.py').read().split('# ===')[1])"

# 2. Fields exist in metadata?
# Check Tooling API for all LG_ fields:
python3 -c "
import requests, os
from dotenv import load_dotenv; load_dotenv()
# ... auth ...
r = requests.get(f'{TOOLING}/query', params={
    'q': \"SELECT DeveloperName FROM CustomField WHERE TableEnumOrId='Lead' AND DeveloperName LIKE 'LG_%'\"
}, headers=h)
print(f'Fields in metadata: {r.json()[\"totalSize\"]}')"

# 3. Fields visible via REST? (FLS check)
# Use the describe verification in Section 6

Execution Order Matters

If you see No such column errors:

1. Create fields (Tooling API)     ← fields exist but invisible
2. Deploy FLS (Metadata SOAP)      ← fields become visible
3. Wait 5 seconds                  ← propagation delay
4. Verify describe shows fields    ← confirm before import
5. Import leads                    ← now it works

Never skip step 2. It’s the most common failure mode.