App Store Review Monitoring: APIs, Limits and Where CAPTCHAs Fit

You can monitor your own App Store and Google Play reviews entirely through official APIs: the App Store Connect customerReviews endpoint and the Google Play Developer API reviews.list method. What shapes the design is what those APIs leave out. Play returns only reviews with text from the last seven days, Apple does not tag a review with an app version, and neither covers competitors. CAPTCHAs show up only at the edges: on third-party review sites behind Cloudflare, and on Google pages where the right response is to slow down.

This page is about getting review data in correctly and alerting on it. For the sentiment scoring and dashboard layer on top, see the review monitoring system build.

Pick the source for each review stream

Every review stream you care about falls into one of five rows, and each row has a different official route:

The first two rows cover most of the value. Start there, and add the others only when a stakeholder actually reads competitor data.

Your iOS app: App Store Connect customer reviews

Setup takes three steps:

1. In App Store Connect, open Users and Access, select the Integrations tab, choose App Store Connect API and, on Team Keys, generate a key. Only an Admin can create team keys, and the role you pick decides what the key can do. Copy the issuer ID shown near the top of the page and the key ID, then download the .p8 private key. The download link works once, and Apple keeps no copy.
2. Sign a JWT with ES256. The header carries kid (your key ID) and typ: JWT. The payload carries iss (issuer ID), iat, exp and aud: appstoreconnect-v1. For this endpoint Apple rejects a token whose lifetime (exp minus iat) is over 20 minutes, and it recommends reusing one token across requests instead of signing a new one each time (token rules).
3. Look up the app's resource ID with GET /v1/apps?filter[bundleId]=..., then page through customerReviews newest first.

"""asc_reviews.py: pull your own app's reviews from the App Store Connect API.
pip install "pyjwt[crypto]" requests  (Python 3.11+)"""
import os
import re
import time
from datetime import datetime

import jwt
import requests

API = "https://api.appstoreconnect.apple.com/v1"
_token = {"value": "", "made": 0}


def auth() -> dict:
    now = int(time.time())
    if now - _token["made"] > 600:  # reuse for 10 min; lifetime must stay <= 20 min
        with open(os.environ["ASC_KEY_PATH"]) as fh:  # AuthKey_<KEY_ID>.p8
            key = fh.read()
        claims = {"iss": os.environ["ASC_ISSUER_ID"], "iat": now,
                  "exp": now + 900, "aud": "appstoreconnect-v1"}
        _token.update(value=jwt.encode(claims, key, algorithm="ES256",
                                       headers={"kid": os.environ["ASC_KEY_ID"]}), made=now)
    return {"Authorization": f"Bearer {_token['value']}"}


def get(url: str, params: dict | None = None) -> dict:
    for _ in range(3):
        resp = requests.get(url, params=params, headers=auth(), timeout=30)
        if resp.status_code == 429:  # RATE_LIMIT_EXCEEDED over a rolling hour
            time.sleep(900)
            continue
        resp.raise_for_status()
        left = re.search(r"user-hour-rem:(\d+)", resp.headers.get("X-Rate-Limit", ""))
        if left and int(left.group(1)) < 100:  # other tools share this key's hour
            print("X-Rate-Limit low:", resp.headers["X-Rate-Limit"])
            time.sleep(60)
        return resp.json()
    raise RuntimeError(f"still rate limited: {url}")


def new_reviews(bundle_id: str, since: datetime) -> list[dict]:
    app_id = get(f"{API}/apps", {"filter[bundleId]": bundle_id})["data"][0]["id"]
    url = f"{API}/apps/{app_id}/customerReviews"
    params, rows = {"sort": "-createdDate", "limit": 200}, []
    while url:
        page = get(url, params)
        for item in page["data"]:
            a = item["attributes"]
            if datetime.fromisoformat(a["createdDate"]) <= since:
                return rows  # newest first, so everything after this is stored
            rows.append({"store": "app_store", "country": a.get("territory"),
                         "app_version": None,  # the endpoint does not expose one
                         "review_id": item["id"], "rating": a["rating"],
                         "title": a.get("title"), "body": a.get("body"),
                         "updated_at": a["createdDate"]})
        url, params = page["links"].get("next"), None  # next carries the query string
    return rows

