2026-09-17, Brand Monitoring · Google · News · Engineering · Tutorials

Online reputation monitoring for search results, news and AI answers: a working pipeline

Online reputation monitoring is the practice of recording, on a schedule, what someone sees when they search for your brand or ask an AI assistant about it, and alerting when that picture changes for the worse. A working pipeline has five parts: a query set that mirrors how people look you up, a schedule that submits those queries as async batches with webhooks, a snapshot store, a diff that turns two snapshots into events, and alert rules that route events to a person for review.

Reputation surface is used here to mean everything shown for a branded query or prompt: organic results, People Also Ask, the knowledge panel, the AI Overview, news results and assistant answers. Mentions on social media are a different dataset; see brand monitoring tools for listening platforms.

What to watch and why

The pipeline below treats every surface, including Google News, as one more task type, so adding a surface needs no new code path.

Step 1: the query set

Reputation queries mirror the intent of someone deciding whether to trust you, so they look different from a keyword research list.

1. Navigational: acme, acme.com, acme login.
2. Evaluation: acme reviews, is acme legit, acme complaints, acme vs globex.
3. Risk: acme scam, acme lawsuit, acme data breach, acme layoffs. These are the ones to watch most closely.
4. People: your CEO’s name, and name plus company. Only monitor executives with their knowledge and within your company’s policy.
5. Products: each flagship product name plus review and problems.
6. Assistant prompts: “Is Acme trustworthy?”, “What are the main complaints about Acme?”, “Should I choose Acme or Globex for payroll?”.
7. News queries: the quoted legal and brand names, "Acme Inc", "Acme Payroll".

Give each entry an id, a tier and the markets it runs in. Store it as data:

QUERY_SET = [
    {"id": "nav-brand",     "engine": "GOOGLE",      "text": "acme",                                 "tier": "daily"},
    {"id": "eval-reviews",  "engine": "GOOGLE",      "text": "acme reviews",                         "tier": "daily"},
    {"id": "risk-scam",     "engine": "GOOGLE",      "text": "acme scam",                            "tier": "watch"},
    {"id": "risk-lawsuit",  "engine": "GOOGLE",      "text": "acme lawsuit",                         "tier": "watch"},
    {"id": "news-legal",    "engine": "GOOGLE_NEWS", "text": "\"Acme Inc\"",                         "tier": "watch"},
    {"id": "ai-trust",      "engine": "CHATGPT",     "text": "Is Acme trustworthy?",                 "tier": "daily"},
    {"id": "ai-complaints", "engine": "GEMINI",      "text": "What are the main complaints about Acme?", "tier": "daily"},
]
MARKETS = [{"country": "US", "hl": "en"}, {"country": "GB", "hl": "en"}]

Google interprets operators in the query, such as quoted phrases and -term exclusions. Test each query by hand before scheduling it.

Step 2: schedule with async batches and webhooks

Two tiers keep cost proportional to risk: watch queries run every 6 hours, daily queries once a day. Each run is a batch of up to 500 tasks with a webhook. The idempotency key encodes the slot, so a cron job that fires twice in the same slot creates nothing new.

import datetime as dt, os, requests

API = "https://api.answerline.dev"
HEADERS = {"Authorization": f"Bearer {os.environ['API_KEY']}"}
HOOK = {"url": "https://hooks.example.com/reputation"}

def payload(entry, market):
    if entry["engine"] == "GOOGLE":
        return {"query": entry["text"], "country": market["country"], "hl": market["hl"],
                "include": {"aioverview": {"markdown": False}}}
    if entry["engine"] == "GOOGLE_NEWS":
        return {"query": entry["text"], "country": market["country"], "hl": market["hl"]}
    return {"prompt": entry["text"], "country": market["country"]}

def submit(tiers):
    now = dt.datetime.now(dt.timezone.utc)
    slot = now.replace(hour=now.hour - now.hour % 6, minute=0, second=0, microsecond=0)
    tasks = [{"taskType": e["engine"], "payload": payload(e, m),
              "priority": 10 if e["tier"] == "watch" else 5,
              "idempotencyKey": f"rep|{e['id']}|{m['country']}|{slot:%Y-%m-%dT%H}",
              "webhook": HOOK}
             for e in QUERY_SET if e["tier"] in tiers for m in MARKETS]
    for i in range(0, len(tasks), 500):
        res = requests.post(f"{API}/v1/async/task/batch", json=tasks[i:i + 500], headers=HEADERS, timeout=60)
        res.raise_for_status()
        for item in res.json()["results"]:
            if not item["success"] and item["error"]["code"] != "RESOURCE_ALREADY_EXISTS":
                print("not queued", tasks[i + item["index"]]["idempotencyKey"], item["error"]["code"])

