How to Automate Stripe Revenue Reports using the API

1import stripe
2import requests
3import csv
4import io
5import os
6import time
7import logging
8from datetime import datetime, timezone, timedelta
9from calendar import monthrange
10
11logging.basicConfig(level=logging.INFO, format='%(asctime)s %(message)s')
12log = logging.getLogger(__name__)
13
14stripe.api_key = os.environ['STRIPE_SECRET_KEY']
15SLACK_WEBHOOK = os.environ.get('SLACK_WEBHOOK_URL')
16
17def get_last_month_range():
18    today = datetime.now(timezone.utc)
19    first_of_month = today.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
20    last_month_end = first_of_month - timedelta(seconds=1)
21    last_month_start = last_month_end.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
22    return int(last_month_start.timestamp()), int(first_of_month.timestamp())
23
24def request_report(start_ts, end_ts):
25    run = stripe.reporting.ReportRun.create(
26        report_type='balance.summary.1',
27        parameters={'interval_start': start_ts, 'interval_end': end_ts}
28    )
29    log.info(f'Report run created: {run.id}')
30    return run.id
31
32def wait_for_report(run_id, max_wait=300):
33    deadline = time.time() + max_wait
34    while time.time() < deadline:
35        run = stripe.reporting.ReportRun.retrieve(run_id)
36        if run.status == 'succeeded':
37            return run.result.url
38        elif run.status == 'failed':
39            raise Exception(f'Report failed: {run.error}')
40        log.info(f'Report status: {run.status}, waiting...')
41        time.sleep(10)
42    raise TimeoutError('Report timed out')
43
44def download_and_parse(url):
45    resp = requests.get(url, auth=(stripe.api_key, ''))
46    resp.raise_for_status()
47    reader = csv.DictReader(io.StringIO(resp.text))
48    return list(reader)
49
50def send_slack(message):
51    if SLACK_WEBHOOK:
52        requests.post(SLACK_WEBHOOK, json={'text': message})
53        log.info('Slack notification sent')
54
55def main():
56    start_ts, end_ts = get_last_month_range()
57    period = datetime.fromtimestamp(start_ts, timezone.utc).strftime('%B %Y')
58    log.info(f'Generating revenue report for {period}')
59
60    run_id = request_report(start_ts, end_ts)
61    download_url = wait_for_report(run_id)
62    rows = download_and_parse(download_url)
63
64    gross = net = fees = 0
65    for row in rows:
66        try:
67            gross += float(row.get('gross', 0) or 0)
68            net += float(row.get('net', 0) or 0)
69            fees += float(row.get('fee', 0) or 0)
70        except (ValueError, KeyError):
71            pass
72
73    currency = rows[0].get('currency', 'usd').upper() if rows else 'USD'
74    message = (
75        f'*Stripe Revenue Report — {period}*\n'
76        f'Gross Revenue: {currency} {gross:,.2f}\n'
77        f'Stripe Fees: {currency} {fees:,.2f}\n'
78        f'Net Revenue: {currency} {net:,.2f}'
79    )
80    log.info(message)
81    send_slack(message)
82
83if __name__ == '__main__':
84    main()