How to Scrape Tripadvisor Reviews With Python

python -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install "selenium>=4.11,<5" "beautifulsoup4>=4.12,<5"

python3 -m venv .venv
source .venv/bin/activate
python -m pip install "selenium>=4.11,<5" "beautifulsoup4>=4.12,<5"

https://www.tripadvisor.com/Hotel_Review-g190327-d264936-Reviews-1926_Le_Soleil_Hotel_Spa-Sliema_Island_of_Malta.html

https://www.tripadvisor.com/Hotel_Review-g190327-d264936-Reviews-or10-1926_Le_Soleil_Hotel_Spa-Sliema_Island_of_Malta.html

const cards = [...document.querySelectorAll('div[data-test-target="HR_CC_CARD"]')];
console.table(cards.slice(0, 3).map(card => ({
  id: card.getAttribute("data-reviewid")
      || card.querySelector("[data-reviewid]")?.getAttribute("data-reviewid"),
  title: card.querySelector('[data-test-target="review-title"] a')?.innerText,
  body: card.querySelector(
    'span[data-automation*="reviewText"], [data-automation="ugcReviewReadMore_Content"]'
  )?.innerText
})));
console.log("Loaded review cards:", cards.length);

$env:ROLA_PROXY_URL = Read-Host "Paste your Rola IP endpoint in http://host:port form"

printf "Paste your Rola IP endpoint in http://host:port form: "
read -r ROLA_PROXY_URL
export ROLA_PROXY_URL

proxy = proxy_endpoint(args.proxy)
if proxy:
    options.add_argument(f"--proxy-server={proxy}")

python tripadvisor_review_scraper.py --check-proxy --max-pages 2 --output-dir output/1926-hotel-rola

python tripadvisor_review_scraper.py --check-proxy --manual-setup --max-pages 2 --output-dir output/1926-hotel-rola

python tripadvisor_review_scraper.py --url "https://www.tripadvisor.com/Hotel_Review-g190327-d264936-Reviews-1926_Le_Soleil_Hotel_Spa-Sliema_Island_of_Malta.html" --max-pages 2 --delay 4 --timeout 30 --output-dir output/1926-hotel-next

output/1926-hotel-rola/
  reviews.csv
  reviews.json
  run.log

import csv
import json
from pathlib import Path

folder = Path("output/1926-hotel-rola")
payload = json.loads((folder / "reviews.json").read_text(encoding="utf-8"))
rows = payload["reviews"]
with (folder / "reviews.csv").open(encoding="utf-8-sig", newline="") as handle:
    csv_rows = list(csv.DictReader(handle))

assert rows, "The export has no reviews"
assert len(rows) == len(csv_rows), "CSV/JSON row counts differ"
assert len({row["review_id"] for row in rows}) == len(rows), "Duplicate review IDs"
for row, csv_row in zip(rows, csv_rows):
    assert row["char_length"] == len(row["review_text"]), "Body length mismatch"
    assert 1 <= row["rating"] <= 5, "Rating outside 1-5"
    for field, value in row.items():
        assert csv_row[field] == str(value), f"CSV/JSON mismatch: {field}"
print("Rows:", len(rows))
print("Stop reason:", payload["run"]["stop_reason"])
print("Error:", payload["run"]["error"])

"""Collect one Tripadvisor hotel's review cards and export CSV/JSON."""

from __future__ import annotations

import argparse
import csv
from datetime import datetime, timezone
import importlib.metadata
import ipaddress
import json
import logging
import os
from pathlib import Path
import platform
import re
import time
from urllib.parse import urljoin, urlsplit, urlunsplit

from bs4 import BeautifulSoup
from selenium import webdriver
from selenium.common.exceptions import StaleElementReferenceException
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait

