How to Automate Review Aggregation and Reporting

When you automate review aggregation into a single reporting dashboard, you stop logging into 5 platforms every morning. I build these for businesses with reviews scattered across Google, Yelp, Facebook, TripAdvisor, and industry-specific sites. One dashboard shows everything: total reviews, average rating, trends, and which platforms need attention.

The weekly report writes itself.

What You Need Before Starting

• Review data from multiple platforms (see system 394 for monitoring)
• Python 3.8+ with Flask and sqlite3
• A scheduling tool for report generation
• Email or Slack for report delivery

Step 1: Normalize Reviews Across Platforms

Each platform formats data differently. Normalize into a standard schema:

def normalize_review(raw_review, platform):
    rating_map = {
        "google": lambda r: {"ONE": 1, "TWO": 2, "THREE": 3, "FOUR": 4, "FIVE": 5}.get(r, r),
        "yelp": lambda r: int(r),
        "facebook": lambda r: int(r),
    }

    return {
        "id": f"{platform}_{raw_review['id']}",
        "platform": platform,
        "author": raw_review.get("author", "Anonymous"),
        "rating": rating_map.get(platform, lambda r: int(r))(raw_review["rating"]),
        "text": raw_review.get("text", ""),
        "date": raw_review["created_at"],
        "responded": raw_review.get("response") is not None,
    }

Step 2: Build Aggregation Queries

import sqlite3

def get_aggregate_stats(days=30):
    conn = sqlite3.connect("reviews_agg.db")
    start = f"-{days} days"

    total = conn.execute("SELECT COUNT(*) FROM reviews WHERE date > datetime('now', ?)", (start,)).fetchone()[0]
    avg_rating = conn.execute("SELECT AVG(rating) FROM reviews WHERE date > datetime('now', ?)", (start,)).fetchone()[0]

    by_platform = conn.execute("""
        SELECT platform, COUNT(*), AVG(rating), 
               SUM(CASE WHEN responded = 1 THEN 1 ELSE 0 END) as responded
        FROM reviews WHERE date > datetime('now', ?)
        GROUP BY platform
    """, (start,)).fetchall()

    by_rating = conn.execute("""
        SELECT rating, COUNT(*) FROM reviews WHERE date > datetime('now', ?)
        GROUP BY rating ORDER BY rating DESC
    """, (start,)).fetchall()

    return {
        "total_reviews": total,
        "avg_rating": round(avg_rating or 0, 2),
        "by_platform": [{"platform": r[0], "count": r[1], "avg": round(r[2], 2), "responded": r[3]} for r in by_platform],
        "by_rating": dict(by_rating)
    }

Step 3: Generate the Weekly Report

import anthropic

client = anthropic.Anthropic()

def generate_weekly_report():
    current = get_aggregate_stats(days=7)
    previous = get_aggregate_stats(days=14)

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=400,
        messages=[{
            "role": "user",
            "content": f"""Write a brief weekly review report. Include the numbers below.
Highlight wins and concerns. Keep it under 200 words.

This Week: {json.dumps(current)}
Last Week (for comparison): {json.dumps(previous)}"""
        }]
    )
    return response.content[0].text

Step 4: Serve the Dashboard API

from flask import Flask, jsonify

app = Flask(__name__)

@app.route("/api/reviews/overview")
def overview():
    return jsonify(get_aggregate_stats(days=30))

@app.route("/api/reviews/trend")
def trend():
    conn = sqlite3.connect("reviews_agg.db")
    rows = conn.execute("""
        SELECT DATE(date) as day, COUNT(*), AVG(rating)
        FROM reviews WHERE date > datetime('now', '-90 days')
        GROUP BY DATE(date) ORDER BY day
    """).fetchall()
    return jsonify([{"date": r[0], "count": r[1], "avg_rating": round(r[2], 2)} for r in rows])

@app.route("/api/reviews/response-rate")
def response_rate():
    conn = sqlite3.connect("reviews_agg.db")
    total = conn.execute("SELECT COUNT(*) FROM reviews WHERE date > datetime('now', '-30 days')").fetchone()[0]
    responded = conn.execute("SELECT COUNT(*) FROM reviews WHERE responded = 1 AND date > datetime('now', '-30 days')").fetchone()[0]
    return jsonify({"total": total, "responded": responded, "rate": round(responded / total * 100, 1) if total else 0})

Step 5: Schedule Automated Delivery

from apscheduler.schedulers.blocking import BlockingScheduler

def deliver_weekly_report():
    report = generate_weekly_report()
    stats = get_aggregate_stats(days=7)

    send_slack_message(
        channel="#reviews",
        text=f"Weekly Review Report\n\n{report}\n\nTotal: {stats['total_reviews']} | Avg: {stats['avg_rating']}"
    )

scheduler = BlockingScheduler()
scheduler.add_job(deliver_weekly_report, "cron", day_of_week="mon", hour=9)
scheduler.start()

What to Build Next

Add review response tracking. Measure how response rate correlates with new review volume and average rating. Most businesses see a 10-15% increase in review volume when they respond to every review consistently.

Related Reading

• AI-Powered Reporting That Actually Gets Read - making review reports people actually open
• The Centralized Brain Concept - centralizing review data into one intelligence layer
• Data Flow Architecture for Non-Engineers - how review data flows from platforms to dashboards