La recherche marketing manuelle engloutit des heures chaque semaine : vérifier les pages concurrentes, suivre les classements de mots-clés, scruter Reddit pour les mentions de marque. Ce tutoriel construit un agent de recherche automatisé qui gère les trois tâches en utilisant l'API Scavio Search, stocke les résultats dans des rapports structurés et coûte moins de 1 $ par mois pour une surveillance quotidienne de 10 concurrents et 50 mots-clés.
Prérequis
- Python 3.11+
- Une clé API Scavio depuis https://scavio.dev
- SQLite3 (inclus avec Python)
- Optionnel : un webhook Slack ou Discord pour les alertes
Parcours
Étape 1: Configurer la configuration de l'agent de recherche
Définissez vos concurrents, mots-clés suivis et cibles de surveillance dans un fichier de configuration. L'agent le lit au démarrage pour savoir quoi rechercher.
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",
]
)Étape 2: Surveiller les concurrents et détecter les modifications de page
Recherchez les pages clés de chaque concurrent et comparez avec les instantanés précédents. Signalez les nouvelles pages, les pages supprimées et les changements de contenu importants.
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Étape 3: Suivre les classements de mots-clés et scanner Reddit
Vérifiez les positions des mots-clés et recherchez Reddit pour les mentions de marque. Les deux utilisent la même API Scavio Search avec différents modèles de requête.
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Étape 4: Générer le rapport de recherche quotidien
Exécutez les trois tâches de recherche et compilez un rapport quotidien avec les nouvelles pages concurrentes, les changements de classement et les mentions Reddit.
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))Exemple 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())Exemple 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();Sortie attendue
Marketing Research Report - 2026-05-17
New competitor pages: 4
New Reddit mentions: 7
Credits: 16 | Cost: $0.080