rank_tracker.py

"""
Bing SERP rank tracker - a real, runnable use case on the Chocodata Bing Scraper API.

Checks where a domain ranks on Bing for a list of keywords, stores every observation
in a local SQLite dataset, and prints what moved since the previous run. Rank tracking
is the single most common reason people scrape a search engine, so it is here end to
end rather than as a snippet.

    pip install requests
    export CHOCODATA_API_KEY="your_key"     # free key (1,000 requests, one-time): https://chocodata.com
    python bing_scraper_api_codes/rank_tracker.py example.com "web scraping" "python tutorial"
    # ... run it again later to see the movements

Export the dataset to CSV any time:
    sqlite3 -header -csv bing_ranks.db "SELECT * FROM ranks ORDER BY ts" > ranks.csv

Cost: 1 request (5 credits) per keyword per run.

Note: Bing server-renders roughly 5 organic results to this endpoint, and there is no
working pagination, so a domain outside the top ~5 reports as "not found" rather than
as rank 27. That makes this useful for tracking top-5 presence, not for full top-100
rank tracking.

Docs: https://chocodata.com/docs
"""
import os
import sqlite3
import sys
import time
from urllib.parse import urlparse

import requests

API = "https://api.chocodata.com/api/v1/bing/search"
KEY = os.environ.get("CHOCODATA_API_KEY")
DB = "bing_ranks.db"

if not KEY:
    sys.exit("Set CHOCODATA_API_KEY first. Free key: https://chocodata.com")


def _check(r) -> None:
    """Map the API's documented errors onto actionable messages instead of a traceback."""
    if r.status_code == 400:
        issues = r.json().get("issues", [])
        detail = "; ".join(f"{'.'.join(str(p) for p in i.get('path', []))}: {i.get('message')}" for i in issues)
        sys.exit(f"400 invalid_params: {detail or 'check your query string'}")
    if r.status_code == 401:
        sys.exit("401 INVALID_API_KEY: key missing or not recognised. Get one: https://chocodata.com")
    if r.status_code == 402:
        sys.exit("402 INSUFFICIENT_CREDITS: balance exhausted. Top up or upgrade: https://chocodata.com/pricing")
    if r.status_code == 429:
        sys.exit("429 RATE_LIMITED: over your plan's concurrency. Back off and retry.")
    if r.status_code == 502:
        sys.exit("502 extraction_failed: Bing did not return a parseable SERP for this request. "
                 "Retryable, and you were not charged.")
    r.raise_for_status()


def fetch(keyword: str, country: str = "us") -> list[dict]:
    """One API call -> the current ranked organic results for this keyword."""
    r = requests.get(API, params={"api_key": KEY, "q": keyword, "country": country}, timeout=90)
    _check(r)
    return r.json().get("organic_results", [])


def host_of(link: str | None) -> str:
    if not link:
        return ""
    return (urlparse(link).hostname or "").lower().removeprefix("www.")


def rank_of(results: list[dict], domain: str) -> tuple[int | None, str | None]:
    """Position of the first result on `domain`, plus the URL that ranked."""
    target = domain.lower().removeprefix("www.")
    for row in results:
        h = host_of(row.get("link"))
        if h == target or h.endswith("." + target):
            return row["position"], row.get("link")
    return None, None


def setup(conn: sqlite3.Connection) -> None:
    conn.execute(
        """CREATE TABLE IF NOT EXISTS ranks (
               domain TEXT, keyword TEXT, position INTEGER, url TEXT, ts INTEGER,
               PRIMARY KEY (domain, keyword, ts)
           )"""
    )


def previous(conn: sqlite3.Connection, domain: str, keyword: str) -> int | None:
    row = conn.execute(
        "SELECT position FROM ranks WHERE domain = ? AND keyword = ? ORDER BY ts DESC LIMIT 1",
        (domain, keyword),
    ).fetchone()
    return row[0] if row else None


def main(domain: str, keywords: list[str]) -> None:
    conn = sqlite3.connect(DB)
    setup(conn)
    now = int(time.time())
    moves = 0

    for kw in keywords:
        results = fetch(kw)
        pos, url = rank_of(results, domain)
        before = previous(conn, domain, kw)
        shown = str(pos) if pos else "-"

        if before is not None and before != pos:
            moves += 1
            if pos is None:
                note = f"DROPPED OUT (was {before})"
            elif before is None:
                note = f"ENTERED at {pos}"
            else:
                arrow = "UP  " if pos < before else "DOWN"
                note = f"{arrow} {before} -> {pos}"
            print(f"{kw[:38]:38} {note}")
        else:
            top = results[0]["link"] if results else "-"
            print(f"{kw[:38]:38} rank {shown:>3}  (of {len(results)} shown; #1 is {host_of(top) or '-'})")

        conn.execute("INSERT OR REPLACE INTO ranks VALUES (?,?,?,?,?)", (domain, kw, pos, url, now))

    conn.commit()
    obs = conn.execute("SELECT COUNT(*) FROM ranks WHERE domain = ?", (domain,)).fetchone()[0]
    conn.close()

    print(f"\n{len(keywords)} keyword(s) checked for {domain} | {moves} movement(s) | {obs} observations in {DB}")
    if moves == 0:
        print("No movement yet. Run it again tomorrow, or schedule it (cron / GitHub Actions).")


if __name__ == "__main__":
    if len(sys.argv) < 3:
        sys.exit('Usage: python bing_scraper_api_codes/rank_tracker.py <domain> "<keyword>" ["<keyword>" ...]')
    main(sys.argv[1], sys.argv[2:])