From a86434552910600f367041dad9995cb5472db7ec Mon Sep 17 00:00:00 2001 From: David Duran Rosich Date: Sun, 19 Jul 2026 12:24:59 +0200 Subject: [PATCH] Moved plugins to their own repo --- joint_account.py | 151 +++++++++++++++++++++++++++++++++++ link_accounts.py | 62 +++++++++++++++ old_split_account.py | 86 ++++++++++++++++++++ optional_balance.py | 41 ++++++++++ split_account.py | 186 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 526 insertions(+) create mode 100644 joint_account.py create mode 100644 link_accounts.py create mode 100644 old_split_account.py create mode 100644 optional_balance.py create mode 100644 split_account.py diff --git a/joint_account.py b/joint_account.py new file mode 100644 index 0000000..3d8e6d0 --- /dev/null +++ b/joint_account.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 + +from beancount.core import data +from beancount.core import account +from beancount.core import getters +from beancount.core import amount +from beancount.core import account_types +from beancount.core import realization +from beancount.core.number import D +from beancount.parser import options + +__plugins__ = ('joint_account_plugin',) + +# DEBT_ACCOUNT = "Deute" +SHARE_ACCOUNT = "Part" + +def debt_account_str(acc: str, debtor: str, accs: account_types.AccountTypes) -> str: + return account.join(accs.liabilities, account.sans_root(acc), debtor) + +def share_account_str(acc: str, debtor: str) -> str: + # return ":".join((account, SHARE_ACCOUNT + "-" + debtor)) + return acc # reimbursment in same account + +def debt_account(posting: data.Posting, debtor: str, joint_accounts: dict, accs: account_types.AccountTypes) -> str: + if debtor in joint_accounts.get(posting.account, {}): + return debt_account_str(posting.account, debtor, accs) + if account_types.is_income_statement_account(posting.account, accs): + return share_account_str(posting.account, debtor) + return "" + +def is_splitable(posting: data.Posting, accs: account_types.AccountTypes): + if posting.meta.get("joint", "") == "nosplit": + return False + return account_types.is_income_statement_account(posting.account, accs) + +def divide_and_flip(posting: data.Posting, n: int = 2) -> data.Posting: + """Return a new posting with n-th the amount and opposite sign.""" + assert(n>0) + old_amount = posting.units + new_amount = amount.div(old_amount, D(-n)) + return posting._replace(units=new_amount) + +def add_tag(txn: data.Transaction, tag: str) -> data.Transaction: + return txn._replace(tags=txn.tags | set([tag])) + +def prepare_posting(posting, user, joint_accounts, accs, n): + return divide_and_flip(posting._replace(account=debt_account(posting, user, joint_accounts, accs), flag=None, meta=None), n) + +def split_among_users(postings, users, joint_accounts, accs): + new_postings = [] + for user in users: + for posting in postings: + new_postings.append(prepare_posting(posting, user, joint_accounts, accs, 1 + len(users))) + return new_postings + + +def debt_transaction(txn: data.Transaction, joint_accounts: dict, new_accounts: set, accs: account_types.AccountTypes) -> None | data.Transaction: + # Only two cases are processed: + # a) all accounts are joint with the same people + # b) a single joint account with splitable/unsplitable accounts + + # get how many joint accounts there are + user_sets = [(idx, set(joint_accounts[posting.account])) for idx, posting in enumerate(txn.postings) if posting.account in joint_accounts] + + new_postings = [] + + # no joint account -> nothing to do + if len(user_sets) == 0: + return None + + # single joint account, assume all expenses/income are split equally between users of the joint account + elif len(user_sets) == 1: + splitable_postings = [posting for posting in txn.postings if is_splitable(posting, accs)] + if len(splitable_postings) == 0: + return None + + # as not every posting may be splittable, debt of joint account might need to be rebalanced + new_balance = realization.compute_postings_balance(splitable_postings) + joint_posting = txn.postings[user_sets[0][0]] + joint_posting = joint_posting._replace(units=-new_balance.get_currency_units(joint_posting.units.currency)) + new_postings = splitable_postings + [joint_posting] + + # all postings are from joint accounts, check they share same users and transfer debt equally + elif len(user_sets) == len(txn.postings): + # check all joint accounts share same users + total_set = set().union(*map(lambda s: s[1], user_sets)) + if any((total_set != s[1] for s in user_sets)): + return None + new_postings = txn.postings + + # just transfer debt equally + new_postings = split_among_users(new_postings, user_sets[0][1], joint_accounts, accs) + new_accounts |= set((p.account for p in new_postings)) + return add_tag(txn._replace(postings=new_postings), "retorn") + +def joint_account_plugin(entries, options_map): + errors = [] + new_entries = [] + new_accounts = set() + + accs = options.get_account_types(options_map) + + joint_accounts = {} + for entry in entries: + if isinstance(entry, data.Open): + if "joint" in entry.meta: + joint_accounts[entry.account] = entry.meta["joint"].strip().split(" ") + + for entry in entries: + if isinstance(entry, data.Transaction): + if any((acc in joint_accounts for acc in getters.get_accounts([entry]))): + new_entry = debt_transaction(entry, joint_accounts, new_accounts, accs) + if new_entry is not None: + new_entries.append(new_entry) + + new_entries.append(entry) + + # add new_accounts at earliest date possible + earliest_date = getters.get_min_max_dates(new_entries)[0] + new_accounts -= set(getters.get_accounts((e for e in entries if isinstance(e, data.Open)))) + + for idx, acc in enumerate(new_accounts): + entry = data.Open(data.new_metadata("", idx), earliest_date, acc, None, None) + new_entries.append(entry) + + # close associated accounts + for entry in entries: + if isinstance(entry, data.Close): + for user in joint_accounts.get(entry.account, []): + e = entry._replace(account=debt_account_str(entry.account, user, accs)) + new_entries.append(e) + + return new_entries, [] + +# def change_to_total_account(txn: data.Transaction, joint_accounts) -> data.Transaction: +# return txn._replace(postings=list(map(lambda p: change_to_total_account_posting(p, joint_accounts), txn.postings))) + +# def change_to_total_account_posting(posting: data.Posting, joint_accounts) -> data.Posting: +# return posting._replace(account=total_account_str(posting.account) if posting.account in joint_accounts else posting.account) + +# def transform_accounts(entry, joint_accounts): +# if hasattr(entry, "account"): +# if entry.account in joint_accounts: +# return entry._replace(account=total_account_str(entry.account)) +# elif isinstance(entry, data.Transaction): +# return change_to_total_account(entry, joint_accounts) +# return entry + + + + diff --git a/link_accounts.py b/link_accounts.py new file mode 100644 index 0000000..bf88096 --- /dev/null +++ b/link_accounts.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python3 + +import re + +from beancount.core import data +from beancount.core import inventory +from beancount.core import account +from beancount.core import amount +from beancount.core import account_types +from beancount.parser import parser +from beancount.core.number import D +from beancount.parser import options +import itertools as it +from collections import namedtuple + +LinkAccount = namedtuple("LinkAccount", ["account", "tags"]) + +__plugins__ = ('link_account_plugin',) + +def add_tags(txn: data.Transaction, *tags: str) -> data.Transaction: + return txn._replace(tags=txn.tags | set(tags)) + +def link_posting(p: data.Posting, link_accounts: dict[str, LinkAccount]) -> list[data.Posting]: + if p.account in link_accounts: + return [ + p._replace(account=link_accounts[p.account].account), + p._replace(units=-p.units) + ] + return None + + +def link_txn(txn: data.Transaction, link_accounts: dict[str, LinkAccount]) -> list[data.Transaction]: + new_postings = [p for p in txn.postings if p.account in link_accounts] + if len(new_postings) == 0: + return [] + + new_txns = [] + + get_account = lambda x: x.account + new_postings = sorted(new_postings, key=get_account) + for k, g in it.groupby(new_postings, key=get_account): + new_postings = [p for o in g for p in link_posting(o, link_accounts)] + new_txns.append(add_tags(txn._replace(postings=new_postings), "link", *link_accounts[k].tags)) + return new_txns + +def link_account_plugin(entries, options_map, config=None): + errors = [] + + link_accounts = {} + for entry in entries: + if isinstance(entry, data.Open): + if "link-account" in entry.meta: + tags = entry.meta["link-tags"].split(" ") if "link-tags" in entry.meta else [] + link_accounts[entry.account] = LinkAccount(account=entry.meta["link-account"], tags=tags) + + new_entries = [] + for entry in entries: + if isinstance(entry, data.Transaction): + new_entries.extend(link_txn(entry, link_accounts)) + new_entries.append(entry) + + return new_entries, errors \ No newline at end of file diff --git a/old_split_account.py b/old_split_account.py new file mode 100644 index 0000000..e14c7ce --- /dev/null +++ b/old_split_account.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 + +import re + +from beancount.core import data +from beancount.core import getters +from beancount.core import amount +from beancount.core import account_types +from beancount.core import realization +from beancount.core.number import D +from beancount.parser import options + +__plugins__ = ('split_account_plugin',) + +def get_group(tag, _pattern=re.compile(r"split-(.*)")): + m = _pattern.match(tag) + if m: + return m.group(1) + return "" + +def is_splitable(posting: data.Posting, accs: account_types.AccountTypes) -> bool: + if posting.meta.get("split", "") == "no": + return False + return account_types.is_income_statement_account(posting.account, accs) + +def add_tag(txn: data.Transaction, tag: str) -> data.Transaction: + return txn._replace(tags=txn.tags | set([tag])) + +def divide_and_flip(posting: data.Posting, n: int = 2) -> data.Posting: + """Return a new posting with n-th the amount and opposite sign.""" + assert(n>0) + old_amount = posting.units + new_amount = amount.div(old_amount, D(-n)) + return posting._replace(units=new_amount) + +def split_txn(txn: data.Transaction, split_accounts: dict, accs: account_types.AccountTypes) -> None | data.Transaction: + # involves_account = any((acc in split_accounts.items() for acc in getters.get_accounts([entry]))) + + groups = [g for g in ( get_group(tag) for tag in txn.tags ) if g in split_accounts] + if len(groups) != 1: + return None + group = groups[0] + + has_ignore_meta = txn.meta.get("split", "") == "no" + if has_ignore_meta: + return None + + # gather all splitable expenses/income, and get their total balance + new_postings = [posting for posting in txn.postings if is_splitable(posting, accs)] + if len(new_postings) == 0: + return None + + new_balance = realization.compute_postings_balance(new_postings) + for position in new_balance: + new_postings.append(data.Posting( + account=split_accounts[group], + units=-position.units, + cost=position.cost, + price=None, + flag=None, + meta=None + )) + + new_postings = [ divide_and_flip(p) for p in new_postings ] + + return add_tag(txn._replace(postings=new_postings), "retorn") + +def split_account_plugin(entries, options_map): + errors = [] + new_entries = [] + accs = options.get_account_types(options_map) + + split_accounts = {} + for entry in entries: + if isinstance(entry, data.Open): + if "split" in entry.meta: + split_accounts[entry.meta["split"]] = entry.account + + for entry in entries: + if isinstance(entry, data.Transaction): + new_txn = split_txn(entry, split_accounts, accs) + if new_txn is not None: + new_entries.append(new_txn) + new_entries.append(entry) + + return new_entries, errors diff --git a/optional_balance.py b/optional_balance.py new file mode 100644 index 0000000..c71ac65 --- /dev/null +++ b/optional_balance.py @@ -0,0 +1,41 @@ +#!/usr/bin/env python3 + +from beancount.core import data +from beancount.core import account +from beancount.core import getters +from beancount.core import amount +from beancount.core import account_types +from beancount.core import realization +from beancount.core.number import D +from beancount.parser import options + +import re + +__plugins__ = ('optional_balance_plugin',) + +def optional_balance_plugin(entries, options_map, config): + try: + match_enabled = re.compile(config) + except Exception as e: + print(f"exception {e}") + return entries, [e] + + errors = [] + new_entries = [] + for entry in entries: + if isinstance(entry, data.Custom): + if entry.type == "optional-balance": + try: + assert(len(entry.values) == 4) + type, acc, units, curr = entry.values + if not match_enabled.match(type.value): + continue + v = amount.Amount(number=D(units.value), currency=curr.value) + entry = data.Balance(meta=entry.meta, date=entry.date, account=acc.value, amount=v, tolerance=None, diff_amount=None) + except Exception as e: + print(f"exception {e}") + errors.append(e) + + new_entries.append(entry) + + return new_entries, errors \ No newline at end of file diff --git a/split_account.py b/split_account.py new file mode 100644 index 0000000..52b05c9 --- /dev/null +++ b/split_account.py @@ -0,0 +1,186 @@ +#!/usr/bin/env python3 + +import re +import collections +from typing import Sequence + +from beancount.core import data +from beancount.core import inventory +from beancount.core import account +from beancount.core import amount +from beancount.core import getters +from beancount.core import account_types +from beancount.parser import parser +from beancount.core.number import D +from beancount.parser import options +import itertools as it + +__plugins__ = ('split_account_plugin',) + +ConfigError = collections.namedtuple("ConfigError", "source message entry") +MultipleAccountError = collections.namedtuple("MultipleAccountError", "source message entry") +SplitAccount = collections.namedtuple("SplitAccount", ["account", "users", "tag"]) + +class MultipleAccountException(Exception): + def __init__(self, split_tags): + self.split_tags = split_tags + +def split_posting( + posting: data.Posting, + split: SplitAccount, + accs: account_types.AccountTypes, + main: None | str + ) -> list[data.Posting]: + weights = None + if "split" in posting.meta: + assert(set(posting.meta["split"].split(" ")) <= set(split.users)) + users = posting.meta["split"].split(" ") + weights = [1 if user in users else 0 for user in split.users] + elif account.leaf(posting.account) in split.users: + user = account.leaf(posting.account) + weights = [1 if u == user else 0 for u in split.users] + else: + weights = [1 for _ in range(len(split.users))] + + total_weight = sum(weights) + assert(total_weight > 0) + + new_postings = [ + posting._replace( + units=amount.mul(posting.units, -D(w) / total_weight), + account=account.join(split.account, split.users[idx]), + ) + for idx, w in enumerate(weights) + ] + if main in split.users: + main_idx = split.users.index(main) + new_postings.extend([ + posting._replace( + units=amount.mul(posting.units, -D(w) / total_weight), + ) + for idx, w in enumerate(weights) + if idx != main_idx + ]) + new_postings.extend([ + posting._replace( + units=amount.mul(posting.units, D(w) / total_weight), + account=account.join("Passiu", "Split", split.users[idx]), + ) + for idx, w in enumerate(weights) + if idx != main_idx + ]) + + return new_postings + +def merge_postings(postings: Sequence[data.Posting]) -> list[data.Posting]: + merged = [] + keyfunc = lambda p: p.account + for acc, g in it.groupby(sorted(postings, key=keyfunc), key=keyfunc): + inv = inventory.Inventory() + for p in g: + inv.add_position(p) + + for pos in inv.get_positions(): + merged.append(data.Posting( + account=acc, + units=pos.units, + cost=pos.cost, + price=None, + flag=None, + meta=None + )) + return merged + + +def split_txn( + txn: data.Transaction, + split_accounts: dict[str, SplitAccount], + split_tags: dict[str, str], + accs: account_types.AccountTypes, + main: None | str + ) -> data.Transaction: + involved_accs = set(p.account for p in txn.postings if p.account in split_accounts) + + tags = list(tag for tag in txn.tags if tag in split_tags) + split_accs = list(split_tags[tag] for tag in tags) + if len(split_accs) > 1: + raise MultipleAccountException(tags) + elif len(split_accs) == 1: + involved_accs.add(split_accs[0]) + if len(involved_accs) == 0: + return txn + + # filter postings that involve parent accounts, this way we can preserve all asset buckets + new_postings = [p for p in txn.postings if p.account not in involved_accs] + + split_postings = [] + for acc in involved_accs: + ps = [(p._replace(account="") if p.account in involved_accs else p) for p in txn.postings if p.account != acc] + split_postings.extend(n for p in ps for n in split_posting(p, split_accounts[acc], accs, main)) + + # merge split_postings grouped by account + new_postings.extend(merge_postings(split_postings)) + + return txn._replace(postings=new_postings) + +def split_account_plugin(entries, options_map, config=None): + errors = [] + new_entries = [] + accs = options.get_account_types(options_map) + + split_accounts = {} + split_tags = {} + for entry in entries: + if isinstance(entry, data.Open): + if "split" in entry.meta: + tag, *users = entry.meta["split"].split(" ") + users = list(set(users)) # remove repeats + if len(users) < 2: + errors.append(ConfigError( + entry.meta, + f"Split account need at least two users, only {len(users)} found: {users}", + entry + )) + continue + + split_accounts[entry.account] = SplitAccount(users=users, account=entry.account, tag=tag) + split_tags[tag] = entry.account + for user in users: + new_entries.append( + data.Open( + meta=data.new_metadata("", 0), + date=entry.date, + account=account.join(entry.account, user), + currencies=entry.currencies, + booking=entry.booking + ) + ) + old_accounts = getters.get_accounts(new_entries) | getters.get_accounts(entries) + + new_accounts = set() + for entry in entries: + if isinstance(entry, data.Transaction): + try: + entry = split_txn(entry, split_accounts, split_tags, accs, config) + new_accounts |= set((p.account for p in entry.postings)) + except MultipleAccountException as e: + errors.append(MultipleAccountError( + entry.meta, + f"Transaction has multiple split tags: {e.split_tags}", + entry + )) + new_entries.append(entry) + + oldest_date = getters.get_min_max_dates(new_entries)[0] + for acc in new_accounts - old_accounts: + new_entries.append( + data.Open( + meta=data.new_metadata("", 1), + date=oldest_date, + account=acc, + currencies=None, + booking=None + ) + ) + + return new_entries, errors