Brand Mention Monitoring With CAPTCHA Handling

You can build brand mention monitoring that watches news, Hacker News, Reddit, Bluesky, Mastodon and newly issued lookalike domains, almost entirely through official APIs. CAPTCHA handling is a small fallback for the few public forums and review sites that put their search form behind Cloudflare Turnstile. The constraint that shapes the design in 2026: both big search-engine APIs are gone or closing, and Google's "unusual traffic" page is a signal to back off, not a puzzle to solve.

Search-engine APIs: what changed

Status, September 2026

• Bing Search APIs: retired. Microsoft retired them on August 11, 2025 and decommissioned every instance. The suggested replacement, Grounding with Bing Search in Azure AI Agents, feeds web data into an LLM agent's answer. It is not a result list you can diff.
• Google Custom Search JSON API: closing. It is closed to new customers, and existing customers have until January 1, 2027 to move. Until then: 100 free queries a day, then $5 per 1,000 up to 10,000 a day. Google points to Vertex AI Search, which covers up to 50 domains, not the whole web.

So "search the web for our name every morning" no longer has a cheap first-party API. Build on sources that publish their own search APIs, and treat web results as a small, separately licensed input; search results data collection covers those mechanics.

Sources that still have real APIs

Start with the free, keyless sources: Hacker News, GDELT and crt.sh (see the lookalike-domain section). Add Reddit and Bluesky once you have credentials. For deeper news coverage see news and media aggregation, and for social platforms see social media research data collection.

The pipeline: collect, normalize, dedupe, match, score, alert

Every source emits the same six fields: source, url, title, snippet, author, published_at. Adding a source means writing one function.

Collectors for Hacker News, GDELT and crt.sh

Save this as collectors.py (Python 3.10+, pip install "httpx>=0.26"). The calls run concurrently, and a failing source is logged and skipped instead of sinking the run.

import asyncio
import time

import httpx

BRAND = "acmecloud"  # the token you search for; one collector run per brand token
HEADERS = {"User-Agent": "brand-monitor/1.0 (contact: [email protected])"}


def mention(source, url, title, snippet="", author=None, published_at=None):
    return {"source": source, "url": url, "title": title or "", "snippet": snippet or "",
            "author": author, "published_at": published_at}


async def hacker_news(client, since):
    params = {"query": BRAND, "tags": "(story,comment)", "hitsPerPage": 100,
              "numericFilters": f"created_at_i>{since}"}
    r = (await client.get("https://hn.algolia.com/api/v1/search_by_date", params=params)).raise_for_status()
    return [mention("hn", f"https://news.ycombinator.com/item?id={h['objectID']}",
                    h.get("title") or h.get("story_title"),
                    h.get("comment_text") or h.get("story_text") or h.get("url"),
                    h.get("author"), h.get("created_at")) for h in r.json()["hits"]]


async def gdelt(client):
    params = {"query": BRAND, "mode": "artlist", "format": "json", "maxrecords": 250,
              "timespan": "1d", "sort": "DateDesc"}
    r = await client.get("https://api.gdeltproject.org/api/v2/doc/doc", params=params)
    if r.status_code != 200 or "json" not in r.headers.get("content-type", ""):
        raise RuntimeError(f"GDELT {r.status_code}: {r.text[:100]}")  # 429s and query errors are text
    return [mention("gdelt", a["url"], a.get("title"), author=a.get("domain"),
                    published_at=a.get("seendate")) for a in r.json().get("articles", [])]


async def crtsh(client):
    params = {"q": f"{BRAND}%", "output": "json", "exclude": "expired", "deduplicate": "Y"}
    # a busy crt.sh answers 502 or times out; collect() logs it and the next run catches up
    rows = (await client.get("https://crt.sh/", params=params, timeout=90)).raise_for_status().json()
    return [mention("crtsh", f"https://crt.sh/?id={c['id']}", " ".join(sorted(set(c["name_value"].split()))),
                    c["issuer_name"], published_at=c["not_before"]) for c in rows]


async def collect(hours=6):
    async with httpx.AsyncClient(headers=HEADERS, timeout=30) as client:
        batches = await asyncio.gather(hacker_news(client, int(time.time()) - hours * 3600),
                                       gdelt(client), crtsh(client), return_exceptions=True)
    for failed in (b for b in batches if isinstance(b, Exception)):
        print("collector failed:", repr(failed))
    return [m for b in batches if isinstance(b, list) for m in b]

