Build a Rank Tracker with SERP API

# requirements.txt
requests>=2.28.0
schedule>=1.1.0
sqlite3  # Built-in with Python

# Install dependencies
pip install requests schedule

import sqlite3
from datetime import datetime

def init_database():
    """Initialize SQLite database for storing rankings"""
    conn = sqlite3.connect('rankings.db')
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS keywords (
            id INTEGER PRIMARY KEY,
            keyword TEXT NOT NULL,
            domain TEXT NOT NULL,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS rankings (
            id INTEGER PRIMARY KEY,
            keyword_id INTEGER,
            position INTEGER,
            url TEXT,
            title TEXT,
            country TEXT DEFAULT 'us',
            tracked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (keyword_id) REFERENCES keywords(id)
        )
    ''')
    
    conn.commit()
    return conn

# Initialize the database
conn = init_database()

import requests
import time

API_KEY = "your_searlo_api_key"
DOMAIN = "yourdomain.com"

def search_google(query, country="us", num_results=100):
    """Search Google using Searlo API"""
    response = requests.get(
        "https://api.searlo.tech/v1/search",
        headers={"Authorization": f"Bearer {API_KEY}"},
        params={
            "q": query,
            "gl": country,
            "num": num_results,
        }
    )
    response.raise_for_status()
    return response.json()

def find_domain_position(results, domain):
    """Find domain position in search results"""
    for result in results.get("organic_results", []):
        url = result.get("link", "")
        if domain.lower() in url.lower():
            return {
                "position": result["position"],
                "url": url,
                "title": result.get("title", ""),
            }
    return None  # Not in top 100

def track_keyword(keyword, domain, country="us"):
    """Track a single keyword and return position data"""
    try:
        results = search_google(keyword, country=country)
        position_data = find_domain_position(results, domain)
        
        return {
            "keyword": keyword,
            "domain": domain,
            "country": country,
            "position": position_data["position"] if position_data else None,
            "url": position_data["url"] if position_data else None,
            "title": position_data["title"] if position_data else None,
            "tracked_at": datetime.now().isoformat(),
            "serp_features": {
                "featured_snippet": results.get("featured_snippet") is not None,
                "ai_overview": results.get("ai_overview") is not None,
                "local_pack": results.get("local_pack") is not None,
            }
        }
    except Exception as e:
        return {"keyword": keyword, "error": str(e)}

def track_keywords_batch(keywords, domain, country="us"):
    """Track multiple keywords with rate limiting"""
    results = []
    
    for i, keyword in enumerate(keywords):
        print(f"Tracking {i+1}/{len(keywords)}: {keyword}")
        
        result = track_keyword(keyword, domain, country)
        results.append(result)
        
        # Respect rate limits (10 req/sec on Starter tier)
        if i < len(keywords) - 1:
            time.sleep(0.15)  # ~6.6 req/sec to stay under limit
    
    return results

def save_rankings(conn, keyword_id, ranking_data):
    """Save ranking data to database"""
    cursor = conn.cursor()
    cursor.execute('''
        INSERT INTO rankings (keyword_id, position, url, title, country)
        VALUES (?, ?, ?, ?, ?)
    ''', (
        keyword_id,
        ranking_data.get("position"),
        ranking_data.get("url"),
        ranking_data.get("title"),
        ranking_data.get("country", "us")
    ))
    conn.commit()

# Track keywords for your domain
keywords = [
    "serp api",
    "google search api",
    "serp scraping api",
    "best serp api 2026",
    "serpapi alternative"
]

rankings = track_keywords_batch(keywords, "searlo.tech")
for r in rankings:
    print(f"{r['keyword']}: Position {r.get('position', 'Not ranked')}")

import schedule
import time

def daily_tracking_job():
    """Run daily tracking job"""
    print(f"Starting daily tracking at {datetime.now()}")
    
    keywords = get_keywords_from_db()  # Implement this
    rankings = track_keywords_batch(keywords, DOMAIN)
    
    # Save to database
    for ranking in rankings:
        save_rankings(conn, ranking["keyword_id"], ranking)
    
    # Optional: Send email report
    send_ranking_report(rankings)
    
    print(f"Completed tracking {len(rankings)} keywords")

def get_keywords_from_db():
    """Get keywords to track from database"""
    cursor = conn.cursor()
    cursor.execute("SELECT id, keyword FROM keywords")
    return [{"id": row[0], "keyword": row[1]} for row in cursor.fetchall()]

# Schedule daily tracking at 6 AM
schedule.every().day.at("06:00").do(daily_tracking_job)

# Run scheduler
print("Rank tracker started. Press Ctrl+C to stop.")
while True:
    schedule.run_pending()
    time.sleep(60)

COUNTRIES = ["us", "uk", "ca", "au", "de"]

def track_multi_country(keyword, domain):
    """Track ranking across multiple countries"""
    results = {}
    
    for country in COUNTRIES:
        result = track_keyword(keyword, domain, country=country)
        results[country] = result.get("position")
        time.sleep(0.15)
    
    return results

# Track "serp api" across countries
multi_rankings = track_multi_country("serp api", "searlo.tech")
print("Rankings by country:")
for country, position in multi_rankings.items():
    print(f"  {country.upper()}: {position or 'Not ranked'}")

def detect_ranking_changes(keyword_id, threshold=5):
    """Detect significant ranking changes"""
    cursor = conn.cursor()
    cursor.execute('''
        SELECT position, tracked_at 
        FROM rankings 
        WHERE keyword_id = ? 
        ORDER BY tracked_at DESC 
        LIMIT 2
    ''', (keyword_id,))
    
    rows = cursor.fetchall()
    if len(rows) < 2:
        return None
    
    current, previous = rows[0][0], rows[1][0]
    
    if current is None or previous is None:
        return {"type": "visibility_change", "from": previous, "to": current}
    
    change = previous - current  # Positive = improvement
    
    if abs(change) >= threshold:
        return {
            "type": "significant_change",
            "from": previous,
            "to": current,
            "change": change,
            "direction": "up" if change > 0 else "down"
        }
    
    return None

# Check for changes after tracking
changes = detect_ranking_changes(keyword_id=1, threshold=3)
if changes:
    print(f"Alert! Ranking changed from {changes['from']} to {changes['to']}")