The customerReviews endpoint gives you more than a firehose. filter[rating]=1,2 builds a low-star queue. filter[territory]=USA narrows to one storefront, and it takes three-letter territory codes. exists[publishedResponse]=false returns reviews without a published reply, which is close to a support team's inbox; Apple notes that a reply you already wrote may still be waiting to publish. Replies go through POST /v1/customerReviewResponses, which creates a response or overwrites your existing one, and they take a while to appear on the App Store.

Every response carries an X-Rate-Limit header such as user-hour-lim:3500;user-hour-rem:500;. The window is a rolling hour per API key, Apple notes that actual limits vary, and exceeding it returns HTTP 429 with RATE_LIMIT_EXCEEDED. An hourly job that stops at the first already-stored review uses a handful of requests. The pause on a low user-hour-rem protects a key that CI jobs or other tools also use.

The gap to plan around: a review's attributes are rating, title, body, reviewerNickname, createdDate and territory. There is no app version. For release tracking on iOS, compare createdDate against your own release timestamps.

Your Android app: Google Play Developer API

Setup:

1. Enable the Google Play Developer API in a Google Cloud project, create a service account and download its JSON key.
2. In Play Console, open Users and permissions, choose Invite new users and add the service account's email with access to the app. Google's reviews guide tells you to enable Reply to reviews for a service account that uses this API, so grant it even if the job only reads.
3. Call reviews.list with the https://www.googleapis.com/auth/androidpublisher scope.

The limits are sharp, and they are all on the reply-to-reviews guide. You can retrieve only reviews created or modified within the last week, only for production versions of the app, and only reviews that include a comment, so star-only ratings and testing-track feedback never appear. The list starts with the most recently created or edited review. Pages hold 10 reviews by default and up to 100 with maxResults; the next page is requested with token set to tokenPagination.nextPageToken. A review with a title has no separate title field: the title and body share text, separated by a tab character. translationLanguage (for example en or en_GB) returns translated text and keeps the original in originalText. The quota is 200 GET requests per hour and 2,000 replies per day, counted per app, and you can ask Google for more.

"""play_reviews.py: your own app's Google Play reviews (last 7 days, text only).
pip install google-auth requests"""
import os
from datetime import datetime, timezone

from google.auth.transport.requests import AuthorizedSession
from google.oauth2 import service_account

SCOPE = "https://www.googleapis.com/auth/androidpublisher"
BASE = "https://androidpublisher.googleapis.com/androidpublisher/v3/applications"


def play_reviews(package: str, since: datetime | None = None) -> list[dict]:
    creds = service_account.Credentials.from_service_account_file(
        os.environ["PLAY_SA_JSON"], scopes=[SCOPE])
    session = AuthorizedSession(creds)
    params, rows = {"maxResults": 100}, []
    while True:
        resp = session.get(f"{BASE}/{package}/reviews", params=params, timeout=30)
        resp.raise_for_status()
        page = resp.json()
        for review in page.get("reviews", []):
            user = next(c["userComment"] for c in review["comments"] if "userComment" in c)
            modified = datetime.fromtimestamp(int(user["lastModified"]["seconds"]), timezone.utc)
            if since and modified <= since:
                return rows  # most recently created or edited first: the rest is stored
            title, tab, body = user.get("text", "").partition("\t")
            if not tab:  # no title: the whole text is the body
                title, body = None, title
            rows.append({"store": "google_play", "language": user.get("reviewerLanguage"),
                         "app_version": user.get("appVersionName"),
                         "review_id": review["reviewId"], "rating": user["starRating"],
                         "title": title or None, "body": body.strip(),
                         "updated_at": modified.isoformat()})
        token = page.get("tokenPagination", {}).get("nextPageToken")
        if not token:
            return rows
        params = {"maxResults": 100, "token": token}


if __name__ == "__main__":
    for r in play_reviews(os.environ["PLAY_PACKAGE"]):
        print(r["updated_at"], r["app_version"], r["rating"], r["body"][:60])

