Files
@ dc2e2d200d70
Branch filter:
Location: NPO-Accounting/conservancy_beancount/conservancy_beancount/filters.py
dc2e2d200d70
4.0 KiB
text/x-python
balance_sheet: Fix logger name.
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 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """filters.py - Common filters for postings"""
# Copyright © 2020 Brett Smith
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import datetime
import re
from beancount.core import data as bc_data
from . import data
from . import rtutil
from typing import (
cast,
Hashable,
Iterable,
Iterator,
Optional,
Pattern,
Set,
TypeVar,
Union,
)
from .beancount_types import (
Directive,
Entries,
MetaKey,
MetaValue,
Transaction,
)
# Saying Optional works around <https://github.com/python/mypy/issues/8768>.
HashT = TypeVar('HashT', bound=Optional[Hashable])
Postings = Iterable[data.Posting]
Regexp = Union[str, Pattern]
def audit_date(entries: Entries) -> Optional[datetime.date]:
for entry in entries:
if (isinstance(entry, bc_data.Custom)
and entry.type == 'conservancy_beancount_audit'): # type:ignore[attr-defined]
return entry.date
return None
def filter_meta_equal(postings: Postings, key: MetaKey, value: MetaValue) -> Postings:
for post in postings:
try:
if post.meta[key] == value:
yield post
except KeyError:
pass
def filter_meta_match(postings: Postings, key: MetaKey, regexp: Regexp) -> Postings:
for post in postings:
try:
if re.search(regexp, post.meta[key]):
yield post
except (KeyError, TypeError):
pass
def filter_for_rt_id(postings: Postings, ticket_id: Union[int, str]) -> Postings:
"""Filter postings with a primary RT ticket
This functions yields postings where the *first* rt-id matches the given
ticket number.
"""
regexp = rtutil.RT.metadata_regexp(ticket_id, first_link_only=True)
return filter_meta_match(postings, 'rt-id', regexp)
def iter_unique(seq: Iterable[HashT]) -> Iterator[HashT]:
seen: Set[HashT] = set()
for item in seq:
if item not in seen:
seen.add(item)
yield item
def remove_opening_balance_txn(entries: Entries) -> Optional[Transaction]:
"""Remove an opening balance transaction from entries returned by Beancount
Returns the removed transaction if found, or None if not.
Note that it modifies the ``entries`` argument in-place.
This function is useful for tools like accrual-report that are more
focused on finding and reporting related transactions than providing
total account balances, etc. Since the opening balance transaction does not
provide the same metadata documentation as typical transactions, it's
typically easiest to filter it out before cross-referencing transactions by
metadata.
Note that this function only removes a single transaction, because that's
fastest for the common case.
"""
for index, entry in enumerate(entries):
if isinstance(entry, bc_data.Transaction):
entry = cast(Transaction, entry)
if data.is_opening_balance_txn(entry):
break
else:
return None
# Deleting from the beginning of a list is O(n) slow in Python:
# <https://wiki.python.org/moin/TimeComplexity>
# So don't do that, and instead replace the transaction with a placeholder
# directive.
# The type:ignore is because of the funny way Beancount builds directives.
entries[index] = bc_data.Custom( # type:ignore[operator]
entry.meta, entry.date, "Removed opening balances", [],
)
return entry
|