"""Base classes for plugin checks""" # 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 abc import datetime import re from .. import config as configmod from .. import data from .. import errors as errormod from .. import ranges from typing import ( Any, Container, Dict, FrozenSet, Generic, Iterable, Iterator, Mapping, Optional, Sequence, Type, TypeVar, ) from ..beancount_types import ( Account, Directive, MetaKey, MetaValue, MetaValueEnum, Transaction, ) ### CONSTANTS # I expect these will become configurable in the future, which is why I'm # keeping them outside of a class, but for now constants will do. # The default start date is the beginning of FY19, minus one day to cover FY19's # opening balances. DEFAULT_START_DATE = datetime.date(2019, 2, 28) # The default stop date leaves a little room after so it's easy to test # dates past the far end of the range. DEFAULT_STOP_DATE = datetime.date(datetime.MAXYEAR, 1, 1) ### TYPE DEFINITIONS HookName = str Entry = TypeVar('Entry', bound=Directive) class Hook(Generic[Entry], metaclass=abc.ABCMeta): DIRECTIVE: Type[Directive] HOOK_GROUPS: FrozenSet[HookName] = frozenset() def __init__(self, config: configmod.Config) -> None: pass # Subclasses that need configuration should override __init__ to check # and store it. @abc.abstractmethod def run(self, entry: Entry) -> errormod.Iter: ... ### HELPER CLASSES class MetadataEnum: """Map acceptable metadata values to their normalized forms. When a piece of metadata uses a set of allowed values, use this class to define them. You can also specify aliases that hooks will normalize to the primary values. """ def __init__(self, key: MetaKey, standard_values: Iterable[MetaValueEnum], aliases_map: Optional[Mapping[MetaValueEnum, MetaValueEnum]]=None, ) -> None: """Specify allowed values and aliases for this metadata. Arguments: * key: The name of the metadata key that uses this enum. * standard_values: A sequence of strings that enumerate the standard values for this metadata. * aliases_map: A mapping of strings to strings. The keys are additional allowed metadata values. The values are standard values that each key will evaluate to. The code asserts that all values are in standard_values. """ self.key = key self._stdvalues = frozenset(standard_values) self._aliases: Dict[MetaValueEnum, MetaValueEnum] = dict(aliases_map or ()) assert self._stdvalues.issuperset(self._aliases.values()) self._aliases.update((v, v) for v in standard_values) def __repr__(self) -> str: return "{}<{}>".format(type(self).__name__, self.key) def __contains__(self, key: MetaValueEnum) -> bool: """Returns true if `key` is a standard value or alias.""" return key in self._aliases def __getitem__(self, key: MetaValueEnum) -> MetaValueEnum: """Return the standard value for `key`. Raises KeyError if `key` is not a known value or alias. """ return self._aliases[key] def __iter__(self) -> Iterator[MetaValueEnum]: """Iterate over standard values.""" return iter(self._stdvalues) def get(self, key: MetaValueEnum, default_key: Optional[MetaValueEnum]=None, ) -> Optional[MetaValueEnum]: """Return self[key], or a default fallback if that doesn't exist. default_key is another key to look up, *not* a default value to return. This helps ensure you always get a standard value. """ try: return self[key] except KeyError: if default_key is None: return None else: return self[default_key] ### HOOK SUBCLASSES class TransactionHook(Hook[Transaction]): DIRECTIVE = Transaction SKIP_FLAGS: Container[str] = frozenset() TXN_DATE_RANGE = ranges.DateRange(DEFAULT_START_DATE, DEFAULT_STOP_DATE) def _run_on_txn(self, txn: Transaction) -> bool: """Check whether we should run on a given transaction This method implements our usual checks for whether or not a hook should run on a given transaction. It's here for subclasses to use in their own implementations. See _PostingHook below for an example. """ return ( txn.flag not in self.SKIP_FLAGS and txn.date in self.TXN_DATE_RANGE and not data.is_opening_balance_txn(txn) ) class _PostingHook(TransactionHook, metaclass=abc.ABCMeta): def __init_subclass__(cls) -> None: cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(['posting']) def _run_on_post(self, txn: Transaction, post: data.Posting) -> bool: return True def run(self, txn: Transaction) -> errormod.Iter: if self._run_on_txn(txn): for post in data.Posting.from_txn(txn): if self._run_on_post(txn, post): yield from self.post_run(txn, post) @abc.abstractmethod def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter: ... class _NormalizePostingMetadataHook(_PostingHook): """Base class to normalize posting metadata from an enum.""" # This class provides basic functionality to filter postings, normalize # metadata values, and set default values. METADATA_KEY: MetaKey VALUES_ENUM: MetadataEnum def __init_subclass__(cls) -> None: super().__init_subclass__() cls.METADATA_KEY = cls.VALUES_ENUM.key cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(['metadata', cls.METADATA_KEY]) # If the posting does not specify METADATA_KEY, the hook calls # _default_value to get a default. This method should either return # a value string from METADATA_ENUM, or else raise InvalidMetadataError. # This base implementation does the latter. def _default_value(self, txn: Transaction, post: data.Posting) -> MetaValueEnum: raise errormod.InvalidMetadataError(txn, self.METADATA_KEY, None, post) def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter: source_value = post.meta.get(self.METADATA_KEY) set_value = source_value error: Optional[errormod.Error] = None if source_value is None: try: set_value = self._default_value(txn, post) except errormod.Error as error_: error = error_ else: try: set_value = self.VALUES_ENUM[source_value] except KeyError: error = errormod.InvalidMetadataError( txn, self.METADATA_KEY, source_value, post, ) if error is None: post.meta[self.METADATA_KEY] = set_value else: yield error class _RequireLinksPostingMetadataHook(_PostingHook): """Base class to require that posting metadata include links""" # This base class confirms that a posting's metadata has one or more links # under one of the metadata keys listed in CHECKED_METADATA. # Most subclasses only need to define CHECKED_METADATA and _run_on_post. CHECKED_METADATA: Sequence[MetaKey] def __init_subclass__(cls) -> None: super().__init_subclass__() cls.HOOK_GROUPS = cls.HOOK_GROUPS.union(cls.CHECKED_METADATA).union('metadata') def _check_metadata(self, txn: Transaction, post: data.Posting, keys: Sequence[MetaKey], ) -> Iterator[errormod.InvalidMetadataError]: have_docs = False for key in keys: try: links = post.meta.get_links(key) except TypeError as error: yield errormod.InvalidMetadataError(txn, key, post.meta[key], post) else: have_docs = have_docs or any(links) if not have_docs: yield errormod.InvalidMetadataError(txn, '/'.join(keys), None, post) def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter: return self._check_metadata(txn, post, self.CHECKED_METADATA)