How to Build an Intent-Based Lead Pipeline with Search Data

import os, requests, time

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
URL = 'https://api.scavio.dev/api/v1/search'
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

def build_intent_queries(product_category: str, competitors: list) -> list:
    """Generate buying-intent search queries."""
    queries = []
    # Direct comparison queries
    for comp in competitors:
        queries.append({'query': f'{comp} alternative 2026', 'intent': 'switching', 'signal': 'high'})
        queries.append({'query': f'{comp} vs', 'intent': 'comparing', 'signal': 'high'})
        queries.append({'query': f'{comp} pricing too expensive', 'intent': 'price_sensitive', 'signal': 'high'})
    # Category queries
    queries.append({'query': f'best {product_category} 2026', 'intent': 'researching', 'signal': 'medium'})
    queries.append({'query': f'{product_category} for enterprise', 'intent': 'enterprise', 'signal': 'medium'})
    # Reddit intent queries
    queries.append({'query': f'site:reddit.com recommend {product_category}', 'intent': 'asking', 'signal': 'high'})
    queries.append({'query': f'site:reddit.com {product_category} frustrating', 'intent': 'pain_point', 'signal': 'high'})
    return queries

queries = build_intent_queries('search API', ['SerpAPI', 'Tavily'])
for q in queries:
    print(f'[{q["signal"]:6s}] {q["intent"]:15s} | {q["query"]}')

import re

def extract_leads_from_results(results: list, query_info: dict) -> list:
    """Extract potential leads from search results."""
    leads = []
    for r in results:
        title = r.get('title', '')
        snippet = r.get('snippet', '')
        url = r.get('link', '')
        # Extract domain as potential lead
        domain_match = re.search(r'https?://(?:www\.)?([\w.-]+)', url)
        domain = domain_match.group(1) if domain_match else ''
        # Skip aggregator/directory sites
        skip_domains = ['reddit.com', 'quora.com', 'wikipedia.org', 'youtube.com',
                       'g2.com', 'capterra.com', 'medium.com']
        if any(d in domain for d in skip_domains):
            # Reddit posts are leads for community outreach, not company leads
            if 'reddit.com' in domain:
                leads.append({
                    'type': 'community',
                    'platform': 'reddit',
                    'title': title,
                    'url': url,
                    'intent': query_info['intent'],
                    'signal': query_info['signal'],
                })
            continue
        leads.append({
            'type': 'company',
            'domain': domain,
            'title': title,
            'snippet': snippet[:200],
            'url': url,
            'intent': query_info['intent'],
            'signal': query_info['signal'],
        })
    return leads

def scan_intent_signals(queries: list) -> list:
    all_leads = []
    for q in queries:
        resp = requests.post(URL, headers=H,
            json={'query': q['query'], 'country_code': 'us', 'num_results': 5})
        results = resp.json().get('organic_results', [])
        leads = extract_leads_from_results(results, q)
        all_leads.extend(leads)
        time.sleep(0.3)
    return all_leads

leads = scan_intent_signals(queries[:5])  # Test with first 5 queries
print(f'Found {len(leads)} intent signals')
for l in leads[:5]:
    print(f'  [{l["signal"]}] {l["type"]}: {l.get("domain", l.get("platform", ""))} - {l["intent"]}')

def score_leads(leads: list) -> list:
    """Score leads by intent strength."""
    intent_scores = {
        'switching': 10, 'price_sensitive': 9, 'pain_point': 8,
        'comparing': 7, 'asking': 6, 'researching': 4, 'enterprise': 5,
    }
    signal_multiplier = {'high': 1.5, 'medium': 1.0, 'low': 0.5}
    for lead in leads:
        base = intent_scores.get(lead['intent'], 3)
        mult = signal_multiplier.get(lead['signal'], 1.0)
        lead['score'] = base * mult
    # Sort by score descending
    leads.sort(key=lambda x: x['score'], reverse=True)
    return leads

def deduplicate_leads(leads: list) -> list:
    """Deduplicate by domain, keeping highest-scored entry."""
    seen = {}
    for lead in leads:
        key = lead.get('domain', lead.get('url', ''))
        if key not in seen or lead['score'] > seen[key]['score']:
            seen[key] = lead
    return sorted(seen.values(), key=lambda x: x['score'], reverse=True)

scored = score_leads(leads)
unique = deduplicate_leads(scored)
print(f'Unique leads: {len(unique)}')
print(f'\nTop intent signals:')
for l in unique[:10]:
    domain = l.get('domain', l.get('platform', ''))
    print(f'  Score: {l["score"]:5.1f} | {l["intent"]:15s} | {domain}')
    print(f'         {l.get("title", "")[:60]}')

import csv

def export_leads(leads: list, filename: str = 'intent_leads.csv'):
    if not leads:
        print('No leads to export')
        return
    fieldnames = ['score', 'type', 'domain', 'intent', 'signal', 'title', 'url']
    with open(filename, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore')
        writer.writeheader()
        writer.writerows(leads)
    high = sum(1 for l in leads if l['score'] >= 10)
    medium = sum(1 for l in leads if 5 <= l['score'] < 10)
    low = sum(1 for l in leads if l['score'] < 5)
    print(f'Exported {len(leads)} leads to {filename}')
    print(f'  High intent (10+): {high}')
    print(f'  Medium intent (5-9): {medium}')
    print(f'  Low intent (<5): {low}')
    print(f'  API cost: {len(queries)} searches x $0.005 = ${len(queries) * 0.005:.3f}')

export_leads(unique)
print(f'\nVolume lead gen: scrapes 1000 emails, 2% conversion = 20 leads')
print(f'Intent pipeline: finds {len([l for l in unique if l["score"] >= 7])} high-intent signals')