Daily Keyword Rank Pipeline

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

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

KEYWORDS = [
    "search API for developers",
    "SERP data API 2026",
    "structured search results API",
    # ... up to 50 keywords
]

def get_rank(keyword: str) -> int | None:
    res = requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers={"x-api-key": API_KEY},
        json={"platform": "google", "query": keyword, "num": 20},
        timeout=15,
    )
    res.raise_for_status()
    for r in res.json().get("organic", []):
        if TARGET_DOMAIN in r.get("link", ""):
            return r.get("position")
    return None

def run():
    date = datetime.utcnow().strftime("%Y-%m-%d")
    history_path = Path("rank_history.json")
    history = json.loads(history_path.read_text()) if history_path.exists() else {}

    today = {}
    alerts = []
    for kw in KEYWORDS:
        pos = get_rank(kw)
        today[kw] = pos
        prev = history.get(kw)
        if pos is not None and prev is not None:
            delta = prev - pos  # positive = improved
            if abs(delta) >= CHANGE_THRESHOLD:
                direction = "up" if delta > 0 else "down"
                alerts.append({"keyword": kw, "previous": prev, "current": pos, "delta": delta, "direction": direction})

    history.update({k: v for k, v in today.items() if v is not None})
    history_path.write_text(json.dumps(history, indent=2))

    # Save daily snapshot
    snapshots_dir = Path("rank_snapshots")
    snapshots_dir.mkdir(exist_ok=True)
    snapshots_dir.joinpath(f"ranks_{date}.json").write_text(json.dumps({"date": date, "ranks": today}, indent=2))

    ranked = sum(1 for v in today.values() if v is not None)
    print(f"Rank check {date}: {ranked}/{len(KEYWORDS)} keywords ranked")
    print(f"  Alerts: {len(alerts)} keywords moved {CHANGE_THRESHOLD}+ positions")
    for a in alerts:
        print(f"  {a['direction'].upper()} {abs(a['delta'])}  #{a['previous']} -> #{a['current']}  {a['keyword']}")

if __name__ == "__main__":
    run()

const API_KEY = "your_scavio_api_key";
const TARGET_DOMAIN = "yourdomain.com";
const THRESHOLD = 3;
const KEYWORDS = ["search API for developers", "SERP data API 2026", "structured search results API"];

async function getRank(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, num: 20 }),
  });
  if (!res.ok) throw new Error(`scavio ${res.status}`);
  const organic = (await res.json()).organic ?? [];
  const found = organic.find((r) => (r.link ?? "").includes(TARGET_DOMAIN));
  return found ? found.position : null;
}

const ranks = {};
const alerts = [];
for (const kw of KEYWORDS) {
  const pos = await getRank(kw);
  ranks[kw] = pos;
}
const ranked = Object.values(ranks).filter((v) => v !== null).length;
console.log(`Rank check: ${ranked}/${KEYWORDS.length} keywords found`);