手动营销研究每周要花费几个小时：检查竞争对手页面、跟踪关键词排名、扫描 Reddit 上的品牌提及信息。本教程构建了一个自动化研究代理，该代理使用 Scavio Search API 处理所有三个问题，将结果存储在结构化报告中，并且每天监控 10 个竞争对手和 50 个关键字的成本低于 1 美元/月。
前置条件
- Python 3.11+
- 来自 https://scavio.dev 的 Scavio API 密钥
- SQLite3（包含在 Python 中）
- 可选：用于警报的 Slack 或 Discord Webhook
操作指南
步骤 1: 设置研究代理配置
在配置文件中定义您的竞争对手、跟踪的关键字和监控目标。代理在启动时读取此内容以了解要研究什么。
Python
import json
from pathlib import Path
from dataclasses import dataclass, field
from datetime import date
@dataclass
class ResearchConfig:
brand: str
competitors: list[str]
keywords: list[str]
reddit_queries: list[str]
domain: str
@classmethod
def from_file(cls, path: str) -> "ResearchConfig":
data = json.loads(Path(path).read_text())
return cls(**data)
# Example config
config = ResearchConfig(
brand="Scavio",
domain="scavio.dev",
competitors=[
"serpapi.com",
"serper.dev",
"brightdata.com",
],
keywords=[
"best search API for agents 2026",
"cheap web scraping API",
"MCP search server",
],
reddit_queries=[
"search API recommendation",
"web scraping API affordable",
"MCP tools search",
]
)步骤 2: 监控竞争对手并检测页面更改
搜索每个竞争对手的关键页面并与以前的快照进行比较。标记新页面、删除的页面和重大内容更改。
Python
import httpx
import sqlite3
SCAVIO_API_KEY = "your-api-key"
DB_PATH = Path("research.db")
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.executescript("""
CREATE TABLE IF NOT EXISTS competitor_pages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
competitor TEXT, url TEXT, title TEXT,
snippet TEXT, first_seen TEXT, last_seen TEXT
);
CREATE TABLE IF NOT EXISTS keyword_ranks (
id INTEGER PRIMARY KEY AUTOINCREMENT,
keyword TEXT, domain TEXT, position INTEGER,
url TEXT, checked_at TEXT
);
CREATE TABLE IF NOT EXISTS reddit_mentions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
query TEXT, url TEXT, title TEXT,
snippet TEXT, found_at TEXT
);
""")
conn.commit()
conn.close()
async def monitor_competitor(client: httpx.AsyncClient, competitor: str) -> list[dict]:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": f"site:{competitor}", "num_results": 15}
)
resp.raise_for_status()
results = resp.json().get("results", [])
conn = sqlite3.connect(DB_PATH)
today = date.today().isoformat()
new_pages = []
for r in results:
url = r.get("url", "")
existing = conn.execute(
"SELECT id FROM competitor_pages WHERE url = ? AND competitor = ?",
(url, competitor)
).fetchone()
if existing:
conn.execute("UPDATE competitor_pages SET last_seen = ? WHERE id = ?", (today, existing[0]))
else:
conn.execute(
"INSERT INTO competitor_pages VALUES (NULL,?,?,?,?,?,?)",
(competitor, url, r.get("title", ""), r.get("description", "")[:300], today, today)
)
new_pages.append({"url": url, "title": r.get("title", "")})
conn.commit()
conn.close()
return new_pages步骤 3: 跟踪关键词排名并扫描 Reddit
检查关键词位置并在 Reddit 上搜索品牌提及情况。两者都使用相同的 Scavio Search API 和不同的查询模式。
Python
async def track_keywords(client: httpx.AsyncClient, keywords: list[str], domain: str):
conn = sqlite3.connect(DB_PATH)
today = date.today().isoformat()
for kw in keywords:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": kw, "num_results": 10}
)
resp.raise_for_status()
position = None
url = None
for i, r in enumerate(resp.json().get("results", [])):
if domain in r.get("url", ""):
position = i + 1
url = r["url"]
break
conn.execute(
"INSERT INTO keyword_ranks VALUES (NULL,?,?,?,?,?)",
(kw, domain, position, url, today)
)
conn.commit()
conn.close()
async def scan_reddit(client: httpx.AsyncClient, queries: list[str]) -> list[dict]:
conn = sqlite3.connect(DB_PATH)
today = date.today().isoformat()
new_mentions = []
for q in queries:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": f"site:reddit.com {q}", "num_results": 10}
)
resp.raise_for_status()
for r in resp.json().get("results", []):
url = r.get("url", "")
existing = conn.execute(
"SELECT id FROM reddit_mentions WHERE url = ?", (url,)
).fetchone()
if not existing:
conn.execute(
"INSERT INTO reddit_mentions VALUES (NULL,?,?,?,?,?)",
(q, url, r.get("title", ""), r.get("description", "")[:300], today)
)
new_mentions.append({"url": url, "title": r.get("title", ""), "query": q})
conn.commit()
conn.close()
return new_mentions步骤 4: 生成每日研究报告
运行所有三项研究任务，并编写包含新竞争对手页面、排名变化和 Reddit 提及的每日报告。
Python
import asyncio
async def daily_research(config: ResearchConfig) -> dict:
init_db()
report = {
"date": date.today().isoformat(),
"new_competitor_pages": [],
"keyword_rankings": [],
"reddit_mentions": [],
"credits_used": 0
}
async with httpx.AsyncClient(timeout=15) as client:
# Monitor competitors
for comp in config.competitors:
new_pages = await monitor_competitor(client, comp)
report["new_competitor_pages"].extend(
[{"competitor": comp, **p} for p in new_pages]
)
report["credits_used"] += 1
# Track keywords
await track_keywords(client, config.keywords, config.domain)
report["credits_used"] += len(config.keywords)
# Scan Reddit
mentions = await scan_reddit(client, config.reddit_queries)
report["reddit_mentions"] = mentions
report["credits_used"] += len(config.reddit_queries)
cost = report["credits_used"] * 0.005
report["cost_usd"] = cost
print(f"Marketing Research Report - {report['date']}")
print(f"New competitor pages: {len(report['new_competitor_pages'])}")
print(f"New Reddit mentions: {len(report['reddit_mentions'])}")
print(f"Credits: {report['credits_used']} | Cost: {cost:.3f}")
for p in report["new_competitor_pages"]:
print(f" NEW [{p['competitor']}]: {p['title']}")
for m in report["reddit_mentions"]:
print(f" REDDIT [{m['query']}]: {m['title']}")
return report
asyncio.run(daily_research(config))Python 示例
Python
import asyncio
import httpx
import sqlite3
from datetime import date
from pathlib import Path
SCAVIO_API_KEY = "your-api-key"
DB = Path("research.db")
async def main():
conn = sqlite3.connect(DB)
conn.execute("""CREATE TABLE IF NOT EXISTS keyword_ranks
(keyword TEXT, position INTEGER, checked_at TEXT)""")
keywords = ["best search API 2026", "cheap scraping API", "MCP search tool"]
today = date.today().isoformat()
async with httpx.AsyncClient(timeout=15) as client:
for kw in keywords:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": kw, "num_results": 10}
)
results = resp.json().get("results", [])
pos = next((i+1 for i, r in enumerate(results) if "scavio.dev" in r.get("url", "")), None)
conn.execute("INSERT INTO keyword_ranks VALUES (?,?,?)", (kw, pos, today))
status = f"#{pos}" if pos else "not found"
print(f" {kw}: {status}")
conn.commit()
cost = len(keywords) * 0.005
print(f"Credits: {len(keywords)} | Cost: {cost:.3f}")
conn.close()
asyncio.run(main())JavaScript 示例
JavaScript
const SCAVIO_API_KEY = "your-api-key";
const KEYWORDS = ["best search API 2026", "cheap scraping API", "MCP search tool"];
const TARGET = "scavio.dev";
async function trackKeyword(keyword) {
const resp = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": SCAVIO_API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ query: keyword, num_results: 10 })
});
const data = await resp.json();
const results = data.results || [];
const idx = results.findIndex(r => (r.url || "").includes(TARGET));
return { keyword, position: idx >= 0 ? idx + 1 : null };
}
async function main() {
const rankings = [];
for (const kw of KEYWORDS) {
const rank = await trackKeyword(kw);
rankings.push(rank);
console.log(` ${kw}: ${rank.position ? "#" + rank.position : "not found"}`);
}
console.log(`Credits: ${KEYWORDS.length} | Cost: $${(KEYWORDS.length * 0.005).toFixed(3)}`);
}
main();预期输出
JSON
Marketing Research Report - 2026-05-17
New competitor pages: 4
New Reddit mentions: 7
Credits: 16 | Cost: $0.080