Edtech Stripe API: Automated Reconciliation Blueprint

import os
import stripe
import json
import csv
from flask import Flask, request, abort

# --- Configuration ---
# Set your Stripe API secret key (use environment variable for security)
stripe.api_key = os.environ.get('STRIPE_SECRET_KEY', 'sk_test_YOUR_SECRET_KEY')

# Set your webhook signing secret (use environment variable for security)
webhook_secret = os.environ.get('STRIPE_WEBHOOK_SECRET', 'whsec_YOUR_WEBHOOK_SECRET')

# Output CSV file for storing transaction data
OUTPUT_CSV = 'stripe_transactions.csv'

# CSV Header
CSV_HEADER = ['event_id', 'event_type', 'created_at', 'customer_id', 'charge_id', 'invoice_id', 'amount', 'currency', 'status']

# --- Flask App Setup ---
app = Flask(__name__)

def initialize_csv():
    """Initializes the CSV file with header if it doesn't exist."""
    if not os.path.exists(OUTPUT_CSV):
        with open(OUTPUT_CSV, 'w', newline='') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(CSV_HEADER)

def log_transaction(event_data):
    """Logs relevant transaction data to the CSV file."""
    try:
        with open(OUTPUT_CSV, 'a', newline='') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow([
                event_data.get('id'),
                event_data.get('type'),
                event_data.get('created'),
                event_data.get('data', {}).get('object', {}).get('customer'),
                event_data.get('data', {}).get('object', {}).get('charge'),
                event_data.get('data', {}).get('object', {}).get('invoice'),
                event_data.get('data', {}).get('object', {}).get('amount'),
                event_data.get('data', {}).get('object', {}).get('currency'),
                event_data.get('data', {}).get('object', {}).get('status')
            ])
        print(f"Logged event {event_data.get('id')}")
    except Exception as e:
        print(f"Error logging to CSV: {e}")

@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.data
    sig_header = request.headers.get('Stripe-Signature')
    event = None

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, webhook_secret
        )
    except ValueError as e:
        # Invalid payload
        print(f"Webhook Error: Invalid payload - {e}")
        return abort(400)
    except stripe.error.SignatureVerificationError as e:
        # Invalid signature
        print(f"Webhook Error: Invalid signature - {e}")
        return abort(400)
    except Exception as e:
        # Other errors
        print(f"Webhook Error: Unexpected error - {e}")
        return abort(400)

    # Handle the event
    if event['type'] in ['charge.succeeded', 'charge.failed', 'invoice.payment_succeeded', 'invoice.payment_failed']:
        # Retrieve the object for further processing
        data = event['data']
        obj = data['object']
        print(f"Received Stripe event: {event['type']} for object ID: {obj.get('id')}")
        
        # Log the relevant data
        log_transaction(event)

    # Return a 200 response to acknowledge receipt of the event
    return json.dumps({'status': 'success'}), 200

if __name__ == '__main__':
    initialize_csv()
    # For local testing, you can use:
    # app.run(port=4242, debug=True)
    # For production, use a WSGI server like Gunicorn:
    # gunicorn -w 4 -b 0.0.0.0:8000 stripe_webhook_listener:app
    print(f"Stripe webhook listener started. Logging to {OUTPUT_CSV}")
    print("Ensure your STRIPE_SECRET_KEY and STRIPE_WEBHOOK_SECRET environment variables are set.")
    print("For local testing, you can run this script directly. For production, use a WSGI server.")
    # Example of running with Flask development server (not for production)
    app.run(port=4242, debug=False) # Set debug=False for more realistic behavior