128 lines
5.1 KiB
Python
128 lines
5.1 KiB
Python
"""meta_project - Validate project metadata"""
|
|
# 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.
|
|
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
import yaml.error
|
|
|
|
from . import core
|
|
from .. import config as configmod
|
|
from .. import data
|
|
from .. import errors as errormod
|
|
from ..beancount_types import (
|
|
MetaValueEnum,
|
|
Transaction,
|
|
)
|
|
|
|
from typing import (
|
|
Any,
|
|
Dict,
|
|
NoReturn,
|
|
Optional,
|
|
Set,
|
|
)
|
|
|
|
class MetaProject(core._NormalizePostingMetadataHook):
|
|
DEFAULT_PROJECT = 'Conservancy'
|
|
PROJECT_DATA_PATH = Path('Projects', 'project-data.yml')
|
|
VALUES_ENUM = core.MetadataEnum('project', {DEFAULT_PROJECT})
|
|
RESTRICTED_FUNDS_ACCT = 'Equity:Funds:Restricted'
|
|
|
|
def __init__(self, config: configmod.Config, source_path: Path=PROJECT_DATA_PATH) -> None:
|
|
repo_path = config.repository_path()
|
|
if repo_path is None:
|
|
self._config_error("no repository configured")
|
|
project_data_path = repo_path / source_path
|
|
source = {'filename': str(project_data_path)}
|
|
try:
|
|
with project_data_path.open() as yaml_file:
|
|
project_data: Dict[str, Dict[str, Any]] = yaml.safe_load(yaml_file)
|
|
names: Set[MetaValueEnum] = {self.DEFAULT_PROJECT}
|
|
aliases: Dict[MetaValueEnum, MetaValueEnum] = {}
|
|
for key, params in project_data.items():
|
|
name = params.get('accountName', key)
|
|
names.add(name)
|
|
human_name = params.get('humanName', name)
|
|
if name != human_name:
|
|
aliases[human_name] = name
|
|
if name != key:
|
|
aliases[key] = name
|
|
except AttributeError:
|
|
self._config_error("loaded YAML data not in project-data format", project_data_path)
|
|
except OSError as error:
|
|
self._config_error(error.strerror, project_data_path)
|
|
except yaml.error.YAMLError as error:
|
|
self._config_error(error.args[0] or "YAML load error", project_data_path)
|
|
else:
|
|
self.VALUES_ENUM = core.MetadataEnum(self.METADATA_KEY, names, aliases)
|
|
|
|
def _config_error(self, msg: str, filename: Optional[Path]=None) -> NoReturn:
|
|
source = {}
|
|
if filename is not None:
|
|
source['filename'] = str(filename)
|
|
raise errormod.ConfigurationError(
|
|
"cannot load project data: " + msg,
|
|
source=source,
|
|
)
|
|
|
|
def _run_on_opening_post(self, txn: Transaction, post: data.Posting) -> bool:
|
|
return post.account.is_under('Equity') is not None
|
|
|
|
def _run_on_other_post(self, txn: Transaction, post: data.Posting) -> bool:
|
|
if post.account.is_under('Liabilities'):
|
|
return not post.account.is_credit_card()
|
|
else:
|
|
return post.account.is_under(
|
|
'Assets:Receivable',
|
|
'Equity',
|
|
'Expenses',
|
|
'Income',
|
|
self.RESTRICTED_FUNDS_ACCT,
|
|
) is not None
|
|
|
|
def _default_value(self, txn: Transaction, post: data.Posting) -> MetaValueEnum:
|
|
if post.account.is_under(
|
|
'Expenses:Payroll',
|
|
'Liabilities:Payable:Vacation',
|
|
):
|
|
return self.DEFAULT_PROJECT
|
|
else:
|
|
raise errormod.InvalidMetadataError(txn, self.METADATA_KEY, None, post)
|
|
|
|
def _run_on_txn(self, txn: Transaction) -> bool:
|
|
return txn.date in self.TXN_DATE_RANGE
|
|
|
|
def post_run(self, txn: Transaction, post: data.Posting) -> errormod.Iter:
|
|
if (post.account.is_under('Equity')
|
|
and not post.account.is_under(self.RESTRICTED_FUNDS_ACCT)):
|
|
# Force all unrestricted Equity accounts to have the default
|
|
# project. This is what our fiscal controls policy says, and
|
|
# setting it here simplifies higher-level queries and reporting.
|
|
post_value = post.meta.get(self.METADATA_KEY)
|
|
txn_value = txn.meta.get(self.METADATA_KEY)
|
|
# Only report an error if the posting specifically had a different
|
|
# value, not if it just inherited it from the transaction.
|
|
if (post_value is not txn_value
|
|
and post_value != self.DEFAULT_PROJECT):
|
|
yield errormod.InvalidMetadataError(
|
|
txn, self.METADATA_KEY, post_value, post,
|
|
)
|
|
post.meta[self.METADATA_KEY] = self.DEFAULT_PROJECT
|
|
else:
|
|
yield from super().post_run(txn, post)
|
|
|
|
def run(self, txn: Transaction) -> errormod.Iter:
|
|
# mypy says we can't assign over a method.
|
|
# I understand why it wants to enforce thas as a blanket rule, but
|
|
# we're substituting in another type-compatible method, so it's pretty
|
|
# safe.
|
|
if data.is_opening_balance_txn(txn):
|
|
self._run_on_post = self._run_on_opening_post # type:ignore[assignment]
|
|
else:
|
|
self._run_on_post = self._run_on_other_post # type:ignore[assignment]
|
|
return super().run(txn)
|