使用搜索数据构建运营仪表板可以让 SEO 和营销团队实时查看 SERP 位置变化、功能片段外观和竞争对手动向，而无需依赖第三方排名跟踪器。 Scavio 搜索 API 会返回任何查询的结构化有机结果、特色片段、People Also Ask 框以及 AI 概述数据。通过按计划轮询目标查询、将结果存储在 SQLite 中并通过轻量级 Flask 仪表板提供服务，您将获得一个自托管的操作视图，该视图会根据预算允许的频率进行更新。本教程构建了从数据收集到可通过浏览器查看的仪表板的完整管道。
前置条件
- 安装了 Python 3.8 或更高版本
- requests、flask 和 sqlite3（标准库）可用
- 来自 scavio.dev 的 Scavio API 密钥
- 要跟踪的目标关键字列表
操作指南
步骤 1: 设置 SQLite 数据库架构
创建一个 SQLite 数据库，其中包含 SERP 快照、位置历史记录和功能外观表。位置表中的每一行都存储查询、URL、位置和时间戳，以便您可以跟踪一段时间内的移动情况。
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()步骤 2: 构建 SERP 数据收集器
编写一个函数，查询 Scavio API 中的每个目标关键字，提取有机位置和 SERP 特征，并将它们插入到 SQLite 中。按照 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')步骤 3: 构建位置变化检测查询
编写 SQL 查询，将最新位置与先前运行进行比较，以检测排名变化。这些查询为仪表板警报面板提供支持。
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]步骤 4: 创建 Flask 仪表板
构建一个最小的 Flask 应用程序，该应用程序提供一个仪表板页面，显示所有跟踪关键字的当前排名、位置变化和功能外观。
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)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)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);预期输出
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