如何在 Python 中使用 Google Search API 构建排名跟踪器

import csv
import os
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse, urlunsplit

import requests

API_URL = "https://api.scrapeless.com/api/v1/scraper/request"


def normalized_host(value: str) -> str:
    candidate = value if "://" in value else urlunsplit(("https", value, "", "", ""))
    host = (urlparse(candidate).hostname or "").lower().rstrip(".")
    return host.removeprefix("www.")


def host_matches(result_url: str, target_domain: str, include_subdomains=True) -> bool:
    result_host = normalized_host(result_url)
    target_host = normalized_host(target_domain)
    if not result_host or not target_host:
        return False
    return result_host == target_host or (
        include_subdomains and result_host.endswith(f".{target_host}")
    )


def organic_results(payload: dict) -> list[dict]:
    if isinstance(payload.get("organic_results"), list):
        return payload["organic_results"]
    data = payload.get("data", {})
    if isinstance(data, dict) and isinstance(data.get("organic_results"), list):
        return data["organic_results"]
    return []


def find_rank(payload: dict, target_domain: str) -> dict:
    for fallback_position, item in enumerate(organic_results(payload), start=1):
        url = item.get("link") or item.get("url") or ""
        if host_matches(url, target_domain):
            return {
                "position": item.get("position", fallback_position),
                "ranking_url": url,
                "title": item.get("title", ""),
            }
    return {"position": None, "ranking_url": "", "title": ""}


def fetch_serp(keyword: str, *, gl="us", hl="en", device="desktop") -> dict:
    api_key = os.environ["SCRAPELESS_API_KEY"]
    response = requests.post(
        API_URL,
        headers={"x-api-token": api_key, "Content-Type": "application/json"},
        json={
            "actor": "scraper.google.search",
            "input": {
                "q": keyword,
                "gl": gl,
                "hl": hl,
                "google_domain": "google.com",
                "device": device,
                "start": 0,
            },
        },
        timeout=60,
    )
    response.raise_for_status()
    return response.json()


def append_snapshot(path: Path, row: dict) -> None:
    fields = [
        "observed_at", "keyword", "target_domain", "position",
        "ranking_url", "title", "gl", "hl", "device", "google_domain",
    ]
    exists = path.exists()
    with path.open("a", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields)
        if not exists:
            writer.writeheader()
        writer.writerow(row)


def track(keyword: str, target_domain: str, output="rank_history.csv") -> dict:
    settings = {"gl": "us", "hl": "en", "device": "desktop"}
    payload = fetch_serp(keyword, **settings)
    match = find_rank(payload, target_domain)
    row = {
        "observed_at": datetime.now(timezone.utc).isoformat(),
        "keyword": keyword,
        "target_domain": normalized_host(target_domain),
        "position": match["position"] or "",
        "ranking_url": match["ranking_url"],
        "title": match["title"],
        **settings,
        "google_domain": "google.com",
    }
    append_snapshot(Path(output), row)
    return row


if __name__ == "__main__":
    print(track("web scraping api", "scrapeless.com"))