Files @ 6deaacb11bdd
Branch filter:

Location: NPO-Accounting/conservancy_beancount/conservancy_beancount/filters.py

bkuhn
Add US:TN:Unemployment as a valid `payroll-type` metadata for taxes
"""filters.py - Common filters for postings"""
# Copyright © 2020  Brett Smith
# License: AGPLv3-or-later WITH Beancount-Plugin-Additional-Permission-1.0
#
# Full copyright and licensing details can be found at toplevel file
# LICENSE.txt in the repository.

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) -> Iterator[data.Posting]:
    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) -> Iterator[data.Posting]:
    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]) -> Iterator[data.Posting]:
    """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