(paypal): Add basic paypal converter

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 *

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())