Google CSE Migration Checklist Workflow

import requests, os, json
from pathlib import Path

SCAVIO_KEY = os.environ["SCAVIO_API_KEY"]
CSE_KEY = os.environ.get("GOOGLE_CSE_KEY", "")
CSE_CX = os.environ.get("GOOGLE_CSE_CX", "")
SH = {"x-api-key": SCAVIO_KEY, "Content-Type": "application/json"}

QUERIES_FILE = Path("cse_queries.json")

def search_scavio(query: str) -> list:
    resp = requests.post(
        "https://api.scavio.dev/api/v1/search",
        headers=SH,
        json={"query": query, "platform": "google"},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json().get("organic", [])

def search_cse(query: str) -> list:
    if not CSE_KEY or not CSE_CX:
        return []
    resp = requests.get(
        "https://www.googleapis.com/customsearch/v1",
        params={"key": CSE_KEY, "cx": CSE_CX, "q": query},
        timeout=15,
    )
    return resp.json().get("items", []) if resp.ok else []

def compare_results(scavio: list, cse: list) -> dict:
    scavio_urls = set(r.get("url", "") for r in scavio[:10])
    cse_urls = set(r.get("link", "") for r in cse[:10])
    overlap = scavio_urls & cse_urls
    return {
        "scavio_count": len(scavio),
        "cse_count": len(cse),
        "overlap": len(overlap),
        "scavio_only": len(scavio_urls - cse_urls),
        "cse_only": len(cse_urls - scavio_urls),
    }

def run_migration():
    queries = json.loads(QUERIES_FILE.read_text())
    report = []
    for query in queries:
        scavio_results = search_scavio(query)
        cse_results = search_cse(query)
        comparison = compare_results(scavio_results, cse_results)
        comparison["query"] = query
        comparison["scavio_ok"] = len(scavio_results) > 0
        report.append(comparison)

    coverage = sum(1 for r in report if r["scavio_ok"]) / max(len(report), 1) * 100
    Path("cse_migration_report.json").write_text(json.dumps(report, indent=2))
    print(f"Migration report: {len(report)} queries tested, {coverage:.0f}% coverage")
    for r in report:
        status = "OK" if r["scavio_ok"] else "GAP"
        print(f"  [{status}] {r['query']}: scavio={r['scavio_count']}, cse={r['cse_count']}, overlap={r['overlap']}")

run_migration()

const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
const CSE_KEY = process.env.GOOGLE_CSE_KEY || '';
const CSE_CX = process.env.GOOGLE_CSE_CX || '';
const SH = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'};
const fs = await import('fs');

const queries = JSON.parse(fs.readFileSync('cse_queries.json', 'utf8'));

async function searchScavio(query) {
  const r = await fetch('https://api.scavio.dev/api/v1/search', {method:'POST', headers:SH, body:JSON.stringify({query, platform:'google'})});
  return (await r.json()).organic || [];
}

async function searchCse(query) {
  if (!CSE_KEY || !CSE_CX) return [];
  try {
    const r = await fetch('https://www.googleapis.com/customsearch/v1?key='+CSE_KEY+'&cx='+CSE_CX+'&q='+encodeURIComponent(query));
    return (await r.json()).items || [];
  } catch { return []; }
}

function compareResults(scavio, cse) {
  const sUrls = new Set(scavio.slice(0,10).map(r=>r.url||''));
  const cUrls = new Set(cse.slice(0,10).map(r=>r.link||''));
  const overlap = [...sUrls].filter(u=>cUrls.has(u)).length;
  return {scavioCount:scavio.length, cseCount:cse.length, overlap, scavioOnly:sUrls.size-overlap, cseOnly:cUrls.size-overlap};
}

const report = [];
for (const query of queries) {
  const scavioResults = await searchScavio(query);
  const cseResults = await searchCse(query);
  const comp = compareResults(scavioResults, cseResults);
  report.push({query, ...comp, scavioOk:scavioResults.length>0});
}

const coverage = Math.round(report.filter(r=>r.scavioOk).length/Math.max(report.length,1)*100);
fs.writeFileSync('cse_migration_report.json', JSON.stringify(report, null, 2));
console.log('Migration report: '+report.length+' queries, '+coverage+'% coverage');
report.forEach(r => console.log('  ['+(r.scavioOk?'OK':'GAP')+'] '+r.query+': scavio='+r.scavioCount+', cse='+r.cseCount+', overlap='+r.overlap));