diff --git a/src/beanify/converters/__init__.py b/src/beanify/converters/__init__.py index f61b013..5b27b3b 100644 --- a/src/beanify/converters/__init__.py +++ b/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 * diff --git a/src/beanify/converters/cbastatement_converter.py b/src/beanify/converters/cbastatement_converter.py new file mode 100644 index 0000000..f51ae38 --- /dev/null +++ b/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())