scripts/ai_lead_agent.py

#!/usr/bin/env python3
"""AI Lead Generation Agent - Main Orchestrator"""
import asyncio
import json
import os
import sys
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
import sqlite3

# Lead scoring system
@dataclass
class LeadFeatures:
    company: str
    job_title: str
    industry: str
    revenue: Optional[str] = None
    employees: Optional[int] = None
    tech_stack: List[str] = None
    engagement: Dict = None
    
    def __post_init__(self):
        if self.tech_stack is None:
            self.tech_stack = []
        if self.engagement is None:
            self.engagement = {}


class LeadScorer:
    """AI-powered lead scoring engine"""
    
    WEIGHTS = {
        "job_seniority": 0.25,
        "company_size": 0.20,
        "digital_footprint": 0.15,
        "engagement_velocity": 0.15,
        "industry_fit": 0.10,
        "historical_signals": 0.15
    }
    
    HIGH_VALUE_TITLES = ['CEO', 'CTO', 'CMO', 'VP', 'Director', 'Head', 'Founder']
    TARGET_INDUSTRIES = ['SaaS', 'Technology', 'Software', 'E-commerce', 'Healthcare', 'Finance', 'Construction', 'Professional Services']
    
    def score_job_seniority(self, title: str) -> float:
        title_upper = title.upper()
        if any(t in title_upper for t in self.HIGH_VALUE_TITLES):
            return 1.0
        return 0.5
    
    def score_company_size(self, revenue: Optional[str], employees: Optional[int]) -> float:
        if revenue:
            try:
                rev = float(revenue.replace('$', '').replace(',', ''))
                if rev >= 10000000:
                    return 1.0
                elif rev >= 1000000:
                    return 0.8
                elif rev >= 100000:
                    return 0.6
            except:
                pass
        if employees:
            if employees >= 500:
                return 1.0
            elif employees >= 50:
                return 0.7
            elif employees >= 10:
                return 0.4
        return 0.2
    
    def score_digital_footprint(self, tech_stack: List[str]) -> float:
        known_tools = {'hubspot', 'salesforce', 'outreach', 'apollo', 'phantombuster', 'clearbit', 'hunter', 'linkedin'}
        matches = sum(1 for t in tech_stack if any(k in t.lower() for k in known_tools))
        return min(1.0, matches * 0.25 + 0.3)
    
    def score_engagement(self, engagement: Dict) -> float:
        score = 0
        score += engagement.get('website_visits', 0) * 0.2
        score += engagement.get('email_opens', 0) * 0.15
        score += engagement.get('page_views', 0) * 0.1
        score += engagement.get('content_downloads', 0) * 0.25
        return min(1.0, score)
    
    def score_industry(self, industry: str) -> float:
        industry_lower = industry.lower()
        matches = sum(1 for t in self.TARGET_INDUSTRIES if t.lower() in industry_lower)
        return min(1.0, matches * 0.2)
    
    def calculate(self, lead: LeadFeatures) -> int:
        """Calculate total lead score 0-100"""
        job_score = self.score_job_seniority(lead.job_title)
        company_score = self.score_company_size(lead.revenue, lead.employees)
        digital_score = self.score_digital_footprint(lead.tech_stack)
        engagement_score = self.score_engagement(lead.engagement)
        industry_score = self.score_industry(lead.industry)
        
        total = (
            job_score * self.WEIGHTS['job_seniority'] +
            company_score * self.WEIGHTS['company_size'] +
            digital_score * self.WEIGHTS['digital_footprint'] +
            engagement_score * self.WEIGHTS['engagement_velocity'] +
            industry_score * self.WEIGHTS['industry_fit'] +
            0.5 * self.WEIGHTS['historical_signals']  # Default historical
        ) * 100
        
        return round(max(0, min(100, total)))
    
    def get_priority(self, score: int) -> str:
        if score >= 85:
            return "immediate"
        elif score >= 70:
            return "high"
        elif score >= 50:
            return "medium"
        return "low"