The seven-day window changes how you operate the job. If it fails silently for a week, those reviews are gone from the API, so alert on failed runs, not only on bad reviews. For history and outage recovery, use the monthly CSV exports under Download reports > Reviews in Play Console. They are named reviews_[package_name]_YYYYMM.csv and also sit in a Cloud Storage bucket whose ID begins with pubsite_prod_rev (report formats). Downloading them needs the account-level permission View app information and download bulk reports (read-only). Two details break naive importers: the files are UTF-16, not UTF-8, and data is posted 3 to 7 days after it is captured. Use them to backfill, not to alert.

Competitor apps: what you can and cannot get

iOS: Apple's public customer-reviews feed

Apple still serves a keyless feed per app and storefront:

https://itunes.apple.com/{country}/rss/customerreviews/page={n}/id={app_id}/sortby=mostrecent/json

Each page holds up to 50 reviews, and page 11 gets HTTP 400 with the body "CustomerReviews RSS page depth is limited to 10". That makes roughly 500 recent reviews per country. Storefronts are separate, so loop over the countries you actually compete in. Unlike App Store Connect, each entry carries im:version, so version-level alerts work for competitors.

Treat the feed as best effort. When we probed five large apps' US storefronts on 29 September 2026, two returned 50 reviews on all ten pages. The other three each had two or three pages that came back as HTTP 200 with well-formed JSON and no entry array, at different page numbers for each app, and a rerun a few minutes later hit the same holes. New reviews push older ones down the list, so polling every few hours with an upsert gives a review more than one chance to land on a page that answers. Probe the feed before you build on it:

#!/usr/bin/env bash
# probe_rss.sh APP_ID [COUNTRY]: count entries on each page of Apple's review feed
set -euo pipefail
app_id="${1:?usage: probe_rss.sh APP_ID [COUNTRY]}"
country="${2:-us}"
for page in $(seq 1 10); do
  url="https://itunes.apple.com/${country}/rss/customerreviews/page=${page}/id=${app_id}/sortby=mostrecent/json"
  count=$(curl -s --max-time 20 "$url" | jq '[.feed.entry // [] | if type == "array" then .[] else . end] | length')
  echo "page ${page}: ${count} entries"
  sleep 2
done

The collector follows the same rules: scan all ten pages, never stop at the first empty one, skip entries without a rating, and deduplicate by review ID.

"""rss_reviews.py APP_ID [COUNTRY ...]: recent public reviews of any iOS app."""
import sqlite3
import sys
import time

import requests

from store import save

FEED = ("https://itunes.apple.com/{c}/rss/customerreviews/page={p}"
        "/id={app}/sortby=mostrecent/json")


def rss_reviews(app_id: str, countries: list[str]) -> list[dict]:
    seen, rows, empty = set(), [], 0
    for country in countries:
        for page in range(1, 11):  # page 11 is refused
            resp = requests.get(FEED.format(c=country, p=page, app=app_id), timeout=30)
            resp.raise_for_status()
            entries = resp.json().get("feed", {}).get("entry", [])
            entries = [entries] if isinstance(entries, dict) else entries
            empty += not entries  # a hole, not the end of the feed
            for e in entries:
                rid = e.get("id", {}).get("label")
                if "im:rating" not in e or rid in seen:
                    continue
                seen.add(rid)
                rows.append({"store": "app_store_rss", "country": country,
                             "app_version": e.get("im:version", {}).get("label"),
                             "review_id": rid, "rating": int(e["im:rating"]["label"]),
                             "title": e["title"]["label"], "body": e["content"]["label"],
                             "updated_at": e["updated"]["label"]})
            time.sleep(2)
    if empty == 10 * len(countries):
        raise RuntimeError("every page was empty: treat as a feed outage, not a quiet week")
    return rows


if __name__ == "__main__":
    found = rss_reviews(sys.argv[1], sys.argv[2:] or ["us"])
    print(save(sqlite3.connect("competitors.db"), found), "new or edited reviews")

Android: no official route

The Google Play Developer API serves only apps in your own Play Console account; there is no official API for another developer's reviews. Collecting them yourself is narrower than it looks. Play's robots.txt leaves app detail pages crawlable, but only a handful of reviews render there, and it disallows /store/getreviews, /store/xhr and every path under /_. Google's terms prohibit automated access that violates those instructions. For more than a sample, the realistic route is a commercial app-intelligence vendor that licenses the data.

