Review Scraper API: Public Review Pages from Any Site

import json, requests

API = "https://api.proxyfield.io/v1"
HEADERS = {"Authorization": "Bearer <api_key>"}

# Review pages you may collect: sites whose terms allow it, or that gave you permission
competitors = {
    "competitor-a": {"site-1": "https://www.example-reviews.com/company/competitor-a?page={n}"},
    "competitor-b": {"site-1": "https://www.example-reviews.com/company/competitor-b?page={n}"},
}
items, meta = [], []
for comp, sites in competitors.items():
    for site, pattern in sites.items():
        for n in range(1, 11):                          # 10 review pages each
            items.append({"url": pattern.format(n=n)})
            meta.append({"competitor": comp, "site": site, "page": n})

jobs = []
for start in range(0, len(items), 1000):              # up to 1,000 items per batch
    r = requests.post(f"{API}/jobs/batch", headers=HEADERS, timeout=60, json={
        "api": "web_scrape",
        "params": {"render": False, "country": "us", "output": "json"},
        "items": items[start:start + 1000],
        "callback_url": "https://your-app.example/hooks/reviews",
        "tag": "competitor-reviews"})
    jobs += [j["id"] for j in r.json()["jobs"]]        # one job per item, in order
json.dump([dict(m, job=j) for m, j in zip(meta, jobs)], open("jobs.json", "w"))

import csv, json, requests
from bs4 import BeautifulSoup

API, HEADERS = "https://api.proxyfield.io/v1", {"Authorization": "Bearer <api_key>"}

def nodes(data):
    """Every JSON-LD object, including those inside @graph, lists and 'review'."""
    if isinstance(data, list):
        for d in data:
            yield from nodes(d)
    elif isinstance(data, dict):
        yield data
        for key in ("@graph", "review", "itemReviewed"):
            yield from nodes(data.get(key, []))

def reviews(html):
    for tag in BeautifulSoup(html, "html.parser").find_all("script", type="application/ld+json"):
        try:
            data = json.loads(tag.string or "")
        except ValueError:
            continue
        for n in nodes(data):
            if n.get("@type") == "Review":
                yield {"rating": (n.get("reviewRating") or {}).get("ratingValue"),
                       "date": n.get("datePublished"),
                       "text": " ".join((n.get("reviewBody") or "").split())[:2000]}

with open("reviews.csv", "w", newline="") as f:
    out = csv.DictWriter(f, ["competitor", "site", "page", "rating", "date", "text"])
    out.writeheader()
    for job in json.load(open("jobs.json")):
        res = requests.get(f"{API}/jobs/{job['job']}", headers=HEADERS, timeout=30).json()
        if res.get("status") != "done":
            print(res.get("status"), job)              # queued, running or failed: fetch it later
            continue
        rows = list(reviews(res["result"].get("content", "")))
        if not rows:
            print("no reviews on", job)                 # save the page, check it, add a site parser
        for row in rows:
            out.writerow({"competitor": job["competitor"], "site": job["site"],
                          "page": job["page"], **row})

competitor,site,page,rating,date,text
competitor-a,site-1,1,5,2026-10-21,"Setup took an afternoon and support answered the same day."
competitor-a,site-1,1,2,2026-10-19,"Reporting is slow once you pass a few thousand rows."
competitor-b,site-1,1,4,2026-10-20,"Good value for a small team; the mobile app lags behind the web."