Convert PayPal CSV to OFX

paypal_csv_2_ofx.py

# TL;RD - This script takes a CSV file exported from PayPal and
#        makes a OFX file.
#
# Warning - I wrote this with the help of an LLM so you may want
#           to review is carefully.
# 
# Warning - I know next to nothing about the OFX format.
#
# Warning - If you do not know anything about python or how to
#           run command line scripts this probably is not the
#           solution for you.
#
# This is script was made to solve a YNAB problem. The integration
# between YNAB and Paypal has never worked very well for me. I blame
# PayPal. I can't imagine PayPal has been very helpful.
#
# The way this script works now is very simple. It takes in the CSV
# in STDIN and send an OFX to STDOUT.
#
# Running in a useful way for me looks like:
# `cat paypal_payments.csv | python paypal_csv_2_ofx.py > paypal_transactions.ofx`
#
# In the future it would be fun to automated more of the steps
# like downloading the CSV from PayPal or importing the OFX to
# YNAB. If I end up using this a lot, maybe I'll work more on it.

import csv
import sys
from datetime import datetime
from io import StringIO

def parse_csv(input_data):
    # Remove BOM if present
    if input_data.startswith('\ufeff'):
        input_data = input_data.lstrip('\ufeff')

    transactions = []
    reader = csv.DictReader(StringIO(input_data))
    for row in reader:
        try:
            date = datetime.strptime(f"{row['Date']} {row['Time']} {row['TimeZone']}", '%m/%d/%Y %H:%M:%S %Z')
        except ValueError:
            date = datetime.strptime(f"{row['Date']} {row['Time']}", '%m/%d/%Y %H:%M:%S')  # For any timezone issues

        transactions.append({
            'date': date,
            'amount': float(row['Net']),
            'name': row['Name'] if row['Name'] else "Unknown",
            'type': row['Type'],
            'id': row['Receipt ID'] if row['Receipt ID'] else f"{date.strftime('%Y%m%d%H%M%S')}_{row['Gross']}",
            'status': row['Status']
        })
    return transactions

def create_ofx(transactions):
    ofx = [
        "OFXHEADER:100",
        "DATA:OFXSGML",
        "VERSION:102",
        "SECURITY:NONE",
        "ENCODING:USASCII",
        "CHARSET:1252",
        "COMPRESSION:NONE",
        "OLDFILEUID:NONE",
        "NEWFILEUID:NONE",
        "",
        "OFX",
        "  <SIGNONMSGSRSV1>",
        "    <SONRS>",
        "      <STATUS>",
        "        <CODE>0",
        "        <SEVERITY>INFO",
        "      </STATUS>",
        "      <DTSERVER>{}</DTSERVER>".format(datetime.now().strftime('%Y%m%d%H%M%S')),
        "      <LANGUAGE>ENG",
        "    </SONRS>",
        "  </SIGNONMSGSRSV1>",
        "  <BANKMSGSRSV1>",
        "    <STMTTRNRS>",
        "      <TRNUID>1",
        "      <STATUS>",
        "        <CODE>0",
        "        <SEVERITY>INFO",
        "      </STATUS>",
        "      <STMTRS>",
        "        <CURDEF>USD",
        "        <BANKTRANLIST>",
        "          <DTSTART>{}</DTSTART>".format(min([t['date'] for t in transactions]).strftime('%Y%m%d')),
        "          <DTEND>{}</DTEND>".format(max([t['date'] for t in transactions]).strftime('%Y%m%d')),
    ]

    for t in transactions:
        ofx.extend([
            "          <STMTTRN>",
            "            <TRNTYPE>{}</TRNTYPE>".format("DEBIT" if t['amount'] < 0 else "CREDIT"),
            "            <DTPOSTED>{}</DTPOSTED>".format(t['date'].strftime('%Y%m%d%H%M%S')),
            "            <TRNAMT>{}</TRNAMT>".format(t['amount']),
            "            <FITID>{}</FITID>".format(t['id']),
            "            <NAME>{}</NAME>".format(t['name']),
            "            <MEMO>{}</MEMO>".format(t['type']),
            "          </STMTTRN>",
        ])

    ofx.extend([
        "        </BANKTRANLIST>",
        "      </STMTRS>",
        "    </STMTTRNRS>",
        "  </BANKMSGSRSV1>",
        "</OFX>",
    ])

    return '\n'.join(ofx)

def write_ofx_file(ofx_content):
    sys.stdout.write(ofx_content)

# Reading CSV input from standard in
input_data = sys.stdin.read()

# Parse the CSV
transactions = parse_csv(input_data)

# Create OFX content
ofx_content = create_ofx(transactions)

# Write OFX output to standard out
write_ofx_file(ofx_content)