1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
|
'''
Plugin that closes an account by transferring its whole balance to another account.
'''
__copyright__ = 'Copyright (c) 2020 Kristóf Marussy <kristof@marussy.com>'
__license__ = 'GNU GPLv2'
from typing import Any, Dict, List, NamedTuple, Optional, Tuple
from beancount.core.data import Balance, Close, Directive, Entries, Meta, Posting, Transaction
from beancount.core.flags import FLAG_OKAY
from beancount.core.number import ZERO
__plugins__ = ('close_with_balance_assertions',)
CLOSE_TO_META = 'close-to'
CLOSING_META = 'closing'
class ClosingBalanceError(NamedTuple):
source: Optional[Meta]
message: str
entry: Directive
def close_with_balance_assertions(entries: Entries,
options_map: Dict[str, Any],
config_str: Optional[str] = None) -> \
Tuple[Entries, List[ClosingBalanceError]]:
new_entries: Entries = []
errors: List[ClosingBalanceError] = []
for entry in entries:
new_entries.append(entry)
if isinstance(entry, Balance) and CLOSE_TO_META in entry.meta:
close_to_account = entry.meta[CLOSE_TO_META]
if not isinstance(close_to_account, str):
errors.append(ClosingBalanceError(
entry.meta,
f'{CLOSE_TO_META} must be a string, got {close_to_account} instead',
entry))
continue
if entry.tolerance is not None and entry.tolerance != ZERO:
errors.append(ClosingBalanceError(
entry.meta,
f'Closing an account requires {ZERO} tolerance, got {entry.tolerance} instead',
entry))
continue
if entry.diff_amount is not None:
errors.append(ClosingBalanceError(
entry.meta,
f'Not closing {entry.account} with {entry.diff_amount} failed balance check',
entry))
continue
new_meta = dict(entry.meta)
del new_meta[CLOSE_TO_META]
if entry.amount.number != ZERO:
new_entries.append(Transaction(
new_meta,
entry.date,
FLAG_OKAY,
None,
f'Closing {entry.account}',
set(),
set(),
[
Posting(entry.account, -entry.amount, None, None, None,
{CLOSING_META: True}),
Posting(close_to_account, entry.amount, None, None, None, None)
]))
new_entries.append(Close(new_meta, entry.date, entry.account))
return new_entries, errors
|
