Add project config for beanify

pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "beanify"
+description = "Interactive converter for beancount accounting with modular converters"
+authors = [{name="Interitio", email="cona@thewisewolf.dev"}]
+readme = "README.md"
+version = "0.1.0a1"
+dependencies = [
+  "pytz",
+  "platformdirs"
+]
+requires-python = ">= 3.10"
+
+[project.optional-dependencies]
+debug = []
+
+[project.scripts]
+beanify = "beanify:run"
+

src/beanify/gui-main.py

@@ -0,0 +1,99 @@
+import asyncio
+import argparse
+import logging
+
+from dataclasses import dataclass
+from tkinter import ttk
+import tomllib
+
+from base.rules import DummyRuleInterface, JSONRuleInterface, RuleSet
+from gui import MainWindow
+from converters import converters_available, converter_factory
+
+logger = logging.getLogger()
+logger.addHandler(logging.StreamHandler())
+logger.setLevel(logging.DEBUG)
+
+
+"""
+Arguments and config needed?
+
+Arguments:
+- What converter to use 
+- The config file path
+- Which rule interface to use (TODO)
+- The rule interface path (or DB arg string)
+- The file(s) to ingest from (multiple uses of --ingest)
+    - window.ingest_files(...) after setup
+- The file(s) to output to (TODO, and maybe not needed)
+
+Config File:
+- converter
+- rule interface 
+- rule interface path 
+
+Probably default accounts etc as well
+Converter config inherits and can override these defaults
+"""
+
+
+parser = argparse.ArgumentParser()
+
+available = ', '.join((f"'{converter.qual_name()}'" for converter in converters_available))
+parser.add_argument(
+    '--converter',
+    dest='converter',
+    help=f"Which converter to use. Available converters: {available}"
+)
+parser.add_argument(
+    '--config',
+    dest='config',
+    help="Ingest configuration file. Should be a toml file with a section for the selected converter."
+)
+parser.add_argument(
+    '--ingest',
+    dest='ingest',
+    help="File to ingest (must be in the correct format for the selected converter). May be provided multiple times to ingest multiple files.",
+    action='append'
+)
+parser.add_argument(
+    '--rules',
+    dest='rules',
+    default=None,
+    help="Path to JSON file with parsing rules. If not given, will use in-memory ruleset, with no saving."
+)
+
+
+async def main():
+    args = parser.parse_args()
+    config_path = args.config
+    converter_name = args.converter
+    ingest_files = args.ingest 
+    rule_path = args.rules
+
+    with open(config_path, "rb") as f:
+        root_config = tomllib.load(f)
+
+    converter_cls = converter_factory(qual_name=converter_name) if converter_name else None
+    if converter_cls is None:
+        raise ValueError("Dynamic converter selection not supported yet.")
+
+    converter_config = converter_cls.config_type.from_dict(root_config[converter_cls.config_field])
+    converter = converter_cls(converter_config)
+
+    if rule_path:
+        rule_interface = JSONRuleInterface(converter_cls.qual_name(), path=rule_path)
+    else:
+        rule_interface = DummyRuleInterface(converter_cls.qual_name())
+    ruleset = RuleSet.load_from(rule_interface)
+
+    logger.debug("Finished configuration loading, creating GUI.")
+
+    window = MainWindow(root_config, converter, ruleset, initial_files=ingest_files, theme='clam')
+    # Note: Arc theme is more modern, might be nicer. But it's also bright. equilux is dark
+    # clam is actually useful
+    window.mainloop()
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

src/beanify/main-simple.py

