La creación de un panel de operaciones con datos de búsqueda brinda a los equipos de SEO y marketing una vista en vivo de los cambios de posición en las SERP, la aparición de fragmentos de funciones y los movimientos de la competencia sin depender de rastreadores de clasificación de terceros. La API de búsqueda de Scavio devuelve resultados orgánicos estructurados, fragmentos destacados, cuadros La gente también pregunta y datos de descripción general de IA para cualquier consulta. Al sondear sus consultas de destino según una programación, almacenar los resultados en SQLite y servirlas a través de un panel liviano de Flask, obtiene una vista de operaciones autohospedada que se actualiza con tanta frecuencia como lo permite su presupuesto. Este tutorial construye el proceso completo desde la recopilación de datos hasta un panel visible en el navegador.
Requisitos previos
- Python 3.8 o superior instalado
- solicitudes, flask y sqlite3 (biblioteca estándar) disponibles
- Una clave API de Scavio de scavio.dev
- Una lista de palabras clave objetivo para realizar un seguimiento
Guia paso a paso
Paso 1: Configurar el esquema de la base de datos SQLite
Cree una base de datos SQLite con tablas para instantáneas SERP, historial de posiciones y apariencias de funciones. Cada fila de la tabla de posiciones almacena una consulta, una URL, una posición y una marca de tiempo para que puedas realizar un seguimiento del movimiento a lo largo del tiempo.
Python
import sqlite3
from pathlib import Path
DB_PATH = Path('serp_dashboard.db')
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.executescript('''
CREATE TABLE IF NOT EXISTS positions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query TEXT NOT NULL,
url TEXT NOT NULL,
position INTEGER NOT NULL,
title TEXT,
fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS features (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query TEXT NOT NULL,
feature_type TEXT NOT NULL,
content TEXT,
fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE INDEX IF NOT EXISTS idx_positions_query ON positions(query, fetched_at);
''')
conn.close()
init_db()Paso 2: Construya el recopilador de datos SERP
Escriba una función que consulte la API de Scavio para cada palabra clave objetivo, extraiga posiciones orgánicas y características SERP y las inserte en SQLite. Ejecute esto en una programación cron.
Python
import requests
import os
API_KEY = os.environ.get('SCAVIO_API_KEY', 'your_scavio_api_key')
TARGET_KEYWORDS = ['best crm software', 'lead scoring tools', 'email outreach automation']
def collect_serp_data():
conn = sqlite3.connect(DB_PATH)
for query in TARGET_KEYWORDS:
response = requests.post(
'https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'query': query, 'country_code': 'us'}
)
data = response.json()
for result in data.get('organic_results', []):
conn.execute(
'INSERT INTO positions (query, url, position, title) VALUES (?, ?, ?, ?)',
(query, result.get('link', ''), result.get('position', 0), result.get('title', ''))
)
if data.get('featured_snippet'):
conn.execute(
'INSERT INTO features (query, feature_type, content) VALUES (?, ?, ?)',
(query, 'featured_snippet', data['featured_snippet'].get('text', ''))
)
if data.get('ai_overview'):
conn.execute(
'INSERT INTO features (query, feature_type, content) VALUES (?, ?, ?)',
(query, 'ai_overview', str(len(data['ai_overview'].get('citations', []))) + ' citations')
)
conn.commit()
conn.close()
print(f'Collected data for {len(TARGET_KEYWORDS)} keywords')Paso 3: Crear consultas de detección de cambios de posición
Escriba consultas SQL que comparen las últimas posiciones con la ejecución anterior para detectar cambios de clasificación. Estas consultas alimentan el panel de alertas del tablero.
Python
def get_position_changes(query: str) -> list[dict]:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
rows = conn.execute('''
WITH latest AS (
SELECT url, position, fetched_at,
ROW_NUMBER() OVER (PARTITION BY url ORDER BY fetched_at DESC) AS rn
FROM positions WHERE query = ?
),
previous AS (
SELECT url, position, fetched_at,
ROW_NUMBER() OVER (PARTITION BY url ORDER BY fetched_at DESC) AS rn
FROM positions WHERE query = ? AND fetched_at < (SELECT MAX(fetched_at) FROM positions WHERE query = ?)
)
SELECT l.url, l.position AS current_pos, p.position AS prev_pos,
p.position - l.position AS change
FROM latest l LEFT JOIN previous p ON l.url = p.url AND p.rn = 1
WHERE l.rn = 1 AND (p.position IS NULL OR l.position != p.position)
ORDER BY ABS(COALESCE(change, 100)) DESC
''', (query, query, query)).fetchall()
conn.close()
return [dict(r) for r in rows]Paso 4: Crear el panel de Flask
Cree una aplicación Flask mínima que proporcione una página de panel que muestre las clasificaciones actuales, los cambios de posición y la apariencia de las funciones para todas las palabras clave rastreadas.
Python
from flask import Flask, render_template_string
app = Flask(__name__)
DASHBOARD_HTML = '''
<html><head><title>SERP Ops Dashboard</title>
<style>
body { font-family: sans-serif; margin: 2rem; }
table { border-collapse: collapse; width: 100%; margin: 1rem 0; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
.up { color: green; } .down { color: red; }
</style></head>
<body>
<h1>SERP Ops Dashboard</h1>
{% for query, changes in data.items() %}
<h2>{{ query }}</h2>
<table><tr><th>URL</th><th>Position</th><th>Change</th></tr>
{% for c in changes %}
<tr>
<td>{{ c.url }}</td><td>{{ c.current_pos }}</td>
<td class="{{ "up" if (c.change or 0) > 0 else "down" }}">{{ c.change if c.change else "NEW" }}</td>
</tr>
{% endfor %}
</table>
{% endfor %}
</body></html>
'''
@app.route('/')
def dashboard():
data = {}
for kw in TARGET_KEYWORDS:
data[kw] = get_position_changes(kw)
return render_template_string(DASHBOARD_HTML, data=data)
if __name__ == '__main__':
app.run(port=5050, debug=True)Ejemplo en Python
Python
import os
import sqlite3
import requests
from pathlib import Path
from flask import Flask, render_template_string
API_KEY = os.environ.get('SCAVIO_API_KEY', 'your_scavio_api_key')
ENDPOINT = 'https://api.scavio.dev/api/v1/search'
DB_PATH = Path('serp_dashboard.db')
TARGET_KEYWORDS = ['best crm software', 'lead scoring tools', 'email outreach automation']
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.executescript('''
CREATE TABLE IF NOT EXISTS positions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query TEXT NOT NULL, url TEXT NOT NULL,
position INTEGER NOT NULL, title TEXT,
fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS features (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query TEXT NOT NULL, feature_type TEXT NOT NULL,
content TEXT, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
''')
conn.close()
def collect_serp_data():
conn = sqlite3.connect(DB_PATH)
for query in TARGET_KEYWORDS:
response = requests.post(
ENDPOINT,
headers={'x-api-key': API_KEY},
json={'query': query, 'country_code': 'us'}
)
data = response.json()
for result in data.get('organic_results', []):
conn.execute(
'INSERT INTO positions (query, url, position, title) VALUES (?, ?, ?, ?)',
(query, result.get('link', ''), result.get('position', 0), result.get('title', ''))
)
if data.get('featured_snippet'):
conn.execute(
'INSERT INTO features (query, feature_type, content) VALUES (?, ?, ?)',
(query, 'featured_snippet', data['featured_snippet'].get('text', ''))
)
conn.commit()
conn.close()
print(f'Collected data for {len(TARGET_KEYWORDS)} keywords')
def get_position_changes(query: str) -> list[dict]:
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
rows = conn.execute('''
WITH ranked AS (
SELECT url, position, fetched_at,
DENSE_RANK() OVER (ORDER BY fetched_at DESC) AS run_rank
FROM positions WHERE query = ?
)
SELECT r1.url, r1.position AS current_pos, r2.position AS prev_pos,
COALESCE(r2.position - r1.position, 0) AS change
FROM ranked r1 LEFT JOIN ranked r2 ON r1.url = r2.url AND r2.run_rank = 2
WHERE r1.run_rank = 1 ORDER BY r1.position
''', (query,)).fetchall()
conn.close()
return [dict(r) for r in rows]
app = Flask(__name__)
DASHBOARD_HTML = '''<html><head><title>SERP Ops Dashboard</title></head><body>
<h1>SERP Ops Dashboard</h1>
{% for query, changes in data.items() %}
<h2>{{ query }}</h2>
<table border="1" cellpadding="6"><tr><th>URL</th><th>Position</th><th>Change</th></tr>
{% for c in changes %}
<tr><td>{{ c.url }}</td><td>{{ c.current_pos }}</td>
<td>{{ c.change if c.change else "NEW" }}</td></tr>
{% endfor %}</table>
{% endfor %}</body></html>'''
@app.route('/')
def dashboard():
data = {kw: get_position_changes(kw) for kw in TARGET_KEYWORDS}
return render_template_string(DASHBOARD_HTML, data=data)
if __name__ == '__main__':
init_db()
collect_serp_data()
app.run(port=5050, debug=True)Ejemplo en JavaScript
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY || 'your_scavio_api_key';
const ENDPOINT = 'https://api.scavio.dev/api/v1/search';
const Database = require('better-sqlite3');
const express = require('express');
const db = new Database('serp_dashboard.db');
const TARGET_KEYWORDS = ['best crm software', 'lead scoring tools', 'email outreach automation'];
db.exec('CREATE TABLE IF NOT EXISTS positions (id INTEGER PRIMARY KEY AUTOINCREMENT, query TEXT, url TEXT, position INTEGER, title TEXT, fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP)');
async function collectData() {
const insert = db.prepare('INSERT INTO positions (query, url, position, title) VALUES (?, ?, ?, ?)');
for (const query of TARGET_KEYWORDS) {
const response = await fetch(ENDPOINT, {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us' })
});
const data = await response.json();
for (const r of (data.organic_results || [])) {
insert.run(query, r.link || '', r.position || 0, r.title || '');
}
}
console.log('Data collected for ' + TARGET_KEYWORDS.length + ' keywords');
}
function getChanges(query) {
return db.prepare('SELECT url, position AS current_pos FROM positions WHERE query = ? ORDER BY fetched_at DESC, position ASC LIMIT 10').all(query);
}
async function main() {
await collectData();
const app = express();
app.get('/', (req, res) => {
let html = '<h1>SERP Ops Dashboard</h1>';
for (const kw of TARGET_KEYWORDS) {
const rows = getChanges(kw);
html += '<h2>' + kw + '</h2><table border="1"><tr><th>URL</th><th>Position</th></tr>';
rows.forEach(r => { html += '<tr><td>' + r.url + '</td><td>' + r.current_pos + '</td></tr>'; });
html += '</table>';
}
res.send(html);
});
app.listen(5050, () => console.log('Dashboard at http://localhost:5050'));
}
main().catch(console.error);Salida esperada
JSON
Collected data for 3 keywords
* Serving Flask app
* Running on http://127.0.0.1:5050
-- Dashboard shows: --
best crm software
hubspot.com/crm Position: 1 Change: 0
salesforce.com Position: 2 Change: +1
zoho.com/crm Position: 3 Change: -1
lead scoring tools
hubspot.com/lead-scoring Position: 1 Change: NEW
email outreach automation
apollo.io Position: 1 Change: 0