如何使用 Scavio Search API 构建代理工具后备链

import asyncio
import httpx
import json
from pathlib import Path

SCAVIO_API_KEY = "your-api-key"
CACHE_FILE = Path("search_cache.json")

async def primary_search(query: str) -> dict:
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.post(
            "https://api.scavio.dev/api/v1/search",
            headers={"x-api-key": SCAVIO_API_KEY},
            json={"query": query, "num_results": 10}
        )
        resp.raise_for_status()
        return resp.json()

async def secondary_search(query: str) -> dict:
    broad_query = " ".join(query.split()[:4])
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.post(
            "https://api.scavio.dev/api/v1/search",
            headers={"x-api-key": SCAVIO_API_KEY},
            json={"query": broad_query, "num_results": 20}
        )
        resp.raise_for_status()
        data = resp.json()
        terms = set(query.lower().split())
        results = data.get("results", [])
        scored = [(sum(1 for t in terms if t in r.get("title", "").lower()), r) for r in results]
        scored.sort(key=lambda x: x[0], reverse=True)
        data["results"] = [r for _, r in scored[:10]]
        return data

def write_cache(query: str, data: dict):
    cache = json.loads(CACHE_FILE.read_text()) if CACHE_FILE.exists() else {}
    cache[query] = data
    CACHE_FILE.write_text(json.dumps(cache, indent=2))

def read_cache(query: str) -> dict | None:
    if not CACHE_FILE.exists():
        return None
    return json.loads(CACHE_FILE.read_text()).get(query)

async def fallback_search(query: str) -> dict:
    try:
        data = await primary_search(query)
        if len(data.get("results", [])) >= 3:
            write_cache(query, data)
            return {"tier": "primary", **data}
    except Exception:
        pass
    try:
        data = await secondary_search(query)
        if data.get("results"):
            write_cache(query, data)
            return {"tier": "secondary", **data}
    except Exception:
        pass
    cached = read_cache(query)
    if cached:
        return {"tier": "cache", **cached}
    return {"tier": "none", "results": []}

result = asyncio.run(fallback_search("best headless CMS 2026"))
print(f"Served from: {result['tier']}")
print(f"Results: {len(result.get('results', []))}")

const SCAVIO_API_KEY = "your-api-key";
const fs = require("fs");
const CACHE_FILE = "search_cache.json";

async function primarySearch(query) {
  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, num_results: 10 })
  });
  if (!resp.ok) throw new Error("Primary search failed: " + resp.status);
  return resp.json();
}

async function secondarySearch(query) {
  const broad = query.split(" ").slice(0, 4).join(" ");
  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: broad, num_results: 20 })
  });
  if (!resp.ok) throw new Error("Secondary search failed");
  const data = await resp.json();
  const terms = new Set(query.toLowerCase().split(" "));
  data.results = data.results
    .map(r => ({ score: [...terms].filter(t => (r.title || "").toLowerCase().includes(t)).length, ...r }))
    .sort((a, b) => b.score - a.score)
    .slice(0, 10);
  return data;
}

function writeCache(query, data) {
  const cache = fs.existsSync(CACHE_FILE) ? JSON.parse(fs.readFileSync(CACHE_FILE)) : {};
  cache[query] = data;
  fs.writeFileSync(CACHE_FILE, JSON.stringify(cache, null, 2));
}

function readCache(query) {
  if (!fs.existsSync(CACHE_FILE)) return null;
  return JSON.parse(fs.readFileSync(CACHE_FILE))[query] || null;
}

async function fallbackSearch(query) {
  try {
    const data = await primarySearch(query);
    if ((data.results || []).length >= 3) { writeCache(query, data); return { tier: "primary", ...data }; }
  } catch {}
  try {
    const data = await secondarySearch(query);
    if (data.results?.length) { writeCache(query, data); return { tier: "secondary", ...data }; }
  } catch {}
  const cached = readCache(query);
  if (cached) return { tier: "cache", ...cached };
  return { tier: "none", results: [] };
}

fallbackSearch("best headless CMS 2026").then(r => {
  console.log("Served from:", r.tier);
  console.log("Results:", (r.results || []).length);
});

Served from: primary
Results: 10