@@ -0,0 +1,376 @@
+"""
+Tired of over-engineering, I just want a script which will actually convert my finances sometime this year.
+"""
+import csv
+import sys
+import asyncio
+import argparse
+from dataclasses import dataclass
+from enum import Enum
+import datetime as dt
+from datetime import datetime
+import logging
+
+from base import Amount, TXNFlag, Transaction, Record, TXNPosting
+
+logger = logging.getLogger(__name__)
+
+"""
+TODO Note:
+- Narration and payee fields should be part of the rule?
+- Ability to add comments in non-batch mode, or change fields
+"""
+
+
+parser = argparse.ArgumentParser()
+parser.add_argument(
+    'input',
+    help="CSV file with WISE transactions to read (note this needs to be in transaction format not statement format)"
+)
+parser.add_argument(
+    '-o', '--output',
+    dest='output',
+    default=None,
+    help="Path to output file to append processed transactions. If not given will use stdout."
+)
+parser.add_argument(
+    '--expenserules',
+    dest='expense_rules',
+    default=None,
+    help="Path to CSV file with expense rules (Wise account name in first column, Bean expense account in second)."
+)
+parser.add_argument(
+    '--incomerules',
+    dest='income_rules',
+    default=None,
+    help="Path to CSV file with income rules (Wise account name in first column, Bean income account in second)."
+)
+
+
+def read_rulemap(path: str | None):
+    """
+    Read the first two columns of a csv file into a dictionary.
+
+    Intended for reading expense and income rules into a map.
+    """
+    rules: dict[str, str] = {}  # data name -> account name
+    if path:
+        with open(path) as f:
+            f.readline()
+            reader = csv.reader(f)
+            for row in reader:
+                rules[row[0]] = row[1]
+
+    return rules
+
+def add_rule(path: str, rule: tuple[str, str]):
+    """
+    Add the provided rule to the csv file at the given path
+    """
+    if path:
+        with open(path, 'a') as f:
+            writer = csv.writer(f)
+            writer.writerow(rule)
+
+class WiseRecordStatus(Enum):
+    COMPLETED = 'COMPLETED'
+    CANCELLED = 'CANCELLED'
+
+class WiseRecordDirection(Enum):
+    OUT = 'OUT'
+    NEUTRAL = 'NEUTRAL'
+    IN = 'IN'
+
+@dataclass(kw_only=True)
+class WiseRecord(Record):
+    id: str
+    status: WiseRecordStatus
+    direction: WiseRecordDirection
+    created_on: datetime
+    finished_on: datetime | None
+    source_fee: Amount | None
+    target_fee: Amount | None
+    exchange_rate: float
+
+    @property
+    def source_currency(self):
+        return self.from_source.currency
+
+    @property
+    def target_currency(self):
+        return self.to_target.currency
+
+    @property
+    def from_source_net(self):
+        """
+        The amount that was taken from the source account not including fees.
+        Use `from_source` to get the amount including fees.
+        """
+        amount = self.from_source
+        if self.source_fee is not None:
+            amount = amount - self.source_fee
+        return amount
+
+    @property
+    def exchanged_amount(self):
+        """
+        The amount that was sent to the target, in the target currency.
+        Includes any target fees.
+        May be used to obtain the amount exchanged to the target currency.
+        """
+        amount = self.to_target
+        if self.target_fee is not None:
+            amount = amount + self.target_fee
+        return amount
+
+    @classmethod
+    def from_row(cls, row):
+            id = row[0]
+            status = row[1]
+            direction = row[2]
+            created_on = row[3]
+            finished_on = row[3]
+            source_fee_amount = float(row[5] or 0)
+            source_fee_currency = row[6]
+            target_fee_amount = float(row[7] or 0)
+            target_fee_currency = row[8]
+            source_name = row[9]
+            source_amount_final = float(row[10] or 0)
+            source_currency = row[11]
+            target_name = row[12]
+            target_amount_final = float(row[13] or 0)
+            target_currency = row[14]
+            exchange_rate = float(row[15] or 0)
+            
+            wise_dt_format = '%Y-%m-%d %H:%M:%S'
+
+            created_on_dt = datetime.strptime(created_on, wise_dt_format)
+            finished_on_dt = datetime.strptime(finished_on, wise_dt_format) if finished_on else None
+            
+            fees = []
+
+            if source_fee_amount:
+                source_fee = Amount(source_fee_amount, source_fee_currency)
+                fees.append(source_fee)
+            else:
+                source_fee = None
+            if target_fee_amount:
+                target_fee = Amount(target_fee_amount, target_fee_currency)
+                fees.append(target_fee)
+            else:
+                target_fee = None
+
+            raw = ','.join(row)
+
+            self = cls(
+                date=created_on_dt.date(),
+                source_account=source_name,
+                target_account=target_name,
+                from_source=Amount(source_amount_final, source_currency),
+                to_target=Amount(target_amount_final, target_currency),
+                fees=fees,
+                raw=raw,
+                id=id,
+                status=WiseRecordStatus(status),
+                direction=WiseRecordDirection(direction),
+                created_on=created_on_dt,
+                finished_on=finished_on_dt,
+                source_fee=source_fee,
+                target_fee=target_fee,
+                exchange_rate=float(exchange_rate),
+            )
+            logger.debug(
+                f"Converted Wise row {raw} to record {self!r}"
+            )
+            return self
+
+
+async def main_wise(record_path, expense_rule_path, income_rule_path, out_path: str | None = None):
+    expense_rules = read_rulemap(expense_rule_path)
+    income_rules = read_rulemap(income_rule_path)
+
+    transactions = []
+    skipped = []
+
+    with open(record_path) as f:
+        f.readline()
+        reader = csv.reader(f)
+        for row in reader:
+            # Convert csv row into a Record
+            record = WiseRecord.from_row(row)
+
+            if record.status is not WiseRecordStatus.COMPLETED:
+                logging.info(f"Skipping record with non-complete status:\n{record}")
+                skipped.append(record)
+                continue
+
+            match record.direction:
+                case WiseRecordDirection.OUT:
+                    # Expense transaction
+                    ruleset = expense_rules
+                    source_account = wise_asset_account(record.source_currency)
+                    if (target_account := ruleset.get(record.target_account)) is None:
+                        # TODO: Factor this out properly
+                        # TODO: Manually associated data can be saved against the row ID
+                        userstr = input(
+                                f"{record!r}\nExpense rule missing for target account '{record.target_account}'.\n"
+                                "(a to Add Rule, s to Skip, or enter an expense account to use for this transaction): "
+                        )
+                        match userstr:
+                            case 's':
+                                print("Skipping record")
+                                skipped.append(record)
+                                continue
+                            case 'a':
+                                userstr = input(f"Enter expense account to associate to target account '{record.target_account}': ")
+                                target_account = userstr.strip()
+                                expense_rules[record.target_account] = target_account
+                                if expense_rule_path:
+                                    add_rule(expense_rule_path, (record.target_account, target_account))
+                                print(f"Associated expense account '{target_account}'")
+                            case _:
+                                target_account = userstr.strip()
+
+                    # Build transaction
+                    transaction = Transaction(
+                        date=record.date,
+                        flag=TXNFlag.INCOMPLETE,
+                        narration="TODO",
+                        payee=record.target_account,
+                        comments=[record.raw],
+                        postings=[]
+                    )
+                case WiseRecordDirection.NEUTRAL:
+                    # Neutral transaction from the asset account to itself
+                    source_account = wise_asset_account(record.source_currency)
+                    target_account = wise_asset_account(record.target_currency)
+                    transaction = Transaction(
+                        date=record.date,
+                        flag=TXNFlag.COMPLETE,
+                        narration="Currency Conversion",
+                        comments=[record.raw],
+                        postings=[]
+                    )
+                case WiseRecordDirection.IN:
+                    # Income transaction
+                    ruleset = income_rules
+                    target_account = wise_asset_account(record.target_currency)
+                    if (source_account := ruleset.get(record.source_account)) is None:
+                        # TODO: Factor this out properly
+                        userstr = input(
+                                f"{record!r}\nIncome rule missing for source account '{record.source_account}'.\n"
+                                "(a to Add Rule, s to Skip, or enter an income account to use for this transaction): "
+                        )
+                        match userstr:
+                            case 's':
+                                print("Skipping record")
+                                skipped.append(record)
+                                continue
+                            case 'a':
+                                userstr = input(f"Enter income account to associate to source account '{record.source_account}': ")
+                                source_account = userstr.strip()
+                                income_rules[record.source_account] = source_account
+                                if income_rule_path:
+                                    add_rule(income_rule_path, (record.source_account, source_account))
+                                print(f"Associated income account '{source_account}'")
+                            case _:
+                                source_account = userstr.strip()
+
+                    # Build transaction
+                    transaction = Transaction(
+                        date=record.date,
+                        flag=TXNFlag.INCOMPLETE,
+                        narration="TODO",
+                        payee=record.source_account,
+                        comments=[record.raw],
+                        postings=[]
+                    )
+
+            # Build postings from record
+
+            postings = []
+            # Source exchange amount, with cost by target total (if exchanging)
+            postings.append(TXNPosting(
+                account=source_account,
+                amount=-record.from_source_net,
+                total_cost=record.exchanged_amount if record.source_currency != record.target_currency else None,
+            ))
+
+            # Source fees: 
+            if record.source_fee:
+                # Source fee asset posting
+                postings.append(TXNPosting(
+                    account=source_account,
+                    amount=-record.source_fee
+                ))
+                # Source fee expense posting
+                postings.append(TXNPosting(
+                    account=wise_fee_account(),
+                    amount=record.source_fee
+                ))
+
+            # Target fees:
+            if record.target_fee:
+                # Note asset posting included in exchange amount 
+                # Target fee expense posting
+                postings.append(TXNPosting(
+                    account=wise_fee_account(),
+                    amount=record.target_fee
+                ))
+
+            # Target amount not including fees
+            postings.append(TXNPosting(
+                account=target_account,
+                amount=record.to_target
+            ))
+
+            transaction.postings = postings
+            if not transaction.check():
+                transaction.comments.append("WARNING: This transaction does not balance!")
+
+            # Finished building the transaction, add it to the list
+            logger.debug(
+                f"Converted Wise record {record!r} to Bean transaction {transaction!r}"
+            )
+            transactions.append(transaction)
+
+    # Print transactions or output to output file given
+    if out_path:
+        with open(out_path, 'a') as outf:
+            for transaction in transactions:
+                outf.write(str(transaction))
+                outf.write('\n\n')
+        # TODO: Add header and footer
+        print(f"Appended {len(transactions)} to {out_path}")
+    else:
+        for transaction in transactions:
+            print(str(transaction))
+            print('\n')
+
+    # Print skipped
+    # Option for skipped entries to be included as comments?
+    if skipped:
+        print(f"The following {len(skipped)} record(s) couldn't be processed:")
+        for record in skipped:
+            print(record.raw)
+
+# TODO: Eventually, move these to the converter configuration
+
+def wise_asset_account(currency):
+    return f"Assets:Adam:Wise:{currency}"
+
+def wise_fee_account():
+    return "Expenses:Financial:Fees:Wise"
+
+
+async def main():
+    args = parser.parse_args()
+    record_path = args.input
+    expense_rule_path = args.expense_rules
+    income_rule_path = args.income_rules
+    out_path = args.output
+    await main_wise(record_path, expense_rule_path, income_rule_path, out_path)
+
+
+if __name__ == '__main__':
+    asyncio.run(main())