How to Build a Brand Monitoring System: Step-by-Step Tutorial

Brand monitoring tracks mentions of your company, products, or competitors across the web. Manual monitoring doesn't scale -- you need automated tools that search, scrape, and alert you in real time.

This tutorial shows you how to build a complete brand monitoring system using Python and SearchHive's APIs. You'll set up automated search, scrape mentions from multiple sources, and generate alerts when something matters.

Key Takeaways

• A functional brand monitoring system needs three components: search, scraping, and alerting
• SearchHive's SwiftSearch API handles web search, while ScrapeForge extracts mention content
• You can build a working monitor in under 100 lines of Python
• Scheduling with APScheduler keeps the system running on autopilot
• Alert routing via webhooks integrates with Slack, Discord, or email

Prerequisites

• Python 3.9+
• A SearchHive API key (free tier works -- 500 credits)
• requests and apscheduler packages

Install dependencies:

pip install requests apscheduler

Step 1: Configure Your API Connection

Start with a reusable API client that wraps SearchHive's endpoints.

import requests
import os
import json
from datetime import datetime

class SearchHiveClient:
    def __init__(self, api_key=None):
        self.api_key = api_key or os.environ.get("SEARCHHIVE_API_KEY")
        self.base_url = "https://api.searchhive.dev/v1"
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}"
        })

    def search(self, query, num=20):
        resp = self.session.get(
            f"{self.base_url}/search",
            params={"q": query, "num": num}
        )
        resp.raise_for_status()
        return resp.json().get("results", [])

    def scrape(self, url):
        resp = self.session.get(
            f"{self.base_url}/scrape",
            params={"url": url}
        )
        resp.raise_for_status()
        return resp.json()

client = SearchHiveClient()

Step 2: Define Your Brand Monitoring Queries

Good brand monitoring uses targeted queries. Don't just search for your brand name -- include modifiers that surface actionable mentions.

BRAND_QUERIES = [
    # Direct brand mentions
    '"YourBrand" review',
    '"YourBrand" vs',
    '"YourBrand" alternative',
    '"YourBrand" pricing',
    # Competitor tracking
    '"CompetitorA" vs "CompetitorB"',
    '"CompetitorA" pricing change',
    # Industry keywords
    'your industry keyword -site:yourbrand.com',
]

# Sentiment indicators to watch for
NEGATIVE_SIGNALS = ["worst", "terrible", "avoid", "scam", "overpriced", "slow"]
POSITIVE_SIGNALS = ["best", "love", "recommend", "excellent", "fast", "cheap"]

Step 3: Search and Collect Mentions

The core monitoring loop searches for each query and collects results.

def search_mentions(client, queries):
    all_mentions = []
    for query in queries:
        try:
            results = client.search(query, num=20)
            for r in results:
                mention = {
                    "title": r.get("title", ""),
                    "url": r.get("url", ""),
                    "snippet": r.get("description", ""),
                    "query": query,
                    "discovered_at": datetime.utcnow().isoformat(),
                    "sentiment": analyze_sentiment(r.get("description", ""))
                }
                all_mentions.append(mention)
        except Exception as e:
            print(f"Error searching '{query}': {e}")
    return all_mentions

def analyze_sentiment(text):
    text_lower = text.lower()
    neg_count = sum(1 for s in NEGATIVE_SIGNALS if s in text_lower)
    pos_count = sum(1 for s in POSITIVE_SIGNALS if s in text_lower)
    if neg_count > pos_count:
        return "negative"
    elif pos_count > neg_count:
        return "positive"
    return "neutral"

Step 4: Scrape Mention Details for Context

Search snippets give you the headline -- scraping gives you the full context around each mention.

def enrich_mentions(client, mentions):
    enriched = []
    for m in mentions[:10]:  # Scrape top 10 to manage credits
        try:
            scraped = client.scrape(m["url"])
            m["full_content"] = scraped.get("markdown", "")[:2000]
            m["scraped_at"] = datetime.utcnow().isoformat()
            enriched.append(m)
        except Exception as e:
            print(f"Error scraping {m['url']}: {e}")
            enriched.append(m)  # Keep unenriched mention
    return enriched

Step 5: Filter and Prioritize Mentions

Not every mention needs an alert. Filter for relevance and severity.

def filter_mentions(mentions, seen_urls):
    actionable = []
    for m in mentions:
        # Skip already-seen URLs
        if m["url"] in seen_urls:
            continue
        # Always alert on negative sentiment
        if m["sentiment"] == "negative":
            actionable.append(m)
        # Alert on competitor comparisons
        elif "vs" in m["query"].lower() or "alternative" in m["query"].lower():
            actionable.append(m)
        # Alert on pricing mentions
        elif "pricing" in m["query"].lower():
            actionable.append(m)
        seen_urls.add(m["url"])
    return actionable

Step 6: Set Up Alerting

Route alerts to Slack, Discord, or any webhook endpoint.

def send_alert(webhook_url, mention):
    emoji = {"negative": "🔴", "positive": "🟢", "neutral": "🟡"}
    payload = {
        "text": (
            f"{emoji.get(mention['sentiment'], '⚪')} Brand Mention Alert\n"
            f"**{mention['title']}**\n"
            f"Sentiment: {mention['sentiment']}\n"
            f"Source: {mention['url']}\n"
            f"Snippet: {mention['snippet'][:200]}"
        )
    }
    try:
        requests.post(webhook_url, json=payload, timeout=10)
    except Exception as e:
        print(f"Alert failed: {e}")

