scripts/ai_lead_scorer.py

#!/usr/bin/env python3
"""AI Lead Scoring Agent for B2B prospect qualification"""
import asyncio
import json
import os
from datetime import datetime
from typing import Dict, List, Optional

# Hermes tools available
try:
    from hermes_tools import web_search, terminal
except ImportError:
    pass

# Lead scoring weights
WEIGHTS = {
    "job_seniority": 0.25,
    "company_size": 0.20,
    "digital_footprint": 0.15,
    "engagement_velocity": 0.15,
    "historical_data": 0.15,
    "industry_fit": 0.10
}


def score_job_seniority(title: str) -> float:
    """Score job title 0-1"""
    high = ['CEO', 'CTO', 'CMO', 'VP', 'Director', 'Head']
    mid = ['Manager', 'Lead', 'Specialist', 'Coordinator']
    
    title_upper = title.upper()
    if any(h in title_upper for h in high):
        return 1.0
    elif any(m in title_upper for m in mid):
        return 0.6
    return 0.3


def score_company_size(revenue: Optional[str] = None, employees: Optional[int] = None) -> float:
    """Score company size 0-1"""
    if revenue:
        try:
            rev = float(revenue.replace('$', '').replace(',', ''))
            if rev >= 10000000:  # $10M+
                return 1.0
            elif rev >= 1000000:  # $1M-$10M
                return 0.8
            elif rev >= 100000:  # $100K-$1M
                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_industry_fit(industry: str, target_industries: List[str]) -> float:
    """Score how well industry matches targets"""
    industry_lower = industry.lower()
    matches = sum(1 for t in target_industries if t.lower() in industry_lower)
    return min(1.0, matches * 0.3)


def calculate_total_score(lead_data: Dict) -> int:
    """Calculate total lead score 0-100"""
    
    job_score = score_job_seniority(lead_data.get('job_title', ''))
    company_score = score_company_size(
        lead_data.get('revenue'), 
        lead_data.get('employees')
    )
    industry_score = score_industry_fit(
        lead_data.get('industry', ''),
        lead_data.get('target_industries', [])
    )
    
    # Engagement signals
    engagement = lead_data.get('engagement', {})
    engagement_score = min(1.0, (
        engagement.get('website_visits', 0) * 0.2 +
        engagement.get('email_opens', 0) * 0.15 +
        engagement.get('content_downloads', 0) * 0.25 +
        engagement.get('social_engagement', 0) * 0.1
    ))
    
    # Historical data
    historical = lead_data.get('historical_data', {})
    historical_score = min(1.0, (
        historical.get('previous_interest', 0) * 0.5 +
        historical.get('demo_requests', 0) * 0.8 +
        historical.get('last_purchase_days', 365) / 365 * -0.5 + 0.5  # Recent = good
    ))
    
    # Digital footprint
    tech_stack = lead_data.get('tech_stack', [])
    digital_score = min(1.0, len(tech_stack) * 0.1 + 
                       (0 if 'none' in tech_stack else 0.3))
    
    # Weighted average
    total = (
        job_score * WEIGHTS['job_seniority'] +
        company_score * WEIGHTS['company_size'] +
        digital_score * WEIGHTS['digital_footprint'] +
        engagement_score * WEIGHTS['engagement_velocity'] +
        historical_score * WEIGHTS['historical_data'] +
        industry_score * WEIGHTS['industry_fit']
    ) * 100
    
    return round(max(0, min(100, total)))


def generate_prediction(score: int, industry: str) -> Dict:
    """Generate prediction metadata"""
    if score >= 80:
        action = "Immediate outreach - scheduling calls"
        timeline = "0-7 days"
    elif score >= 65:
        action = "Send personalized content package"
        timeline = "7-14 days"
    elif score >= 50:
        action = "Monitor for engagement signals"
        timeline = "14-30 days"
    else:
        action = "Nurture with educational content"
        timeline = "30+ days"
    
    return {
        "recommended_action": action,
        "timeline": timeline,
        "confidence": "high" if score > 70 else "medium" if score > 50 else "low"
    }


async def process_lead(lead_data: Dict) -> Dict:
    """Process single lead and return score"""
    score = calculate_total_score(lead_data)
    prediction = generate_prediction(score, lead_data.get('industry', ''))
    
    return {
        "lead_id": lead_data.get('id'),
        "company": lead_data.get('company'),
        "job_title": lead_data.get('job_title'),
        "lead_score": score,
        "prediction": prediction,
        "processed_at": datetime.utcnow().isoformat()
    }


async def main():
    """Example usage"""
    sample_leads = [
        {
            "id": 1,
            "company": "TechStart Inc",
            "job_title": "Marketing Director",
            "industry": "SaaS",
            "revenue": "$25000000",
            "target_industries": ["SaaS", "Technology", "Software"],
            "engagement": {
                "website_visits": 5,
                "email_opens": 3
            }
        },
        {
            "id": 2,
            "company": "Local Bakery",
            "job_title": "Owner",
            "industry": "Food & Beverage",
            "employees": 15,
            "engagement": {
                "website_visits": 1
            }
        }
    ]
    
    for lead in sample_leads:
        result = await process_lead(lead)
        print(json.dumps(result, indent=2))


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