holo/beanify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add package and config

Browse Source
This commit is contained in:
Interitio
2025-12-03 21:25:00 +10:00
parent ff4da24123
commit 052f473855
12 changed files with 405 additions and 669 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pyproject.toml
View File

@@ -10,7 +10,9 @@ readme = "README.md"
version = "0.1.0a1"
dependencies = [
"pytz",
"platformdirs"
"platformdirs",
"toml",
"ttkthemes"
]
requires-python = ">= 3.10"

54
src/beanify/base/rules.py
View File

@@ -1,8 +1,9 @@
from typing import Self
import os
class Rule:
__slots__ = ('conditions', 'values')
__slots__ = ("conditions", "values")
def __init__(self, conditions: dict[str, str], values: dict[str, str]):
self.conditions = conditions
@@ -12,7 +13,9 @@ class Rule:
"""
Check whether this rule applies to the given record fields.
"""
return all(record.get(key, None) == value for key, value in self.conditions.items())
return all(
record.get(key, None) == value for key, value in self.conditions.items()
)
class RuleInterface:
@@ -39,14 +42,17 @@ class JSONRuleInterface(RuleInterface):
Schema:
{
'rules': [
{
'record_fields': {},
'transaction_fields': {},
}
]
'convertername': {
'rules': [
{
'record_fields': {},
'transaction_fields': {},
}
]
}
}
"""
def __init__(self, converter: str, path: str, **kwargs):
self.path = path
super().__init__(converter, **kwargs)
@@ -58,12 +64,13 @@ class JSONRuleInterface(RuleInterface):
if not os.path.exists(self.path):
self.save_rules([])
with open(self.path, 'r') as f:
with open(self.path, "r") as f:
data = json.load(f)
for rule_data in data.get('rules', []):
for rule_data in data.get(self.converter, {}).get("rules", []):
rule = Rule(
conditions=rule_data['record_fields'],
values=rule_data['transaction_fields'],
conditions=rule_data["record_fields"],
values=rule_data["transaction_fields"],
)
rules.append(rule)
@@ -74,12 +81,22 @@ class JSONRuleInterface(RuleInterface):
rule_data = []
for rule in rules:
rule_data.append({
'record_fields': rule.conditions,
'transaction_fields': rule.values,
})
data = json.dumps({'rules': rule_data}, indent=2)
with open(self.path, 'w') as f:
rule_data.append(
{
"record_fields": rule.conditions,
"transaction_fields": rule.values,
}
)
existing = {}
if os.path.exists(self.path):
with open(self.path, "r") as f:
existing = json.load(f)
existing.setdefault(self.converter, {})
existing[self.converter]["rules"] = rule_data
data = json.dumps(existing, indent=2)
with open(self.path, "w") as f:
f.write(data)
@@ -88,6 +105,7 @@ class DummyRuleInterface(RuleInterface):
Dummy plug for the rule interface.
Can be used for testing or if the rules are otherwise loaded internally.
"""
def load_rules(self):
return []

5
src/beanify/base/transaction.py
View File

@@ -42,7 +42,10 @@ class Amount:
return NotImplemented
def __eq__(self, other):
return (self.value == other.value) and (self.currency == other.currency)
if isinstance(other, Amount):
return (self.value == other.value) and (self.currency == other.currency)
else:
return NotImplemented
def __rsub__(self, other):
return self - other

2
src/beanify/converters/__init__.py
View File

@@ -1,6 +1,6 @@
from typing import TYPE_CHECKING, Type
from base.converter import Converter
from ..base.converter import Converter
converters_available: list[Type[Converter]] = []

14
src/beanify/converters/cba_converter.py
View File

@@ -6,10 +6,10 @@ 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 ..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
@@ -75,8 +75,8 @@ class CBAConfig(ConverterConfig):
asset_currency: str
required = {
"asset_account",
"asset_currency",
"asset-account",
"asset-currency",
}
@classmethod
@@ -85,7 +85,7 @@ class CBAConfig(ConverterConfig):
raise ValueError(
f"CBA CSV Converter Configuration missing required field: {f}"
)
return cls(**data)
return cls(data["asset-account"], data["asset-currency"])
@available_converter

14
src/beanify/converters/wise_converter.py
View File

