google_search_scraper.py

"""Run the hosted Google Search SERP Scraper and print Dataset items."""

from __future__ import annotations

import json
import os
from collections.abc import Iterator
from pathlib import Path
from typing import Any

from apify_client import ApifyClient


ACTOR_ID = "datascraperes/google-serp-scraper"
ROOT = Path(__file__).resolve().parents[2]
INPUT_PATH = ROOT / "data" / "sample-input.json"


def run_google_search(run_input: dict[str, Any]) -> Iterator[dict[str, Any]]:
    """Run the hosted Actor and yield default Dataset items."""
    token = os.environ.get("APIFY_API_TOKEN")
    if not token:
        raise RuntimeError("Set APIFY_API_TOKEN before running this example.")

    client = ApifyClient(token)
    run = client.actor(ACTOR_ID).call(run_input=run_input)
    yield from client.dataset(run["defaultDatasetId"]).iterate_items()


def main() -> None:
    run_input = json.loads(INPUT_PATH.read_text(encoding="utf-8"))
    for item in run_google_search(run_input):
        print(json.dumps(item, ensure_ascii=False))


if __name__ == "__main__":
    main()