If Google answers with "Our systems have detected unusual traffic from your computer network", treat it as a rate signal, not a puzzle. Google's help page says users can search normally once the automated traffic stops, so cut concurrency, add delay, cache what you already fetched and try later. For comparing competitors once the data is in, see the competitor analysis dashboard build.

Reviews outside the app stores: Google and Trustpilot

If you arrived from our old Google and Trustpilot aggregator page, the official routes are narrower than most tutorials suggest:

• Trustpilot. The APIs need a Trustpilot for Business account with the API module, which is sold in tiers of 200,000 calls a year, 1,000,000 a year, or unlimited. Resolve your business unit ID from your domain with GET /v1/business-units/find?name=..., then page through GET /v1/business-units/{businessUnitId}/reviews with your key in an apikey header; one call returns up to 100 reviews or 1 MB. That public endpoint omits customer emails and order IDs, which need the private endpoint and a business-user OAuth token. If you store the reviews, Trustpilot asks you to mirror its deletions through the Deletions API every 28 days.
• Google Business Profile. Reviews for locations you manage come from GET https://mybusiness.googleapis.com/v4/accounts/{accountId}/locations/{locationId}/reviews, up to 50 per page. Access is by application: the profile must be verified and active for 60+ days, you apply through the GBP API contact form ("Application for Basic API Access"), and the project's quota reads 0 QPM until Google approves it and 300 QPM after.
• Places API. Place details return at most 5 reviews, sorted by relevance. That is a sample for display, not a monitoring feed.

Store every review once and keep every edit

One table serves every source: store, country, language, app_version, review_id, rating, title, body and updated_at, keyed on (store, review_id). Apple's territory fills country, while Play offers only reviewerLanguage, which is the device language, not a country. Keep a history table, because Play reviews change: lastModified moves when a user edits, and a one-star review that becomes four stars after your fix is a signal worth keeping.

"""store.py: upsert on (store, review_id); keep the previous version on edits."""
import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS reviews (
  store TEXT NOT NULL, review_id TEXT NOT NULL, country TEXT, language TEXT,
  app_version TEXT, rating INTEGER, title TEXT, body TEXT, updated_at TEXT,
  first_seen TEXT DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (store, review_id));
CREATE TABLE IF NOT EXISTS review_history (
  store TEXT, review_id TEXT, rating INTEGER, title TEXT, body TEXT,
  updated_at TEXT, replaced_at TEXT DEFAULT CURRENT_TIMESTAMP);
"""
UPSERT = """
INSERT INTO reviews (store, review_id, country, language, app_version,
                     rating, title, body, updated_at)
VALUES (:store, :review_id, :country, :language, :app_version,
        :rating, :title, :body, :updated_at)
ON CONFLICT (store, review_id) DO UPDATE SET
  rating = excluded.rating, title = excluded.title, body = excluded.body,
  app_version = excluded.app_version, updated_at = excluded.updated_at