@@ -5,10 +5,10 @@ import datetime as dt
from datetime import datetime
import logging
from base import Converter, PartialTXN, PartialPosting, Record, Amount
from base.converter import ConverterConfig
from base.rules import RuleSet
from base.transaction import TXNFlag
from ..base import Converter, PartialTXN, PartialPosting, Record, Amount
from ..base.converter import ConverterConfig
from ..base.rules import RuleSet
from ..base.transaction import TXNFlag
from . import available_converter
@@ -201,15 +201,15 @@ class WiseConfig(ConverterConfig):
fee_account: str
required = {
"asset_account",
"fee_account",
"asset-account",
"fee-account",
}
@classmethod
def from_dict(cls, data: dict) -> "WiseConfig":
if (f := next((f for f in cls.required if f not in data), None)) is not None:
raise ValueError(f"Wise Configuration missing required field: {f}")
return cls(**data)
return cls(data["asset-account"], data["fee-account"])
@available_converter

99
src/beanify/gui-main.py
View File

@@ -1,99 +0,0 @@
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())

8
src/beanify/gui/mainwindow.py
View File

@@ -3,9 +3,9 @@ from tkinter import StringVar, filedialog, messagebox, ttk
from ttkthemes import ThemedTk
from base.converter import Converter
from base.partial import PartialTXN
from base.rules import Rule, RuleSet
from ..base.converter import Converter
from ..base.partial import PartialTXN
from ..base.rules import Rule, RuleSet
from . import logger
from .roweditor import RowEditor
@@ -70,7 +70,7 @@ class MainWindow(ThemedTk):
menu_file.add_command(label="Export Transactions", command=self.do_export_txn)
menu_file.add_separator()
menu_file.add_command(label="Save Rules", command=self.do_save_rules)
menu_file.add_command(label="Reload Rules", command=self.do_save_rules)
# menu_file.add_command(label="Reload Rules", command=self.do_save_rules)
menu_file.add_separator()
menu_file.add_command(label="Exit", command=lambda: self.destroy())
self.menubar.add_cascade(menu=menu_file, label="File")

110
src/beanify/gui/roweditor.py
View File

