ideas-generator/server/scripts/import-remaining-agents.py

#!/usr/bin/env python3

import csv
import json
import re
import sys
import urllib.request
import urllib.parse
import urllib.error
from pathlib import Path

# Get current directory and CSV file path
script_dir = Path(__file__).parent
csv_file = script_dir.parent.parent / "I-gen-assistant-instructions.csv"
api_base = "http://localhost:3000/api"

def slugify(text):
    """Convert text to slug format"""
    # Remove special characters and convert to lowercase
    text = re.sub(r'[^\w\s-]', '', text).strip().lower()
    # Replace spaces and multiple dashes with single dash
    return re.sub(r'[-\s]+', '-', text)

def extract_starter_message(system_instructions):
    """Extract starter message from system instructions"""
    # Look for common patterns of starter messages
    patterns = [
        r'STARTER MESSAGE:\s*"([^"]+)"',
        r'INITIAL MESSAGE:\s*"([^"]+)"',
        r'GREETING:\s*"([^"]+)"',
        r'Hello[^.]*\.',
        r'Hi[^.]*\.',
        r'Welcome[^.]*\.'
    ]
    
    for pattern in patterns:
        match = re.search(pattern, system_instructions, re.IGNORECASE | re.MULTILINE)
        if match:
            if '"' in pattern:
                return match.group(1).strip()
            else:
                return match.group(0).strip()
    
    # If no pattern found, return None
    return None

def categorize_agent(name, instructions):
    """Determine category based on name and instructions"""
    name_lower = name.lower()
    instructions_lower = instructions.lower()
    
    # Business category keywords
    if any(word in name_lower or word in instructions_lower for word in [
        'business', 'marketing', 'sales', 'strategy', 'consultant', 'finance', 
        'management', 'entrepreneur', 'profit', 'revenue', 'market'
    ]):
        return 'business'
    
    # Creative category keywords
    if any(word in name_lower or word in instructions_lower for word in [
        'creative', 'design', 'art', 'writer', 'writing', 'content', 'story',
        'brand', 'advertisement', 'creative', 'innovation', 'idea'
    ]):
        return 'creative'
    
    # Technical category keywords
    if any(word in name_lower or word in instructions_lower for word in [
        'technical', 'code', 'programming', 'software', 'development', 'api',
        'database', 'algorithm', 'technology', 'digital'
    ]):
        return 'technical'
    
    # Educational category keywords
    if any(word in name_lower or word in instructions_lower for word in [
        'teach', 'learn', 'education', 'training', 'tutorial', 'explain',
        'knowledge', 'study', 'academic'
    ]):
        return 'educational'
    
    # Health category keywords
    if any(word in name_lower or word in instructions_lower for word in [
        'health', 'fitness', 'wellness', 'medical', 'therapy', 'mental',
        'physical', 'nutrition', 'exercise'
    ]):
        return 'health'
    
    # Personal category keywords
    if any(word in name_lower or word in instructions_lower for word in [
        'personal', 'life', 'relationship', 'goal', 'motivation', 'self',
        'habit', 'productivity', 'organization'
    ]):
        return 'personal'
    
    # Default to creative
    return 'creative'

def get_existing_agents():
    """Get list of existing agent original IDs"""
    try:
        url = f"{api_base}/assistants?admin=true"
        with urllib.request.urlopen(url) as response:
            if response.status == 200:
                data = json.loads(response.read().decode('utf-8'))
                return {agent.get('metadata', {}).get('originalId') for agent in data.get('agents', [])}
            else:
                print(f"Failed to get existing agents: {response.status}")
                return set()
    except Exception as e:
        print(f"Error getting existing agents: {e}")
        return set()

def create_agent(agent_data):
    """Create new agent via API"""
    try:
        url = f"{api_base}/assistants"
        data = json.dumps(agent_data).encode('utf-8')
        
        req = urllib.request.Request(
            url, 
            data=data,
            headers={'Content-Type': 'application/json'}
        )
        
        with urllib.request.urlopen(req) as response:
            if response.status == 201:
                result = json.loads(response.read().decode('utf-8'))
                return True, result
            else:
                return False, f"HTTP {response.status}"
                
    except urllib.error.HTTPError as e:
        error_body = e.read().decode('utf-8') if hasattr(e, 'read') else str(e)
        return False, f"HTTP {e.code}: {error_body}"
    except Exception as e:
        return False, str(e)