GDELT's JSON carries url, title, seendate, domain, language and sourcecountry but no snippet, and a match can come from the article text rather than the title. A query with no matches returns {}, hence .get("articles", []). GDELT throttles by IP, so on a shared egress IP (a CI runner, a NAT gateway) other callers can use up the allowance. In our tests, requests spaced 20 seconds apart from such an IP still got 429 for over a minute. Treat a logged GDELT 429 as a skipped run, not a bug. The HN Search API matches loosely: in our test, a query for captchaai also returned comments about "captchas". So the next stage re-checks every hit.

Normalize, dedupe and match

pipeline.py strips HTML and entities and builds a canonical URL: https, lowercase host without www., no fragment or trailing slash, tracking parameters dropped, the rest sorted. The SHA-256 of that URL is a SQLite primary key, so INSERT OR IGNORE doubles as the "seen it?" check. The insert stays uncommitted until the alert goes out: if Slack is down, the run exits, SQLite discards the rows, and the next run alerts on the same mentions instead of losing them. Certificate rows dedupe on their hostname set, so renewals stay quiet.

import hashlib
import html
import re
import sqlite3
import time
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit

INCLUDE = re.compile(r"\bacme[\s-]?cloud\b", re.I)                    # brand terms and spellings
EXCLUDE = re.compile(r"\bacme[\s-]?cloud\s+(gaming|anvils?)\b", re.I)  # known false positives
OWN_ZONES = ("acmecloud.com", "acmecloud.io")                          # your own certificates
TRACKING = ("utm_", "fbclid", "gclid", "mc_")
db = sqlite3.connect("mentions.db")
db.execute("CREATE TABLE IF NOT EXISTS mentions (k TEXT PRIMARY KEY, seen_at INTEGER, "
           "source TEXT, url TEXT, title TEXT)")


def normalize(m):
    def clean(s):
        return " ".join(re.sub(r"<[^>]+>", " ", html.unescape(s)).split())
    return {**m, "title": clean(m["title"]), "snippet": clean(m["snippet"])[:500]}


def canonical(url):
    p = urlsplit(url)
    query = sorted((k, v) for k, v in parse_qsl(p.query) if not k.lower().startswith(TRACKING))
    host = p.netloc.lower().removeprefix("www.")
    return urlunsplit(("https", host, p.path.rstrip("/"), urlencode(query), ""))


def is_new(m):
    # crt.sh rows dedupe on their hostname set; everything else on the canonical URL
    key = m["title"] if m["source"] == "crtsh" else canonical(m["url"])
    cur = db.execute("INSERT OR IGNORE INTO mentions VALUES (?, ?, ?, ?, ?)",
                     (hashlib.sha256(key.encode()).hexdigest(), int(time.time()),
                      m["source"], m["url"], m["title"]))
    return cur.rowcount == 1  # uncommitted: alerts.py commits once the digest is delivered


def own_name(name):
    name = name.lstrip("*.").lower()
    return any(name == zone or name.endswith("." + zone) for zone in OWN_ZONES)


def relevant(m):
    if m["source"] == "crtsh":
        return not all(own_name(n) for n in m["title"].split())
    text = f"{m['title']} {m['snippet']}"
    # GDELT returns no snippet and can match on article text, so trust its match
    return not EXCLUDE.search(text) and (m["source"] == "gdelt" or bool(INCLUDE.search(text)))

Matching is Boolean: at least one brand pattern, no exclusion. If your brand is also a common word, the exclusions do most of the work. Grow them from the false positives your team marks in the first two weeks.

Sentiment and a rate-limited Slack digest

alerts.py scores new mentions with VADER (pip install vaderSentiment) and flags a compound score of -0.05 or lower as negative, the typical threshold in VADER's documentation. New domains and negative mentions sort first. Slack's incoming webhooks allow about one message per second and answer HTTP 429 with Retry-After, so the script sends one digest per run, capped at 20 lines, and escapes &, < and > in titles and URLs as Slack's message format requires (a GDELT article URL often carries & in its query string).

import asyncio
import os
import time

import httpx
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer

from collectors import collect
from pipeline import db, is_new, normalize, relevant

vader = SentimentIntensityAnalyzer()


def label(m):
    if m["source"] == "crtsh":
        return "NEW DOMAIN"
    score = vader.polarity_scores(f"{m['title']} {m['snippet']}")["compound"]
    return "NEGATIVE" if score <= -0.05 else m["source"]