class PersonalizationEngine:
    """Generate personalized outreach content"""
    
    TEMPLATES = {
        "email_subject": [
            "{company}'s {job_title} deserves better results",
            "3 ideas for {company} from AI Lead Gen",
            "Quick win for {company}'s lead pipeline?",
            "{job_title} at {company} - relevant insights"
        ],
        "email_body": """
Hi {first_name},

I noticed {company} is working on {topic}. Many teams struggle with {pain_point}, especially when dealing with {industry_specific}.

Our AI system helped {similar_company} increase qualified leads by 340% in 90 days. The approach:

1. **{step1}** - Identifies high-intent signals
2. **{step2}** - Delivers personalized outreach
3. **{step3}** - Optimizes in real-time

Would you be open to a 15-minute conversation this week?

Best,
{your_name}
""",
        "linkedin_message": """
Hi {first_name},

Saw your profile and thought you might be interested in how AI is transforming lead generation for {industry} companies.

Happy to share a relevant case study if you're open to it.

Best,
{your_name}
"""
    }
    
    def personalize_email(self, lead: LeadFeatures) -> Dict:
        import random
        
        first_name = lead.company.split()[0] if lead.company else "there"
        
        topics = {
            'SaaS': 'scaling faster',
            'Technology': 'product adoption',
            'Healthcare': 'patient acquisition',
            'Finance': 'client onboarding',
            'E-commerce': 'revenue growth',
            'Construction': 'project leads',
            'default': 'results'
        }
        topic = topics.get(lead.industry, topics['default'])
        
        pain_points = {
            'SaaS': 'manual lead qualification',
            'Technology': 'sales cycle length',
            'Healthcare': 'patient scheduling',
            'Finance': 'client retention',
            'E-commerce': 'customer acquisition cost',
            'Construction': 'project pipeline',
            'default': 'lead quality'
        }
        pain_point = pain_points.get(lead.industry, pain_points['default'])
        
        return {
            "subject": random.choice(self.TEMPLATES['email_subject']).format(
                company=lead.company,
                job_title=lead.job_title,
                first_name=first_name
            ),
            "body": self.TEMPLATES['email_body'].format(
                first_name=first_name,
                company=lead.company,
                topic=topic,
                pain_point=pain_point,
                industry_specific=lead.industry.lower(),
                similar_company="a similar company",
                step1="Predictive scoring",
                step2="Personalized sequences",
                step3="Performance analytics",
                your_name="Your Lead Gen AI"
            )
        }


class CampaignExecutor:
    """Execute automated campaigns"""
    
    def __init__(self, db_path: str = "leads.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        conn = sqlite3.connect(self.db_path)
        conn.execute('''CREATE TABLE IF NOT EXISTS leads (
            id INTEGER PRIMARY KEY,
            company TEXT,
            job_title TEXT,
            industry TEXT,
            score INTEGER,
            priority TEXT,
            status TEXT DEFAULT 'new',
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )''')
        conn.execute('''CREATE TABLE IF NOT EXISTS outreach_log (
            id INTEGER PRIMARY KEY,
            lead_id INTEGER,
            channel TEXT,
            content TEXT,
            sent_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            engagement INTEGER DEFAULT 0
        )''')
        conn.commit()
        conn.close()


async def process_leads_batch(leads_data: List[Dict]) -> List[Dict]:
    """Process batch of leads"""
    scorer = LeadScorer()
    engine = PersonalizationEngine()
    results = []
    
    for lead_dict in leads_data:
        lead = LeadFeatures(
            company=lead_dict.get('company', ''),
            job_title=lead_dict.get('job_title', ''),
            industry=lead_dict.get('industry', ''),
            revenue=lead_dict.get('revenue'),
            employees=lead_dict.get('employees'),
            tech_stack=lead_dict.get('tech_stack', []),
            engagement=lead_dict.get('engagement', {})
        )
        
        score = scorer.calculate(lead)
        priority = scorer.get_priority(score)
        personalized = engine.personalize_email(lead)
        
        results.append({
            "lead": {"company": lead.company, "job_title": lead.job_title},
            "score": score,
            "priority": priority,
            "next_action": "Send email" if score > 50 else "Monitor",
            "email": personalized
        })
    
    return results


async def main():
    """Demo execution"""
    sample_leads = [
        {
            "company": "TechStart Inc",
            "job_title": "Marketing Director",
            "industry": "SaaS",
            "revenue": "$25000000",
            "tech_stack": ["HubSpot", "Loom", "Zoom"],
            "engagement": {"website_visits": 5, "email_opens": 3}
        },
        {
            "company": "MediCare Hospital",
            "job_title": "CTO",
            "industry": "Healthcare",
            "revenue": "$150000000",
            "tech_stack": ["Epic", "Cerner", "Salesforce"],
            "engagement": {"website_visits": 12, "content_downloads": 2}
        },
        {
            "company": "Local Bakery",
            "job_title": "Owner",
            "industry": "Food & Beverage",
            "employees": 15,
            "engagement": {"website_visits": 1}
        }
    ]
    
    print("🚀 AI Lead Generation Processing...")
    print("=" * 50)
    
    results = await process_leads_batch(sample_leads)
    
    for r in results:
        print(f"\n{r['lead']['company']} - {r['lead']['job_title']}")
        print(f"  Score: {r['score']} | Priority: {r['priority']} | Action: {r['next_action']}")
    
    print("\n✅ Complete - processed", len(results), "leads")


if __name__ == "__main__":
    asyncio.run(main())