Last quarter, I spent 6 hours every Monday compiling business reports from Stripe, Google Analytics, and our CRM. Now a Python script does it in 90 seconds and emails the PDF to my team. Here's the complete implementation.
System Architecture
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ Data Sources │ │ AI Summary │ │ Output │
│ │ │ │ │ │
│ • Stripe API │───▶│ • Claude API │───▶│ • PDF Report │
│ • GA4 API │ │ • Trend │ │ • Email via │
│ • CRM API │ │ Analysis │ │ SendGrid │
│ • Database │ │ • Anomaly │ │ • Slack Post │
│ │ │ Detection │ │ │
└──────────────┘ └──────────────┘ └──────────────┘
Step 1: Data Collection Layer
import stripe
import anthropic
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import (
RunReportRequest, DateRange, Metric, Dimension
)
from dataclasses import dataclass, field
from datetime import datetime, timedelta
@dataclass
class RevenueData:
total_revenue: float
transaction_count: int
avg_transaction: float
mrr: float
churn_rate: float
new_customers: int
top_products: list[dict] = field(default_factory=list)
@dataclass
class TrafficData:
total_sessions: int
unique_users: int
bounce_rate: float
avg_session_duration: float
top_pages: list[dict] = field(default_factory=list)
traffic_sources: list[dict] = field(default_factory=list)
@dataclass
class BusinessReport:
period_start: datetime
period_end: datetime
revenue: RevenueData
traffic: TrafficData
ai_summary: str = ""
anomalies: list[str] = field(default_factory=list)
recommendations: list[str] = field(default_factory=list)
def fetch_stripe_data(
start_date: datetime, end_date: datetime
) -> RevenueData:
stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
# Fetch charges in date range
charges = stripe.Charge.list(
created={
"gte": int(start_date.timestamp()),
"lte": int(end_date.timestamp()),
},
limit=100,
)
successful = [c for c in charges.data if c.status == "succeeded"]
total = sum(c.amount for c in successful) / 100 # cents to dollars
# Fetch subscriptions for MRR
subscriptions = stripe.Subscription.list(status="active", limit=100)
mrr = sum(
s.items.data[0].price.unit_amount * s.items.data[0].quantity
for s in subscriptions.data
) / 100
# New customers this period
customers = stripe.Customer.list(
created={
"gte": int(start_date.timestamp()),
"lte": int(end_date.timestamp()),
}
)
# Top products by revenue
product_revenue: dict[str, float] = {}
for charge in successful:
desc = charge.description or "Other"
product_revenue[desc] = product_revenue.get(desc, 0) + charge.amount / 100
top_products = [
{"name": k, "revenue": v}
for k, v in sorted(
product_revenue.items(), key=lambda x: x[1], reverse=True
)[:5]
]
return RevenueData(
total_revenue=total,
transaction_count=len(successful),
avg_transaction=total / len(successful) if successful else 0,
mrr=mrr,
churn_rate=0.0, # Calculate from subscription cancellations
new_customers=len(customers.data),
top_products=top_products,
)
def fetch_ga4_data(
property_id: str, start_date: datetime, end_date: datetime
) -> TrafficData:
client = BetaAnalyticsDataClient()
request = RunReportRequest(
property=f"properties/{property_id}",
date_ranges=[DateRange(
start_date=start_date.strftime("%Y-%m-%d"),
end_date=end_date.strftime("%Y-%m-%d"),
)],
metrics=[
Metric(name="sessions"),
Metric(name="totalUsers"),
Metric(name="bounceRate"),
Metric(name="averageSessionDuration"),
],
)
response = client.run_report(request)
row = response.rows[0]
return TrafficData(
total_sessions=int(row.metric_values[0].value),
unique_users=int(row.metric_values[1].value),
bounce_rate=float(row.metric_values[2].value),
avg_session_duration=float(row.metric_values[3].value),
)
Step 2: AI Analysis Engine
This is where Claude transforms raw numbers into actionable insights:
import json
import os
def analyze_with_ai(report: BusinessReport) -> BusinessReport:
client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
data_summary = {
"period": f"{report.period_start:%Y-%m-%d} to {report.period_end:%Y-%m-%d}",
"revenue": {
"total": report.revenue.total_revenue,
"transactions": report.revenue.transaction_count,
"avg_transaction": report.revenue.avg_transaction,
"mrr": report.revenue.mrr,
"new_customers": report.revenue.new_customers,
"top_products": report.revenue.top_products,
},
"traffic": {
"sessions": report.traffic.total_sessions,
"users": report.traffic.unique_users,
"bounce_rate": report.traffic.bounce_rate,
"avg_duration": report.traffic.avg_session_duration,
},
}
message = client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2000,
system=(
"You are a business analyst. Analyze the data and return JSON with: "
"summary (3-paragraph executive summary), "
"anomalies (array of unusual patterns), "
"recommendations (array of actionable next steps). "
"Be specific with numbers. Flag anything unusual."
),
messages=[{
"role": "user",
"content": f"Analyze this business data:\n{json.dumps(data_summary, indent=2)}",
}],
)
analysis = json.loads(message.content[0].text)
report.ai_summary = analysis["summary"]
report.anomalies = analysis["anomalies"]
report.recommendations = analysis["recommendations"]
return report
Step 3: PDF Report Generation
from reportlab.lib.pagesizes import letter
from reportlab.platypus import (
SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle, HRFlowable
)
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.colors import HexColor
from reportlab.lib.units import inch
def generate_pdf(report: BusinessReport, output_path: str) -> str:
doc = SimpleDocTemplate(output_path, pagesize=letter)
styles = getSampleStyleSheet()
story = []
# Header
title_style = ParagraphStyle(
name="ReportTitle", fontSize=22, spaceAfter=5,
textColor=HexColor("#0f172a"), fontName="Helvetica-Bold",
)
story.append(Paragraph("Weekly Business Report", title_style))
story.append(Paragraph(
f"{report.period_start:%B %d} — {report.period_end:%B %d, %Y}",
styles["Normal"]
))
story.append(HRFlowable(width="100%", color=HexColor("#e2e8f0")))
story.append(Spacer(1, 0.3 * inch))
# KPI Summary Table
kpi_data = [
["Metric", "Value"],
["Total Revenue", f"${report.revenue.total_revenue:,.2f}"],
["Transactions", str(report.revenue.transaction_count)],
["MRR", f"${report.revenue.mrr:,.2f}"],
["New Customers", str(report.revenue.new_customers)],
["Sessions", f"{report.traffic.total_sessions:,}"],
["Unique Users", f"{report.traffic.unique_users:,}"],
["Bounce Rate", f"{report.traffic.bounce_rate:.1f}%"],
]
table = Table(kpi_data, colWidths=[3 * inch, 3 * inch])
table.setStyle(TableStyle([
("BACKGROUND", (0, 0), (-1, 0), HexColor("#1e293b")),
("TEXTCOLOR", (0, 0), (-1, 0), HexColor("#ffffff")),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
("GRID", (0, 0), (-1, -1), 0.5, HexColor("#cbd5e1")),
("PADDING", (0, 0), (-1, -1), 10),
("ROWBACKGROUNDS", (0, 1), (-1, -1), [HexColor("#f8fafc"), HexColor("#ffffff")]),
]))
story.append(table)
story.append(Spacer(1, 0.3 * inch))
# AI Summary
story.append(Paragraph("Executive Summary", styles["Heading2"]))
story.append(Paragraph(report.ai_summary, styles["BodyText"]))
# Anomalies
if report.anomalies:
story.append(Paragraph("Anomalies Detected", styles["Heading2"]))
for anomaly in report.anomalies:
story.append(Paragraph(f"⚠ {anomaly}", styles["BodyText"]))
# Recommendations
story.append(Paragraph("Recommendations", styles["Heading2"]))
for i, rec in enumerate(report.recommendations, 1):
story.append(Paragraph(f"{i}. {rec}", styles["BodyText"]))
doc.build(story)
return output_path
Step 4: Email Delivery
import sendgrid
from sendgrid.helpers.mail import (
Mail, Attachment, FileContent, FileName, FileType, Disposition
)
import base64
def email_report(report_path: str, recipients: list[str]) -> bool:
sg = sendgrid.SendGridAPIClient(api_key=os.environ["SENDGRID_API_KEY"])
with open(report_path, "rb") as f:
encoded_pdf = base64.b64encode(f.read()).decode()
attachment = Attachment(
FileContent(encoded_pdf),
FileName("weekly-report.pdf"),
FileType("application/pdf"),
Disposition("attachment"),
)
message = Mail(
from_email="reports@yourdomain.com",
to_emails=recipients,
subject=f"Weekly Business Report — {datetime.now():%B %d, %Y}",
html_content=(
"<h2>Your weekly business report is attached.</h2>"
"<p>Key highlights are included in the executive summary.</p>"
),
)
message.attachment = attachment
response = sg.send(message)
return response.status_code == 202
Scheduling with Cron
# Run every Monday at 8 AM
0 8 * * 1 cd /path/to/project && python report_generator.py >> /var/log/reports.log 2>&1
Or with a Python scheduler for more control:
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
scheduler = BlockingScheduler()
@scheduler.scheduled_job(CronTrigger(day_of_week="mon", hour=8))
def weekly_report():
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
report = BusinessReport(
period_start=start_date,
period_end=end_date,
revenue=fetch_stripe_data(start_date, end_date),
traffic=fetch_ga4_data("GA4_PROPERTY_ID", start_date, end_date),
)
report = analyze_with_ai(report)
pdf_path = generate_pdf(report, f"report_{end_date:%Y%m%d}.pdf")
email_report(pdf_path, ["team@company.com", "ceo@company.com"])
scheduler.start()
What This Actually Saved
- 6 hours/week of manual report compilation → 90 seconds automated
- Consistency — every report has the same structure and depth
- AI catches things I miss — anomaly detection flagged a 40% drop in one traffic source that I would have overlooked
The complete system with 30 automation blueprints — covering CRM sync, invoice processing, customer onboarding, and more — is in my AI Automation Playbook for SMB. Each blueprint includes the full source code and deployment instructions.
What business process are you automating with AI? I'm always looking for new use cases to build.
Top comments (0)