Dex 🔜 cleanup

dex_cleanup.py

#Designed to strip 🔜 [something] from the last name of my Dex contacts imported from LinkedIn, because game industry people 
#LOVE to add that and it causes false duplicates. Pass `--api-key your_api_key` to do a dry run, and pass `--execute` after to 
#actually run it against your Dex database.

import requests
import re
import time
import argparse
import os
from typing import List, Dict, Any

class DexContactCleaner:
    def __init__(self, api_key: str):
        self.base_url = "https://api.getdex.com/api/rest"
        self.headers = {
            "Content-Type": "application/json",
            "x-hasura-dex-api-key": api_key,
        }
    
    def fetch_all_contacts(self, limit: int = 200) -> List[Dict[str, Any]]:
        """Fetch all contacts from Dex API"""
        all_contacts = []
        offset = 0
        
        print("Fetching contacts...")
        while True:
            try:
                response = requests.get(
                    f"{self.base_url}/contacts",
                    params={"limit": limit, "offset": offset},
                    headers=self.headers,
                    timeout=30,
                )
                response.raise_for_status()
                
                batch = response.json().get("contacts", [])
                if not batch:
                    break
                
                all_contacts.extend(batch)
                print(f"Fetched {len(all_contacts)} contacts so far...")
                
                if len(batch) < limit:
                    break
                    
                offset += limit
                time.sleep(0.1)  # Be nice to the API
                
            except requests.exceptions.RequestException as e:
                print(f"Error fetching contacts: {e}")
                break
        
        print(f"Total contacts fetched: {len(all_contacts)}")
        return all_contacts
    
    def find_contacts_with_emoji(self, contacts: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Find contacts with 🔜 emoji in their last name"""
        emoji_pattern = r'🔜.*'
        problematic_contacts = []
        
        for contact in contacts:
            last_name = contact.get("last_name", "")
            if last_name and "🔜" in last_name:
                problematic_contacts.append(contact)
        
        print(f"Found {len(problematic_contacts)} contacts with 🔜 emoji in last name")
        return problematic_contacts
    
    def clean_last_name(self, last_name: str) -> str:
        """Remove 🔜 emoji and everything after it from last name"""
        if "🔜" in last_name:
            # Split on the emoji and take only the part before it, then strip whitespace
            cleaned = last_name.split("🔜")[0].strip()
            return cleaned
        return last_name
    
    def update_contact(self, contact_id: str, first_name: str, last_name: str) -> bool:
        """Update a contact's name via Dex API"""
        try:
            update_data = {
                "contactId": contact_id,
                "changes": {
                    "first_name": first_name,
                    "last_name": last_name,
                }
            }
            
            response = requests.put(
                f"{self.base_url}/contacts/{contact_id}",
                json=update_data,
                headers=self.headers,
                timeout=30,
            )
            response.raise_for_status()
            return True
            
        except requests.exceptions.RequestException as e:
            print(f"Error updating contact {contact_id}: {e}")
            return False
    
    def clean_contacts(self, dry_run: bool = True) -> None:
        """Main method to find and clean contacts"""
        # Fetch all contacts
        contacts = self.fetch_all_contacts()
        
        # Find contacts with the emoji
        problematic_contacts = self.find_contacts_with_emoji(contacts)
        
        if not problematic_contacts:
            print("No contacts found with 🔜 emoji in their last name.")
            return
        
        print(f"\nContacts to be cleaned:")
        print("-" * 60)
        
        for contact in problematic_contacts:
            contact_id = contact["id"]
            first_name = contact.get("first_name", "")
            original_last_name = contact.get("last_name", "")
            cleaned_last_name = self.clean_last_name(original_last_name)
            
            print(f"ID: {contact_id}")
            print(f"Name: {first_name} {original_last_name}")
            print(f"Will become: {first_name} {cleaned_last_name}")
            print("-" * 40)
        
        if dry_run:
            print(f"\n🔍 DRY RUN MODE: No changes made. Use --execute to apply changes.")
            return
        
        # Confirm before proceeding
        user_input = input(f"\nProceed to update {len(problematic_contacts)} contacts? (y/N): ")
        if user_input.lower() != 'y':
            print("Operation cancelled.")
            return
        
        # Update contacts
        print("\nUpdating contacts...")
        success_count = 0
        
        for contact in problematic_contacts:
            contact_id = contact["id"]
            first_name = contact.get("first_name", "")
            original_last_name = contact.get("last_name", "")
            cleaned_last_name = self.clean_last_name(original_last_name)
            
            print(f"Updating: {first_name} {original_last_name} -> {first_name} {cleaned_last_name}")
            
            if self.update_contact(contact_id, first_name, cleaned_last_name):
                success_count += 1
                print("✅ Updated successfully")
            else:
                print("❌ Update failed")
            
            time.sleep(0.2)  # Rate limiting
        
        print(f"\n🎉 Successfully updated {success_count}/{len(problematic_contacts)} contacts")

def main():
    parser = argparse.ArgumentParser(description='Clean up Dex contacts with 🔜 emoji in last names')
    parser.add_argument('--execute', action='store_true', 
                       help='Actually make changes (default is dry-run mode)')
    parser.add_argument('--api-key', type=str, 
                       help='Your Dex API key (or set as environment variable DEX_API_KEY)')
    
    args = parser.parse_args()
    
    # Get API key from argument or environment variable
    API_KEY = args.api_key or os.getenv('DEX_API_KEY')
    
    if not API_KEY:
        print("⚠️  Please provide your Dex API key using:")
        print("   --api-key YOUR_KEY")
        print("   or set environment variable: export DEX_API_KEY=your_key")
        return
    
    cleaner = DexContactCleaner(API_KEY)
    
    # Determine mode based on command line flag
    dry_run = not args.execute
    
    if dry_run:
        print("Running in DRY RUN mode... (use --execute to make actual changes)")
    else:
        print("EXECUTE mode: Will make actual changes to contacts")
    
    cleaner.clean_contacts(dry_run=dry_run)

if __name__ == "__main__":
    main()