TARGET_URL = (
    "https://www.tripadvisor.com/Hotel_Review-g190327-d264936-Reviews-"
    "1926_Le_Soleil_Hotel_Spa-Sliema_Island_of_Malta.html"
)
SELECTORS = {
    "card": 'div[data-test-target="HR_CC_CARD"]',
    "id": '[data-reviewid]',
    "title": 'div[data-test-target="review-title"] a span span, '
             'div[data-test-target="review-title"] a',
    "text": 'span[data-automation*="reviewText"], '
            'span[data-automation*="reviewText"] span, '
            '[data-automation="ugcReviewReadMore_Content"]',
    "rating": 'div[data-test-target="review-rating"] > span, '
              'svg[data-automation="bubbleRatingImage"]',
    "published": 'span.scrVn, div[data-test-target="review-date"]',
    "stay": 'span.teHYY, div[data-test-target="review-title"] + div',
    "next": 'a[data-smoke-attr="pagination-next-arrow"]',
}
EXPAND_XPATH = (
    ".//*[@data-test-target='expand-review'] | "
    ".//button[normalize-space(.)='Read more'] | "
    ".//a[normalize-space(.)='Read more'] | "
    ".//*[@role='button' and normalize-space(.)='Read more']"
)
FIELDS = [
    "review_id", "property_name", "page_num", "rating", "title",
    "review_text", "char_length", "published_date", "stay_date",
    "review_url", "source_page_url",
]
LOG = logging.getLogger("tripadvisor")


def proxy_endpoint(value):
    """Chrome proxy-server uses an endpoint; this workflow uses IP whitelisting."""
    if not value:
        return None
    try:
        parsed = urlsplit(value.strip())
        port = parsed.port
        valid = (parsed.scheme == "http" and parsed.hostname and port
                 and parsed.username is None and parsed.password is None
                 and parsed.path in ("", "/") and not parsed.query and not parsed.fragment)
    except ValueError:
        valid = False
    if not valid:
        raise ValueError("Use an HTTP endpoint in http://host:port form with IP-whitelist authentication; omit credentials")
    return f"http://{parsed.netloc}"


def check_proxy_ip(driver, timeout):
    driver.get("https://api.ipify.org?format=json")

    def read_ip(browser):
        try:
            body = browser.find_element(By.TAG_NAME, "body").text
            value = json.loads(body).get("ip", "")
            return str(ipaddress.ip_address(value))
        except (ValueError, TypeError, AttributeError):
            return False

    return WebDriverWait(driver, timeout).until(read_ip)


def clean(value):
    return re.sub(r"\s+", " ", value or "").strip()


def longest_text(card, selector):
    # Alternatives can select both a parent and its child. Do not concatenate.
    return max((clean(n.get_text(" ", strip=True))
                for n in card.select(selector)), key=len, default="")


def parse_rating(card):
    for node in card.select(SELECTORS["rating"]):
        nodes = [node, *node.select("[aria-label], [class], title")]
        for item in nodes:
            value = " ".join([
                item.get("aria-label", ""),
                " ".join(item.get("class", [])),
                item.get_text(" ", strip=True),
            ])
            bubble = re.search(r"\bbubble_(\d{2})\b", value)
            label = re.search(r"(\d(?:\.\d+)?)\s*(?:of|out of|/)\s*5\b", value, re.I)
            score = (int(bubble.group(1)) / 10 if bubble else
                     float(label.group(1)) if label else None)
            if score is not None and 1 <= score <= 5:
                return score
    return None


