import operator

import re

import sys

import urllib.parse as urlparse

from pathlib import Path

from typing import (

    cast,

    Any,

    BinaryIO,

    Callable,

    Dict,

    Iterable,

    Iterator,

    FrozenSet,

    List,

    Mapping,

    NamedTuple,

    Optional,

    Sequence,

    Set,

    TextIO,

    Tuple,

    Union,

)

from ..beancount_types import (

    Entries,

    Error,

    Errors,

    MetaKey,

    MetaValue,

    Transaction,

)

import odf.style  # type:ignore[import]

import odf.table  # type:ignore[import]

import rt

from beancount.parser import printer as bc_printer

from . import core

from .. import cliutil

from .. import config as configmod

from .. import data

from .. import filters

from .. import rtutil

PROGNAME = 'accrual-report'

STANDARD_PATH = Path('-')

PostGroups = Mapping[Optional[MetaValue], 'AccrualPostings']

RTObject = Mapping[str, str]

logger = logging.getLogger('conservancy_beancount.reports.accrual')

class Sentinel:

    pass

class Account(NamedTuple):

    name: str

    aging_thresholds: Sequence[int]

class AccrualAccount(enum.Enum):

    # Note the aging report uses the same order accounts are defined here.

    # See AgingODS.start_spreadsheet().

    RECEIVABLE = Account(

        'Assets:Receivable', lambda bal: bal, [365, 120, 90, 60],

    )

    PAYABLE = Account(

        'Liabilities:Payable', operator.neg, [365, 90, 60, 30],

    )

    @classmethod

    def account_names(cls) -> Iterator[str]:

        return (acct.value.name for acct in cls)

    @classmethod

    def by_account(cls, name: data.Account) -> 'AccrualAccount':

        for account in cls:

            if name.is_under(account.value.name):

                return account

        raise ValueError(f"unrecognized account {name!r}")

    @classmethod

    def classify(cls, related: core.RelatedPostings) -> 'AccrualAccount':

        for account in cls:

            account_name = account.value.name

            if all(post.account.is_under(account_name) for post in related):

                return account

        raise ValueError("unrecognized account set in related postings")

