From 42751393ad6794bbf82280df30c30ed48e1ae28c Mon Sep 17 00:00:00 2001 From: Interitio Date: Thu, 24 Sep 2026 11:15:20 +1000 Subject: [PATCH] (paypal): Add basic paypal converter --- src/beanify/converters/__init__.py | 1 + src/beanify/converters/paypal_converter.py | 398 +++++++++++++++++++++ 2 files changed, 399 insertions(+) create mode 100644 src/beanify/converters/paypal_converter.py diff --git a/src/beanify/converters/__init__.py b/src/beanify/converters/__init__.py index 5b27b3b..f77b878 100644 --- a/src/beanify/converters/__init__.py +++ b/src/beanify/converters/__init__.py @@ -34,3 +34,4 @@ def available_converter(converter_cls): from .wise_converter import * from .cba_converter import * from .cbastatement_converter import * +from .paypal_converter import * diff --git a/src/beanify/converters/paypal_converter.py b/src/beanify/converters/paypal_converter.py new file mode 100644 index 0000000..5c7c9a4 --- /dev/null +++ b/src/beanify/converters/paypal_converter.py @@ -0,0 +1,398 @@ +""" +Beanify Converter implementation for the hot mess which is paypal csv records. +""" + +from typing import NamedTuple, Optional +from dataclasses import dataclass, field +from enum import Enum +import datetime as dt +from datetime import date, datetime, time +import logging +import csv + +from ..base import Converter, PartialTXN, PartialPosting, Record, Amount +from ..base.converter import ConverterConfig +from ..base.rules import RuleSet +from ..base.transaction import TXNFlag, TXNPosting, Transaction +from ..base.partial import UserInputError + +from . import available_converter + +logger = logging.getLogger(__name__) + +__all__ = [ + "PaypalRecord", + "PaypalRecordDirection", + "PaypalCSVRow", + "PaypalPartialTXN", + "PaypalConverter", +] + + +class PaypalCSVRow(NamedTuple): + date: str + time: str + timezone: str + name: str + txntype: str + status: str + currency: str + amount: str + fees: str + total: str + ex_rate: str + receipt_id: str + balance: str + txnid: str + item_title: str + + +class PaypalRecordDirection(Enum): + IGNORE = -1 + TRANSFERIN = 0 + IN = 1 + CONVERT = 2 + OUT = 3 + TRANSFEROUT = 0 + + +def get_direction(typestr): + match typestr: + case "Reversal of General Account Hold": + return PaypalRecordDirection.IGNORE + case "General Currency Conversion": + return PaypalRecordDirection.CONVERT + case "Subscription Payment": + return PaypalRecordDirection.OUT + case "Express Checkout Payment": + return PaypalRecordDirection.OUT + case "Pre-approved Payment Bill User Payment": + return PaypalRecordDirection.OUT + case "Mass payment": + return PaypalRecordDirection.IN + case "Transfer to PayPal account": + return PaypalRecordDirection.TRANSFERIN + case "Account Hold for Open Authorisation": + return PaypalRecordDirection.IGNORE + case "User Initiated Withdrawal": + # TODO: Probably don't actually want to ignore this + # return PaypalRecordDirection.TRANSFEROUT + return PaypalRecordDirection.IGNORE + case "Reversal of ACH Withdrawal Transaction": + return PaypalRecordDirection.IGNORE + case "BillPay transaction": + return PaypalRecordDirection.OUT + case "Payment Refund": + return PaypalRecordDirection.IN + case "Reversal of ACH Deposit": + return PaypalRecordDirection.IGNORE + case "Mobile Payment": + return PaypalRecordDirection.IN + case _: + raise ValueError(f"Unknown Paypal typestr {typestr}") + + +@dataclass(kw_only=True, frozen=True, eq=False) +class PaypalRecord(Record): + # With time zone, for a change + created_on: datetime + + txnid: str + name: str + txntype: str + direction: PaypalRecordDirection + balance_effect: Amount + + ex_target_amount: Amount | None = None + + # Each source amount is (Amount, exchange rate) + ex_source_amounts: list[tuple[Amount, float]] = field(default_factory=list) + + transferred_in: Amount | None = None + + _display_fields = [ + ("txnid", "ID"), + ("txntype", "Type"), + ("direction", "Direction"), + ("created_on", "Created"), + ("name", "Name"), + ("source_account", "Source"), + ("target_account", "Target"), + ("balance_effect", "Balance Change"), + ("transferred_in", "Transferred In"), + ] + _match_fields = [ + "txnid", + "txntype", + "direction", + "balance_effect", + "name", + ] + + @classmethod + def sample_record(cls): + self = cls( + date=dt.date.today(), + source_account="John Doe", + target_account="Jane Austen", + from_source=Amount(314, "CROC"), + to_target=Amount(314, "CROC"), + created_on=datetime.now(), + txnid="0", + name="Sample", + txntype="Payment Refund", + direction=PaypalRecordDirection.IN, + balance_effect=Amount(314, "CROC"), + transferred_in=Amount(314, "CROC"), + ex_target_amount=None, + ex_source_amounts=[], + ) + return self + + +@dataclass +class PaypalConfig(ConverterConfig): + asset_account: str + + required = { + "asset-account", + } + + @classmethod + def from_dict(cls, data: dict) -> "PaypalConfig": + if (f := next((f for f in cls.required if f not in data), None)) is not None: + raise ValueError(f"Paypal Configuration missing required field: {f}") + return cls(data["asset-account"]) + + +@dataclass(kw_only=True) +class PaypalPartialTXN(PartialTXN): + # Hard-coded postings that will be added to the final transaction + # Very much a hack, will fix later + extra_postings: list[TXNPosting] | None = None + + def upgrade(self, defaults=...) -> Transaction: + txn = super().upgrade(defaults) + if self.extra_postings: + txn.postings.extend(self.extra_postings) + return txn + + +@available_converter +class PaypalConverter(Converter[PaypalRecord, PaypalPartialTXN, PaypalConfig]): + record_type = PaypalRecord + config_type = PaypalConfig + txn_type = PaypalPartialTXN + converter_name = "paypal" + version = "0" + display_name = "Paypal CSV converter v0" + config_field = "PAYPAL" + + def __init__(self, config: PaypalConfig, *kwargs): + self.config = config + + def convert(self, record: PaypalRecord, ruleset: RuleSet) -> PaypalPartialTXN: + fields = {} + + match record.direction: + case PaypalRecordDirection.TRANSFERIN: + fields["target_account"] = self.config.asset_account + case PaypalRecordDirection.IN: + fields["target_account"] = self.config.asset_account + case PaypalRecordDirection.OUT: + fields["source_account"] = self.config.asset_account + case PaypalRecordDirection.TRANSFEROUT: + fields["source_account"] = self.config.asset_account + + fields |= ruleset.apply(record.match_fields()) + args = {} + args["date"] = record.date + + for name in { + "payee", + "narration", + "comment", + "document", + "tags", + "links", + "flag", + }: + if name in fields: + args[name] = fields[name] + + # source posting + # target posting + + args["source_posting"] = PartialPosting( + account=fields.get("source_account", None), + amount=record.from_source, + total_cost=abs(record.to_target) + if abs(record.from_source) != abs(record.to_target) + else None, + ) + + args["target_posting"] = PartialPosting( + account=fields.get("target_account", None), + amount=record.to_target, + ) + + # comment + args.setdefault( + "comment", + f"{record.txntype} | {record.name} | {record.txnid} | Transferred {record.transferred_in}", + ) + + currency_postings = [] + if record.ex_target_amount is not None: + for source_amount, exrate in record.ex_source_amounts: + source_posting = TXNPosting( + account=self.config.asset_account, + amount=source_amount, + price=Amount(exrate, record.ex_target_amount.currency), + ) + currency_postings.append(source_posting) + target_posting = TXNPosting( + account=self.config.asset_account, amount=record.ex_target_amount + ) + currency_postings.append(target_posting) + + txn = PaypalPartialTXN(extra_postings=currency_postings, **args) + logger.debug(f"Converted Paypal CSV Record {record!r} to PaypalTXN {txn!r}") + return txn + + def _make_record(self, *rows: PaypalCSVRow) -> PaypalRecord | None: + # Find main, which is an in or out + main_txns = [] + convert_rows = [] + transfer_rows = [] + for row in rows: + dir = get_direction(row.txntype) + if dir is PaypalRecordDirection.IGNORE: + continue + elif dir in (PaypalRecordDirection.IN, PaypalRecordDirection.OUT): + main_txns.append(row) + elif dir is PaypalRecordDirection.TRANSFERIN: + transfer_rows.append(row) + elif dir is PaypalRecordDirection.CONVERT: + convert_rows.append(row) + + # Some sanity checks + if not main_txns: + # No main transaction + if not convert_rows: + return None + raise ValueError( + f"Paypal Row block with convert by no main transactions: {rows}" + ) + if len(main_txns) > 1: + raise ValueError(f"Paypal Row block with multiple main txns: {rows}") + if len(convert_rows) == 1: + raise ValueError(f"Paypal Row block with orphan convert: {rows}") + if convert_rows and not any(not row.ex_rate for row in rows): + raise ValueError(f"Paypal Row block without main convert: {rows}") + if any(float(row.fees) for row in rows): + raise ValueError(f"Paypal Row block with fees: {rows}") + + # We guarantee that we are somewhat sane + mainrow = main_txns[0] + direction = get_direction(mainrow.txntype) + dt_format = "%d/%m/%Y %H:%M:%S" + created_on = datetime.strptime(f"{mainrow.date} {mainrow.time}", dt_format) + balance_effect = Amount(float(mainrow.amount), mainrow.currency) + + # transfer rows + if transfer_rows: + assert len({row.currency for row in transfer_rows}) == 1 + transferred_in = Amount( + sum(float(row.amount) for row in transfer_rows), + transfer_rows[0].currency, + ) + else: + transferred_in = None + + # Currency conversion + source_amounts = [] + target_amount = None + for row in convert_rows: + if row.ex_rate: + # If the row has an exchange rate, it's a source row + # Note the amount will be negative + source_details = (Amount(row.amount, row.currency), float(row.ex_rate)) + source_amounts.append(source_details) + else: + # Otherwise, main row + target_amount = Amount(row.amount, row.currency) + + if direction is PaypalRecordDirection.IN: + record = PaypalRecord( + date=created_on.date(), + source_account=mainrow.name, + target_account="ACCOUNT", + from_source=-balance_effect, + to_target=balance_effect, + created_on=created_on, + txnid=mainrow.txnid, + name=mainrow.name, + txntype=mainrow.txntype, + direction=direction, + balance_effect=balance_effect, + transferred_in=transferred_in, + ex_target_amount=target_amount, + ex_source_amounts=source_amounts, + ) + else: + record = PaypalRecord( + date=created_on.date(), + target_account=mainrow.name, + source_account="ACCOUNT", + from_source=balance_effect, + to_target=-balance_effect, + created_on=created_on, + txnid=mainrow.txnid, + name=mainrow.name, + txntype=mainrow.txntype, + direction=direction, + balance_effect=balance_effect, + transferred_in=transferred_in, + ex_target_amount=target_amount, + ex_source_amounts=source_amounts, + ) + return record + + def ingest_string(self, data: str) -> list[PaypalRecord]: + """ + Parse a string of Paypal CSV input into a list of records + """ + reader = csv.reader(data.splitlines()) + + records = [] + thisblock = [] + thisblock_key = None + for i, row in enumerate(reader): + if not i: + # Skip header row + continue + row = PaypalCSVRow(*row) + key = (row.date, row.time) + if key != thisblock_key: + if thisblock: + # Flush + record = self._make_record(*thisblock) + if record: + records.append(record) + thisblock.clear() + thisblock_key = key + thisblock.append(row) + + if thisblock: + # Flush + record = self._make_record(*thisblock) + if record: + records.append(record) + thisblock.clear() + return records + + def ingest_file(self, path) -> list[PaypalRecord]: + with open(path) as f: + f.readline() + return self.ingest_string(f.read())