def esc(text):  # Slack reads &, < and > as control characters
    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


def post_digest(fresh, cap=20):
    urgent = {"NEW DOMAIN": 0, "NEGATIVE": 1}
    rows = sorted(((label(m), m) for m in fresh), key=lambda r: urgent.get(r[0], 2))
    lines = [f"[{tag}] {esc(m['title'][:140])} {esc(m['url'])}" for tag, m in rows[:cap]]
    if len(rows) > cap:
        lines.append(f"...and {len(rows) - cap} more in mentions.db")
    hook = os.environ.get("SLACK_WEBHOOK_URL")
    if not lines or not hook:
        print("\n".join(lines) or "no new mentions")
        return
    for _ in range(3):  # one digest per run keeps you far below the webhook limit
        r = httpx.post(hook, json={"text": "\n".join(lines)}, timeout=10)
        if r.status_code != 429:
            r.raise_for_status()
            return
        time.sleep(int(r.headers.get("Retry-After", "1")))
    raise RuntimeError("Slack still answers 429; the mentions stay unsent for the next run")


if __name__ == "__main__":
    mentions = [normalize(m) for m in asyncio.run(collect())]
    post_digest([m for m in mentions if relevant(m) and is_new(m)])
    db.commit()  # only now are these mentions marked as seen

Run it from cron every 30-60 minutes. For several brands, run the collectors one after another so GDELT sees at most one call per 5 seconds. Mentions are new URLs; to learn when a page you already know changes, such as a review profile or a partner's listing of your product, a content change monitoring bot diffs snapshots instead.

Lookalike domains: Certificate Transparency, zone files and aftermarket listings

A lookalike domain that wants HTTPS needs a certificate, and publicly trusted certificates land in Certificate Transparency logs that anyone can search. crt.sh indexes them and returns JSON with output=json. The query form decides what you catch:

• %.acmecloud.com lists every name under your own zone, including subdomains nobody told you about.
• acmecloud% is a prefix search. It finds acmecloud-login.com, acmecloud.io, and names like acmecloud.some-host.io where someone uses your brand as a subdomain label (our test run turned up exactly that).
• %acmecloud% looks like a substring search, but in our tests it returned an empty list where the prefix query found results. Don't rely on it.

Add exclude=expired for live certificates only and deduplicate=Y to collapse each precertificate and final-certificate pair. Rows carry id, issuer_name, common_name, name_value (one name per line), not_before, not_after and serial_number. crt.sh often answers 502 or times out under load, but since each call returns every live certificate, the next run catches up. Because the substring form fails, every other shape needs its own prefix query: typosquats that don't start with your exact brand (acmec1oud, acme-cloud) and names that put a word in front (login-acmecloud.com). A permutation generator such as dnstwist, fed a short word list (login, secure, support), builds that list.

Domains that never get HTTPS never reach the logs. ICANN's Centralized Zone Data Service (CZDS) gives approved requesters the zone files of participating generic TLDs, one registry approval per TLD; diff each day's file for your brand token. Country-code TLDs are not included.

Before escalating a hit, check the registrar, creation date and MX records: a lookalike that can send mail is a phishing risk. Domain WHOIS lookup automation covers those lookups. Also watch aftermarkets such as Sedo and Afternic for listings containing your brand, preferring any partner feed and reading the marketplace's terms before automating its search.

SERP features, briefly

Keep this narrow. Once a week, check a handful of brand queries ("acmecloud", "acmecloud reviews", "acmecloud login", "is acmecloud legit") and record which other domains rank on page one (review sites, forums, lookalikes), whether a knowledge panel, sitelinks, Top stories, People also ask or an AI Overview appears and whom it cites, and what changed since last week. Search Console's Performance report, grouped by search appearance, shows the features your own site earns. Third-party results need a licensed SERP source; see search results data collection. If Google redirects you to google.com/sorry, stop querying from that setup. This pipeline does not try to solve that page.

Where CAPTCHA handling fits, and where it does not

Cloudflare calls cf-mitigated: challenge the reliable way to detect a challenge page, whatever the challenge type; detecting Cloudflare Challenge vs Turnstile goes further.

Turnstile on a forum or review-site search form

