"""Enhanced Beancount data structures for Conservancy

The classes in this module are interface-compatible with Beancount's core data

structures, and provide additional business logic that we want to use

throughout Conservancy tools.

"""

# 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 collections

import datetime

import decimal

import functools

import re

from beancount.core import account as bc_account

from beancount.core import amount as bc_amount

from beancount.core import convert as bc_convert

from beancount.core import data as bc_data

from beancount.core import position as bc_position

from beancount.parser import options as bc_options

from typing import (

    cast,

    overload,

    Callable,

    Hashable,

    Iterable,

    Iterator,

    MutableMapping,

    Optional,

    Pattern,

    Sequence,

    TypeVar,

    Union,

)

from .beancount_types import (

    Close,

    Currency,

    Directive,

    Meta,

    MetaKey,

    MetaValue,

    Open,

    OptionsMap,

    Posting as BasePosting,

    Transaction,

)

DecimalCompat = Union[decimal.Decimal, int]

EQUITY_ACCOUNTS = frozenset([

    'Equity',

    'Expenses',

    'Income',

])

FUND_ACCOUNTS = EQUITY_ACCOUNTS | frozenset([

    'Assets:Prepaid',

    'Assets:Receivable',

    'Liabilities:Payable',

    'Liabilities:UnearnedIncome',

])

LINK_METADATA = frozenset([

    'approval',

    'bank-statement',

    'check',

    'contract',

    'invoice',

    'purchase-order',

    'receipt',

    'rt-id',

    'statement',

    'tax-statement',

])

class AccountMeta(MutableMapping[MetaKey, MetaValue]):

    """Access account metadata

    This class provides a consistent interface to all the metadata provided by

    Beancount's ``open`` and ``close`` directives: open and close dates,

    used currencies, booking method, and the metadata associated with each.

    For convenience, you can use this class as a Mapping to access the ``open``

    directive's metadata directly.

    """

    __slots__ = ('_opening', '_closing')

    def __init__(self, opening: Open, closing: Optional[Close]=None) -> None:

        self._opening = opening

        self._closing: Optional[Close] = None

        if closing is not None:

            self.add_closing(closing)

    def add_closing(self, closing: Close) -> None:

        if self._closing is not None and self._closing is not closing:

            raise ValueError(f"{self.account} already closed by {self._closing!r}")

        elif closing.account != self.account:

            raise ValueError(f"cannot close {self.account} with {closing.account}")

        elif closing.date < self.open_date:

            raise ValueError(f"close date {closing.date} predates open date {self.open_date}")

        else:

            self._closing = closing

    def __iter__(self) -> Iterator[MetaKey]:

        return iter(self._opening.meta)

    def __len__(self) -> int:

        return len(self._opening.meta)

    def __getitem__(self, key: MetaKey) -> MetaValue:

        return self._opening.meta[key]

    def __setitem__(self, key: MetaKey, value: MetaValue) -> None:

        self._opening.meta[key] = value

    def __delitem__(self, key: MetaKey) -> None:

        del self._opening.meta[key]

    @property

    def account(self) -> 'Account':

        return Account(self._opening.account)

    @property

    def booking(self) -> Optional[bc_data.Booking]:

        return self._opening.booking

    @property

    def close_date(self) -> Optional[datetime.date]:

        return None if self._closing is None else self._closing.date

    @property

    def close_meta(self) -> Optional[Meta]:

        return None if self._closing is None else self._closing.meta

    @property

    def currencies(self) -> Optional[Sequence[Currency]]:

        return self._opening.currencies

    @property

    def open_date(self) -> datetime.date:

        return self._opening.date

    @property

    def open_meta(self) -> Meta:

        return self._opening.meta

