How to Automate Review Monitoring Across Platforms

When you automate review monitoring across google yelp and facebook, every new review lands in one place within minutes of being posted. I build these for multi-location businesses that cannot watch 5 platforms per location manually. The system polls each platform, detects new reviews, and sends alerts with the full review text.

Negative reviews get immediate alerts. Positive ones get batched daily.

What You Need Before Starting

• API access to Google Business Profile, Yelp, and Facebook
• Python 3.8+ with requests and scheduling
• A database to track seen reviews
• Slack or email for notifications

Step 1: Build Platform Connectors

Each platform has its own API. Create a connector for each:

import requests
import os

def fetch_google_reviews(location_id):
    url = f"https://mybusiness.googleapis.com/v4/accounts/{os.getenv('GOOGLE_ACCOUNT_ID')}/locations/{location_id}/reviews"
    headers = {"Authorization": f"Bearer {os.getenv('GOOGLE_ACCESS_TOKEN')}"}
    response = requests.get(url, headers=headers)
    reviews = response.json().get("reviews", [])
    return [{
        "platform": "google",
        "review_id": r["reviewId"],
        "author": r["reviewer"]["displayName"],
        "rating": r["starRating"],
        "text": r.get("comment", ""),
        "created_at": r["createTime"]
    } for r in reviews]

def fetch_facebook_reviews(page_id):
    url = f"https://graph.facebook.com/v18.0/{page_id}/ratings"
    params = {"access_token": os.getenv("FB_PAGE_TOKEN"), "fields": "reviewer,rating,review_text,created_time"}
    response = requests.get(url, params=params)
    reviews = response.json().get("data", [])
    return [{
        "platform": "facebook",
        "review_id": r["id"],
        "author": r["reviewer"]["name"],
        "rating": r["rating"],
        "text": r.get("review_text", ""),
        "created_at": r["created_time"]
    } for r in reviews]

Step 2: Detect New Reviews

Track what you have seen and flag new ones:

import sqlite3

def get_new_reviews(reviews):
    conn = sqlite3.connect("reviews.db")
    new_reviews = []

    for review in reviews:
        exists = conn.execute(
            "SELECT 1 FROM seen_reviews WHERE platform = ? AND review_id = ?",
            (review["platform"], review["review_id"])
        ).fetchone()

        if not exists:
            new_reviews.append(review)
            conn.execute(
                "INSERT INTO seen_reviews (platform, review_id, seen_at) VALUES (?, ?, datetime('now'))",
                (review["platform"], review["review_id"])
            )
    conn.commit()
    return new_reviews

Step 3: Route Alerts by Sentiment

def process_new_reviews(reviews):
    urgent = []
    normal = []

    for review in reviews:
        rating = convert_rating(review["rating"])
        if rating <= 2:
            urgent.append(review)
        else:
            normal.append(review)

    if urgent:
        send_urgent_alerts(urgent)

    if normal:
        queue_daily_digest(normal)

def convert_rating(rating):
    """Normalize ratings to 1-5 scale across platforms."""
    if isinstance(rating, str):
        star_map = {"ONE": 1, "TWO": 2, "THREE": 3, "FOUR": 4, "FIVE": 5}
        return star_map.get(rating, 3)
    return int(rating)

Step 4: Send Notifications

def send_urgent_alerts(reviews):
    for review in reviews:
        message = (
            f"New {review['rating']}-star review on {review['platform']}:\n"
            f"From: {review['author']}\n"
            f"\"{review['text'][:300]}\"\n"
            f"Respond ASAP to mitigate."
        )
        requests.post(os.getenv("SLACK_URGENT_WEBHOOK"), json={"text": message})

def send_daily_digest(reviews):
    summary = f"New reviews today: {len(reviews)}\n\n"
    for r in reviews:
        summary += f"{'*' * convert_rating(r['rating'])} - {r['author']} ({r['platform']})\n"
        summary += f"  {r['text'][:100]}...\n\n"
    requests.post(os.getenv("SLACK_REVIEWS_WEBHOOK"), json={"text": summary})

Step 5: Run the Monitor

Schedule checks every 30 minutes:

from apscheduler.schedulers.blocking import BlockingScheduler

def check_all_platforms():
    all_reviews = []
    all_reviews.extend(fetch_google_reviews(os.getenv("GOOGLE_LOCATION_ID")))
    all_reviews.extend(fetch_facebook_reviews(os.getenv("FB_PAGE_ID")))

    new = get_new_reviews(all_reviews)
    if new:
        process_new_reviews(new)
        save_reviews(new)

scheduler = BlockingScheduler()
scheduler.add_job(check_all_platforms, "interval", minutes=30)
scheduler.start()

What to Build Next

Add competitor review monitoring. Track what customers say about your competitors on the same platforms. That competitive intelligence helps you address gaps in your own service before customers point them out.

Related Reading

• Competitive Intelligence with AI - monitoring competitors alongside your own reviews
• Why Monitoring Is Not Optional - review monitoring as a business critical system
• The Feedback Loop That Powers Everything - reviews feeding back into service improvement