#

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

    Iterable,

    Optional,

    Pattern,

    Union,

)

from .beancount_types import (

    Directive,

    Entries,

    MetaKey,

    MetaValue,

    Transaction,

)

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 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:

        self.account = self._single_item(post.account for post in self)

        if isinstance(self.account, Sentinel):

            self.accrual_type: Optional[AccrualAccount] = None

            norm_func: Callable[[T], T] = lambda x: x

            entity_pred: Callable[[data.Posting], bool] = bool

        else:

            self.accrual_type = AccrualAccount.by_account(self.account)

            norm_func = self.accrual_type.normalize_amount

            entity_pred = lambda post: norm_func(post.units).number > 0

        self.entity = self._single_item(self.entities(entity_pred))

        self.invoice = self._single_item(self.first_links('invoice'))

        self.end_balance = norm_func(self.balance_at_cost())

    def _single_item(self, seq: Iterable[T]) -> Union[T, Sentinel]:

        items = iter(seq)

        try:

            item1 = next(items)

        except StopIteration:

            all_same = False

        else:

            all_same = all(item == item1 for item in items)

        return item1 if all_same else self.INCONSISTENT

    def entities(self, pred: Callable[[data.Posting], bool]=bool) -> Iterator[MetaValue]:

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

        return (post.meta.first_link(key, default) for post in self)

    def make_consistent(self) -> Iterator[Tuple[MetaValue, 'AccrualPostings']]:

        account_ok = isinstance(self.account, str)

        entity_ok = isinstance(self.entity, str)

        # `'/' in self.invoice` is just our heuristic to ensure that the

        # invoice metadata is "unique enough," and not just a placeholder

        # value like "FIXME". It can be refined if needed.

        invoice_ok = isinstance(self.invoice, str) and '/' in self.invoice

        if account_ok and entity_ok and invoice_ok:

            yield (self.invoice, self)

            return

        groups = collections.defaultdict(list)

        for post in self:

            post_invoice = self.invoice if invoice_ok else (

                post.meta.get('invoice') or 'BlankInvoice'

            )

            post_entity = self.entity if entity_ok else (

                post.meta.get('entity') or 'BlankEntity'

            )

            groups[f'{post.account} {post_invoice} {post_entity}'].append(post)

        type_self = type(self)

        for post in getattr(entry, 'postings', ())

    )

@pytest.mark.parametrize('entry', [

    bc_data.Custom({}, testutil.FY_START_DATE, 'conservancy_beancount_audit', []),

    None,

])

def test_audit_date(entry):

    dates = testutil.date_seq()

    entries = [

        bc_data.Open({}, next(dates), 'Income:Donations', ['USD'], None),

        bc_data.Open({}, next(dates), 'Assets:Cash', ['USD'], None),

        testutil.Transaction(postings=[

            ('Income:Donations', -10),

            ('Assets:Cash', 10),

        ]),

    ]

    if entry is not None:

        entries.append(entry)

    actual = filters.audit_date(entries)

    if entry is None:

        assert actual is None

    else:

        assert actual == entry.date