class Account(str):

    """Account name string

    This is a string that names an account, like Assets:Bank:Checking

    or Income:Donations. This class provides additional methods for common

    account name parsing and queries.

    """

    __slots__ = ()

    ACCOUNT_RE: Pattern

    SEP = bc_account.sep

    _meta_map: MutableMapping[str, AccountMeta] = {}

    _options_map: OptionsMap

    @classmethod

    def load_options_map(cls, options_map: OptionsMap) -> None:

        cls._options_map = options_map

        roots: Sequence[str] = bc_options.get_account_types(options_map)

        cls.ACCOUNT_RE = re.compile(

            r'^(?:{})(?:{}[A-Z0-9][-A-Za-z0-9]*)+$'.format(

                '|'.join(roots), cls.SEP,

            ))

    @classmethod

    def load_opening(cls, opening: Open) -> None:

        cls._meta_map[opening.account] = AccountMeta(opening)

    @classmethod

    def load_closing(cls, closing: Close) -> None:

        try:

            cls._meta_map[closing.account].add_closing(closing)

        except KeyError:

            raise ValueError(

                f"tried to load {closing.account} close directive before open",

            ) from None

    @classmethod

    def load_openings_and_closings(cls, entries: Iterable[Directive]) -> None:

        for entry in entries:

            # type ignores because Beancount's directives aren't type-checkable.

            if isinstance(entry, bc_data.Open):

                cls.load_opening(entry)  # type:ignore[arg-type]

            elif isinstance(entry, bc_data.Close):

                cls.load_closing(entry)  # type:ignore[arg-type]

    @classmethod

    def load_from_books(cls, entries: Iterable[Directive], options_map: OptionsMap) -> None:

        cls.load_options_map(options_map)

        cls.load_openings_and_closings(entries)

    @classmethod

    def is_account(cls, s: str) -> bool:

        return cls.ACCOUNT_RE.fullmatch(s) is not None

    @classmethod

    def iter_accounts_by_classification(cls, s: str) -> Iterator['Account']:

        class_re = re.compile(f'^{re.escape(s)}(?::|$)')

        for name, meta in cls._meta_map.items():

            try:

                match = class_re.match(meta['classification'])

            except KeyError:

                match = None

            if match:

                yield cls(name)

    @classmethod

    def iter_accounts_by_hierarchy(cls, s: str) -> Iterator['Account']:

        for name in cls._meta_map:

            account = cls(name)

            if account.is_under(s):

                yield account

    @classmethod

    def iter_accounts(cls, s: Optional[str]=None) -> Iterator['Account']:

        """Iterate account objects by name or classification

        With no argument, returns an iterator of all known account names.

        If you pass in a root account name, or a valid account string, returns

        an iterator of all accounts under that account in the hierarchy.

        Otherwise, returns an iterator of all accounts with the given

        ``classification`` metadata.

        """

        if s is None:

            return (cls(acct) for acct in cls._meta_map)

        # We append a stub subaccount to match root accounts.

        elif cls.is_account(f'{s}:RootsOK'):

            return cls.iter_accounts_by_hierarchy(s)

        else:

            return cls.iter_accounts_by_classification(s)

    @property

    def meta(self) -> AccountMeta:

        return self._meta_map[self]

    def is_cash_equivalent(self) -> bool:

        return (

            self.is_under('Assets:') is not None

            and self.is_under('Assets:Prepaid', 'Assets:Receivable') is None

        )

    def is_checking(self) -> bool:

        return self.is_cash_equivalent() and ':Check' in self

    def is_credit_card(self) -> bool:

        return self.is_under('Liabilities:CreditCard') is not None

    def is_open_on_date(self, date: datetime.date) -> Optional[bool]:

        """Return true if this account is open on the given date.

        This method considers the dates on the account's open and close

        directives. If there is no close directive, it just checks the date is

        on or after the opening date. If neither exists, returns None.

        """

        try:

            meta = self.meta

        except KeyError:

            return None

        close_date = meta.close_date

        if close_date is None:

            close_date = date + datetime.timedelta(days=1)

        return meta.open_date <= date < close_date

    def is_opening_equity(self) -> bool:

        return self.is_under('Equity:Funds', 'Equity:OpeningBalance') is not None

    def is_under(self, *acct_seq: str) -> Optional[str]:

        """Return a match if this account is "under" a part of the hierarchy

        Pass in any number of account name strings as arguments. If this

        account is under one of those strings in the account hierarchy, the

        first matching string will be returned. Otherwise, None is returned.

        You can use the return value of this method as a boolean if you don't

        care which account string is matched.

        An account is considered to be under itself:

          Account('Expenses:Tax').is_under('Expenses:Tax') # returns 'Expenses:Tax'

        To do a "strictly under" search, end your search strings with colons:

          Account('Expenses:Tax').is_under('Expenses:Tax:') # returns None

          Account('Expenses:Tax').is_under('Expenses:') # returns 'Expenses:'

        This method does check that all the account boundaries match:

          Account('Expenses:Tax').is_under('Exp') # returns None

        """

        for prefix in acct_seq:

            if self.startswith(prefix) and (

                prefix.endswith(self.SEP)

                or self == prefix

                or self[len(prefix)] == self.SEP

            ):

                return prefix

        return None

    def keeps_balance(self) -> bool:

        return self.is_under(

            self._options_map['name_assets'],

            self._options_map['name_liabilities'],

        ) is not None

    def _find_part_slice(self, index: int) -> slice:

        if index < 0:

            raise ValueError(f"bad part index {index!r}")

        start = 0

        for _ in range(index):

            try:

                start = self.index(self.SEP, start) + 1

            except ValueError:

                raise IndexError("part index {index!r} out of range") from None

        try:

            stop = self.index(self.SEP, start + 1)

        except ValueError:

            stop = len(self)

        return slice(start, stop)

    def count_parts(self) -> int:

        return self.count(self.SEP) + 1

    @overload

    def slice_parts(self, start: None=None, stop: None=None) -> Sequence[str]: ...

    @overload

    def slice_parts(self, start: slice, stop: None=None) -> Sequence[str]: ...

    @overload

    def slice_parts(self, start: int, stop: int) -> Sequence[str]: ...

    @overload

    def slice_parts(self, start: int, stop: None=None) -> str: ...

    def slice_parts(self,

                    start: Optional[Union[int, slice]]=None,

                    stop: Optional[int]=None,

    ) -> Sequence[str]:

        """Slice the account parts like they were a list

        Given a single index, return that part of the account name as a string.

        Otherwise, return a list of part names sliced according to the arguments.

        """

        if start is None:

            part_slice = slice(None)

        elif isinstance(start, slice):

            part_slice = start

        elif stop is None:

            return self[self._find_part_slice(start)]

        else:

            part_slice = slice(start, stop)

        return self.split(self.SEP)[part_slice]

    def root_part(self, count: int=1) -> str:

        """Return the first part(s) of the account name as a string"""

        try:

            stop = self._find_part_slice(count - 1).stop

        except IndexError:

            return self

        else:

            return self[:stop]

