PayPal Activity CSV to GnuCash-ready splits (GBP, USD, FX reference)

#!/usr/bin/env python3
r"""
Split a PayPal "Activity Download" CSV into:
  1) GBP purchases/refunds (excluding funding and FX rows)
  2) USD purchases/refunds (excluding FX rows)
  3) FX reference table (GBP->USD conversions for USD purchases)

Works with the PayPal CSV format that has columns like:
Date, Time, Time Zone, Description, Currency, Gross, Fee, Net, Balance, Transaction ID, Reference Txn ID, Name, ...

Usage (Windows):
  py -3 paypal_split_by_currency.py "C:\path\PayPal.CSV"

Outputs files next to the input CSV by default.
"""
import argparse
import csv
from collections import Counter
from pathlib import Path

EXCLUDE_GBP_DESCRIPTIONS = {
    "Bank Deposit to PP Account",
    "User Initiated Withdrawal",
    "General Currency Conversion",
}

EXCLUDE_USD_DESCRIPTIONS = {
    "General Currency Conversion",
}

USD_PAYMENT_DESCRIPTIONS = {
    "Pre-approved Payment Bill User Payment",
    "Express Checkout Payment",
    "General Payment",
}

def parse_float(s: str) -> float:
    s = (s or "").strip()
    if s == "":
        return 0.0
    return float(s)

def read_rows(csv_path: Path):
    with csv_path.open("r", encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
        if not rows:
            raise SystemExit("CSV appears to be empty.")
        return reader.fieldnames, rows

def write_rows(out_path: Path, fieldnames, rows):
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        w.writeheader()
        w.writerows(rows)

def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("csv", type=Path, help="PayPal activity CSV file")
    ap.add_argument("--outdir", type=Path, default=None, help="Output directory (default: same folder as input)")
    ap.add_argument("--prefix", type=str, default=None, help="Filename prefix (default: input filename stem)")
    args = ap.parse_args()

    csv_path: Path = args.csv
    outdir: Path = args.outdir if args.outdir else csv_path.parent
    prefix: str = args.prefix if args.prefix else csv_path.stem

    fieldnames, rows = read_rows(csv_path)

    # Basic sanity check for columns we rely on
    required = {"Currency", "Description", "Net", "Transaction ID", "Reference Txn ID", "Date", "Time"}
    missing_cols = required - set(fieldnames or [])
    if missing_cols:
        raise SystemExit(f"Missing expected columns: {', '.join(sorted(missing_cols))}")

    # Split for import
    gbp_rows = [
        r for r in rows
        if (r.get("Currency") == "GBP") and (r.get("Description") not in EXCLUDE_GBP_DESCRIPTIONS)
    ]
    usd_rows = [
        r for r in rows
        if (r.get("Currency") == "USD") and (r.get("Description") not in EXCLUDE_USD_DESCRIPTIONS)
    ]

    # Build quick lookup for FX components keyed by purchase transaction id
    # For a USD purchase with Transaction ID = PID, PayPal records:
    #   GBP "Bank Deposit to PP Account"      Reference Txn ID == PID
    #   GBP "General Currency Conversion"     Reference Txn ID == PID (Net negative)
    #   USD "General Currency Conversion"     Reference Txn ID == PID (Net positive)
    fx_lookup = {}
    for r in rows:
        ref = (r.get("Reference Txn ID") or "").strip()
        if not ref:
            continue
        fx_lookup.setdefault(ref, []).append(r)

    fx_ref_rows = []
    fx_missing = 0
    for r in rows:
        if r.get("Currency") != "USD":
            continue
        if r.get("Description") not in USD_PAYMENT_DESCRIPTIONS:
            continue

        pid = (r.get("Transaction ID") or "").strip()
        if not pid:
            continue

        bundle = fx_lookup.get(pid, [])
        gbp_deposit = next((x for x in bundle if x.get("Currency") == "GBP" and x.get("Description") == "Bank Deposit to PP Account"), None)
        gbp_conv    = next((x for x in bundle if x.get("Currency") == "GBP" and x.get("Description") == "General Currency Conversion"), None)
        usd_conv    = next((x for x in bundle if x.get("Currency") == "USD" and x.get("Description") == "General Currency Conversion"), None)

        if not (gbp_deposit and gbp_conv and usd_conv):
            fx_missing += 1
            continue

        gbp_topup = parse_float(gbp_deposit.get("Net"))
        gbp_converted = abs(parse_float(gbp_conv.get("Net")))
        usd_received = parse_float(usd_conv.get("Net"))
        usd_spent = abs(parse_float(r.get("Net")))
        rate = (usd_received / gbp_converted) if gbp_converted else 0.0

        merchant = (r.get("Name") or "").strip() or "(blank name)"
        fx_ref_rows.append({
            "Date": r.get("Date"),
            "Time": r.get("Time"),
            "Merchant": merchant,
            "GBP topup (bank->PayPal GBP)": f"{gbp_topup:.2f}",
            "GBP converted (PayPal GBP->USD)": f"{gbp_converted:.2f}",
            "USD received (PayPal USD)": f"{usd_received:.2f}",
            "USD spent at merchant": f"{usd_spent:.2f}",
            "Implied rate (USD per GBP)": f"{rate:.6f}",
            "Purchase TxID": pid,
        })

    # Output paths
    gbp_out = outdir / f"{prefix}_GBP_purchases.csv"
    usd_out = outdir / f"{prefix}_USD_purchases.csv"
    fx_out  = outdir / f"{prefix}_FX_reference.csv"

    write_rows(gbp_out, fieldnames, gbp_rows)
    write_rows(usd_out, fieldnames, usd_rows)

    fx_fields = ["Date","Time","Merchant","GBP topup (bank->PayPal GBP)","GBP converted (PayPal GBP->USD)","USD received (PayPal USD)","USD spent at merchant","Implied rate (USD per GBP)","Purchase TxID"]
    write_rows(fx_out, fx_fields, fx_ref_rows)

    # Summary
    ccy_counts = Counter(r.get("Currency") for r in rows)
    desc_counts = Counter(r.get("Description") for r in rows)
    print("Input:", csv_path)
    print("Rows:", len(rows), "| Currencies:", dict(ccy_counts))
    print("Wrote:")
    print("  ", gbp_out, f"({len(gbp_rows)} rows)")
    print("  ", usd_out, f"({len(usd_rows)} rows)")
    print("  ", fx_out,  f"({len(fx_ref_rows)} FX rows; {fx_missing} missing bundles skipped)")
    print("Top descriptions:", ", ".join([f"{k}={v}" for k,v in desc_counts.most_common(6)]))

if __name__ == "__main__":
    main()