Export TikTok Shop reviews to CSV: API and Python script

curl "https://api.toolzerhub.com/v1/tiktok-shop/product/reviews?product_id=1729718141631565851&sort=most_recent" \
  -H "x-api-key: $TOOLZERHUB_API_KEY"

{
  "data": {
    "product_reviews": [
      {
        "review_id": "7702390543619507981",
        "product_id": "1729718141631565851",
        "review_rating": 5,
        "review_time": "1793352606606",
        "is_verified_purchase": true,
        "is_incentivized_review": false,
        "reviewer_name": "N**k",
        "review_text": "Used for a day. And skin felt so good all day. No dryness. Can’t wait to see my outcome in week two",
        "review_images": [],
        "sku_specification": "Default",
        "review_country": "US"
      }
    ],
    "total_reviews": "80585",
    "has_more": true,
    "next_page": 2
  }
}

import csv
import os
import re
import sys
import time
from datetime import datetime, timezone

import requests

API = "https://api.toolzerhub.com/v1/tiktok-shop/product/reviews"
HEADERS = {"x-api-key": os.environ.get("TOOLZERHUB_API_KEY", "")}
FIELDS = ["product_id", "review_id", "rating", "posted", "verified", "incentivized", "sku",
          "text", "images", "country"]
calls = 0


def cell(value):
    # Spreadsheets run text that starts with = + - or @ as a formula; an apostrophe keeps it text.
    if isinstance(value, str) and value[:1] in ("=", "+", "-", "@", "\t", "\r"):
        return "'" + value
    return value


def get(params, tries=5):
    # 429 and 503 are retryable; the error body may say how long to wait.
    global calls
    for attempt in range(tries):
        r = requests.get(API, headers=HEADERS, params=params, timeout=90)
        calls += 1
        if r.status_code not in (429, 503) or attempt == tries - 1:
            r.raise_for_status()
            return r.json().get("data") or {}
        try:
            wait = r.json()["error"]["details"]["retryAfter"]
        except (ValueError, KeyError, TypeError):
            wait = 5 * (attempt + 1)
        time.sleep(min(wait, 60))


def row(review):
    ms = int(review.get("review_time") or 0)
    return {
        "product_id": review["product_id"],
        "review_id": review["review_id"],
        "rating": review.get("review_rating"),
        "posted": datetime.fromtimestamp(ms / 1000, timezone.utc).date().isoformat() if ms else "",
        "verified": review.get("is_verified_purchase"),
        "incentivized": review.get("is_incentivized_review"),
        "sku": cell(review.get("sku_specification") or ""),
        "text": cell((review.get("review_text") or "").replace("\n", " ")),
        "images": " ".join(review.get("review_images") or []),
        "country": review.get("review_country"),
    }


def export(writer, product_id, max_pages, sort, region):
    params = {"product_id": product_id, "sort": sort, "region": region}
    seen, total = set(), None
    for _ in range(max_pages):
        data = get(params)
        total = total or data.get("total_reviews")
        for review in data.get("product_reviews") or []:
            if review["review_id"] not in seen:  # pages can repeat a review
                seen.add(review["review_id"])
                writer.writerow(row(review))
        if not data.get("has_more") or not data.get("next_page"):
            break
        params["page"] = data["next_page"]
    print(f"{product_id}: {len(seen)} reviews saved of {total} listed")
    return len(seen)


def main(products, max_pages="10", sort="most_recent", region="US", out="tiktok_shop_reviews.csv"):
    # Accept product IDs or product page URLs; the ID is the long number in the URL.
    ids = [m for p in products.split(",") for m in re.findall(r"\d{15,30}", p)]
    with open(out, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=FIELDS)
        writer.writeheader()
        saved = sum(export(writer, pid, int(max_pages), sort, region) for pid in ids)
    print(f"wrote {saved} rows to {out} in {calls} calls")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        sys.exit("usage: python tiktok_shop_reviews.py PRODUCT_ID[,PRODUCT_ID] [max_pages]"
                 " [most_recent|recommended] [region] [file.csv]")
    if not HEADERS["x-api-key"]:
        sys.exit("set TOOLZERHUB_API_KEY first")
    main(*sys.argv[1:])

export TOOLZERHUB_API_KEY=your_key
python tiktok_shop_reviews.py "1729718141631565851,1729717811293949979,1729717561093951515" 10 most_recent US tts_reviews.csv

1729718141631565851: 200 reviews saved of 80585 listed
1729717811293949979: 68 reviews saved of 68 listed
1729717561093951515: 200 reviews saved of 16132 listed
wrote 468 rows to tts_reviews.csv in 24 calls