# cron: every 6 hours -> submit({"watch"}); once a day at 06:00 UTC -> submit({"watch", "daily"})

The key uses | as a separator so the webhook handler can recover the query id and market from task.idempotencyKey without a lookup. See async tasks for statuses and webhooks for delivery rules.

Step 3: receive and store snapshots

Each delivery is a signed POST of {task, credits, response}. Verify the signature on the raw body, skip test deliveries, deduplicate by task.id, and reduce the response to a snapshot you can diff.

import hashlib, hmac, json, time
from urllib.parse import urlsplit
from flask import Flask, abort, request

app = Flask(__name__)

def verify(raw, header, secret, tolerance=300):
    parts = [p.split("=", 1) for p in (header or "").split(",") if "=" in p]
    ts = [v for k, v in parts if k == "t"]
    sigs = [v for k, v in parts if k == "v1"]
    if len(ts) != 1 or not sigs or not ts[0].isdigit() or abs(time.time() - int(ts[0])) > tolerance:
        return False
    mac = hmac.new(secret.encode(), f"{ts[0]}.".encode() + raw, hashlib.sha256).hexdigest()
    return any(hmac.compare_digest(mac, s) for s in sigs)

def host(url):
    return (urlsplit(url or "").hostname or "").removeprefix("www.").lower()

def snapshot(task_type, r):
    if task_type == "GOOGLE":
        kg = r.get("knowledgeGraph") or {}
        aio = r.get("aioverview") or {}
        return {
            "top10": [{"pos": o["position"], "url": o["link"], "host": host(o["link"]), "title": o.get("title")}
                      for o in r.get("organicResults", [])[:10]],
            "paa": [p["question"] for p in r.get("peopleAlsoAsk", [])],
            "kg": {"description": kg.get("description"), "website": kg.get("website"),
                   "profiles": sorted(p.get("link", "") for p in kg.get("profiles", []))},
            "aio_text": aio.get("text"),
            "aio_sources": sorted({host(s["url"]) for s in aio.get("sources", [])}),
        }
    if task_type == "GOOGLE_NEWS":
        return {"news": [{"url": n["link"], "source": n.get("source"), "title": n.get("title")}
                         for n in r.get("newsResults", [])]}
    return {"text": r.get("text"), "sources": sorted({host(s["url"]) for s in r.get("sources", [])})}

@app.post("/reputation")
def reputation():
    raw = request.get_data()
    if not verify(raw, request.headers.get("Webhook-Signature"), os.environ["WEBHOOK_SECRET"]):
        abort(400)
    d = json.loads(raw)
    task = d["task"]
    if d.get("test") or task["status"] != "COMPLETED":
        return "", 204
    _, query_id, country, slot = task["idempotencyKey"].split("|")
    snap = snapshot(task["taskType"], d["response"]["result"])
    store(task["id"], query_id, country, slot, snap, d["response"])   # insert ... on conflict (task_id) do nothing
    enqueue_diff(query_id, country)                                    # do the diff off the request path
    return "", 204

Keep the full response next to the snapshot. When a reviewer asks what the page said on Tuesday, the snapshot is too lossy; the stored response is the evidence. For Google Search, include.html also returns the page HTML.

store and enqueue_diff are yours: a table keyed by task_id with query_id, country, slot, snapshot jsonb, response jsonb, and a queue. The endpoint must answer 2xx within 15 seconds or the delivery is retried, so do nothing slow here.

Step 4: diff snapshots into events

A diff compares the latest snapshot for a query and market with the previous one and emits typed events.

NEGATIVE = ("scam", "lawsuit", "fraud", "complaint", "breach", "layoff", "fine", "recall", "class action")
OWNED = {"acme.com", "help.acme.com", "status.acme.com"}

def negative(s):
    return any(w in (s or "").lower() for w in NEGATIVE)