def main():
    print("🚀 Starting agent import from CSV...")
    
    if not csv_file.exists():
        print(f"❌ CSV file not found: {csv_file}")
        sys.exit(1)
    
    # Get existing agents
    print("📋 Getting existing agents...")
    existing_ids = get_existing_agents()
    print(f"Found {len(existing_ids)} existing agents")
    
    # Parse CSV and import new agents
    imported_count = 0
    skipped_count = 0
    error_count = 0
    
    print("📖 Reading CSV file...")
    
    with open(csv_file, 'r', encoding='utf-8') as file:
        # Handle potential CSV parsing issues with multi-line fields
        content = file.read()
        
        # Split into lines but handle quoted multi-line fields
        lines = []
        current_line = ""
        in_quotes = False
        
        for char in content:
            if char == '"' and (len(current_line) == 0 or current_line[-1] != '\\'):
                in_quotes = not in_quotes
            elif char == '\n' and not in_quotes:
                lines.append(current_line)
                current_line = ""
                continue
            current_line += char
        
        if current_line:
            lines.append(current_line)
        
        # Parse header
        header = lines[0].split(',')
        print(f"CSV columns: {header}")
        
        # Process each agent
        for i, line in enumerate(lines[1:], 1):
            try:
                # Handle CSV parsing for quoted fields
                fields = []
                current_field = ""
                in_quotes = False
                
                j = 0
                while j < len(line):
                    char = line[j]
                    if char == '"':
                        in_quotes = not in_quotes
                    elif char == ',' and not in_quotes:
                        fields.append(current_field.strip('"'))
                        current_field = ""
                        j += 1
                        continue
                    current_field += char
                    j += 1
                
                if current_field:
                    fields.append(current_field.strip('"'))
                
                if len(fields) < 3:  # Need at least ID, name, and instructions
                    print(f"⚠️ Skipping line {i}: insufficient fields")
                    continue
                
                assistant_id = fields[0].strip()
                assistant_name = fields[1].strip()
                system_instructions = fields[2].strip() if len(fields) > 2 else ""
                
                # Skip if already imported
                if assistant_id in existing_ids:
                    skipped_count += 1
                    continue
                
                # Extract starter message
                starter_message = extract_starter_message(system_instructions)
                
                # Clean system instructions (remove starter message if found)
                if starter_message:
                    clean_instructions = re.sub(
                        r'STARTER MESSAGE:\s*"[^"]+"', 
                        '', 
                        system_instructions, 
                        flags=re.IGNORECASE | re.MULTILINE
                    ).strip()
                else:
                    clean_instructions = system_instructions
                    # Generate default starter message
                    starter_message = f"Hello! I'm {assistant_name}. How can I help you today?"
                
                # Create agent data
                agent_key = slugify(assistant_name)
                category = categorize_agent(assistant_name, system_instructions)
                
                agent_data = {
                    "key": agent_key,
                    "name": assistant_name,
                    "description": f"AI assistant specialized in {assistant_name.lower()}",
                    "category": category,
                    "systemPrompt": clean_instructions,
                    "starterMessage": starter_message,
                    "model": "gpt-4o",
                    "temperature": 0.7,
                    "maxTokens": 4000,
                    "metadata": {
                        "originalId": assistant_id
                    }
                }
                
                # Create the agent
                success, result = create_agent(agent_data)
                
                if success:
                    print(f"✅ Created agent: {assistant_name}")
                    imported_count += 1
                else:
                    print(f"❌ Failed to create {assistant_name}: {result}")
                    error_count += 1
                
            except Exception as e:
                print(f"❌ Error processing line {i}: {e}")
                error_count += 1
    
    print(f"\n🎉 Import complete!")
    print(f"✅ Imported: {imported_count}")
    print(f"⏭️ Skipped (already exists): {skipped_count}")
    print(f"❌ Errors: {error_count}")

if __name__ == "__main__":
    main()