🛠️ merge_payment_ledger: Merge Gumroad, Stripe, and PayPal CSVs into a unified transaction ledg

Merge Payment Ledger

A simple Python tool to consolidate transaction data from Gumroad, Stripe, and PayPal into a single, consistent CSV ledger. This is useful for creators and small businesses receiving payments across multiple platforms who want a unified view for accounting or tax reporting.

Features

• Automatically detects and parses CSV files from Gumroad, Stripe, and PayPal
• Normalizes disparate column names and date formats
• Outputs a clean, merged CSV with consistent schema
• Handles duplicate transactions using ID and timestamp
• Lightweight, no database required

Usage

Run the script via command line:

python merge_payment_ledger.py -i ./transactions/ -o merged_ledger.csv

Input directory should contain CSV files from Gumroad, Stripe, and/or PayPal. The script auto-detects the source based on headers.

Output CSV includes the following columns:

• platform (source: gumroad, stripe, paypal)
• transaction_id
• date (standardized ISO format)
• amount_usd
• currency
• description
• buyer_email

Requirements

• Python 3.6+
• Standard libraries only: csv, os, datetime, argparse

No external dependencies. Safe for financial data.

Why This Tool?

Manually reconciling income across platforms is error-prone. This script ensures consistency and transparency by applying deterministic merging logic. Perfect for solo creators, indie hackers, or freelancers using multiple payment tools.

Notes

• Always back up original files before processing
• Review output for edge cases (e.g., refunds, fees)
• Extendable: add new providers by updating the parsing logic

Maintained for simplicity and auditability.

import csv
import os
import argparse
from datetime import datetime


def parse_gumroad_row(row):
    return {
        'platform': 'gumroad',
        'transaction_id': row.get('sale_id', ''),
        'date': row.get('sale_date', ''),
        'amount_usd': float(row.get('amount_usd', 0) or 0),
        'currency': row.get('currency', 'USD'),
        'description': row.get('product_name', ''),
        'buyer_email': row.get('customer_email', '')
    }


def parse_stripe_row(row):
    return {
        'platform': 'stripe',
        'transaction_id': row.get('Transaction ID', ''),
        'date': row.get('Created', '').split()[0],
        'amount_usd': float(row.get('Amount', 0) or 0),
        'currency': row.get('Currency', 'USD').upper(),
        'description': row.get('Description', ''),
        'buyer_email': row.get('Customer Email', '')
    }


def parse_paypal_row(row):
    return {
        'platform': 'paypal',
        'transaction_id': row.get('Transaction ID', ''),
        'date': row.get('Date', '').split()[0],
        'amount_usd': abs(float(row.get('Gross', 0) or 0)),
        'currency': row.get('Currency', 'USD'),
        'description': row.get('Name', ''),
        'buyer_email': row.get('Name', '')
    }


def detect_platform(headers):
    if 'sale_id' in headers and 'product_name' in headers:
        return 'gumroad'
    elif 'Transaction ID' in headers and 'Customer Email' in headers:
        return 'stripe'
    elif 'Transaction ID' in headers and 'Gross' in headers:
        return 'paypal'
    return None


def merge_csvs(input_dir, output_file):
    transactions = []
    seen = set()

    for file in os.listdir(input_dir):
        if not file.lower().endswith('.csv'):
            continue
        path = os.path.join(input_dir, file)
        with open(path, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            platform = detect_platform(reader.fieldnames)

            for row in reader:
                tx = None
                if platform == 'gumroad':
                    tx = parse_gumroad_row(row)
                elif platform == 'stripe':
                    tx = parse_stripe_row(row)
                elif platform == 'paypal':
                    tx = parse_paypal_row(row)

                if tx and tx['transaction_id'] not in seen:
                    seen.add(tx['transaction_id'])
                    transactions.append(tx)

    transactions.sort(key=lambda x: x['date'], reverse=True)

    if transactions:
        keys = transactions[0].keys()
        with open(output_file, 'w', newline='', encoding='utf-8') as f:
            writer = csv.DictWriter(f, fieldnames=keys)
            writer.writeheader()
            writer.writerows(transactions)

    print(f'Merged {len(transactions)} transactions into {output_file}')


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Merge Gumroad, Stripe, PayPal CSVs into one ledger.')
    parser.add_argument('-i', '--input', required=True, help='Input directory containing CSV files')
    parser.add_argument('-o', '--output', default='merged_ledger.csv', help='Output CSV file')
    args = parser.parse_args()
    merge_csvs(args.input, args.output)