Scrape Trustpilot Category Rankings (Top Companies + Ratings) with ProxiesAPI

python3 -m venv .venv
source .venv/bin/activate

pip install requests beautifulsoup4 lxml python-dotenv

export PROXIESAPI_PROXY_URL="http://YOUR_USERNAME:YOUR_PASSWORD@gw.proxiesapi.com:8080"

import os
import time
import random
import requests

PROXY_URL = os.getenv("PROXIESAPI_PROXY_URL")
TIMEOUT = (10, 30)  # connect, read

session = requests.Session()


def fetch(url: str) -> str:
    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/123.0.0.0 Safari/537.36"
        ),
        "Accept-Language": "en-US,en;q=0.9",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
    }

    proxies = None
    if PROXY_URL:
        proxies = {"http": PROXY_URL, "https": PROXY_URL}

    r = session.get(url, headers=headers, proxies=proxies, timeout=TIMEOUT)
    r.raise_for_status()
    return r.text


def polite_sleep(min_s: float = 1.0, max_s: float = 2.2) -> None:
    time.sleep(random.uniform(min_s, max_s))

import re
from urllib.parse import urljoin
from bs4 import BeautifulSoup

BASE = "https://www.trustpilot.com"

RATING_RE = re.compile(r"(\d+(?:\.\d+)?)")
REVIEWS_RE = re.compile(r"([\d,]+)\s+reviews", re.IGNORECASE)


def parse_rating_and_reviews(text: str) -> tuple[float | None, int | None]:
    rating = None
    reviews = None

    if text:
        m = RATING_RE.search(text)
        if m:
            rating = float(m.group(1))

        m = REVIEWS_RE.search(text)
        if m:
            reviews = int(m.group(1).replace(",", ""))

    return rating, reviews


def parse_category_page(html: str) -> list[dict]:
    soup = BeautifulSoup(html, "lxml")

    rows = []

    # Trustpilot category pages contain many /review/ links.
    # We dedupe by href and only keep the first occurrence.
    seen = set()

    for a in soup.select('a[href^="/review/"]'):
        href = a.get("href")
        if not href or href in seen:
            continue
        seen.add(href)

        # Card-ish container: climb a few parents.
        card = a
        for _ in range(6):
            if not getattr(card, "parent", None):
                break
            card = card.parent

        card_text = " ".join((card.get_text(" ", strip=True) if card else a.get_text(" ", strip=True)).split())

        rating, reviews = parse_rating_and_reviews(card_text)

        name = a.get_text(" ", strip=True)
        review_url = urljoin(BASE, href)

        # The website domain usually appears as plain text near the company name.
        dm = re.search(r"\b([a-z0-9-]+\.[a-z]{2,})\b", card_text, re.IGNORECASE)
        domain = dm.group(1).lower() if dm else None

        rows.append(
            {
                "name": name,
                "review_url": review_url,
                "website": domain,
                "rating": rating,
                "review_count": reviews,
            }
        )

    return rows

from urllib.parse import urljoin


def find_next_page(html: str) -> str | None:
    soup = BeautifulSoup(html, "lxml")
    nav = soup.select_one('nav[aria-label="Pagination"]')
    if not nav:
        return None

    a = nav.select_one('a[aria-label="Next page"], a[rel="next"]')
    if not a:
        return None

    href = a.get("href")
    if not href:
        return None

    return urljoin(BASE, href)


def crawl_category(category_slug: str, max_pages: int = 5) -> list[dict]:
    url = f"{BASE}/categories/{category_slug}"

    all_rows: list[dict] = []
    page = 1

    while url and page <= max_pages:
        html = fetch(url)

        rows = parse_category_page(html)
        for r in rows:
            r["category_slug"] = category_slug
            r["page"] = page
        all_rows.extend(rows)

        url = find_next_page(html)
        page += 1
        polite_sleep()

    return all_rows

import csv


def write_csv(path: str, rows: list[dict]) -> None:
    if not rows:
        raise ValueError("no rows")

    fieldnames = list(rows[0].keys())
    with open(path, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        w.writeheader()
        w.writerows(rows)


if __name__ == "__main__":
    category = "travel_insurance_company"
    rows = crawl_category(category_slug=category, max_pages=3)
    write_csv("trustpilot_category_rankings.csv", rows)
    print(f"wrote {len(rows)} rows")