"""


def save(db: sqlite3.Connection, rows: list[dict]) -> int:
    db.executescript(SCHEMA)
    changed = 0
    for row in rows:
        r = {"country": None, "language": None, "title": None, "app_version": None, **row}
        old = db.execute("SELECT rating, title, body, updated_at FROM reviews "
                         "WHERE store = ? AND review_id = ?",
                         (r["store"], r["review_id"])).fetchone()
        if old == (r["rating"], r["title"], r["body"], r["updated_at"]):
            continue  # unchanged
        if old:  # edited by the reviewer: archive the previous text and rating
            db.execute("INSERT INTO review_history (store, review_id, rating, title, body,"
                       " updated_at) VALUES (?, ?, ?, ?, ?, ?)", (r["store"], r["review_id"], *old))
        db.execute(UPSERT, r)
        changed += 1
    db.commit()
    return changed

Alerts that catch a bad release

Three rules catch most incidents, and each maps to a question someone will actually ask:

• Low-star spike. One- and two-star reviews in the last 24 hours against the daily average of the seven days before.
• Version health. A version's average rating in its first 72 hours. This needs app_version, so it works for Play and the competitor feed; for your iOS app, compare against release time instead.
• Keyword clusters. Counts of reviews mentioning crash, login, payment or subscription terms in the last day, which often name the broken flow before the crash dashboard does.

"""alerts.py [DB]: three review signals worth paging someone for."""
import sqlite3
import sys

CLUSTERS = {"crash": ["crash", "freez", "force close"],
            "login": ["login", "log in", "sign in", "password"],
            "payment": ["payment", "charged", "refund"],
            "subscription": ["subscription", "cancel", "renew"]}
RECENT = "(datetime(updated_at) >= datetime('now', '-1 day'))"


def alerts(db: sqlite3.Connection) -> list[str]:
    out = []
    recent, older = db.execute(f"""
        SELECT COALESCE(SUM({RECENT}), 0), COALESCE(SUM(NOT {RECENT}), 0) FROM reviews
        WHERE rating <= 2 AND datetime(updated_at) >= datetime('now', '-8 days')""").fetchone()
    if recent >= 5 and recent > 2 * older / 7:
        out.append(f"low-star spike: {recent} in 24 h vs {older / 7:.1f}/day before")
    for store, version, avg, n in db.execute("""
        SELECT store, app_version, AVG(rating), COUNT(*) FROM reviews r
        WHERE app_version IS NOT NULL AND datetime(updated_at) <= (
          SELECT datetime(MIN(datetime(updated_at)), '+72 hours') FROM reviews
          WHERE store = r.store AND app_version = r.app_version)
        GROUP BY store, app_version
        HAVING MIN(datetime(updated_at)) >= datetime('now', '-72 hours')"""):
        if n >= 10 and avg < 3.5:
            out.append(f"{store} {version}: {avg:.2f} stars from {n} reviews in 72 h")
    for name, words in CLUSTERS.items():
        match = " OR ".join("lower(body) LIKE ?" for _ in words)
        (count,) = db.execute(f"SELECT COUNT(*) FROM reviews WHERE {RECENT} AND ({match})",
                              [f"%{w}%" for w in words]).fetchone()
        if count >= 3:
            out.append(f"{name}: {count} reviews in 24 h")
    return out


if __name__ == "__main__":
    print("\n".join(alerts(sqlite3.connect(sys.argv[1] if len(sys.argv) > 1 else "reviews.db")))
          or "no alerts")

The thresholds (5 reviews, 2x baseline, 10 reviews below 3.5 stars, 3 keyword hits) are starting points for a mid-sized app. Tune them after two weeks of history. For Play, updated_at is the last edit time, so an old review that a user rewrites lands back in the 24-hour window, which is usually what you want.

Run it on a schedule

Own apps run hourly. Each run starts one day before the newest review already stored for that store and stops as soon as it reaches older reviews, so it sits far below Apple's rolling-hour quota and Play's 200 GET requests per hour, and the upsert makes the overlap harmless. Starting from stored data rather than from a fixed "last two days" also means a job that was down for a while catches up on its next run, as far back as each API allows:

"""run_own.py: import your own apps' reviews, then evaluate the alert rules."""
import os
import sqlite3
from datetime import datetime, timedelta, timezone

from alerts import alerts
from asc_reviews import new_reviews
from play_reviews import play_reviews
from store import SCHEMA, save


def resume_from(db: sqlite3.Connection, store: str) -> datetime:
    (newest,) = db.execute("SELECT MAX(datetime(updated_at)) FROM reviews WHERE store = ?",
                           (store,)).fetchone()
    if newest is None:  # first run: take everything the API still serves
        return datetime(2008, 1, 1, tzinfo=timezone.utc)
    return datetime.fromisoformat(newest).replace(tzinfo=timezone.utc) - timedelta(days=1)


db = sqlite3.connect("reviews.db")
db.executescript(SCHEMA)
rows = (new_reviews(os.environ["ASC_BUNDLE_ID"], resume_from(db, "app_store"))
        + play_reviews(os.environ["PLAY_PACKAGE"], resume_from(db, "google_play")))
print(save(db, rows), "new or edited reviews")
for line in alerts(db):
    print("ALERT", line)  # forward to Slack, PagerDuty or email here