def parse_card(card, page_num, page_url, property_name):
    id_node = card if card.has_attr("data-reviewid") else card.select_one(SELECTORS["id"])
    review_id = id_node.get("data-reviewid", "") if id_node else ""
    review_url = ""
    for link in card.select('a[href*="/ShowUserReviews-"]'):
        match = re.search(r"-r(\d+)-", link["href"])
        if match and (not review_id or match.group(1) == review_id):
            review_id = review_id or match.group(1)
            review_url = urljoin(page_url, link["href"])
            break
    text = longest_text(card, SELECTORS["text"])
    rating = parse_rating(card)
    if not re.fullmatch(r"\d+", review_id) or not text or rating is None:
        raise ValueError("Missing review ID, body, or individual rating; inspect the review card selectors")
    published = longest_text(card, SELECTORS["published"])
    published = re.sub(r"^.*?\bwrote a review\s*", "", published, flags=re.I)
    published = re.sub(r"^(?:Written|Reviewed)\s*:?\s*", "", published, flags=re.I)
    stay = ""
    for node in card.select(SELECTORS["stay"]) + card.select("span"):
        value = clean(node.get_text(" ", strip=True))
        match = re.search(r"Date of stay:\s*([A-Za-z]+\s+\d{4})", value, re.I)
        if match:
            stay = match.group(1)
            break
    return {
        "review_id": review_id, "property_name": property_name,
        "page_num": page_num, "rating": rating,
        "title": longest_text(card, SELECTORS["title"]),
        "review_text": text, "char_length": len(text),
        "published_date": clean(published), "stay_date": stay,
        "review_url": review_url, "source_page_url": page_url,
    }


def check_page(driver):
    # Restriction detection uses headings, never guest review bodies.
    headings = " ".join(n.text for n in driver.find_elements(By.CSS_SELECTOR, "h1"))
    value = (driver.title + " " + headings).lower()
    if any(term in value for term in (
        "access denied", "verify you are human", "captcha", "too many requests",
    )):
        raise RuntimeError("Access restriction detected; stop and inspect the browser")


def wait_for_cards(driver, timeout, previous_ids):
    last = None
    since = time.monotonic()

    def ready(browser):
        nonlocal last, since
        check_page(browser)
        try:
            cards = browser.find_elements(By.CSS_SELECTOR, SELECTORS["card"])
            signature = tuple(c.get_attribute("outerHTML") for c in cards)
            if not signature:
                return False
            if signature != last:
                last, since = signature, time.monotonic()
                return False
            if time.monotonic() - since < 1:
                return False
            if previous_ids:
                soup = BeautifulSoup("".join(signature), "html.parser")
                ids = {n.get("data-reviewid") for n in soup.select("[data-reviewid]")}
                ids.update(re.findall(r"-r(\d+)-", str(soup)))
                if not ids - previous_ids:
                    return False
            return cards
        except StaleElementReferenceException:
            return False

    return WebDriverWait(driver, timeout, poll_frequency=0.5).until(ready)


def body_text(card):
    soup = BeautifulSoup(card.get_attribute("outerHTML"), "html.parser")
    return longest_text(soup, SELECTORS["text"])


def expansion_controls(card):
    return [n for n in card.find_elements(By.XPATH, EXPAND_XPATH)
            if n.is_displayed() and n.is_enabled()
            and n.get_attribute("aria-expanded") != "true"
            and clean(n.text).lower() != "read less"]


def expand_card(driver, card_index, timeout):
    clicks = 0
    for _ in range(3):
        card = driver.find_elements(By.CSS_SELECTOR, SELECTORS["card"])[card_index]
        controls = expansion_controls(card)
        if not controls:
            return clicks
        before = body_text(card)
        button = controls[0]
        driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", button)
        button.click()
        clicks += 1
        last, since = None, time.monotonic()

        def expanded(browser):
            nonlocal last, since
            try:
                current = browser.find_elements(By.CSS_SELECTOR, SELECTORS["card"])[card_index]
                text = body_text(current)
                if not text or text != last:
                    last, since = text, time.monotonic()
                    return False
                return (time.monotonic() - since >= 1
                        and (len(text) > len(before) or not expansion_controls(current)))
            except (StaleElementReferenceException, IndexError):
                return False

        WebDriverWait(driver, timeout, poll_frequency=0.5).until(expanded)
    card = driver.find_elements(By.CSS_SELECTOR, SELECTORS["card"])[card_index]
    if expansion_controls(card):
        raise RuntimeError("Read more control remains active; inspect expansion before continuing")
    return clicks


