Migre 60.000 consultas de Google CSE a una API SERP exportando su registro de consultas existente, asignando parámetros de CSE a campos de API, ejecutando pruebas de validación paralelas, cambiando el tráfico de producción y desmantelando el proyecto CSE. Google CSE tiene un límite de 10.000 consultas por día en niveles pagos y devuelve estructuras JSON inconsistentes en todas las versiones de esquema. Una API de búsqueda dedicada elimina esos límites y devuelve JSON normalizado en todos los tipos de consultas. A $0,005/crédito con Scavio, 60.000 consultas mensuales cuestan $300 en comparación con los $5/1.000 de CSE (USD 300), pero con resultados estructurados, soporte multiplataforma y sin límites diarios.
Requisitos previos
- Python 3.8+ instalado
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
- Acceso a sus registros de consultas o análisis de Google CSE
Guia paso a paso
Paso 1: Exportar consultas CSE existentes
Extraiga su registro de consultas de los análisis de CSE o de la base de datos de su aplicación. Agrupe las consultas por frecuencia para priorizar las pruebas de migración.
import os, json, csv
# Export from your app DB or CSE analytics CSV
def load_cse_queries(csv_path: str) -> list:
queries = []
with open(csv_path) as f:
reader = csv.DictReader(f)
for row in reader:
queries.append({
'query': row['query'],
'frequency': int(row.get('count', 1)),
'cse_params': {'cx': row.get('cx', ''), 'num': row.get('num', '10')},
})
queries.sort(key=lambda q: q['frequency'], reverse=True)
print(f'Loaded {len(queries)} unique queries ({sum(q["frequency"] for q in queries)} total calls)')
return queries
# Example with mock data:
queries = [{'query': 'best crm 2026', 'frequency': 500},
{'query': 'python async tutorial', 'frequency': 200}]
print(f'Top query: {queries[0]["query"]} ({queries[0]["frequency"]} calls/mo)')Paso 2: Asignar campos CSE a parámetros API
Asigne parámetros de Google CSE como cx, num, start y lr a equivalentes de API de Scavio. La mayoría de los parámetros CSE se asignan directamente o se vuelven innecesarios.
import requests
API_KEY = os.environ['SCAVIO_API_KEY']
def map_cse_to_scavio(cse_query: dict) -> dict:
"""Map CSE parameters to Scavio API format."""
return {
'platform': 'google',
'query': cse_query['query'],
# CSE 'num' -> results come full by default
# CSE 'cx' -> not needed, full Google index
# CSE 'lr' -> use country/language params if needed
}
def test_single(query: dict) -> dict:
payload = map_cse_to_scavio(query)
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json=payload, timeout=15)
resp.raise_for_status()
results = resp.json().get('organic_results', [])
return {'query': query['query'], 'results': len(results), 'status': 'ok'}
print(test_single({'query': 'best crm 2026'}))Paso 3: Ejecutar pruebas de validación paralelas
Pruebe las 100 consultas principales en paralelo para validar la calidad de los resultados y medir la latencia antes de cambiar de producción.
from concurrent.futures import ThreadPoolExecutor
import time
def validate_batch(queries: list, max_workers: int = 5) -> dict:
start = time.monotonic()
results = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
results = list(pool.map(test_single, queries[:100]))
elapsed = time.monotonic() - start
passed = sum(1 for r in results if r['results'] > 0)
failed = [r for r in results if r['results'] == 0]
print(f'Validated {len(results)} queries in {elapsed:.1f}s')
print(f'Passed: {passed}, Empty: {len(failed)}')
if failed:
print(f'Empty queries: {[f["query"] for f in failed[:5]]}')
return {'passed': passed, 'failed': len(failed), 'elapsed': round(elapsed, 1)}
validate_batch([{'query': 'best crm 2026'}, {'query': 'python async'}])Paso 4: Cambiar el tráfico de producción
Reemplace la llamada HTTP CSE en su código de producción con el punto final de Scavio. Utilice una marca de función para una implementación gradual.
# Production switch with feature flag
def search(query: str, use_scavio: bool = True) -> list:
if use_scavio:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'platform': 'google', 'query': query}, timeout=10)
return resp.json().get('organic_results', [])
else:
# Legacy CSE call (keep as fallback during migration)
resp = requests.get('https://www.googleapis.com/customsearch/v1',
params={'key': os.environ.get('CSE_KEY', ''), 'cx': os.environ.get('CSE_CX', ''), 'q': query})
return resp.json().get('items', [])
results = search('best crm 2026', use_scavio=True)
print(f'{len(results)} results from Scavio')Paso 5: Desmantelar el proyecto CSE
Después de 7 días de tráfico estable de Scavio, desactive la facturación de CSE, elimine las credenciales de CSE y archive el registro de migración.
# Post-migration checklist:
# 1. Verify 7 days of zero CSE calls in your logs
# 2. Disable CSE API key in Google Cloud Console
# 3. Remove CSE environment variables
# 4. Archive migration validation results
def migration_report(validation: dict, daily_queries: int = 2000) -> str:
monthly = daily_queries * 30
cse_cost = monthly * 0.005 # $5/1K queries
scavio_cost = monthly * 0.005 # $0.005/credit
report = f'Migration complete: {monthly} queries/month\n'
report += f'CSE cost was: ${cse_cost:.0f}/mo\n'
report += f'Scavio cost: ${scavio_cost:.0f}/mo\n'
report += f'Validation: {validation["passed"]} passed, {validation["failed"]} empty\n'
report += 'CSE project can be decommissioned.'
return report
print(migration_report({'passed': 98, 'failed': 2}))Ejemplo en Python
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def search(query):
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}).json()
return data.get('organic_results', [])
# Replaces: requests.get('googleapis.com/customsearch/v1', params={...})
results = search('best crm 2026')
print(f'{len(results)} results')Ejemplo en JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function search(query) {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
return (await r.json()).organic_results || [];
}
// Replaces: fetch(`googleapis.com/customsearch/v1?key=...&cx=...&q=${query}`)
search('best crm 2026').then(r => console.log(r.length + ' results'));Salida esperada
A fully migrated search pipeline serving 60K+ monthly queries through Scavio instead of Google CSE, with validated result quality and a decommission checklist.