Marketing Stack Migration: From 11 Fragmented Tools to Unified System

Annual Tool Costs (before migration):
- Mailchimp: $8,400
- ActiveCampaign: $12,000
- Unbounce: $3,600
- Webflow: $4,800
- Hotjar: $3,900
- Mixpanel: $10,800
- Zapier: $9,600
- Calendly: $1,200
- Typeform: $3,600
- Clearbit: $36,000
- Google Analytics 360: $0 (free tier)
Total: $94,000/year

# Marketing Stack Migration - Contact Data Consolidation Script
# Merges contacts from Mailchimp, ActiveCampaign, Typeform into HubSpot

import hubspot
from hubspot import HubSpot
import pandas as pd
import hashlib

class ContactConsolidator:
    def __init__(self, hubspot_api_key):
        self.hubspot = HubSpot(access_token=hubspot_api_key)
        self.staging_db = []

    def extract_mailchimp_contacts(self, mailchimp_export):
        """Extract contacts from Mailchimp export CSV"""
        df = pd.read_csv(mailchimp_export)
        contacts = []

        for _, row in df.iterrows():
            contact = {
                'email': row['Email Address'],
                'firstname': row['First Name'],
                'lastname': row['Last Name'],
                'mailchimp_status': row['Status'],
                'mailchimp_rating': row['Member Rating'],
                'mailchimp_tags': row['Tags'].split(',') if pd.notna(row['Tags']) else [],
                'source_system': 'mailchimp',
                'email_hash': hashlib.md5(row['Email Address'].lower().encode()).hexdigest()
            }
            contacts.append(contact)

        return contacts

    def extract_activecampaign_contacts(self, ac_export):
        """Extract contacts from ActiveCampaign export"""
        df = pd.read_csv(ac_export)
        contacts = []

        for _, row in df.iterrows():
            contact = {
                'email': row['email'],
                'firstname': row['first_name'],
                'lastname': row['last_name'],
                'phone': row['phone'],
                'company': row['orgname'],
                'ac_score': row['score'],
                'ac_tags': row['tags'].split(',') if pd.notna(row['tags']) else [],
                'source_system': 'activecampaign',
                'email_hash': hashlib.md5(row['email'].lower().encode()).hexdigest()
            }
            contacts.append(contact)

        return contacts

    def merge_duplicate_contacts(self, contacts_list):
        """
        Merge contacts from multiple systems based on email hash
        Strategy: Most recent data wins, but preserve all source tags
        """
        merged = {}

        for contact in contacts_list:
            email_hash = contact['email_hash']

            if email_hash not in merged:
                # First occurrence - add to merged dict
                merged[email_hash] = contact
                merged[email_hash]['all_tags'] = contact.get('mailchimp_tags', []) + contact.get('ac_tags', [])
                merged[email_hash]['source_systems'] = [contact['source_system']]
            else:
                # Duplicate found - merge intelligently
                existing = merged[email_hash]

                # Use most complete data (prefer non-null values)
                for key in ['firstname', 'lastname', 'phone', 'company']:
                    if key in contact and contact[key] and not existing.get(key):
                        existing[key] = contact[key]

                # Combine tags from all sources
                existing['all_tags'].extend(contact.get('mailchimp_tags', []))
                existing['all_tags'].extend(contact.get('ac_tags', []))
                existing['all_tags'] = list(set(existing['all_tags']))  # Deduplicate

                # Track which systems this contact existed in
                existing['source_systems'].append(contact['source_system'])

                # Preserve highest engagement metrics
                if 'ac_score' in contact and contact['ac_score']:
                    existing['engagement_score'] = max(
                        existing.get('engagement_score', 0),
                        contact['ac_score']
                    )

        return list(merged.values())

    def map_to_hubspot_properties(self, contact):
        """Map consolidated contact data to HubSpot property schema"""
        hubspot_contact = {
            'email': contact['email'],
            'firstname': contact.get('firstname', ''),
            'lastname': contact.get('lastname', ''),
            'phone': contact.get('phone', ''),
            'company': contact.get('company', ''),
            'migration_source_systems': ';'.join(contact['source_systems']),
            'migration_date': datetime.now().isoformat(),
            'legacy_tags': ';'.join(contact['all_tags']),
            'legacy_engagement_score': contact.get('engagement_score', 0)
        }

        # Map engagement score to HubSpot score
        if contact.get('engagement_score'):
            hubspot_contact['hs_lead_score'] = self.convert_score_to_hubspot(
                contact['engagement_score']
            )

        return hubspot_contact

    def batch_import_to_hubspot(self, contacts, batch_size=100):
        """
        Import contacts to HubSpot in batches
        Handles rate limiting and error recovery
        """
        total = len(contacts)
        imported = 0
        errors = []

        for i in range(0, total, batch_size):
            batch = contacts[i:i+batch_size]

            try:
                # Use HubSpot batch API
                response = self.hubspot.crm.contacts.batch_api.create(
                    batch_input_simple_public_object_input={
                        'inputs': [
                            {'properties': contact} for contact in batch
                        ]
                    }
                )

                imported += len(batch)
                print(f"Imported {imported}/{total} contacts...")

            except Exception as e:
                print(f"Error importing batch {i}-{i+batch_size}: {str(e)}")
                errors.append({
                    'batch_range': f"{i}-{i+batch_size}",
                    'error': str(e),
                    'contacts': batch
                })

        return {
            'total': total,
            'imported': imported,
            'errors': errors
        }