class AccrualPostings(core.RelatedPostings):

    def _meta_getter(key: MetaKey) -> Callable[[data.Posting], MetaValue]:  # type:ignore[misc]

        def meta_getter(post: data.Posting) -> MetaValue:

            return post.meta.get(key)

        return meta_getter

    _FIELDS: Dict[str, Callable[[data.Posting], MetaValue]] = {

        'account': operator.attrgetter('account'),

        'contract': _meta_getter('contract'),

        'invoice': _meta_getter('invoice'),

        'purchase_order': _meta_getter('purchase-order'),

    }

    INCONSISTENT = Sentinel()

    __slots__ = (

        'accrual_type',

        'end_balance',

        'account',

        'accounts',

        'contract',

        'contracts',

        'invoice',

        'invoices',

        'purchase_order',

        'purchase_orders',

    )

    def __init__(self,

                 source: Iterable[data.Posting]=(),

                 *,

                 _can_own: bool=False,

    ) -> None:

        super().__init__(source, _can_own=_can_own)

        # The following type declarations tell mypy about values set in the for

        # loop that are important enough to be referenced directly elsewhere.

        self.account: Union[data.Account, Sentinel]

        self.invoice: Union[MetaValue, Sentinel]

        for name, get_func in self._FIELDS.items():

            values = frozenset(get_func(post) for post in self)

            setattr(self, f'{name}s', values)

            if len(values) == 1:

                one_value = next(iter(values))

            else:

                one_value = self.INCONSISTENT

            setattr(self, name, one_value)

        if self.account is self.INCONSISTENT:

            self.accrual_type: Optional[AccrualAccount] = None

            self.end_balance = self.balance_at_cost()

        else:

            self.accrual_type = AccrualAccount.classify(self)

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

        account_ok = isinstance(self.account, 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 invoice_ok:

            yield (self.invoice, self)

            return

        groups = collections.defaultdict(list)

        for post in self:

            if invoice_ok:

                key = f'{self.invoice} {post.account}'

            else:

                key = f'{post.account} {post.meta.get("entity")} {post.meta.get("invoice")}'

            groups[key].append(post)

        type_self = type(self)

        for group_key, posts in groups.items():

            yield group_key, type_self(posts, _can_own=True)

    def report_inconsistencies(self) -> Iterable[Error]:

        for field_name, get_func in self._FIELDS.items():

            if getattr(self, field_name) is self.INCONSISTENT:

                href = f'../{href}'

            if text is None:

                href_path = Path(urlparse.urlparse(href).path)

                text = urlparse.unquote(href_path.name)

            yield (href, text)

    def write_row(self, row: AccrualPostings) -> None:

        age = (self.date - row[0].meta.date).days

        if row.end_balance.ge_zero():

            for index, threshold in enumerate(self.age_thresholds):

                if age >= threshold:

                    self.age_balances[index] += row.end_balance

                    break

            else:

                return

        raw_balance = row.balance()

        if row.accrual_type is not None:

            raw_balance = row.accrual_type.value.norm_func(raw_balance)

        if raw_balance == row.end_balance:

            amount_cell = odf.table.TableCell()

        else:

            amount_cell = self.balance_cell(raw_balance)

        self.add_row(

            self.date_cell(row[0].meta.date),

            amount_cell,

            self.balance_cell(row.end_balance),

            self.multilink_cell(self._link_seq(row, 'rt-id')),

            self.multilink_cell(self._link_seq(row, 'invoice')),

        )

class AgingReport(BaseReport):

    def __init__(self,

                 rt_client: rt.Rt,

                 out_file: BinaryIO,

                 date: Optional[datetime.date]=None,

    ) -> None:

        if date is None:

            date = datetime.date.today()

        self.out_bin = out_file

        self.logger = logger.getChild(type(self).__name__)

        self.ods = AgingODS(rt_client, date, self.logger)

    def run(self, groups: PostGroups) -> None:

        rows = list(group for group in groups.values() if not group.is_zero())

        rows.sort(key=lambda related: (

            related.account,

            related[0].meta.date,

        ))

        self.ods.write(rows)

        self.ods.save_file(self.out_bin)

class BalanceReport(BaseReport):

    def _report(self,

                invoice: str,

                posts: AccrualPostings,

                index: int,

    ) -> Iterable[str]:

        posts = posts.since_last_nonzero()

        date_s = posts[0].meta.date.strftime('%Y-%m-%d')

        if index:

            yield ""

        yield f"{invoice}:"

        yield f"  {posts.balance_at_cost()} outstanding since {date_s}"

class OutgoingReport(BaseReport):

    def __init__(self, rt_client: rt.Rt, out_file: TextIO) -> None:

        super().__init__(out_file)

        self.rt_client = rt_client

        self.rt_wrapper = rtutil.RT(rt_client)

2010-03-01 * "Opening balances"

  Equity:Funds:Opening  -1000 USD

  Assets:Receivable:Accounts  6000 USD

  Liabilities:Payable:Accounts  -5000 USD

2010-03-05 * "EarlyBird" "Payment for receivable from previous FY"

  rt-id: "rt:40"

  invoice: "rt:40/400"

  Assets:Receivable:Accounts  -500 USD

  Assets:Checking  500 USD

2010-03-06 * "EarlyBird" "Payment for payment from previous FY"

  rt-id: "rt:44"

  invoice: "rt:44/440"

  Liabilities:Payable:Accounts  125 USD

  Assets:Checking  -125 USD

2010-03-30 * "EarlyBird" "Travel reimbursement"

  rt-id: "rt:490"

  invoice: "rt:490/4900"

  Liabilities:Payable:Accounts  -75 USD

  Expenses:Travel  75 USD

2010-04-30 ! "Vendor" "Travel reimbursement"

  rt-id: "rt:310"

  contract: "rt:310/3100"

  invoice: "FIXME"  ; still waiting on them to send it

  Liabilities:Payable:Accounts  -200 USD

  Expenses:Travel  200 USD

2010-05-05 * "DonorA" "Donation pledge"

  rt-id: "rt:505"

  invoice: "rt:505/5050"

  approval: "rt:505/5040"

  Income:Donations  -2,500 EUR {1.100 USD}

  Assets:Receivable:Accounts  2,500 EUR {1.100 USD}

2010-05-10 * "Lawyer" "April legal services"

  rt-id: "rt:510"

  invoice: "rt:510/5100"

  contract: "rt:510/4000"

  Expenses:Services:Legal  200.00 USD

  Liabilities:Payable:Accounts  -200.00 USD

2010-05-15 * "MatchingProgram" "May matched donations"

  invoice: "rt://ticket/515/attachments/5150"

  approval: "rt://ticket/515/attachments/5140"

from conservancy_beancount.reports import core

_accruals_load = bc_loader.load_file(testutil.test_path('books/accruals.beancount'))

ACCRUAL_TXNS = [

    entry for entry in _accruals_load[0]

    if hasattr(entry, 'narration')

    and entry.narration != 'Opening balances'

]

ACCRUALS_COUNT = sum(

    1

    for txn in ACCRUAL_TXNS

    for post in txn.postings

    if post.account.startswith(('Assets:Receivable:', 'Liabilities:Payable:'))

)

ACCOUNTS = [

    'Assets:Receivable:Accounts',

    'Assets:Receivable:Loans',

    'Liabilities:Payable:Accounts',

    'Liabilities:Payable:Vacation',

]

CONSISTENT_METADATA = [

    'contract',

    'purchase-order',

]

class AgingRow(NamedTuple):

    date: datetime.date

    entity: Sequence[str]

    amount: Optional[Sequence[bc_data.Amount]]

    at_cost: bc_data.Amount

    rt_id: Sequence[str]

    invoice: Sequence[str]

    @classmethod

    def make_simple(cls, date, entity, at_cost, invoice, rt_id=None, orig_amount=None):

        if isinstance(date, str):

            date = datetime.datetime.strptime(date, '%Y-%m-%d').date()

        if not isinstance(at_cost, tuple):

            at_cost = testutil.Amount(at_cost)

        if rt_id is None:

            rt_id, _, _ = invoice.partition('/')

        return cls(date, [entity], orig_amount, at_cost, [rt_id], [invoice])

    def check_row_match(self, sheet_row):

        cells = testutil.ODSCell.from_row(sheet_row)

        assert len(cells) == len(self)

    ])

    related = accrual.AccrualPostings(data.Posting.from_txn(txn))

    assert related.account == acct_name

    assert related.accounts == {acct_name}

