许多应用程序需要来自多个搜索平台的数据：谷歌用于网络上下文，亚马逊用于产品定价，YouTube用于视频内容，沃尔玛用于零售替代品。管理四个独立的 API 集成（每个集成具有不同的身份验证、速率限制和响应模式）既复杂又昂贵。 Scavio API 提供了一个端点，将所有四个平台统一在一个一致的接口下。本教程构建了一个统一的搜索功能，可以查询所有四个平台并返回标准化结果集。
前置条件
- Python 3.10 或更高版本
- 请求已安装库
- Scavio API 密钥
- 对Python并发编程的基本了解
操作指南
步骤 1: 定义平台请求配置
创建一个帮助程序，从单个查询字符串为每个平台构建正确的请求正文。
Python
def platform_body(platform: str, query: str) -> dict:
base = {"platform": platform, "query": query}
if platform == "google":
return {"query": query, "country_code": "us"}
if platform == "amazon":
return {**base, "marketplace": "US"}
return base步骤 2: 同时扇出请求
使用 ThreadPoolExecutor 同时查询所有四个平台以最大限度地减少延迟。
Python
from concurrent.futures import ThreadPoolExecutor
PLATFORMS = ["google", "amazon", "youtube", "walmart"]
def search_all(query: str) -> dict:
def fetch(platform):
r = requests.post(ENDPOINT, headers={"x-api-key": API_KEY},
json=platform_body(platform, query))
r.raise_for_status()
return r.json()
with ThreadPoolExecutor(max_workers=4) as ex:
futs = {p: ex.submit(fetch, p) for p in PLATFORMS}
return {p: f.result() for p, f in futs.items()}步骤 3: 每个平台的提取结果计数
构建一个摘要，显示每个平台为查询返回的结果数量。
Python
def summarize(results: dict) -> dict:
return {
"google": len(results["google"].get("organic_results", [])),
"amazon": len(results["amazon"].get("products", [])),
"youtube": len(results["youtube"].get("videos", [])),
"walmart": len(results["walmart"].get("products", [])),
}步骤 4: 打印统一摘要
显示每个平台的结果计数和热门项目。
Python
results = search_all("portable bluetooth speaker")
counts = summarize(results)
for platform, count in counts.items():
print(f"{platform}: {count} results")Python 示例
Python
import os
import requests
from concurrent.futures import ThreadPoolExecutor
API_KEY = os.environ.get("SCAVIO_API_KEY", "your_scavio_api_key")
ENDPOINT = "https://api.scavio.dev/api/v1/search"
def body(platform: str, query: str) -> dict:
if platform == "google": return {"query": query, "country_code": "us"}
if platform == "amazon": return {"platform": platform, "query": query, "marketplace": "US"}
return {"platform": platform, "query": query}
def search_all(query: str) -> dict:
def fetch(p):
r = requests.post(ENDPOINT, headers={"x-api-key": API_KEY}, json=body(p, query))
r.raise_for_status()
return r.json()
with ThreadPoolExecutor(max_workers=4) as ex:
futs = {p: ex.submit(fetch, p) for p in ["google", "amazon", "youtube", "walmart"]}
return {p: f.result() for p, f in futs.items()}
if __name__ == "__main__":
data = search_all("portable bluetooth speaker")
print(f"Google: {len(data['google'].get('organic_results', []))} results")
print(f"Amazon: {len(data['amazon'].get('products', []))} products")
print(f"YouTube: {len(data['youtube'].get('videos', []))} videos")
print(f"Walmart: {len(data['walmart'].get('products', []))} products")JavaScript 示例
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY || "your_scavio_api_key";
const ENDPOINT = "https://api.scavio.dev/api/v1/search";
function body(platform, query) {
if (platform === "google") return { query, country_code: "us" };
if (platform === "amazon") return { platform, query, marketplace: "US" };
return { platform, query };
}
async function fetchPlatform(platform, query) {
const res = await fetch(ENDPOINT, {
method: "POST",
headers: { "x-api-key": API_KEY, "Content-Type": "application/json" },
body: JSON.stringify(body(platform, query))
});
return [platform, await res.json()];
}
async function searchAll(query) {
const platforms = ["google", "amazon", "youtube", "walmart"];
const results = await Promise.all(platforms.map(p => fetchPlatform(p, query)));
return Object.fromEntries(results);
}
searchAll("portable bluetooth speaker").then(data => {
console.log(`Google: ${data.google.organic_results?.length || 0} results`);
console.log(`Amazon: ${data.amazon.products?.length || 0} products`);
console.log(`YouTube: ${data.youtube.videos?.length || 0} videos`);
console.log(`Walmart: ${data.walmart.products?.length || 0} products`);
}).catch(console.error);预期输出
JSON
Google: 10 results
Amazon: 20 products
YouTube: 10 videos
Walmart: 16 products
Top Amazon: JBL Flip 6 Portable Speaker — $79.95
Top Walmart: Anker Soundcore 3 — $35.99