# Migration execution
consolidator = ContactConsolidator(hubspot_api_key='YOUR_KEY')

# Extract from all sources
mailchimp_contacts = consolidator.extract_mailchimp_contacts('mailchimp_export.csv')
ac_contacts = consolidator.extract_activecampaign_contacts('activecampaign_export.csv')

# Merge duplicates intelligently
all_contacts = mailchimp_contacts + ac_contacts
merged_contacts = consolidator.merge_duplicate_contacts(all_contacts)

print(f"Original contacts: {len(all_contacts)}")
print(f"After deduplication: {len(merged_contacts)}")
print(f"Duplicates eliminated: {len(all_contacts) - len(merged_contacts)}")

# Map to HubSpot schema
hubspot_contacts = [
    consolidator.map_to_hubspot_properties(contact)
    for contact in merged_contacts
]

# Import to HubSpot
result = consolidator.batch_import_to_hubspot(hubspot_contacts)
print(f"Migration complete: {result['imported']} contacts imported")

// Example: Migrating ActiveCampaign "Lead Nurture Sequence" to HubSpot Workflow

// OLD: ActiveCampaign automation (pseudo-code representation)
// Trigger: Tag added "Downloaded Whitepaper"
// Wait 2 days → Email 1
// Wait 3 days → Email 2
// Wait 4 days → Email 3
// If clicked link in any email → Tag "Engaged", notify sales

// NEW: HubSpot Workflow (custom code action for complex logic)

const contact = enrollment.contact;

// Track engagement across nurture sequence
let engagementScore = 0;

// Email 1: Educational content (Day 2)
if (contact.nurture_email_1_opened) engagementScore += 5;
if (contact.nurture_email_1_clicked) engagementScore += 15;

// Email 2: Product use case (Day 5)
if (contact.nurture_email_2_opened) engagementScore += 5;
if (contact.nurture_email_2_clicked) engagementScore += 20;

// Email 3: Case study + CTA (Day 9)
if (contact.nurture_email_3_opened) engagementScore += 5;
if (contact.nurture_email_3_clicked) engagementScore += 25;

// Progressive scoring - engaged leads get higher scores
if (engagementScore >= 30) {
  contact.lifecycle_stage = 'Marketing Qualified Lead';
  notifySalesTeam(contact, 'High engagement in nurture sequence');
}

// Track sequence completion
contact.nurture_sequence_completed = true;
contact.nurture_sequence_score = engagementScore;

// Segment integration - send HubSpot events to data warehouse

// HubSpot Webhook → Segment → Data Warehouse (for advanced reporting)

const segmentClient = require('@segment/analytics-node');
const analytics = new segmentClient('SEGMENT_WRITE_KEY');

