Custom SEO Dashboard: Raw API Architecture

# Architecture: SERP API -> SQLite -> Transform -> Dashboard
#
# [Cron/n8n scheduler]
#     |
#     v
# [SERP API] (DataForSEO queue or Scavio)
#     |
#     v
# [SQLite / PostgreSQL]
#     |
#     v
# [Transform: rank deltas, feature detection]
#     |
#     v
# [Looker Studio / Grafana / Custom UI]

import requests, os, sqlite3, json
from datetime import datetime

DB_PATH = "serp_data.db"

def init_db():
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS serp_results (
            id INTEGER PRIMARY KEY,
            keyword TEXT,
            domain TEXT,
            position INTEGER,
            serp_features TEXT,
            captured_at TEXT
        )
    """)
    conn.commit()
    return conn

def collect_serp(keyword: str, conn: sqlite3.Connection):
    """Pull SERP data via Scavio and store positions."""
    resp = requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers={"x-api-key": os.environ["SCAVIO_API_KEY"]},
        json={"query": keyword, "platform": "google", "num_results": 50},
        timeout=15,
    )
    data = resp.json()
    now = datetime.utcnow().isoformat()

    features = []
    if data.get("ai_overview"):
        features.append("ai_overview")
    if data.get("people_also_ask"):
        features.append("paa")
    if data.get("local_pack"):
        features.append("local_pack")

    for i, result in enumerate(data.get("organic_results", []), 1):
        domain = result.get("displayed_link", "").split("/")[0]
        conn.execute(
            "INSERT INTO serp_results (keyword, domain, position, serp_features, captured_at) VALUES (?, ?, ?, ?, ?)",
            (keyword, domain, i, json.dumps(features), now),
        )
    conn.commit()

# Daily collection: 200 keywords = 200 credits = $1.00/day on Scavio
# Or $0.12/day on DataForSEO queue

def compute_rank_deltas(conn: sqlite3.Connection, keyword: str) -> list:
    """Compare today's positions to yesterday's."""
    rows = conn.execute("""
        SELECT domain, position, captured_at
        FROM serp_results
        WHERE keyword = ?
        ORDER BY captured_at DESC
    """, (keyword,)).fetchall()

    # Group by date
    by_date = {}
    for domain, pos, captured in rows:
        date = captured[:10]
        if date not in by_date:
            by_date[date] = {}
        by_date[date][domain] = pos

    dates = sorted(by_date.keys(), reverse=True)
    if len(dates) < 2:
        return []

    today, yesterday = by_date[dates[0]], by_date[dates[1]]
    deltas = []
    for domain, pos in today.items():
        prev = yesterday.get(domain)
        if prev:
            delta = prev - pos  # Positive = improved
            deltas.append({"domain": domain, "position": pos, "delta": delta})

    return sorted(deltas, key=lambda x: abs(x["delta"]), reverse=True)

# Export to CSV for Looker Studio / Sheets import
import csv

def export_for_dashboard(conn: sqlite3.Connection, output_path: str):
    rows = conn.execute("""
        SELECT keyword, domain, position, serp_features,
               DATE(captured_at) as date
        FROM serp_results
        ORDER BY date DESC, keyword, position
    """).fetchall()

    with open(output_path, "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["keyword", "domain", "position", "features", "date"])
        writer.writerows(rows)

# For automated pipelines, push directly to Google Sheets API
# or use n8n's Google Sheets node