GEO Audit Weekly Pipeline

import requests
import json
from pathlib import Path
from datetime import datetime

API_KEY = "your_scavio_api_key"
TARGET_DOMAIN = "yourdomain.com"

KEYWORDS = [
    "best search API 2026",
    "SERP API for AI agents",
    "web search structured data API",
    # ... up to 50 keywords
]

def audit_keyword(keyword: str) -> dict:
    res = requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers={"x-api-key": API_KEY},
        json={"platform": "google", "query": keyword, "ai_overview": True, "num": 10},
        timeout=15,
    )
    res.raise_for_status()
    data = res.json()
    aio = data.get("ai_overview") or {}
    aio_text = aio.get("text", "")
    sources = aio.get("sources", [])
    cited = any(TARGET_DOMAIN in (s.get("link", "") or s.get("url", "")) for s in sources)

    # Determine organic rank
    organic_rank = None
    for r in data.get("organic", []):
        if TARGET_DOMAIN in r.get("link", ""):
            organic_rank = r.get("position")
            break

    if aio_text and cited:
        status = "cited"
        priority = 1  # Maintain
    elif aio_text and not cited:
        status = "opportunity"
        priority = 3  # High priority to optimize
    else:
        status = "no_aio"
        priority = 2  # Traditional SEO

    return {
        "keyword": keyword,
        "has_ai_overview": bool(aio_text),
        "cited": cited,
        "organic_rank": organic_rank,
        "status": status,
        "priority": priority,
        "aio_sources": len(sources),
    }

def run():
    date = datetime.utcnow().strftime("%Y-%m-%d")
    results = [audit_keyword(kw) for kw in KEYWORDS]

    # Sort by priority (opportunities first)
    results.sort(key=lambda x: (-x["priority"], x["keyword"]))

    aio_count = sum(1 for r in results if r["has_ai_overview"])
    cited_count = sum(1 for r in results if r["cited"])
    opp_count = sum(1 for r in results if r["status"] == "opportunity")

    report = {
        "date": date,
        "keywords_audited": len(KEYWORDS),
        "ai_overviews_present": aio_count,
        "citations": cited_count,
        "opportunities": opp_count,
        "results": results,
    }

    audit_dir = Path("geo_audits")
    audit_dir.mkdir(exist_ok=True)
    audit_dir.joinpath(f"audit_{date}.json").write_text(json.dumps(report, indent=2))

    print(f"GEO Audit {date}: {aio_count}/{len(KEYWORDS)} have AI Overviews")
    print(f"  Cited: {cited_count} | Opportunities: {opp_count} | No AIO: {len(KEYWORDS) - aio_count}")

if __name__ == "__main__":
    run()

const API_KEY = "your_scavio_api_key";
const TARGET_DOMAIN = "yourdomain.com";
const KEYWORDS = ["best search API 2026", "SERP API for AI agents", "web search structured data API"];

async function auditKeyword(keyword) {
  const res = await fetch("https://api.scavio.dev/api/v1/search", {
    method: "POST",
    headers: { "x-api-key": API_KEY, "content-type": "application/json" },
    body: JSON.stringify({ platform: "google", query: keyword, ai_overview: true, num: 10 }),
  });
  if (!res.ok) throw new Error(`scavio ${res.status}`);
  const data = await res.json();
  const aio = data.ai_overview ?? {};
  const sources = aio.sources ?? [];
  const cited = sources.some((s) => (s.link ?? s.url ?? "").includes(TARGET_DOMAIN));
  const hasAio = !!aio.text;
  return { keyword, hasAio, cited, status: hasAio && cited ? "cited" : hasAio ? "opportunity" : "no_aio" };
}

const results = [];
for (const kw of KEYWORDS) results.push(await auditKeyword(kw));
const aio = results.filter((r) => r.hasAio).length;
const cited = results.filter((r) => r.cited).length;
const opps = results.filter((r) => r.status === "opportunity").length;
console.log(`GEO Audit: ${aio}/${KEYWORDS.length} AIO, ${cited} cited, ${opps} opportunities`);