HubSpot to Twenty CRM Migration Guide: Exporting Deals & Contacts

import csv
import json
import os
import sys
import time
from datetime import datetime
import requests
from dateutil import parser

# Configuration & API Credentials
TWENTY_GRAPHQL_ENDPOINT = os.getenv("TWENTY_API_URL", "http://localhost:3000/graphql")
TWENTY_API_KEY = os.getenv("TWENTY_API_KEY", "your-twenty-api-bearer-token")

HEADERS = {
    "Authorization": f"Bearer {TWENTY_API_KEY}",
    "Content-Type": "application/json"
}

LEDGER_FILE = "id_mapping_ledger.json"

def load_ledger():
    if os.path.exists(LEDGER_FILE):
        with open(LEDGER_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    return {"companies": {}, "people": {}, "opportunities": {}}

def save_ledger(ledger):
    with open(LEDGER_FILE, "w", encoding="utf-8") as f:
        json.dump(ledger, f, indent=2)

def execute_graphql(query, variables=None):
    payload = {"query": query, "variables": variables or {}}
    for attempt in range(5):
        try:
            resp = requests.post(TWENTY_GRAPHQL_ENDPOINT, headers=HEADERS, json=payload, timeout=30)
            if resp.status_code == 200:
                data = resp.json()
                if "errors" in data:
                    print(f"[ERROR] GraphQL Error: {data['errors']}")
                    return None
                return data.get("data")
            elif resp.status_code in [429, 502, 503, 504]:
                wait_time = (2 ** attempt) * 1.5
                print(f"[WARN] HTTP {resp.status_code}. Retrying in {wait_time:.1f}s...")
                time.sleep(wait_time)
            else:
                print(f"[FATAL] Ingestion rejected with status {resp.status_code}: {resp.text}")
                return None
        except requests.RequestException as e:
            time.sleep(2)
    return None

# Step 1: Migrate Companies
def migrate_companies(csv_path, ledger):
    print("--> Migrating Companies...")
    mutation = """
    mutation CreateCompany($input: CompanyCreateInput!) {
        createCompany(data: $input) {
            id
            name
            domainName
        }
    }
    """
    with open(csv_path, mode="r", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        for row in reader:
            hs_id = row.get("Record ID") or row.get("hs_object_id")
            name = row.get("Company Name") or row.get("name")
            domain = (row.get("Company Domain Name") or row.get("domain") or "").replace("https://", "").replace("http://", "").strip("/")

            if not name or hs_id in ledger["companies"]:
                continue

            input_data = {
                "name": name,
                "domainName": domain if domain else None
            }

            res = execute_graphql(mutation, {"input": input_data})
            if res and "createCompany" in res:
                twenty_id = res["createCompany"]["id"]
                ledger["companies"][hs_id] = twenty_id
                print(f" [Company] Migrated: {name} -> {twenty_id}")
                save_ledger(ledger)

# Step 2: Migrate People (Contacts)
def migrate_people(csv_path, ledger):
    print("--> Migrating Contacts (People)...")
    mutation = """
    mutation CreatePerson($input: PersonCreateInput!) {
        createPerson(data: $input) {
            id
            name { firstName lastName }
        }
    }
    """
    with open(csv_path, mode="r", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        for row in reader:
            hs_id = row.get("Record ID") or row.get("hs_object_id")
            first_name = row.get("First Name", "").strip()
            last_name = row.get("Last Name", "").strip()
            email = row.get("Email", "").strip().lower()
            phone = row.get("Phone Number", "").strip()
            hs_comp_id = row.get("Associated Company ID", "").strip()

            if (not first_name and not last_name and not email) or hs_id in ledger["people"]:
                continue

            input_data = {
                "name": {"firstName": first_name, "lastName": last_name},
                "emails": {"primaryEmail": email, "additionalEmails": []} if email else None,
                "phones": {"primaryPhoneNumber": phone, "primaryPhoneCallingCode": "+1"} if phone else None,
            }

            if hs_comp_id and hs_comp_id in ledger["companies"]:
                input_data["companyId"] = ledger["companies"][hs_comp_id]

            res = execute_graphql(mutation, {"input": input_data})
            if res and "createPerson" in res:
                twenty_id = res["createPerson"]["id"]
                ledger["people"][hs_id] = twenty_id
                print(f" [Person] Migrated: {first_name} {last_name} ({email}) -> {twenty_id}")
                save_ledger(ledger)

# Step 3: Migrate Deals (Opportunities)
STAGE_MAP = {
    "appointmentscheduled": "NEW",
    "qualifiedtobuy": "SCREENING",
    "presentationscheduled": "MEETING",
    "decisionmakerboughtin": "PROPOSAL",
    "contractsent": "NEGOTIATION",
    "closedwon": "CLOSED_WON",
    "closedlost": "CLOSED_LOST"
}

def migrate_deals(csv_path, ledger):
    print("--> Migrating Deals (Opportunities)...")
    mutation = """
    mutation CreateOpportunity($input: OpportunityCreateInput!) {
        createOpportunity(data: $input) {
            id
            name
            amount { amountMicros currencyCode }
            stage
        }
    }
    """
    with open(csv_path, mode="r", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        for row in reader:
            hs_id = row.get("Record ID") or row.get("hs_object_id")
            deal_name = row.get("Deal Name", "Untitled Deal").strip()
            amount_str = row.get("Amount", "0").replace("$", "").replace(",", "").strip()
            stage_raw = row.get("Deal Stage", "appointmentscheduled").strip().lower()
            close_date_raw = row.get("Close Date", "").strip()
            comp_id = row.get("Associated Company ID", "").strip()
            contact_id = row.get("Associated Contact ID", "").strip()

            if hs_id in ledger["opportunities"]:
                continue

            try:
                amount_micros = int(float(amount_str) * 1_000_000) if amount_str else 0
            except ValueError:
                amount_micros = 0

            stage = STAGE_MAP.get(stage_raw, "NEW")
            close_date = None
            if close_date_raw:
                try:
                    close_date = parser.parse(close_date_raw).isoformat()
                except Exception:
                    pass

            input_data = {
                "name": deal_name,
                "amount": {"amountMicros": amount_micros, "currencyCode": "USD"},
                "stage": stage,
                "closeDate": close_date
            }

            if comp_id and comp_id in ledger["companies"]:
                input_data["companyId"] = ledger["companies"][comp_id]
            if contact_id and contact_id in ledger["people"]:
                input_data["pointOfContactId"] = ledger["people"][contact_id]

            res = execute_graphql(mutation, {"input": input_data})
            if res and "createOpportunity" in res:
                twenty_id = res["createOpportunity"]["id"]
                ledger["opportunities"][hs_id] = twenty_id
                print(f" [Deal] Migrated: {deal_name} (${amount_str}) -> {twenty_id}")
                save_ledger(ledger)

if __name__ == "__main__":
    ledger = load_ledger()
    # Execute sequential dependency chain
    migrate_companies("hubspot_companies.csv", ledger)
    migrate_people("hubspot_contacts.csv", ledger)
    migrate_deals("hubspot_deals.csv", ledger)
    print("\n[SUCCESS] Migration completed successfully. Ledger saved.")