def canonical_page(url, property_id):
    parsed = urlsplit(url)
    host = (parsed.hostname or "").lower()
    if (parsed.scheme != "https" or not (host == "tripadvisor.com" or host.endswith(".tripadvisor.com"))
            or not parsed.path.startswith("/Hotel_Review-")
            or f"-d{property_id}-" not in parsed.path):
        raise ValueError("Pagination must stay on the same Tripadvisor hotel property")
    return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, parsed.query, ""))


def next_page(driver, property_id):
    for button in driver.find_elements(By.CSS_SELECTOR, SELECTORS["next"]):
        if not button.is_displayed():
            continue
        if button.get_attribute("aria-disabled") == "true" or button.get_attribute("disabled"):
            return None, "next_disabled"
        href = button.get_attribute("href")
        if not href:
            raise RuntimeError("Next control has no URL; inspect pagination")
        return canonical_page(urljoin(driver.current_url, href), property_id), None
    return None, "next_missing_check_page"


def export_files(folder, records, run):
    folder.mkdir(parents=True, exist_ok=True)
    with (folder / "reviews.csv").open("w", newline="", encoding="utf-8-sig") as handle:
        writer = csv.DictWriter(handle, fieldnames=FIELDS)
        writer.writeheader()
        writer.writerows(records)
    (folder / "reviews.json").write_text(
        json.dumps({"run": run, "reviews": records}, ensure_ascii=False, indent=2), encoding="utf-8")


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--url", default=TARGET_URL)
    parser.add_argument("--max-pages", type=int, default=2)
    parser.add_argument("--delay", type=float, default=4)
    parser.add_argument("--timeout", type=float, default=30)
    parser.add_argument("--output-dir", type=Path)
    parser.add_argument("--headless", action="store_true")
    parser.add_argument("--manual-setup", action="store_true",
                        help="Pause on the first hotel page for consent and review-section setup")
    parser.add_argument("--proxy", default=os.environ.get("ROLA_PROXY_URL") or None,
                        help="IP-whitelisted HTTP proxy endpoint; defaults to ROLA_PROXY_URL")
    parser.add_argument("--check-proxy", action="store_true",
                        help="Check the proxied browser's public IP before collecting reviews")
    args = parser.parse_args(argv)
    if args.manual_setup and args.headless:
        parser.error("Use --manual-setup with a visible browser; omit --headless")
    if args.max_pages < 1 or args.delay < 2 or args.timeout < 1:
        parser.error("Use --max-pages >= 1, --delay >= 2, and --timeout >= 1")
    try:
        proxy = proxy_endpoint(args.proxy)
    except ValueError as exc:
        parser.error(str(exc))
    if args.check_proxy and not proxy:
        parser.error("--check-proxy requires --proxy or ROLA_PROXY_URL")
    match = re.search(r"-d(\d+)-", args.url)
    if not match:
        parser.error("URL must contain a hotel property ID")
    property_id = match.group(1)
    url = canonical_page(args.url, property_id)
    folder = args.output_dir or Path("tripadvisor_output") / datetime.now().strftime("%Y%m%d-%H%M%S-%f")
    folder.mkdir(parents=True, exist_ok=True)
    handlers = [logging.StreamHandler(), logging.FileHandler(folder / "run.log", encoding="utf-8")]
    LOG.setLevel(logging.INFO)
    LOG.propagate = False
    for handler in handlers:
        handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
        LOG.addHandler(handler)
    options = webdriver.ChromeOptions()
    options.add_argument("--lang=en-US")
    options.add_experimental_option("prefs", {"intl.accept_languages": "en-US,en"})
    if proxy:
        options.add_argument(f"--proxy-server={proxy}")
    if args.headless:
        options.add_argument("--headless=new")
    records, seen_ids, visited = [], set(), set()
    run = {
        "started_at": datetime.now(timezone.utc).isoformat(), "target_url": url,
        "requested_locale": "en-US", "max_pages": args.max_pages,
        "delay_seconds": args.delay, "timeout_seconds": args.timeout,
        "proxy_enabled": bool(proxy), "proxy_egress_ip": None,
        "python": platform.python_version(), "os": platform.platform(),
        "selenium": importlib.metadata.version("selenium"),
        "beautifulsoup4": importlib.metadata.version("beautifulsoup4"),
        "page_breakdown": [], "stop_reason": "error", "error": None,
    }
    driver = None
    exit_code = 0
    try:
        driver = webdriver.Chrome(options=options)
        run["browser_version"] = driver.capabilities.get("browserVersion")
        driver.set_page_load_timeout(args.timeout)
        driver.execute_cdp_cmd("Network.enable", {})
        driver.execute_cdp_cmd("Network.setExtraHTTPHeaders", {"headers": {"Accept-Language": "en-US,en;q=0.9"}})
        if args.check_proxy:
            run["proxy_egress_ip"] = check_proxy_ip(driver, args.timeout)
            LOG.info("Browser proxy egress IP: %s", run["proxy_egress_ip"])
        for page_num in range(1, args.max_pages + 1):
            if url in visited:
                raise RuntimeError("Repeated page URL detected")
            visited.add(url)
            LOG.info("Loading page %d: %s", page_num, url)
            driver.get(url)
            if page_num == 1 and args.manual_setup:
                input("In Chrome, handle any consent dialog and scroll to the reviews. "
                      "Press Enter here to continue: ")
            cards = wait_for_cards(driver, args.timeout, seen_ids)
            actual_url = canonical_page(driver.current_url, property_id)
            page = {"page_num": page_num, "page_url": actual_url, "cards_found": len(cards),
                    "expand_clicks": 0, "new_reviews": 0, "duplicates": 0}
            run["page_breakdown"].append(page)
            for index in range(len(cards)):
                page["expand_clicks"] += expand_card(driver, index, args.timeout)
            soup = BeautifulSoup(driver.page_source, "html.parser")
            heading = soup.select_one("h1")
            property_name = clean(heading.get_text(" ", strip=True)) if heading else f"Tripadvisor property {property_id}"
            for card in soup.select(SELECTORS["card"]):
                row = parse_card(card, page_num, actual_url, property_name)
                if row["review_id"] in seen_ids:
                    page["duplicates"] += 1
                    continue
                seen_ids.add(row["review_id"])
                records.append(row)
                page["new_reviews"] += 1
                if not row["title"] or not row["published_date"] or not row["stay_date"]:
                    LOG.warning("Review %s has an empty title or date field", row["review_id"])
            LOG.info("Page %d: %d new, %d duplicate; total %d", page_num,
                     page["new_reviews"], page["duplicates"], len(records))
            if not page["new_reviews"]:
                raise RuntimeError("Page produced no new reviews")
            if page_num == args.max_pages:
                run["stop_reason"] = "max_pages_reached"
                break
            url, reason = next_page(driver, property_id)
            if url is None:
                run["stop_reason"] = reason
                break
            time.sleep(args.delay)
    except Exception as exc:
        exit_code = 1
        run["error"] = f"{type(exc).__name__}: {exc}"
        LOG.exception("Collection stopped; retaining rows already parsed")
    finally:
        run["finished_at"] = datetime.now(timezone.utc).isoformat()
        run["unique_reviews_exported"] = len(records)
        run["exit_code"] = exit_code
        if driver:
            try:
                driver.quit()
            except Exception:
                LOG.warning("Browser cleanup failed", exc_info=True)
        try:
            export_files(folder, records, run)
            LOG.info("Saved %d rows to %s; stop reason: %s", len(records), folder, run["stop_reason"])
        finally:
            for handler in handlers:
                LOG.removeHandler(handler)
                handler.close()
    return exit_code


if __name__ == "__main__":
    raise SystemExit(main())