La création d'un tableau de bord opérationnel avec des données de recherche offre aux équipes SEO et marketing une vue en direct des changements de position SERP, des apparitions de feature snippets et des mouvements des concurrents, sans dépendre de trackers de position tiers. L'API de recherche Scavio renvoie des résultats organiques structurés, des featured snippets, des boîtes People Also Ask et des données AI Overview pour toute requête. En interrogeant vos requêtes cibles selon un calendrier, en stockant les résultats dans SQLite et en les servant via un tableau de bord Flask léger, vous obtenez une vue opérationnelle auto-hébergée qui se met à jour aussi souvent que votre budget le permet. Ce didacticiel construit le pipeline complet, de la collecte des données à un tableau de bord visualisable dans le navigateur.
Prérequis
- Python 3.8 ou version ultérieure installé
- requests, flask et sqlite3 (bibliothèque standard) disponibles
- Une clé API Scavio depuis scavio.dev
- Une liste de mots-clés cibles à suivre
Parcours
Étape 1: Configurer le schéma de base de données SQLite
Créez une base de données SQLite avec des tables pour les instantanés SERP, l'historique des positions et les apparitions de features. Chaque ligne de la table positions stocke une requête, une URL, une position et un horodatage afin de suivre les mouvements dans le temps.
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()Étape 2: Construire le collecteur de données SERP
Écrivez une fonction qui interroge l'API Scavio pour chaque mot-clé cible, extrait les positions organiques et les features SERP, et les insère dans SQLite. Exécutez cela selon un planning cron.
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')Étape 3: Construire des requêtes de détection de changement de position
Écrivez des requêtes SQL qui comparent les dernières positions avec l'exécution précédente pour détecter les changements de classement. Ces requêtes alimentent le panneau d'alertes du tableau de bord.
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]Étape 4: Créer le tableau de bord Flask
Construisez une application Flask minimale qui sert une page de tableau de bord montrant les classements actuels, les changements de position et les apparitions de features pour tous les mots-clés suivis.
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)Exemple 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)Exemple 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);Sortie attendue
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