// Webhook listener for HubSpot workflow events
app.post('/webhooks/hubspot/workflow-completion', (req, res) => {
  const event = req.body;

  // Send to Segment for data warehouse loading
  analytics.track({
    userId: event.contactId,
    event: 'Workflow Completed',
    properties: {
      workflowId: event.workflowId,
      workflowName: event.workflowName,
      completionDate: event.completedAt,
      engagementScore: event.engagementScore,
      leadSource: event.originalSource
    }
  });

  res.status(200).send('Event tracked');
});

- HubSpot Marketing Hub Professional: $18,000
- HubSpot Sales Hub Professional: $8,400 (already using)
- Webflow: $4,800 (kept)
- Clearbit: $0 (native HubSpot integration, included)
- Google Analytics: $0 (free tier)
Total: $34,000/year (was $94,000)

Annual Savings: $60,000 (64% reduction)
3-year TCO Savings: $180,000

def handle_conflicting_data(contact_a, contact_b):
    """
    When two contact records have conflicting data, choose the best value
    """
    # For name fields, prefer capitalized over lowercase
    if contact_a.get('firstname') and contact_b.get('firstname'):
        if contact_a['firstname'].istitle():
            return contact_a['firstname']
        else:
            return contact_b['firstname']

    # For engagement metrics, take the maximum
    if 'engagement_score' in contact_a and 'engagement_score' in contact_b:
        return max(contact_a['engagement_score'], contact_b['engagement_score'])

    # For timestamps, take the most recent
    if 'last_activity_date' in contact_a and 'last_activity_date' in contact_b:
        return max(contact_a['last_activity_date'], contact_b['last_activity_date'])

# During migration transition period, write to both systems

class DualWriteContactSync:
    def __init__(self, legacy_client, hubspot_client):
        self.legacy = legacy_client  # ActiveCampaign
        self.hubspot = hubspot_client
        self.errors = []

    def sync_contact_update(self, email, properties):
        """
        Write contact updates to both systems during migration
        Ensures no data is lost during transition period
        """
        results = {'legacy': None, 'hubspot': None}

        try:
            # Write to legacy system first (existing dependency)
            results['legacy'] = self.legacy.update_contact(email, properties)
        except Exception as e:
            self.errors.append({'system': 'legacy', 'error': str(e), 'email': email})

        try:
            # Write to HubSpot (new system)
            results['hubspot'] = self.hubspot.update_contact(email, properties)
        except Exception as e:
            self.errors.append({'system': 'hubspot', 'error': str(e), 'email': email})

        # Log sync status for monitoring
        self.log_sync_status(email, results)

        return results

class MigrationValidator:
    def validate_contact_migration(self, legacy_contact, hubspot_contact):
        """
        Validate that migrated contact data matches source
        """
        validation_results = {
            'passed': True,
            'errors': []
        }

        # Critical field validation
        critical_fields = ['email', 'firstname', 'lastname']
        for field in critical_fields:
            if legacy_contact.get(field) != hubspot_contact.get(field):
                validation_results['passed'] = False
                validation_results['errors'].append({
                    'field': field,
                    'expected': legacy_contact.get(field),
                    'actual': hubspot_contact.get(field)
                })

        # Tag preservation validation
        legacy_tags = set(legacy_contact.get('tags', []))
        hubspot_tags = set(hubspot_contact.get('legacy_tags', '').split(';'))

        if not legacy_tags.issubset(hubspot_tags):
            missing_tags = legacy_tags - hubspot_tags
            validation_results['passed'] = False
            validation_results['errors'].append({
                'field': 'tags',
                'issue': 'missing_tags',
                'missing': list(missing_tags)
            })

        return validation_results

# Run validation on sample of migrated contacts
validator = MigrationValidator()
sample_size = 1000
sample_contacts = random.sample(migrated_contacts, sample_size)

validation_failures = []
for contact in sample_contacts:
    legacy = get_legacy_contact(contact['email'])
    hubspot = get_hubspot_contact(contact['email'])

    result = validator.validate_contact_migration(legacy, hubspot)
    if not result['passed']:
        validation_failures.append(result)

print(f"Validation: {sample_size - len(validation_failures)}/{sample_size} passed")
print(f"Accuracy: {((sample_size - len(validation_failures)) / sample_size) * 100}%")