@pytest.mark.parametrize('meta_key,acct_name', testutil.combine_values(

    CONSISTENT_METADATA,

    ACCOUNTS,

))

def test_accrual_postings_consistent_metadata(meta_key, acct_name):

    meta_value = f'{meta_key}.pdf'

    meta = {

        meta_key: meta_value,

        'invoice': f'invoice with {meta_key}.pdf',

    }

    txn = testutil.Transaction(postings=[

        (acct_name, 70, meta),

        (acct_name, 35, meta),

    ])

    related = accrual.AccrualPostings(data.Posting.from_txn(txn))

    attr_name = meta_key.replace('-', '_')

    assert getattr(related, attr_name) == meta_value

    assert getattr(related, f'{attr_name}s') == {meta_value}

def test_accrual_postings_inconsistent_account():

    meta = {'invoice': 'invoice.pdf'}

    txn = testutil.Transaction(postings=[

        (acct_name, index, meta)

        for index, acct_name in enumerate(ACCOUNTS)

    ])

    related = accrual.AccrualPostings(data.Posting.from_txn(txn))

    assert related.account is related.INCONSISTENT

    assert related.accounts == set(ACCOUNTS)

@pytest.mark.parametrize('meta_key,acct_name', testutil.combine_values(

    CONSISTENT_METADATA,

    ACCOUNTS,

))

def test_accrual_postings_inconsistent_metadata(meta_key, acct_name):

    invoice = 'invoice with {meta_key}.pdf'

    meta_value = f'{meta_key}.pdf'

    txn = testutil.Transaction(postings=[

        (acct_name, 20, {'invoice': invoice, meta_key: meta_value}),

        (acct_name, 35, {'invoice': invoice}),

    ])

    related = accrual.AccrualPostings(data.Posting.from_txn(txn))

    attr_name = meta_key.replace('-', '_')

    assert getattr(related, attr_name) is related.INCONSISTENT

    assert getattr(related, f'{attr_name}s') == {meta_value, None}

@pytest.mark.parametrize('meta_key,account', testutil.combine_values(

    CONSISTENT_METADATA,

    ACCOUNTS,

))

def test_consistency_check_when_consistent(meta_key, account):

    invoice = f'test-{meta_key}-invoice'

    meta_value = f'test-{meta_key}-value'

    meta = {

        'invoice': invoice,

        meta_key: meta_value,

    }

    txn = testutil.Transaction(postings=[

        (account, 100, meta),

        (account, -100, meta),

    ])

    related = accrual.AccrualPostings(data.Posting.from_txn(txn))

    assert not list(related.report_inconsistencies())

@pytest.mark.parametrize('meta_key,account', testutil.combine_values(

    ACCOUNTS,

))

def test_consistency_check_ignored_metadata(meta_key, account):

    invoice = f'test-{meta_key}-invoice'

    txn = testutil.Transaction(postings=[

        (account, 100, {'invoice': invoice, meta_key: 'credit'}),

        (account, -100, {'invoice': invoice, meta_key: 'debit'}),

    ])

    related = accrual.AccrualPostings(data.Posting.from_txn(txn))

    assert not list(related.report_inconsistencies())

@pytest.mark.parametrize('meta_key,account', testutil.combine_values(

    CONSISTENT_METADATA,

    ACCOUNTS,

))

def test_consistency_check_when_inconsistent(meta_key, account):

    invoice = f'test-{meta_key}-invoice'

    txn = testutil.Transaction(postings=[

        (account, 100, {'invoice': invoice, meta_key: 'credit', 'lineno': 1}),

        (account, -100, {'invoice': invoice, meta_key: 'debit', 'lineno': 2}),

    ])

    related = accrual.AccrualPostings(data.Posting.from_txn(txn))

    errors = list(related.report_inconsistencies())

    for exp_lineno, (actual, exp_msg) in enumerate(itertools.zip_longest(errors, [