def diff(prev, cur):
    events = []
    if "top10" in cur:
        before = {o["url"] for o in prev.get("top10", [])}
        for o in cur["top10"]:
            if o["url"] not in before and o["host"] not in OWNED:
                events.append(("serp_new_third_party", o["pos"], o["url"], o["title"]))
        prev_owned = [o["pos"] for o in prev.get("top10", []) if o["host"] in OWNED]
        cur_owned = [o["pos"] for o in cur["top10"] if o["host"] in OWNED]
        if prev_owned and (not cur_owned or min(cur_owned) > min(prev_owned) + 2):
            events.append(("serp_owned_dropped", min(prev_owned), cur_owned[:1], None))
        for q in set(cur["paa"]) - set(prev.get("paa", [])):
            events.append(("paa_new_question", None, None, q))
        for k in ("description", "website", "profiles"):
            if prev.get("kg", {}).get(k) != cur["kg"].get(k):
                events.append(("kg_changed", None, k, cur["kg"].get(k)))
        for h in set(cur["aio_sources"]) - set(prev.get("aio_sources", [])):
            events.append(("aio_new_source", None, h, None))
        if cur["aio_text"] and negative(cur["aio_text"]) and not negative(prev.get("aio_text")):
            events.append(("aio_turned_negative", None, None, cur["aio_text"][:300]))
    elif "news" in cur:
        seen = {n["url"] for n in prev.get("news", [])}
        for n in cur["news"]:
            if n["url"] not in seen:
                events.append(("news_new_article", None, n["url"], f"{n['source']}: {n['title']}"))
    else:
        for h in set(cur["sources"]) - set(prev.get("sources", [])):
            events.append(("ai_new_source", None, h, None))
        if negative(cur["text"]) and not negative(prev.get("text")):
            events.append(("ai_turned_negative", None, None, (cur["text"] or "")[:300]))
    return events

Three details make diffs trustworthy:

• Compare like with like. Same query, market, device and engine. Mixing mobile and desktop snapshots produces phantom changes.
• Skip the first snapshot. With no previous snapshot, everything is “new”. Store it as the baseline and emit nothing.
• Normalise URLs before comparing (lowercase host, drop fragments and utm_* parameters). Otherwise tracking parameters look like new pages.

AI answers vary from run to run even when nothing has changed; see AI answer volatility. Treat assistant events as signals to aggregate, not as individual alarms, unless the answer contains a specific false claim.

A keyword list is a crude negativity test. It catches “lawsuit” and misses sarcasm. Use it to route events, and let a person or a classifier you have evaluated make the call.

Step 5: alert rules

Map event types to severity and routing. A starting table:

Search pages fluctuate. Requiring the same event in two consecutive runs removes much of the noise at the cost of one interval of delay. For watch-tier queries on a 6-hour cadence that is at most 12 hours from appearance to alert; accept it for low severity and skip the gate for knowledge panel website changes, which are rarely transient.

Deduplicate events: an open event with the same type, query, market and URL is updated, not re-sent.

Step 6: human review

Every high or medium event becomes a review item with:

1. The before and after snapshots side by side.
2. Links to the stored responses for both runs.
3. The query, market, engine and timestamps.
4. A verdict: no_action, monitor, respond, escalate.
5. For respond: the owner and the chosen response, such as updating your own page, replying on a review platform, correcting a knowledge panel through Google’s feedback process, or contacting a publisher about a factual error.

Record verdicts. After a month, the share of no_action verdicts per rule tells you which rules to tighten.

Reporting metrics

Cost

Credits per async task: Google Search 3 for one page, plus 2 with include.aioverview; Google News 3; ChatGPT 5; Gemini 4. A failed request is charged nothing (credits).

The example set above has 4 Google queries (2 daily, 2 watch), 1 news query (watch), 1 ChatGPT and 1 Gemini prompt (daily), in 2 markets:

That is about 4,260 credits over 30 days. A realistic program with 40 queries and 10 prompts scales linearly from there. See pricing for plan sizes and AI monitoring cost planning for budgeting larger sets.

This API returns what pages and answers showed. It does not return search volume, traffic, review scores from review platforms, or social mentions.

Pitfalls

• Results are requested for a country, language and device, without a signed-in user. Treat them as a consistent reference view; individual searchers may see something different.
• Agree executive name monitoring with the executive and limit it to their public professional role.
• One odd AI answer is noise. Use the persistence gate and rates over several runs.
• Reputation damage can be your own page sliding down with no new negative page appearing. Keep the serp_owned_dropped rule.
• Set a retention period for raw responses long enough to support any response you might make.

For enforcement against ads, sellers and lookalike domains, see the brand protection use case; for coverage datasets and velocity, the news monitoring use case. During an active incident, brand crisis AI monitoring covers faster cadences. Request fields are listed on the Google Search engine page.

Keep reading