Account.load_options_map(bc_options.OPTIONS_DEFAULTS)

class Amount(bc_amount.Amount):

    """Beancount amount after processing

    Beancount's native Amount class declares number to be Optional[Decimal],

    because the number is None when Beancount first parses a posting that does

    not have an amount, because the user wants it to be automatically balanced.

    As part of the loading process, Beancount replaces those None numbers

    with the calculated amount, so it will always be a Decimal. This class

    overrides the type declaration accordingly, so the type checker knows

    that our code doesn't have to consider the possibility that number is

    None.

    """

    number: decimal.Decimal

    # beancount.core._Amount is the plain namedtuple.

    # beancore.core.Amount adds instance methods to it.

    # b.c.Amount.__New__ calls `b.c._Amount.__new__`, which confuses type

    # checking. See <https://github.com/python/mypy/issues/1279>.

    # It works fine if you use super(), which is better practice anyway.

    # So we override __new__ just to call _Amount.__new__ this way.

    def __new__(cls, number: decimal.Decimal, currency: str) -> 'Amount':

        return super(bc_amount.Amount, Amount).__new__(cls, number, currency)

class Metadata(MutableMapping[MetaKey, MetaValue]):

    """Transaction or posting metadata

    This class wraps a Beancount metadata dictionary with additional methods

    for common parsing and query tasks.

    """

    __slots__ = ('meta',)

    _HUMAN_NAMES: MutableMapping[MetaKey, str] = {

        # Initialize this dict with special cases.

        # We use it as a cache for other metadata names as they're queried.

        'check-id': 'Check Number',

        'paypal-id': 'PayPal ID',

        'rt-id': 'Ticket',

    }

    def __init__(self, source: MutableMapping[MetaKey, MetaValue]) -> None:

        self.meta = source

    def __iter__(self) -> Iterator[MetaKey]:

        return iter(self.meta)

    def __len__(self) -> int:

        return len(self.meta)

    def __getitem__(self, key: MetaKey) -> MetaValue:

        return self.meta[key]

    def __setitem__(self, key: MetaKey, value: MetaValue) -> None:

        self.meta[key] = value

    def __delitem__(self, key: MetaKey) -> None:

        del self.meta[key]

    def get_links(self, key: MetaKey) -> Sequence[str]:

        try:

            value = self.meta[key]

        except KeyError:

            return ()

        if isinstance(value, str):

            return value.split()

        else:

            raise TypeError("{} metadata is a {}, not str".format(

                key, type(value).__name__,

            ))

    def report_links(self, key: MetaKey) -> Sequence[str]:

        """Return a sequence of link strings under the named metadata key

        get_links raises a TypeError if the metadata is not a string.

        This method simply returns the empty sequence.

        Validation code (like in the plugin) usually uses get_links()

        while reporting code uses report_links().

        """

        try:

            return self.get_links(key)

        except TypeError:

            return ()

    @overload

    def first_link(self, key: MetaKey, default: None=None) -> Optional[str]: ...

