Google Maps 評論抓取工具

import os
import time
import uuid

import requests

API = "https://mapsscraper.dev/api/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['MAPSSCRAPER_API_KEY']}"}


def run(task, units):
    # Keep the key: a retry of the same page must send it again.
    key = str(uuid.uuid4())
    job = requests.post(
        f"{API}/jobs",
        json=task,
        headers={**HEADERS, "Idempotency-Key": key, "X-Max-Units": str(units)},
        timeout=30,
    )
    job.raise_for_status()
    job_id = job.json()["id"]
    while True:
        status = requests.get(f"{API}/jobs/{job_id}", headers=HEADERS, timeout=30).json()["status"]
        if status not in ("queued", "running"):
            return job_id, status
        time.sleep(5)


task = {"kind": "reviews", "business_id": "sample-business-id", "limit": 100,
        "review_options": {"sort_by": "newest"}}
units = requests.post(f"{API}/estimate", json=task, headers=HEADERS, timeout=30).json()["units"]
job_id, status = run(task, units)

if status in ("completed", "partial"):
    page = requests.get(
        f"{API}/jobs/{job_id}/results",
        params={"offset": 0, "limit": 100},
        headers=HEADERS,
        timeout=30,
    ).json()
    for review in page["items"]:
        print(review.get("rating"), review.get("review_text"))

    # The next source page is a separate paid task: preview it, then submit
    # its exact task only if you need it.
    preview = requests.get(f"{API}/jobs/{job_id}/continuation", headers=HEADERS, timeout=30).json()
    if preview["available"]:
        run(preview["task"], preview["units"])