Combine GSC, GA4, and SERP Data in One Analytics Pipeline

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

API_KEY = "your_scavio_api_key"

def get_serp_position(keyword: str, domain: str) -> dict:
    """Get current SERP position for a keyword."""
    res = requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers={"x-api-key": API_KEY},
        json={"platform": "google", "query": keyword, "num": 30},
        timeout=15,
    )
    res.raise_for_status()
    data = res.json()
    position = None
    for item in data.get("organic", []):
        if domain in item.get("link", ""):
            position = item.get("position")
            break
    return {
        "keyword": keyword,
        "position": position,
        "has_ai_overview": bool(data.get("ai_overview")),
        "has_featured_snippet": bool(data.get("featured_snippet")),
    }

def build_combined_report(keywords: list[str], domain: str, gsc_data: dict, ga4_data: dict) -> dict:
    """Combine SERP positions with GSC and GA4 data."""
    rows = []
    for kw in keywords:
        serp = get_serp_position(kw, domain)
        gsc = gsc_data.get(kw, {})
        ga4 = ga4_data.get(kw, {})
        rows.append({
            "keyword": kw,
            "serp_position": serp["position"],
            "has_ai_overview": serp["has_ai_overview"],
            "gsc_impressions": gsc.get("impressions", 0),
            "gsc_clicks": gsc.get("clicks", 0),
            "gsc_ctr": gsc.get("ctr", 0),
            "ga4_sessions": ga4.get("sessions", 0),
            "ga4_conversions": ga4.get("conversions", 0),
        })
    # Flag keywords with drops across all three signals
    for row in rows:
        row["alert"] = (
            row["serp_position"] is not None
            and row["serp_position"] > 10
            and row["gsc_clicks"] > 0
        )
    return {
        "date": datetime.utcnow().strftime("%Y-%m-%d"),
        "domain": domain,
        "keywords": len(rows),
        "alerts": sum(1 for r in rows if r["alert"]),
        "rows": rows,
    }

# Example: GSC and GA4 data would come from their respective APIs
gsc_data = {"your keyword": {"impressions": 1200, "clicks": 85, "ctr": 0.07}}
ga4_data = {"your keyword": {"sessions": 80, "conversions": 5}}
report = build_combined_report(["your keyword"], "yourdomain.com", gsc_data, ga4_data)
print(f"Report: {report['keywords']} keywords, {report['alerts']} alerts")

const API_KEY = "your_scavio_api_key";

async function getSerpPosition(keyword, domain) {
  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: 30 }),
  });
  if (!res.ok) throw new Error(`scavio ${res.status}`);
  const data = await res.json();
  let position = null;
  for (const item of data.organic ?? []) {
    if (item.link?.includes(domain)) { position = item.position; break; }
  }
  return { keyword, position, hasAiOverview: !!data.ai_overview };
}

async function buildReport(keywords, domain, gscData, ga4Data) {
  const rows = [];
  for (const kw of keywords) {
    const serp = await getSerpPosition(kw, domain);
    const gsc = gscData[kw] ?? {};
    const ga4 = ga4Data[kw] ?? {};
    rows.push({ keyword: kw, serpPosition: serp.position, hasAiOverview: serp.hasAiOverview, gscImpressions: gsc.impressions ?? 0, gscClicks: gsc.clicks ?? 0, ga4Sessions: ga4.sessions ?? 0, ga4Conversions: ga4.conversions ?? 0, alert: serp.position > 10 && (gsc.clicks ?? 0) > 0 });
  }
  return { date: new Date().toISOString().slice(0, 10), keywords: rows.length, alerts: rows.filter((r) => r.alert).length, rows };
}

const report = await buildReport(["your keyword"], "yourdomain.com", { "your keyword": { impressions: 1200, clicks: 85 } }, { "your keyword": { sessions: 80, conversions: 5 } });
console.log(`Report: ${report.keywords} keywords, ${report.alerts} alerts`);