"""Beancount plugin entry point for Conservancy"""

# 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 importlib

import logging

import beancount.core.data as bc_data

from typing import (

    AbstractSet,

    Any,

    Dict,

    Iterator,

    List,

    Optional,

    Set,

    Tuple,

    Type,

)

from ..beancount_types import (

    ALL_DIRECTIVES,

    Directive,

    Entries,

    Errors,

    OptionsMap,

)

from .. import config as configmod

from .core import (

    Hook,

    HookName,

)

from ..errors import (

    ConfigurationError,

    Error,

)

__plugins__ = ['run']

logger = logging.getLogger('conservancy_beancount.plugin')

class HookRegistry:

    INCLUDED_HOOKS: Dict[str, Optional[List[str]]] = {

        '.meta_approval': None,

        '.meta_entity': None,

        '.meta_expense_type': None,

        '.meta_income_type': None,

        '.meta_invoice': None,

        # Enforcing this hook would be premature as of May 2020.  --brett

        # '.meta_payable_documentation': None,

        '.meta_paypal_id': ['MetaPayPalID'],

        '.meta_payroll_type': [

            'HealthInsuranceHook',

            'OtherBenefitsHook',

            'SalaryHook',

            'TaxHook',

        ],

        '.txn_date': ['TransactionDate'],

    }

    def __init__(self) -> None:

        self.group_name_map: Dict[HookName, Set[Type[Hook]]] = {

            t.__name__: set() for t in ALL_DIRECTIVES

        }

        self.group_name_map['all'] = set()

    def add_hook(self, hook_cls: Type[Hook]) -> Type[Hook]:

        self.group_name_map['all'].add(hook_cls)

        self.group_name_map[hook_cls.DIRECTIVE.__name__].add(hook_cls)

        for key in hook_cls.HOOK_GROUPS:

            self.group_name_map.setdefault(key, set()).add(hook_cls)

        return hook_cls  # to allow use as a decorator

    def import_hooks(self,

                     mod_name: str,

                     *hook_names: str,

                     package: Optional[str]=None,

    ) -> None:

        if not hook_names:

            _, _, hook_name = mod_name.rpartition('.')

            hook_names = (hook_name.title().replace('_', ''),)

        module = importlib.import_module(mod_name, package)

        for hook_name in hook_names:

            self.add_hook(getattr(module, hook_name))

    def load_included_hooks(self) -> None:

        for mod_name, hook_names in self.INCLUDED_HOOKS.items():

            self.import_hooks(mod_name, *(hook_names or []), package=self.__module__)

    def group_by_directive(self, config_str: str='') -> Iterator[Tuple[HookName, Type[Hook]]]:

        config_str = config_str.strip()

        if not config_str:

            config_str = 'all'

        elif config_str.startswith('-'):

            config_str = 'all ' + config_str

        available_hooks: Set[Type[Hook]] = set()

        for token in config_str.split():

            if token.startswith('-'):

                update_available = available_hooks.difference_update

                key = token[1:]

            else:

                update_available = available_hooks.update

                key = token

            try:

                update_set = self.group_name_map[key]

            except KeyError:

                raise ValueError("configuration refers to unknown hooks {!r}".format(key)) from None

            else:

                update_available(update_set)

        for directive in ALL_DIRECTIVES:

            key = directive.__name__

            for hook in self.group_name_map[key] & available_hooks:

                yield key, hook

