Add cbastatement converter

src/beanify/converters/__init__.py

@@ -33,3 +33,4 @@ def available_converter(converter_cls):
 
 from .wise_converter import *
 from .cba_converter import *
+from .cbastatement_converter import *

src/beanify/converters/cbastatement_converter.py

@@ -0,0 +1,205 @@
+import csv
+from dataclasses import dataclass
+import datetime as dt
+from datetime import datetime
+import logging
+from typing import NamedTuple
+from enum import Enum
+
+from ..base import Converter, PartialTXN, PartialPosting, Record, Amount
+from ..base.converter import ConverterConfig
+from ..base.rules import Rule, RuleSet
+from ..base.transaction import TXNFlag
+
+from . import available_converter
+
+__all__ = [
+    "CBASTRecord",
+    "CBASTConfig",
+    "CBASTConverter",
+]
+
+logger = logging.getLogger(__name__)
+
+
+class CBASTRow(NamedTuple):
+    date: str
+    description: str
+    debit: str
+    credit: str
+    balance: str
+
+
+class RecordDirection(Enum):
+    OUT = "OUT"
+    IN = "IN"
+
+
+@dataclass(kw_only=True, frozen=True, eq=False)
+class CBASTRecord(Record):
+    description: str
+    direction: RecordDirection
+    balance: Amount
+
+    _display_fields = [
+        ("date", "Date"),
+        ("source_account", "Source Account"),
+        ("target_account", "Target Account"),
+        ("from_source", "From Source"),
+        ("to_target", "To Target"),
+        ("balance", "Account Balance"),
+        ("description", "Description"),
+    ]
+
+    _match_fields = [
+        "source_account",
+        "target_account",
+        "description",
+    ]
+
+    @classmethod
+    def sample_record(cls):
+        self = cls(
+            date=dt.date.today(),
+            source_account="John Doe",
+            target_account="Jane Austen",
+            from_source=Amount(314, "USD"),
+            to_target=Amount(314, "USD"),
+            balance=Amount(2, "USD"),
+            fees=(),
+            raw="Raw Data",
+            description="Raw Description",
+            direction=RecordDirection.OUT,
+        )
+        return self
+
+
+@dataclass
+class CBASTConfig(ConverterConfig):
+    asset_account: str
+    asset_currency: str
+
+    required = {
+        "asset-account",
+        "asset-currency",
+    }
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "CBASTConfig":
+        if (f := next((f for f in cls.required if f not in data), None)) is not None:
+            raise ValueError(
+                f"CBA ST Converter Configuration missing required field: {f}"
+            )
+        return cls(data["asset-account"], data["asset-currency"])
+
+
+@available_converter
+class CBASTConverter(Converter[CBASTRecord, CBASTConfig]):
+    record_type = CBASTRecord
+    config_type = CBASTConfig
+    converter_name = "cbast"
+    version = "0"
+    display_name = "CBAST converter v0"
+    config_field = "CBAST"
+
+    def convert(self, record: CBASTRecord, ruleset: RuleSet) -> PartialTXN:
+        fields = {}
+
+        match record.direction:
+            case RecordDirection.OUT:
+                fields["source_account"] = self.config.asset_account
+            case RecordDirection.IN:
+                fields["target_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]
+
+        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,
+        )
+        args.setdefault("comment", record.description)
+
+        txn = PartialTXN(**args)
+        logger.debug(f"Converted CBA Statement Record {record!r} to PartialTXN {txn!r}")
+        return txn
+
+    def _make_record(self, row: CBASTRow) -> CBASTRecord | None:
+        dt_format = "%Y-%m-%d"
+
+        # FIXME:: Ignoring currency conversions for now.
+
+        created_on = datetime.strptime(row.date, dt_format).date()
+        if row.credit and not row.debit:
+            # Into account
+            direction = RecordDirection.IN
+            source_account = row.description
+            target_account = "ACCOUNT"
+            rawvalue = row.credit
+        elif row.debit and not row.credit:
+            # Out of account
+            direction = RecordDirection.OUT
+            source_account = "ACCOUNT"
+            target_account = row.description
+            rawvalue = row.debit
+        elif not row.debit and not row.credit:
+            # Nothing changing
+            # Could make this a balance posting, but for now make a noop
+            return None
+        else:
+            # TODO: Unhandled case for now
+            raise ValueError(
+                "Converter does not support transactions with both credit and debit."
+            )
+
+        amount = Amount(float(rawvalue.strip('"')), self.config.asset_currency)
+        record = CBASTRecord(
+            date=created_on,
+            source_account=source_account,
+            target_account=target_account,
+            from_source=amount if direction is RecordDirection.OUT else -amount,
+            to_target=amount if direction is RecordDirection.IN else -amount,
+            raw=",".join(row),
+            description=row.description,
+            direction=direction,
+            balance=Amount(float(row.balance or "0"), self.config.asset_currency),
+        )
+        return record
+
+    def ingest_string(self, data: str) -> list[CBASTRecord]:
+        reader = csv.reader(data.splitlines())
+        # Discard header
+        next(reader)
+
+        records = []
+        for row in reader:
+            record = self._make_record(CBASTRow(*row))
+            if record:
+                records.append(record)
+        return records
+
+    def ingest_file(self, path) -> list[CBASTRecord]:
+        with open(path) as f:
+            return self.ingest_string(f.read())