Cron starts jobs with an almost empty environment, so the credentials the scripts read (ASC_KEY_PATH, ASC_ISSUER_ID, ASC_KEY_ID, ASC_BUNDLE_ID, PLAY_SA_JSON, PLAY_PACKAGE) go in a monitor.env file of NAME=value lines that the cron entry loads:

#!/usr/bin/env bash
# install_schedule.sh: own apps hourly, competitor feeds every six hours
set -euo pipefail
dir=/opt/review-monitor
mkdir -p "$dir/logs"
test -r "$dir/monitor.env" || { echo "create $dir/monitor.env first" >&2; exit 1; }
{
  crontab -l 2>/dev/null | grep -v "$dir" || true
  echo "5 * * * * cd $dir && set -a && . ./monitor.env && set +a && python3 run_own.py >> logs/own.log 2>&1"
  echo "40 */6 * * * cd $dir && python3 rss_reviews.py 123456789 us gb de >> logs/rss.log 2>&1"
} | crontab -
crontab -l

Replace 123456789 with the competitor's App Store ID, the number after id in its App Store URL. Both scripts raise on failure, so the traceback lands in the log file. Point a log alert at those files: a run that breaks quietly has to be noticed before Play's seven-day window closes on the reviews it missed.

Where CAPTCHAs actually appear

Nothing above touches a CAPTCHA. The official APIs are authenticated, and Google's "unusual traffic" page gets a back-off, not a solve. That leaves third-party review sites that put Cloudflare in front of public review pages you are permitted to collect. Check the site's terms first and prefer its API or a data licence. When Cloudflare does appear, it takes one of two shapes, and telling Challenge and Turnstile apart decides which CaptchaAI method to call:

• A Turnstile widget on a form, such as a review search or "load more" request. Send method=turnstile with sitekey (the widget's data-sitekey) and pageurl. The token comes back in request and is single-use; submit it once as the form's cf-turnstile-response field. The Turnstile API walkthrough covers the injection step.
• A full Challenge page in place of the content. Cloudflare sets cf-mitigated: challenge on every challenge page response (detection docs). Send method=cloudflare_challenge with pageurl, proxy and proxytype; the proxy is mandatory. Poll with json=1: result holds the cf_clearance value and user_agent the browser string it is bound to. Reuse both, through the same proxy, as the cf_clearance cookie guide explains.

Some review widgets use hCaptcha instead. CaptchaAI does not support hCaptcha, so the code below skips those sources rather than guessing. Proxy use is disabled on CaptchaAI accounts by default, so ask support to enable it before relying on the Challenge path. If you have not made a first call yet, the quickstart takes five minutes.

"""captchaai_client.py: submit a task to CaptchaAI and poll res.php for the answer."""
import os
import time

import requests

BASE = "https://ocr.captchaai.com"
KEY = os.environ.get("CAPTCHAAI_KEY", "")
WAIT = {"CAPCHA_NOT_READY", "ERROR_INTERNAL_SERVER_ERROR"}  # keep polling the same ID


def _json(resp: requests.Response) -> dict:
    try:
        return resp.json()
    except ValueError:  # a plain-text error code can arrive even with json=1
        return {"status": 0, "request": resp.text.strip()}


def solve(fields: dict, first_wait: int, verb: str = "GET") -> dict:
    if len(KEY) != 32:
        raise SystemExit("Set CAPTCHAAI_KEY to your 32-character CaptchaAI API key")
    body = {"key": KEY, "json": 1, **fields}
    sent = _json(requests.request(verb, f"{BASE}/in.php", timeout=30,
                                  **({"params": body} if verb == "GET" else {"data": body})))
    if sent.get("status") != 1:  # key and parameter errors need a fix, not a retry loop
        raise RuntimeError(f"in.php refused the task: {sent.get('request')}")
    time.sleep(first_wait)
    for _ in range(36):  # every 5 s for about three more minutes
        got = _json(requests.get(f"{BASE}/res.php", timeout=30, params={
            "key": KEY, "action": "get", "id": sent["request"], "json": 1}))
        if got.get("status") == 1:
            return got
        reply = str(got.get("request", ""))
        if reply not in WAIT and not reply.startswith("<"):  # "<": a rare HTML 5xx page
            raise RuntimeError(f"res.php: {reply}")  # e.g. ERROR_CAPTCHA_UNSOLVABLE
        time.sleep(5 if reply == "CAPCHA_NOT_READY" else 10)
    raise TimeoutError(f"task {sent['request']} was not answered")

"""cf_fetch.py URL: fetch a public review page, handling Cloudflare only when it appears."""
import os
import re
import sys
from urllib.parse import urlparse

import requests

from captchaai_client import solve

PROXY = os.environ.get("REVIEW_PROXY")  # login:password@IP:PORT, as CaptchaAI expects it


class BackOff(Exception):
    """Slow down: fewer workers, longer gaps, cached pages."""


class Unsupported(Exception):
    """A CAPTCHA type CaptchaAI does not solve: skip the source."""


def session() -> requests.Session:
    s = requests.Session()
    if PROXY:  # every request leaves from the IP the solver will use
        s.proxies = {"http": f"http://{PROXY}", "https": f"http://{PROXY}"}
    return s


def fetch(s: requests.Session, url: str) -> requests.Response:
    resp = s.get(url, timeout=30)
    if resp.status_code == 429 or "detected unusual traffic" in resp.text:
        raise BackOff(url)
    if resp.headers.get("cf-mitigated") == "challenge":
        if not PROXY:
            raise BackOff("Cloudflare Challenge needs one proxy for the solve and the fetch")
        got = solve({"method": "cloudflare_challenge", "pageurl": url,
                     "proxy": PROXY, "proxytype": "HTTP"}, first_wait=20, verb="POST")
        s.headers["User-Agent"] = got["user_agent"]  # cf_clearance is bound to it
        s.cookies.set("cf_clearance", got["result"], domain=urlparse(url).hostname)
        resp = s.get(url, timeout=30)
        if resp.headers.get("cf-mitigated") == "challenge":  # UA or IP drifted from the solve
            raise BackOff(f"still challenged after a solve: {url}")
    return resp


def turnstile_token(url: str, html: str) -> str | None:
    """Single-use token for a Turnstile-guarded form: send it as cf-turnstile-response."""
    widget = re.search(r"<div[^>]*cf-turnstile[^>]*>", html)
    if not widget:
        if "hcaptcha.com" in html:
            raise Unsupported("hCaptcha is not supported by CaptchaAI")
        return None
    sitekey = re.search(r'data-sitekey="([^"]+)"', widget.group(0))
    if not sitekey:
        return None
    return solve({"method": "turnstile", "sitekey": sitekey.group(1), "pageurl": url},
                 first_wait=12)["request"]


if __name__ == "__main__":
    page = fetch(session(), sys.argv[1])
    print(page.status_code, len(page.text), "bytes")

The error codes that matter here: ERROR_BAD_PROXY and ERROR_PROXY_CONNECTION_FAILED mean switch proxies. ERROR_ZERO_BALANCE means no free thread or no active plan. After ERROR_CAPTCHA_UNSOLVABLE, never re-poll the same ID; make one fresh attempt at most. Key errors mean stop and fix the key. CaptchaAI bills per concurrent thread with unlimited solves, and a monitor that meets a few challenge pages per run rarely needs more than BASIC ($15/month, 5 threads).

Troubleshooting

FAQ

Can I reply to reviews through the same APIs?

Yes. On iOS, POST /v1/customerReviewResponses creates a response or replaces your existing one. On Play, POST .../reviews/{reviewId}:reply takes a replyText of up to 350 characters, within the default 2,000 replies per day. Google discourages automated replies that you plan to rewrite by hand later, and the user is notified only of your first reply to each version of their review, so draft replies automatically but let a person send them.

Do my own app reviews ever require solving a CAPTCHA?

No. Both stores give you authenticated APIs for your own apps. A CAPTCHA appears only when you collect third-party review pages, and on Google properties it means you are sending too much traffic.

How many storefronts should I poll for a competitor?

Every country where you compete. Each storefront is a separate feed with its own ten-page limit, so a US-only poll misses most reviews for an app that sells mainly in Germany or Japan.