Cómo encontrar clientes de agencias de automatización con datos de búsqueda

import os, requests, time, re

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_discovery_queries(niche: str) -> list:
    return [
        # Job postings for manual roles (automation opportunity)
        {'query': f'{niche} data entry hiring 2026', 'signal': 'hiring_manual'},
        {'query': f'{niche} manual reporting job', 'signal': 'hiring_manual'},
        # Reddit pain points
        {'query': f'site:reddit.com {niche} spreadsheet nightmare', 'signal': 'pain_point'},
        {'query': f'site:reddit.com {niche} automate manual process', 'signal': 'seeking_automation'},
        # Tool research (actively looking for solutions)
        {'query': f'{niche} automation tools 2026', 'signal': 'tool_research'},
        {'query': f'{niche} workflow automation software', 'signal': 'tool_research'},
        # Industry-specific
        {'query': f'{niche} agency needs automation', 'signal': 'direct_need'},
    ]

queries = build_discovery_queries('real estate')
for q in queries:
    print(f'[{q["signal"]:20s}] {q["query"]}')

def search_for_prospects(queries: list) -> list:
    prospects = []
    for q in queries:
        resp = requests.post(URL, headers=H,
            json={'query': q['query'], 'country_code': 'us', 'num_results': 5})
        for r in resp.json().get('organic_results', []):
            domain = re.search(r'https?://(?:www\.)?([\w.-]+)', r.get('link', ''))
            prospects.append({
                'title': r['title'],
                'url': r['link'],
                'domain': domain.group(1) if domain else '',
                'snippet': r.get('snippet', '')[:200],
                'signal': q['signal'],
                'query': q['query'],
            })
        time.sleep(0.3)
    return prospects

def score_prospects(prospects: list) -> list:
    scores = {
        'direct_need': 10, 'seeking_automation': 9, 'pain_point': 8,
        'hiring_manual': 7, 'tool_research': 5,
    }
    for p in prospects:
        p['score'] = scores.get(p['signal'], 3)
        # Boost if snippet mentions specific pain
        snippet_lower = p['snippet'].lower()
        if any(w in snippet_lower for w in ['manual', 'tedious', 'hours', 'spreadsheet']):
            p['score'] += 2
    prospects.sort(key=lambda x: x['score'], reverse=True)
    return prospects

prospects = search_for_prospects(queries)
scored = score_prospects(prospects)
print(f'Found {len(scored)} prospect signals')
for p in scored[:5]:
    print(f'  [{p["score"]:2d}] {p["signal"]:20s} | {p["domain"]}')
    print(f'       {p["title"][:60]}')

def enrich_prospect(domain: str) -> dict:
    """Get company details via search."""
    resp = requests.post(URL, headers=H,
        json={'query': domain, 'country_code': 'us', 'num_results': 3})
    data = resp.json()
    kg = data.get('knowledge_graph', {})
    organic = data.get('organic_results', [])[:2]
    return {
        'domain': domain,
        'description': kg.get('description', organic[0].get('snippet', '') if organic else ''),
        'website': kg.get('website', f'https://{domain}'),
        'type': kg.get('type', ''),
    }

def build_prospect_profiles(scored_prospects: list, max_enrich: int = 10) -> list:
    """Enrich top prospects with company details."""
    # Deduplicate by domain
    seen = {}
    for p in scored_prospects:
        d = p['domain']
        if d and d not in seen:
            seen[d] = p
    unique = sorted(seen.values(), key=lambda x: x['score'], reverse=True)[:max_enrich]
    profiles = []
    for p in unique:
        enrichment = enrich_prospect(p['domain'])
        p.update(enrichment)
        profiles.append(p)
        time.sleep(0.3)
    return profiles

profiles = build_prospect_profiles(scored)
print(f'\nEnriched {len(profiles)} prospect profiles')
for p in profiles[:5]:
    print(f'\n  {p["domain"]} (Score: {p["score"]})')
    print(f'  Signal: {p["signal"]}')
    print(f'  Description: {p["description"][:80]}')

import csv

def generate_outreach(profiles: list) -> list:
    signal_pitches = {
        'hiring_manual': 'You are hiring for manual {niche} roles. Automation could handle this at 10x speed for the cost of one hire.',
        'pain_point': 'Your team mentioned struggling with manual processes. We automate exactly this type of workflow.',
        'seeking_automation': 'You are actively looking for automation solutions. We specialize in {niche} workflow automation.',
        'tool_research': 'You are evaluating automation tools. We build custom solutions that integrate your existing stack.',
        'direct_need': 'You have expressed a direct need for automation. We can scope a solution in a 30-minute call.',
    }
    for p in profiles:
        p['pitch'] = signal_pitches.get(p['signal'], 'We help companies automate manual workflows.').format(niche='your')
    # Export
    with open('agency_prospects.csv', 'w', newline='') as f:
        w = csv.DictWriter(f, fieldnames=['domain', 'score', 'signal', 'description', 'pitch', 'url'],
                          extrasaction='ignore')
        w.writeheader()
        w.writerows(profiles)
    total_searches = len(queries) + len(profiles)  # discovery + enrichment
    print(f'Exported {len(profiles)} prospects to agency_prospects.csv')
    print(f'Total searches: {total_searches}')
    print(f'Cost: ${total_searches * 0.005:.3f}')
    return profiles

generate_outreach(profiles)