@@ -1,8 +1,8 @@
from tkinter import BooleanVar, StringVar, ttk
from base.partial import PartialTXN
from base.record import Record
from ..base.partial import PartialTXN
from ..base.record import Record
from . import logger
@@ -13,7 +13,7 @@ class RowEditor(ttk.Frame):
# Data state
self.record: Record | None = None
self.txn: PartialTXN | None = None
self.txn: PartialTXN | None = None
self.acmpl_cache: dict[str, list[str]] = acmpl_cache
# UI State
@@ -29,7 +29,6 @@ class RowEditor(ttk.Frame):
self.txn_vars: dict[str, StringVar] = {}
self.txn_rows: dict[str, tuple[ttk.Checkbutton, ttk.Label, ttk.Entry]] = {}
self.layout()
def layout(self):
@@ -38,32 +37,39 @@ class RowEditor(ttk.Frame):
# self.rowconfigure(2, weight=1)
self.rowconfigure(3, weight=1)
self.configure(relief='sunken')
self.configure(relief="sunken")
self.record_frame = ttk.LabelFrame(self, text="Record")
self.txn_frame = ttk.LabelFrame(self, text="Bean Transaction")
self.record_frame.grid(row=1, column=0, columnspan=2, padx=5, pady=5, sticky='nsew')
self.record_frame.grid(
row=1, column=0, columnspan=2, padx=5, pady=5, sticky="nsew"
)
self.record_frame.columnconfigure(2, weight=1)
self.txn_frame.grid(row=2, column=0, columnspan=2, padx=5, pady=5, sticky='nsew')
self.txn_frame.grid(
row=2, column=0, columnspan=2, padx=5, pady=5, sticky="nsew"
)
self.txn_frame.columnconfigure(2, weight=1)
self.match_var = BooleanVar()
self.match_toggle = ttk.Checkbutton(self, variable=self.match_var, command=self.do_match_toggle)
self.match_label = ttk.Label(
self,
text="Rule creation mode"
self.match_toggle = ttk.Checkbutton(
self, variable=self.match_var, command=self.do_match_toggle
)
self.match_toggle.grid(row=0, column=0, padx=11, pady=10, sticky='nw')
self.match_label.grid(row=0, column=1, padx=7, pady=10, sticky='nw')
self.match_label = ttk.Label(self, text="Rule creation mode")
self.match_toggle.grid(row=0, column=0, padx=11, pady=10, sticky="nw")
self.match_label.grid(row=0, column=1, padx=7, pady=10, sticky="nw")
self.button_frame = ttk.Frame(self, padding=(2, 2, 3, 3))
self.button_frame.grid(row=3, column=0, columnspan=2, sticky='s')
self.button_frame.grid(row=3, column=0, columnspan=2, sticky="s")
self.save_button = ttk.Button(self.button_frame, text="Save Transaction", command=self.do_save_txn)
self.save_button = ttk.Button(
self.button_frame, text="Save Transaction", command=self.do_save_txn
)
self.save_button.grid(row=0, column=0, padx=20)
self.rule_button = ttk.Button(self.button_frame, text="Create Rule", command=self.do_create_rule)
self.rule_button = ttk.Button(
self.button_frame, text="Create Rule", command=self.do_create_rule
)
self.rule_button.grid(row=0, column=1, padx=20)
self.reset_button = ttk.Button(self.button_frame, text="Reset TXN")
self.reset_button.grid(row=0, column=2, padx=20)
@@ -94,7 +100,7 @@ class RowEditor(ttk.Frame):
if var is None:
relayout = True
break
var.set(str(field.value if field.value is not None else '-'))
var.set(str(field.value if field.value is not None else "-"))
if relayout:
# Record fields changed, maybe the type of record changed?
# This shouldn't happen, but re-layout should fix it.
@@ -115,7 +121,7 @@ class RowEditor(ttk.Frame):
if label:
label.destroy()
# If the record is None, show the placeholder
# If the record is None, show the placeholder
# If the record is not None, hide the placeholder
# And build the record vars and labels and grid them
@@ -123,10 +129,9 @@ class RowEditor(ttk.Frame):
# Show placeholder label
if (label := self.record_placeholder) is None:
label = self.record_placeholder = ttk.Label(
self.record_frame,
text="Select a row to view record details."
self.record_frame, text="Select a row to view record details."
)
label.grid(row=0, column=0, columnspan=2, sticky='w')
label.grid(row=0, column=0, columnspan=2, sticky="w")
else:
# Hide placeholder if it exists
logger.debug(f"Laying out {record!r}")
@@ -142,12 +147,12 @@ class RowEditor(ttk.Frame):
self.record_frame,
text=field.display_name,
)
name_label.grid(row=i, column=1, sticky='w', padx=10)
name_label.grid(row=i, column=1, sticky="w", padx=10)
value_label = ttk.Label(
self.record_frame,
textvariable=var,
)
value_label.grid(row=i, column=2, sticky='w', padx=5)
value_label.grid(row=i, column=2, sticky="w", padx=5)
if field.matchable:
matchbox = ttk.Checkbutton(
@@ -166,11 +171,11 @@ class RowEditor(ttk.Frame):
if value and self.record and self.txn:
# Enable match boxes
for i, (box, _, _) in enumerate(self.txn_rows.values()):
box.grid(row=i, column=0, sticky='ew', padx=5)
box.grid(row=i, column=0, sticky="ew", padx=5)
for i, (box, _, _) in enumerate(self.record_rows):
if box:
box.grid(row=i, column=0, sticky='ew', padx=5)
box.grid(row=i, column=0, sticky="ew", padx=5)
self.rule_button.configure(state="enabled")
else:
# Disable match boxes
@@ -187,9 +192,11 @@ class RowEditor(ttk.Frame):
if self.txn is None:
raise ValueError("Cannot save with no transaction")
# Convert variables to dict of txn fields -> values
# Convert variables to dict of txn fields -> values
visible = [field.name for field in self.txn.display_fields()]
input_fields = {name: var.get() for name, var in self.txn_vars.items() if name in visible}
input_fields = {
name: var.get() for name, var in self.txn_vars.items() if name in visible
}
# Pass to PartialTXN.parse_input
update_fields = self.txn.parse_input(input_fields)
@@ -198,11 +205,11 @@ class RowEditor(ttk.Frame):
new_txn = self.txn.copy()
new_txn.update(**update_fields)
self.update_txn(new_txn)
self.event_generate('<<RowUpdated>>')
self.event_generate("<<RowUpdated>>")
def do_create_rule(self):
# Signal parent to create rule and rebuild
self.event_generate('<<RuleCreated>>')
self.event_generate("<<RuleCreated>>")
def make_rule(self):
if not self.match_var.get():
@@ -234,30 +241,26 @@ class RowEditor(ttk.Frame):
TODO: Warning icon and error? Change background colour on fields which don't validate?
"""
self.txn_placeholder = ttk.Label(
self.txn_frame,
text="Select a row to view txn details."
self.txn_frame, text="Select a row to view txn details."
)
# The txn display fields will always be a subset of this
fieldnames = PartialTXN.fields
# field -> stringvar
# field -> stringvar
txnvars = {name: StringVar() for name in fieldnames}
# field -> boolvar
txnmatching = {name: BooleanVar() for name in fieldnames}
# field -> (match box, key label, value entrybox)
txnrows = {}
for name, dname in (fieldnames.items()):
for name, dname in fieldnames.items():
matchbox = ttk.Checkbutton(self.txn_frame, variable=txnmatching[name])
keylabel = ttk.Label(self.txn_frame, text=dname)
if name in self.acmpl_cache:
entrybox = self._make_account_entrybox(name, txnvars[name])
else:
entrybox = ttk.Entry(
self.txn_frame,
textvariable=txnvars[name]
)
entrybox = ttk.Entry(self.txn_frame, textvariable=txnvars[name])
# TODO!: If we have a list of accounts, we could use a ComboBox here instead? For autocompletion.
# TODO!: Key shortcuts for prev/next record, and hints on the buttons
# TODO!: Radio Buttons on the flag
@@ -266,7 +269,7 @@ class RowEditor(ttk.Frame):
# TODO: Labels for the preview? The annotation? The errors?
self.txn_rows = txnrows
self.txn_matchvars = txnmatching
self.txn_matchvars = txnmatching
self.txn_vars = txnvars
def _make_account_entrybox(self, accountname: str, var: StringVar):
@@ -274,40 +277,44 @@ class RowEditor(ttk.Frame):
self.txn_frame,
textvariable=var,
values=list(self.acmpl_cache[accountname]),
postcommand=lambda: box.configure(values=list(self.acmpl_cache[accountname]))
postcommand=lambda: box.configure(
values=list(self.acmpl_cache[accountname])
),
)
return box
def layout_txn(self, txn: PartialTXN | None):
# If the txn is None, hide all the labels and replace with placeholder
# If the txn is None, hide all the labels and replace with placeholder
# If the txn is not None, re-grid the labels depending on whether the field is in the txn.
if txn is not None:
self.txn_placeholder.grid_remove()
self.save_button.configure(state='enabled')
self.save_button.configure(state="enabled")
txnfields = txn.display_fields()
visible = {field.name for field in txnfields}
for i, (field, (matchbox, keylabel, entrybox)) in enumerate(self.txn_rows.items()):
for i, (field, (matchbox, keylabel, entrybox)) in enumerate(
self.txn_rows.items()
):
# matchbox.grid(row=i, column=0, padx=5)
keylabel.grid(row=i, column=1, sticky='e', padx=3)
entrybox.grid(row=i, column=2, sticky='ew', padx=10)
keylabel.grid(row=i, column=1, sticky="e", padx=3)
entrybox.grid(row=i, column=2, sticky="ew", padx=10)
if field not in visible:
# Make sure hidden fields are not included in the match
# Otherwise we will have unexpected behaviour with hidden/garbage data
self.txn_matchvars[field].set(False)
self.txn_vars[field].set('')
matchbox.configure(state='disabled')
self.txn_vars[field].set("")
matchbox.configure(state="disabled")
# keylabel.grid_remove()
entrybox.configure(state='disabled')
entrybox.configure(state="disabled")
else:
matchbox.configure(state='enabled')
entrybox.configure(state='enabled')
matchbox.configure(state="enabled")
entrybox.configure(state="enabled")
else:
for labels in self.txn_rows.values():
for label in labels:
label.grid_remove()
self.txn_placeholder.grid(row=0, column=0, rowspan=3, sticky='ew')
self.save_button.configure(state='disabled')
self.txn_placeholder.grid(row=0, column=0, rowspan=3, sticky="ew")
self.save_button.configure(state="disabled")
def update_txn(self, txn: PartialTXN | None):
self.txn = txn
@@ -323,4 +330,3 @@ class RowEditor(ttk.Frame):
self.update_record(record)
self.update_txn(txn)
self.do_match_toggle()

121
src/beanify/gui/rowtree.py
View File

@@ -3,9 +3,9 @@ from datetime import datetime, date
from functools import partial
from tkinter import BooleanVar, ttk
from base.partial import PartialTXN
from base.record import Record
from base.transaction import Amount
from ..base.partial import PartialTXN
from ..base.record import Record
from ..base.transaction import Amount
from . import logger
@@ -19,6 +19,7 @@ def datetime_sort_key(datestr: datetime | str):
value = datetime.strptime(datestr, "%Y-%m-%d %H:%M:%S")
return value
def date_sort_key(datestr: date | str):
value = datetime.fromtimestamp(0).date()
if isinstance(datestr, date):
@@ -28,6 +29,7 @@ def date_sort_key(datestr: date | str):
value = datetime.strptime(datestr, "%Y-%m-%d").date()
return value
def amount_sort_key(amount: Amount | str):
if isinstance(amount, Amount):
value = (amount.currency, amount.value)
@@ -35,7 +37,7 @@ def amount_sort_key(amount: Amount | str):
amountstr, currency = amount.strip().split()
value = (currency, float(amountstr))
else:
value = ('', 0)
value = ("", 0)
return value
@@ -43,15 +45,16 @@ class SortingTree(ttk.Treeview):
"""
Treeview with column sorting and column selection
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.bind('<Button-3>', self.handle_rightclick)
self.bind("<Button-3>", self.handle_rightclick)
self.showvars = {col: BooleanVar() for col in self['columns']}
self.showvars = {col: BooleanVar() for col in self["columns"]}
for var in self.showvars.values():
# Default columns to on
var.set(True)
for col in self['columns']:
for col in self["columns"]:
self.column(col, stretch=True)
# self.set_columns(self['columns'])
@@ -60,18 +63,18 @@ class SortingTree(ttk.Treeview):
def heading_columns(self):
heading_map = {}
for col in self["columns"]:
heading_map[self.heading(col)['text']] = col
heading_map[self.heading(col)["text"]] = col
return heading_map
def reset_column_sizes(self):
displayed = self['displaycolumns']
displayed = self["displaycolumns"]
max_widths = [0] * len(displayed)
fontname = ttk.Style(self).lookup('TLabel', "font")
fontname = ttk.Style(self).lookup("TLabel", "font")
# treefont = font.nametofont(fontname)
# treefont = font.Font(family="TkDefaultFont")
magic = 7
for child in self.get_children(''):
for (i, col) in enumerate(displayed):
for child in self.get_children(""):
for i, col in enumerate(displayed):
data = self.set(child, col)
# length = treefont.measure(str(data))
length = len(str(data)) * magic
@@ -82,28 +85,28 @@ class SortingTree(ttk.Treeview):
self.column(col, width=width)
def heading(self, column, option=None, *, sort_key=None, **kwargs):
if sort_key and not hasattr(kwargs, 'command'):
if sort_key and not hasattr(kwargs, "command"):
command = partial(self._sort, column, False, sort_key)
kwargs['command'] = command
kwargs["command"] = command
return super().heading(column, option, **kwargs)
def set_columns(self, columns: list[str]):
self['displaycolumns'] = columns
self["displaycolumns"] = columns
for col, var in self.showvars.items():
var.set(col in columns)
def _sort(self, column, reverse, key):
l = [(self.set(k, column), k) for k in self.get_children('')]
l = [(self.set(k, column), k) for k in self.get_children("")]
l.sort(key=lambda t: key(t[0]), reverse=reverse)
for index, (_, k) in enumerate(l):
self.move(k, '', index)
self.move(k, "", index)
self.heading(column, command=partial(self._sort, column, not reverse, key))
def handle_rightclick(self, event):
logger.debug(f"Received right click on SortingTree: {event!r}")
region = self.identify_region(event.x, event.y)
logger.debug(f"REGION: {region}")
if region == 'heading':
if region == "heading":
self.do_column_select(event)
def do_column_select(self, event):
@@ -112,7 +115,11 @@ class SortingTree(ttk.Treeview):
menu = tk.Menu(self, tearoff=1)
for heading, column in self.heading_columns.items():
menu.add_checkbutton(variable=self.showvars[column], label=heading, command=self._show_columns)
menu.add_checkbutton(
variable=self.showvars[column],
label=heading,
command=self._show_columns,
)
try:
menu.tk_popup(event.x_root, event.y_root)
finally:
@@ -139,11 +146,14 @@ class RowTree(ttk.Frame):
self.layout()
def generate_columns(self, record: Record,):
def generate_columns(
self,
record: Record,
):
record_fields = record.display_fields()
columns = {
'record.date': ("Date", date_sort_key),
'record.from_source': ("Amount (from source)", amount_sort_key),
"record.date": ("Date", date_sort_key),
"record.from_source": ("Amount (from source)", amount_sort_key),
}
for field in record_fields:
name = f"record.{field.name}"
@@ -151,41 +161,41 @@ class RowTree(ttk.Frame):
if isinstance(value, Amount):
sorter = amount_sort_key
elif isinstance(value, datetime):
sorter = datetime_sort_key
sorter = datetime_sort_key
else:
sorter = str
columns[name] = (field.display_name, sorter)
columns |= {
'txn.flag': ("Bean Status", str),
'txn.payee': ("Bean Payee", str),
'txn.narration': ("Bean Narration", str),
'txn.comment': ("Bean Comment", str),
'txn.document': ("Bean Document", str),
'txn.tags': ("Bean Tags", str),
'txn.links': ("Bean Links", str),
'txn.source_account': ("Bean Source", str),
'txn.source_fee_asset_account': ("Bean Source Fee Asset Acc", str),
'txn.source_fee_expense_account': ("Bean Source Fee Expense Acc", str),
'txn.target_account': ("Bean Target", str),
'txn.target_fee_expense_account': ("Bean Target Fee Acc", str),
"txn.flag": ("Bean Status", str),
"txn.payee": ("Bean Payee", str),
"txn.narration": ("Bean Narration", str),
"txn.comment": ("Bean Comment", str),
"txn.document": ("Bean Document", str),
"txn.tags": ("Bean Tags", str),
"txn.links": ("Bean Links", str),
"txn.source_account": ("Bean Source", str),
"txn.source_fee_asset_account": ("Bean Source Fee Asset Acc", str),
"txn.source_fee_expense_account": ("Bean Source Fee Expense Acc", str),
"txn.target_account": ("Bean Target", str),
"txn.target_fee_expense_account": ("Bean Target Fee Acc", str),
}
return columns
def make_tree(self):
self.columns = self.generate_columns(self.base_record)
initially_enabled = [
'record.date',
'record.source_account',
'record.target_account',
'record.from_source',
'txn.source_account',
'txn.target_account',
"record.date",
"record.source_account",
"record.target_account",
"record.from_source",
"txn.source_account",
"txn.target_account",
]
self.tree = SortingTree(
self,
columns=tuple(self.columns.keys()),
show='headings',
show="headings",
)
for col, (dname, sort_key) in self.columns.items():
self.tree.heading(col, text=dname, sort_key=sort_key)
@@ -195,23 +205,23 @@ class RowTree(ttk.Frame):
self.rowconfigure(0, weight=1)
self.columnconfigure(0, weight=1)
self.tree.grid(row=0, column=0, sticky='NSEW')
self.tree.grid(row=0, column=0, sticky="NSEW")
self.tree.rowconfigure(0, weight=1)
self.tree.columnconfigure(0, weight=1)
self.scrollyframe = ttk.Frame(self, relief='groove')
self.scrollyframe = ttk.Frame(self, relief="groove")
self.scrolly = ttk.Scrollbar(
self,
orient='vertical',
orient="vertical",
command=self.tree.yview,
)
# self.scrollyframe.grid(row=0, column=1, sticky='NS')
self.scrolly.grid(row=0, column=1, sticky='NS')
self.scrolly.grid(row=0, column=1, sticky="NS")
self.tree.configure(yscrollcommand=self.scrolly.set)
self.scrollx = ttk.Scrollbar(
self,
orient='horizontal',
orient="horizontal",
command=self.tree.xview,
)
@@ -235,16 +245,11 @@ class RowTree(ttk.Frame):
for record, txn in rows.items():
if record in self.record_items:
itemid = self.record_items[record]
self.tree.item(
itemid,
values=self.row_values(record, txn)
)
self.tree.item(itemid, values=self.row_values(record, txn))
updated += 1
else:
itemid = self.tree.insert(
parent='',
index='end',
values=self.row_values(record, txn)
parent="", index="end", values=self.row_values(record, txn)
)
self.record_items[record] = itemid
added += 1
@@ -259,20 +264,18 @@ class RowTree(ttk.Frame):
def row_values(self, record, txn):
values = []
for col in self.columns:
match col.split('.', maxsplit=1):
match col.split(".", maxsplit=1):
case ["txn", "flag"]:
value = txn.flag.value
case ["txn", field]:
value = getattr(txn, field)
case ["record", "date"]:
value = record.date.strftime('%Y-%m-%d')
value = record.date.strftime("%Y-%m-%d")
case ["record", field]:
value = getattr(record, field)
# print(f"{col=} {value=} {type(value)=}")
case _:
raise ValueError(f"Unexpected column {col}")
value = value if value is not None else ''
value = value if value is not None else ""
values.append(value)
return values

376
src/beanify/main-simple.py
View File

@@ -1,376 +0,0 @@
"""
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())

267
src/beanify/main.py
View File

@@ -1,82 +1,261 @@
import os
import asyncio
import argparse
import logging
from dataclasses import dataclass
from pathlib import Path
import tomllib
import toml
from tkinter import ttk
from platformdirs import PlatformDirs
from base.rules import DummyRuleInterface, JSONRuleInterface, RuleSet
from converters.wise_converter import WiseConverter
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
"""
DEFAULTCONFIG = """\
[beanify]
default-converter = "wise"
# rule-interface = "json"
[beanify.gui]
# theme = "arc"
[rules.json]
# rulepath = "{datadir}/rules.json"
[converters.wise_v0]
asset-account = "Asset:Wise:{{currency}}"
fee-account = "Expenses:Wise:Fees"
[converters.cbacsv_v0]
asset-account = "Liabilities:CreditCard"
asset-currency = "AUD"
[converters.customcba]
base-converter = "cbacsv_v0"
asset-account = "Assets:CBA:Custom"
"""
parser = argparse.ArgumentParser()
parser.add_argument(
'converter',
help="Which converter to use. Available converters: 'wise'"
available = ", ".join(
(f"'{converter.qual_name()}'" for converter in converters_available)
)
parser.add_argument(
'--config',
dest='config',
help="Ingest configuration file. Should be a toml file with a section for the selected converter."
"--converter",
dest="converter",
help="Which converter to use.",
)
parser.add_argument(
'--ingest',
dest='ingest',
help="File to ingest (must be in the correct format for the selected converter)"
"--config",
dest="config",
help="Custom configuration file. Should be a toml file with a section for the selected converter.",
)
parser.add_argument(
'-o', '--output',
dest='output',
dest="ingest",
help="File(s) to ingest (must be in the correct format for the selected converter).",
action="append",
)
parser.add_argument(
"--rules",
dest="rules",
default=None,
help="Path to output file to append processed transactions. If not given will use stdout."
help="Path to JSON file with parsing rules. If not given, will use in-memory ruleset, with no saving.",
)
parser.add_argument(
'--rules',
dest='rules',
"--ruleinterface",
dest="ruleinterface",
default=None,
help="Path to JSON file with parsing rules. If not given, will use in-memory ruleset, with no saving."
help="'json' or 'dummy'",
)
"""
TODO:
- [ ] Add parser arguments
- [ ] Find config file, or create and close
- [ ] Read available converters, read requested converter, make sure requested converter exists
- [ ] Create rule interface for requested converter, warn if dummy (support using system path)
- [ ] Load converter with options from config
"""
async def main():
args = parser.parse_args()
converter = args.converter
ingest_file = args.ingest
out_path = args.output
rule_path = args.rules
config_path = args.config
dirs = PlatformDirs("beanify", "Interitio")
configdir = Path(dirs.user_config_dir)
datadir = Path(dirs.user_data_dir)
# Open configuration file
configpath: Path | None = None
if args.config:
# --config choc.toml should search for configdir/choc.toml and ./choc.toml
if os.path.exists(args.config):
configpath = Path(args.config)
elif os.path.exists(configdir / args.config):
configpath = configdir / args.config
else:
print("Couldn't find configuration file '{args.config}'")
return
else:
# If configuration file isn't found, create default and exit
configpath = configdir / "config.toml"
if not os.path.exists(configpath):
if not os.path.exists(configdir):
os.makedirs(configdir)
with open(configpath, "w") as f:
f.write(DEFAULTCONFIG.format(datadir=datadir))
print(
f"Created new configuration file at '{configpath}'. Please configure desired converters."
)
return
config = toml.load(configpath)
# Figure out requested converter
if args.converter:
requested_converter = args.converter
elif defconv := config["beanify"].get("default-converter"):
requested_converter = defconv
else:
print("ERROR: No converter selected, and no default converter configured.")
return
# Get config for requested converter
confconfig = config.get("converters", {})
at_root = False
currconv = requested_converter
confstack = []
seen = set()
while not at_root:
if currconv not in confconfig:
print(f"ERROR: Missing configuration section for converter '{currconv}'")
return
confsec = confconfig[currconv]
confstack.append(confsec)
seen.add(currconv)
if "base-converter" in confsec:
currconv = confsec["base-converter"]
if currconv in seen:
print(f"ERROR: Configuration loop detected on '{currconv}'")
return
else:
at_root = True
convconf = {} # The actual convert configuration
for conf in reversed(confstack):
convconf |= conf
# Build and initialise the converter
converter_cls = converter_factory(qual_name=currconv)
converter_config = converter_cls.config_type.from_dict(convconf)
converter = converter_cls(converter_config)
# Figure out requested rule interface
if args.ruleinterface:
requested_interface = args.ruleinterface
elif defint := config["beanify"].get("rule-interface"):
requested_interface = defint
else:
requested_interface = "json"
if requested_interface == "json":
# Get or create the rule path
if args.rules:
rulepath = Path(args.rules)
elif confrulepath := config.get("rules", {}).get("json", {}).get("rulepath"):
rulepath = Path(confrulepath)
else:
rulepath = datadir / "rules.json"
if not os.path.exists(datadir):
os.makedirs(datadir)
if not os.path.exists(rulepath):
with open(rulepath, "w") as f:
f.write("{}")
print(f"Created new json rule file at '{rulepath}'.")
rule_interface = JSONRuleInterface(requested_converter, path=rulepath)
elif requested_interface == "dummy":
rule_interface = DummyRuleInterface(requested_converter)
else:
print(f"Unsupported rule interface: '{requested_interface}'.")
return
ruleset = RuleSet.load_from(rule_interface)
window = MainWindow(
config, converter, ruleset, initial_files=args.ingest, theme="clam"
)
window.mainloop()
async def _main():
args = parser.parse_args()
config_path = args.config
converter_name = args.converter
ingest_files = args.ingest
rule_path = args.rules
# Create converter config
# TODO: Will need to move to tomlkit or other write-capable library
with open(config_path, "rb") as f:
root_config = tomllib.load(f)
# TODO: converter factory
converter_cls = WiseConverter
converter_config = converter_cls.config_type.from_dict(root_config[converter_cls.config_field])
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)
# TODO: rule interface factory
# Also dummy plug when no rules given
if rule_path:
rule_interface = JSONRuleInterface(converter_cls.qual_name(), path=rule_path)
else:
rule_interface = DummyRuleInterface(converter_cls.qual_name())
ruleset = await RuleSet.load_from(rule_interface)
ruleset = RuleSet.load_from(rule_interface)
print("Finished configuration. Loading records")
records = converter.ingest_file(ingest_file)
print(f"Loaded {len(records)} records.")
partials = [converter.convert(record, ruleset) for record in records]
print(f"Converted {len(partials)} partial transactions.")
for partial in partials:
print(repr(partial))
for record, partial in zip(records, partials):
print('\n'.join(record.display_table()))
print(repr(partial))
print('-'*10)
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__':
def run():
"""
Script entrance to beanify.
"""
asyncio.run(main())
if __name__ == "__main__":
asyncio.run(main())