Cómo crear una canalización de datos de directorio B2B con n8n y una API de búsqueda

// n8n HTTP Request Node Configuration
// Method: POST
// URL: https://api.scavio.dev/api/v1/search
// Headers:
//   x-api-key: {{ $env.SCAVIO_API_KEY }}
//   Content-Type: application/json
// Body (JSON):
{
  "query": "site:clutch.co {{ $json.niche }} companies",
  "country_code": "us",
  "num_results": 10
}

// Test with Python equivalent:
import os, requests

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']

def search_directory(niche: str, directory: str = 'clutch.co') -> list:
    resp = requests.post('https://api.scavio.dev/api/v1/search',
        headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
        json={'query': f'site:{directory} {niche} companies',
              'country_code': 'us', 'num_results': 10})
    return resp.json().get('organic_results', [])

results = search_directory('web development agency')
print(f'Found {len(results)} directory listings')
for r in results[:3]:
    print(f"  {r['title']} -> {r['link']}")

import re

def extract_directory_data(results: list, directory: str) -> list:
    """Extract structured company data from directory search results."""
    companies = []
    for r in results:
        title = r.get('title', '')
        snippet = r.get('snippet', '')
        link = r.get('link', '')
        # Skip non-company pages (category pages, blog posts)
        if any(skip in link for skip in ['/blog/', '/resources/', '/press/']):
            continue
        # Extract rating from snippet
        rating_match = re.search(r'(\d+\.\d+)\s*(?:/5|stars?|rating)', snippet, re.I)
        rating = float(rating_match.group(1)) if rating_match else None
        # Extract review count
        review_match = re.search(r'(\d+)\s*(?:reviews?|ratings?)', snippet, re.I)
        reviews = int(review_match.group(1)) if review_match else None
        # Clean company name from title
        name = title.split(' - ')[0].split(' | ')[0].strip()
        name = re.sub(r'\s*(?:Reviews?|Profile|Company).*$', '', name, flags=re.I)
        companies.append({
            'name': name,
            'rating': rating,
            'reviews': reviews,
            'profile_url': link,
            'directory': directory,
            'snippet': snippet[:200],
        })
    return companies

results = search_directory('web development agency', 'clutch.co')
companies = extract_directory_data(results, 'clutch.co')
for c in companies[:3]:
    print(f"{c['name']} | Rating: {c['rating']} | Reviews: {c['reviews']}")
    print(f"  {c['profile_url']}")

import time

DIRECTORIES = ['clutch.co', 'g2.com', 'capterra.com']

def multi_directory_search(niche: str) -> list:
    all_companies = []
    for directory in DIRECTORIES:
        results = search_directory(niche, directory)
        companies = extract_directory_data(results, directory)
        all_companies.extend(companies)
        time.sleep(0.3)
    # Deduplicate by normalized name
    seen = {}
    for c in all_companies:
        key = c['name'].lower().strip()
        if key in seen:
            # Merge: keep the one with more reviews
            existing = seen[key]
            if (c.get('reviews') or 0) > (existing.get('reviews') or 0):
                seen[key] = c
            # Track which directories list them
            dirs = existing.get('directories', [existing['directory']])
            if c['directory'] not in dirs:
                dirs.append(c['directory'])
                existing['directories'] = dirs
        else:
            seen[key] = c
    return list(seen.values())

companies = multi_directory_search('marketing automation software')
print(f'Found {len(companies)} unique companies across {len(DIRECTORIES)} directories')
print(f'Cost: {len(DIRECTORIES)} searches = ${len(DIRECTORIES) * 0.005:.3f}')
for c in companies[:5]:
    dirs = c.get('directories', [c['directory']])
    print(f"  {c['name']} (on {', '.join(dirs)})")

import csv, json

def export_to_csv(companies: list, filename: str = 'directory_companies.csv'):
    """Export companies to CSV (n8n equivalent: Google Sheets Append node)."""
    fieldnames = ['name', 'rating', 'reviews', 'directory', 'profile_url', 'snippet']
    with open(filename, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction='ignore')
        writer.writeheader()
        writer.writerows(companies)
    print(f'Exported {len(companies)} companies to {filename}')

def export_to_json(companies: list, filename: str = 'directory_companies.json'):
    """Export as JSON (n8n equivalent: write to webhook or database)."""
    with open(filename, 'w') as f:
        json.dump(companies, f, indent=2)
    print(f'Exported {len(companies)} companies to {filename}')

# n8n workflow summary:
# 1. Schedule Trigger (daily/weekly)
# 2. Set Node: define niches to search
# 3. Loop: for each niche x directory
# 4. HTTP Request: Scavio search API
# 5. Code Node: extract_directory_data()
# 6. Google Sheets: append rows

companies = multi_directory_search('CRM software')
export_to_csv(companies)
print(f'\nn8n workflow cost per run: ${len(DIRECTORIES) * 0.005:.3f} per niche')