Step 7: Schedule Automatic Monitoring

Run the monitoring loop on a schedule using APScheduler.

from apscheduler.schedulers.blocking import BlockingScheduler

seen_urls = set()
WEBHOOK_URL = os.environ.get("ALERT_WEBHOOK_URL", "")

def monitoring_job():
    print(f"[{datetime.utcnow()}] Running brand monitor...")
    mentions = search_mentions(client, BRAND_QUERIES)
    enriched = enrich_mentions(client, mentions)
    actionable = filter_mentions(enriched, seen_urls)
    for m in actionable:
        print(f"  Alert: {m['sentiment']} - {m['title']}")
        if WEBHOOK_URL:
            send_alert(WEBHOOK_URL, m)
    print(f"  Found {len(mentions)} mentions, {len(actionable)} actionable")

scheduler = BlockingScheduler()
scheduler.add_job(monitoring_job, "interval", hours=4)
print("Brand monitor started. Running every 4 hours.")
scheduler.start()

Step 8: Persist Mentions to Disk

For historical tracking, save mentions to a JSONL file.

import os

MENTIONS_FILE = "mentions.jsonl"

def save_mentions(mentions):
    with open(MENTIONS_FILE, "a") as f:
        for m in mentions:
            f.write(json.dumps(m) + "\n")

def load_seen_urls():
    urls = set()
    if os.path.exists(MENTIONS_FILE):
        with open(MENTIONS_FILE) as f:
            for line in f:
                data = json.loads(line)
                urls.add(data.get("url", ""))
    return urls

Add save_mentions(enriched) to your monitoring job and initialize seen_urls = load_seen_urls() on startup.

Complete Code Example

Here's the full script combined:

import requests
import os
import json
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

# Configuration
BRAND_QUERIES = [
    '"SearchHive" review',
    '"SearchHive" vs',
    '"SearchHive" alternative',
    '"SearchHive" pricing',
]
NEGATIVE_SIGNALS = ["worst", "terrible", "avoid", "scam"]
POSITIVE_SIGNALS = ["best", "love", "recommend", "excellent"]
MENTIONS_FILE = "mentions.jsonl"

class BrandMonitor:
    def __init__(self):
        self.api_key = os.environ.get("SEARCHHIVE_API_KEY")
        self.webhook = os.environ.get("ALERT_WEBHOOK_URL", "")
        self.seen = self._load_seen()

    def _headers(self):
        return {"Authorization": f"Bearer {self.api_key}"}

    def search(self, query):
        resp = requests.get(
            "https://api.searchhive.dev/v1/search",
            headers=self._headers(), params={"q": query, "num": 20}
        )
        return resp.json().get("results", [])

    def scrape(self, url):
        resp = requests.get(
            "https://api.searchhive.dev/v1/scrape",
            headers=self._headers(), params={"url": url}
        )
        return resp.json()

    def run(self):
        mentions = []
        for q in BRAND_QUERIES:
            for r in self.search(q):
                if r["url"] in self.seen:
                    continue
                mentions.append({"title": r["title"], "url": r["url"],
                                 "snippet": r.get("description", "")})
                self.seen.add(r["url"])

        actionable = [m for m in mentions if any(
            s in m["snippet"].lower() for s in NEGATIVE_SIGNALS
        )]

        for m in actionable[:5]:
            try:
                data = self.scrape(m["url"])
                m["content"] = data.get("markdown", "")[:500]
            except Exception:
                pass

        self._save(mentions)
        for m in actionable:
            self._alert(m)
        return len(mentions), len(actionable)

    def _alert(self, m):
        if not self.webhook:
            return
        requests.post(self.webhook, json={
            "text": f"🚨 Brand Alert: {m['title']}\n{m['url']}"
        }, timeout=10)

    def _save(self, mentions):
        with open(MENTIONS_FILE, "a") as f:
            for m in mentions:
                f.write(json.dumps(m) + "\n")

    def _load_seen(self):
        s = set()
        if os.path.exists(MENTIONS_FILE):
            with open(MENTIONS_FILE) as f:
                for line in f:
                    s.add(json.loads(line).get("url", ""))
        return s

monitor = BrandMonitor()
sched = BlockingScheduler()
sched.add_job(monitor.run, "interval", hours=4)
sched.start()

Common Issues

Rate limits: The free tier gives 500 credits. Each search costs ~1 credit, each scrape costs ~1-3 credits. Running 4 queries every 4 hours = ~24 searches/day + ~12 scrapes = ~36 credits/day. Well within the free tier.

Duplicate alerts: The seen_urls set prevents duplicate alerts. Persist it to disk between restarts using the JSONL file.

False positives: The keyword-based sentiment filter is basic. For production use, consider passing snippets through an LLM for more accurate classification. Or use SearchHive's DeepDive for analysis of complex mentions.

Webhook failures: Add retry logic with exponential backoff. The example above silently catches errors -- in production, log them and retry.

Next Steps

• Scale up: Move from local scheduling to a cloud function (AWS Lambda, Cloudflare Workers) triggered by cron expression generator
• Add competitors: Expand BRAND_QUERIES to track competitor launches and pricing changes
• Build a dashboard: Store mentions in a database and build a simple web UI for the team
• Use DeepDive: Replace simple scraping with SearchHive's DeepDive API for deeper analysis of complex mentions

Get started with SearchHive -- 500 free credits, no credit card required. Read the full API documentation for advanced configuration options.

/tutorials/brand-monitoring | /blog/cheapest-serp-api-for-developers | /compare/serpapi