def run(

        entries: Entries,

        options_map: OptionsMap,

        config: str='',

        hook_registry: Optional[HookRegistry]=None,

) -> Tuple[Entries, Errors]:

    if hook_registry is None:

        hook_registry = HookRegistry()

        hook_registry.load_included_hooks()

    errors: Errors = []

    hooks: Dict[HookName, List[Hook]] = {

        # mypy thinks NamedTuples don't have __name__ but they do at runtime.

        t.__name__: [] for t in bc_data.ALL_DIRECTIVES  # type:ignore[attr-defined]

    }

    user_config = configmod.Config()

    try:

        user_config.load_file()

    except OSError as error:

        logger.debug("error reading configuration file %s: %s",

                     error.filename, error.strerror, exc_info=True)

        errors.append(ConfigurationError(

            f"error reading configuration file {error.filename}: {error.strerror}",

            source={'filename': error.filename},

        ))

    for key, hook_type in hook_registry.group_by_directive(config):

        try:

            hook = hook_type(user_config)

        except Error as error:

            errors.append(error)

        else:

            hooks[key].append(hook)

    for entry in entries:

        entry_type = type(entry).__name__

        for hook in hooks[entry_type]:

            errors.extend(hook.run(entry))

    return entries, errors

#!/usr/bin/env python3

from setuptools import setup

setup(

    name='conservancy_beancount',

    description="Plugin, library, and reports for reading Conservancy's books",

    author='Software Freedom Conservancy',

    author_email='info@sfconservancy.org',

    license='GNU AGPLv3+',

    install_requires=[

        'babel>=2.6',  # Debian:python3-babel

        'beancount>=2.2',  # Debian:beancount

        'GitPython>=2.0',  # Debian:python3-git

        # 1.4.1 crashes when trying to save some documents.

        'odfpy>=1.4.0,!=1.4.1',  # Debian:python3-odf

        'pdfminer.six>=20200101',

        'PyYAML>=3.0',  # Debian:python3-yaml

        'regex',  # Debian:python3-regex

        'rt>=2.0',

    ],

    setup_requires=[

        'pytest-mypy',

        'pytest-runner',  # Debian:python3-pytest-runner

    ],

    tests_require=[

        'mypy>=0.770',  # Debian:python3-mypy

        'pytest',  # Debian:python3-pytest

    ],

    packages=[

        'conservancy_beancount',

        'conservancy_beancount.pdfforms',

        'conservancy_beancount.pdfforms.extract',

        'conservancy_beancount.plugin',

        'conservancy_beancount.reconcile',

        'conservancy_beancount.reports',

        'conservancy_beancount.tools',

    ],

    entry_points={

        'console_scripts': [

            'accrual-report = conservancy_beancount.reports.accrual:entry_point',

            'assemble-audit-reports = conservancy_beancount.tools.audit_report:entry_point',

            'balance-sheet-report = conservancy_beancount.reports.balance_sheet:entry_point',

            'budget-report = conservancy_beancount.reports.budget:entry_point',

            'bean-sort = conservancy_beancount.tools.sort_entries:entry_point',

            'extract-odf-links = conservancy_beancount.tools.extract_odf_links:entry_point',

            'fund-report = conservancy_beancount.reports.fund:entry_point',

            'ledger-report = conservancy_beancount.reports.ledger:entry_point',

            'opening-balances = conservancy_beancount.tools.opening_balances:entry_point',

            'pdfform-extract = conservancy_beancount.pdfforms.extract:entry_point',

            'pdfform-extract-irs990scheduleA = conservancy_beancount.pdfforms.extract.irs990scheduleA:entry_point',

            'pdfform-fill = conservancy_beancount.pdfforms.fill:entry_point',

            'reconcile-paypal = conservancy_beancount.reconcile.paypal:entry_point',

            'reconcile-statement = conservancy_beancount.reconcile.statement:entry_point',

            'split-ods-links = conservancy_beancount.tools.split_ods_links:entry_point',

        ],

    },

)