solve() follows CaptchaAI's documented flow: POST the task to https://ocr.captchaai.com/in.php with json=1, which returns {"status":1,"request":"<task id>"}; wait 15 seconds; call res.php with action=get every 5 seconds while the answer is CAPCHA_NOT_READY; give up at 120 seconds. The Turnstile token comes back in request. Cloudflare tokens are valid for 300 seconds and single-use, so solve right before submitting. Save the code as captcha_fallback.py, put your key in CAPTCHAAI_KEY, and call search_with_turnstile() with two httpx.AsyncClient instances: one for the forum, which keeps its cookies, and one for CaptchaAI.

import asyncio
import os
import re

import httpx

API = "https://ocr.captchaai.com"
KEY = os.environ["CAPTCHAAI_KEY"]  # YOUR_API_KEY: the 32-character key from your dashboard
WIDGET = re.compile(r"<div[^>]+cf-turnstile[^>]*>")
SLOTS = asyncio.Semaphore(5)  # BASIC plan: 5 threads


def parse(r):
    try:
        return r.json()
    except ValueError:  # some error codes arrive as plain text even with json=1
        return {"status": 0, "request": r.text.strip()}


async def solve(api, params, files=None, first_wait=15):
    async with SLOTS:  # never more solves in flight than your plan has threads
        sub = parse(await api.post(f"{API}/in.php", data={"key": KEY, "json": 1, **params}, files=files))
        if sub.get("status") != 1:
            raise RuntimeError(f"in.php: {sub.get('request')}")
        await asyncio.sleep(first_wait)
        for _ in range((120 - first_wait) // 5):  # last poll at 115 s, give up at 120 s
            res = parse(await api.get(f"{API}/res.php", params={
                "key": KEY, "action": "get", "id": sub["request"], "json": 1}))
            if res.get("status") == 1:
                return res
            if str(res.get("request")).startswith(("ERROR_", "IP_BANNED")):
                raise RuntimeError(f"res.php: {res.get('request')}")
            await asyncio.sleep(5)  # CAPCHA_NOT_READY, or a rare HTML 5xx page: poll again
        raise TimeoutError(f"task {sub['request']} not solved within 120 s")


async def search_with_turnstile(site, api, form_url, action_url, fields):
    page = await site.get(form_url)
    if page.headers.get("cf-mitigated") == "challenge":
        raise PermissionError("full-page Cloudflare challenge: use the cloudflare_challenge path")
    widget = WIDGET.search(page.text)
    # data-sitekey and data-action map one-to-one onto CaptchaAI's sitekey and action
    attrs = dict(re.findall(r'data-(sitekey|action)="([^"]+)"', widget.group(0))) if widget else {}
    if "sitekey" in attrs:
        task = {"method": "turnstile", "pageurl": str(page.url), **attrs}
        answer = await solve(api, task)  # solve just before submitting: tokens are single-use
        fields = {**fields, "cf-turnstile-response": answer["request"]}
    elif "challenges.cloudflare.com/turnstile" in page.text:
        raise LookupError("no data-sitekey in the HTML: read the sitekey from turnstile.render()")
    return await site.post(action_url, data=fields)

Some errors arrive as plain text even with json=1, hence parse(). While polling, anything that is not an ERROR_ code or IP_BANNED, such as the rare HTML 500 or 502 page, means poll again in 5 seconds. ERROR_WRONG_USER_KEY and ERROR_KEY_DOES_NOT_EXIST mean stop and fix the key, since repeated bad keys can get your IP banned. ERROR_ZERO_BALANCE means no free thread or no active plan. After ERROR_CAPTCHA_UNSOLVABLE, stop polling that ID and submit at most once more with a fresh page. If the sitekey only appears in a turnstile.render() call, see how to solve Cloudflare Turnstile using the API.

Full-page challenges and legacy image CAPTCHAs

Both paths import solve() from captcha_fallback.py. A Cloudflare Challenge needs your own proxy as login:password@host:port (no scheme), with proxytype set to HTTP, HTTPS, SOCKS4 or SOCKS5. Proxy use is off by default on CaptchaAI accounts, so ask support to enable it. The answer is result (the cf_clearance value) plus user_agent, and the cookie only works from the same IP with that User-Agent, so use a fixed exit IP. ERROR_BAD_PROXY or ERROR_PROXY_CONNECTION_FAILED means CaptchaAI could not use that proxy (a failed proxy is marked bad for about 10 minutes), so switch proxies before you submit again. The cf_clearance cookie guide explains the binding.

import os

import httpx

from captcha_fallback import solve

PROXY = os.environ["MONITOR_PROXY"]  # login:password@host:port with a fixed exit IP


async def cleared_client(api, pageurl):
    """Solve a full-page Cloudflare challenge and return a client that can reuse the clearance."""
    answer = await solve(api, {"method": "cloudflare_challenge", "pageurl": pageurl,
                               "proxy": PROXY, "proxytype": "HTTP"})
    client = httpx.AsyncClient(proxy=f"http://{PROXY}", timeout=30,
                               headers={"User-Agent": answer["user_agent"]})
    client.cookies.set("cf_clearance", answer["result"], domain=httpx.URL(pageurl).host)
    return client  # same proxy, same User-Agent, or the cookie is rejected


async def solve_forum_image(site, api, image_url, numeric=False, case_sensitive=False):
    """Read a legacy forum's image CAPTCHA; fetch it with the session that submits the answer."""
    image = (await site.get(image_url)).raise_for_status()
    if not 100 <= len(image.content) <= 100_000:
        raise ValueError(f"{len(image.content)} bytes: CaptchaAI takes 100 bytes to 100 KB")
    kind = image.headers.get("content-type", "image/png").split(";")[0]  # image/jpeg, png or gif
    flags = {"numeric": int(numeric), "regsense": int(case_sensitive)}  # 0 is the default for both
    answer = await solve(api, {"method": "post", **flags}, first_wait=5,
                         files={"file": ("captcha." + kind.split("/")[-1], image.content, kind)})
    return answer["request"]

Image files must be 100 bytes to 100 KB, in JPG, PNG or GIF; the upload keeps the forum's own content type so the file extension matches. Fetch the image in the session that submits the search, because the forum ties it to your cookie. Image answers are usually ready sooner, hence the 5-second first wait.

Thread sizing: why BASIC is usually enough

CaptchaAI bills concurrent threads, with unlimited solves per thread. Threads needed = solves per second × seconds per solve. Say the fallback checks 8 Turnstile-protected forums hourly, one solve each: 192 solves a day. Turnstile typically clears in under 10 seconds, so the average load is 192 / 86,400 × 10 ≈ 0.02 threads.

The peak decides the plan. If the hourly job hits all 8 forums at once, 8 solves are in flight. The SLOTS semaphore in solve() caps the fallback at 5, so BASIC ($15/month, 5 threads) covers it; the month's 5,760 solves are a small fraction of the roughly 259,000 one busy Turnstile thread could clear. STANDARD ($30/month, 15 threads) only makes sense above 5 concurrent solves. How to pick the right thread count covers other workloads.

Terms, robots.txt and retention

• Collect public content only. Use credentials only with the official APIs above (Reddit, Bluesky, Mastodon, X), never to scrape pages behind a login, and never use the fallback to get around a ban.
• Read robots.txt and the terms of every site you add; drop sites that forbid automated search.
• Reddit grants Data API access on request and wants a separate agreement for commercial use. X bills every post read under pay-per-usage pricing, so cap max_results and query frequency; a post you read again within the same UTC day is not billed twice, so overlapping query windows cost less than they look.
• Store the minimum (URL, title, short snippet, author handle, timestamp). Keep dedupe hashes, expire snippets and author fields on a schedule, and remove items their authors delete. Reddit's API rules require deleting content removed on Reddit and recommend purging stored user data within 48 hours.

Troubleshooting

FAQ

Can this replace Mention or Brand24?

For developer-heavy audiences, often yes: Hacker News, Reddit, Bluesky, news and lookalike domains cover most of what matters, and you own the rules. Commercial tools typically license sources you cannot query cheaply and handle multilingual sentiment, so many teams run both.

Why not solve Google's "unusual traffic" page?

Google shows it when it has decided the traffic is automated. A monitor running a few dozen brand queries a week is better served by a licensed SERP source, so this pipeline treats the page as a stop signal.

How far back can I backfill?

GDELT covers the last 3 months. X's recent search covers 7 days; its full-archive endpoint, /2/tweets/search/all, is open to pay-per-usage and Enterprise accounts. Hacker News goes back to the start if you walk created_at_i windows. crt.sh returns every logged certificate if you drop exclude=expired.

Ready to add the fallback? Compare CaptchaAI plans and start with BASIC; the Turnstile and image paths need nothing more, and the challenge path adds only your own proxy, once support has enabled proxy use on your account.