def _primary_rt_id(self, posts: AccrualPostings) -> rtutil.TicketAttachmentIds:

        rt_ids = {url for url in posts.first_links('rt-id') if url is not None}

        rt_ids_count = len(rt_ids)

        if rt_ids_count != 1:

            raise ValueError(f"{rt_ids_count} rt-id links found")

        parsed = rtutil.RT.parse(rt_ids.pop())

        if parsed is None:

            raise ValueError("rt-id is not a valid RT reference")

        else:

            return parsed

    def _report(self, posts: AccrualPostings, index: int) -> Iterable[str]:

        posts = posts.since_last_nonzero()

        try:

            ticket_id, _ = self._primary_rt_id(posts)

            ticket = self.rt_client.get_ticket(ticket_id)

            # Note we only use this when ticket is None.

            errmsg = f"ticket {ticket_id} not found"

        except (ValueError, rt.RtError) as error:

            ticket = None

            errmsg = error.args[0]

        if ticket is None:

            self.logger.error(

                "can't generate outgoings report for %s because no RT ticket available: %s",

                posts.invoice, errmsg,

            )

            return

        try:

            rt_requestor = self.rt_client.get_user(ticket['Requestors'][0])

        except (IndexError, rt.RtError):

            rt_requestor = None

        if rt_requestor is None:

            requestor = ''

            requestor_name = ''

        else:

            requestor_name = (

                rt_requestor.get('RealName')

                or ticket.get('CF.{payment-to}')

                or ''

            )

            requestor = f'{requestor_name} <{rt_requestor["EmailAddress"]}>'.strip()

        balance_s = posts.end_balance.format(None)

        raw_balance = -posts.balance()

        if raw_balance != posts.end_balance:

            balance_s = f'{raw_balance} ({balance_s})'

        if contract_links:

            contract_s = ' , '.join(self.rt_wrapper.iter_urls(

                contract_links, missing_fmt='<BROKEN RT LINK: {}>',

            ))

        else:

            contract_s = "NO CONTRACT GOVERNS THIS TRANSACTION"

        projects = [v for v in posts.meta_values('project')

                    if isinstance(v, str)]

        yield "PAYMENT FOR APPROVAL:"

        yield f"REQUESTOR: {requestor}"

        yield f"TOTAL TO PAY: {balance_s}"

        yield f"AGREEMENT: {contract_s}"

        yield f"PAYMENT TO: {ticket.get('CF.{payment-to}') or requestor_name}"

        yield f"PAYMENT METHOD: {ticket.get('CF.{payment-method}', '')}"

        yield f"PROJECT: {', '.join(projects)}"

        yield "\nBEANCOUNT ENTRIES:\n"

        last_txn: Optional[Transaction] = None

        for post in posts:

            txn = post.meta.txn

            if txn is not last_txn:

                last_txn = txn

                txn = self.rt_wrapper.txn_with_urls(txn, '{}')

                yield bc_printer.format_entry(txn)

class ReportType(enum.Enum):

    AGING = AgingReport

    BALANCE = BalanceReport

    OUTGOING = OutgoingReport

    AGE = AGING

    BAL = BALANCE

    OUT = OUTGOING

    OUTGOINGS = OUTGOING

    @classmethod

    def by_name(cls, name: str) -> 'ReportType':

        try:

            return cls[name.upper()]

        except KeyError:

            raise ValueError(f"unknown report type {name!r}") from None

    @classmethod

    def default_for(cls, groups: PostGroups) -> 'ReportType':

        if len(groups) == 1 and all(

                group.accrual_type is AccrualAccount.PAYABLE

                and not group.is_paid()

"""core.py - Common data classes for reporting functionality"""

# Copyright © 2020  Brett Smith

#

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

import collections

import datetime

import itertools

import operator

import re

import babel.core  # type:ignore[import]

import babel.numbers  # type:ignore[import]

import odf.config  # type:ignore[import]

import odf.element  # type:ignore[import]

import odf.number  # type:ignore[import]

import odf.opendocument  # type:ignore[import]

import odf.style  # type:ignore[import]

import odf.table  # type:ignore[import]

import odf.text  # type:ignore[import]

from decimal import Decimal

from pathlib import Path

from beancount.core import amount as bc_amount

from .. import data

from typing import (

    cast,

    overload,

    Any,

    BinaryIO,

    Callable,

    DefaultDict,

    Dict,

    Generic,

    Iterable,

    Iterator,

    List,

    Mapping,

    MutableMapping,

    Optional,

    Sequence,

    Set,

    Tuple,

    Type,

    TypeVar,

    Union,

)

from ..beancount_types import (

    MetaKey,

    MetaValue,

)

DecimalCompat = data.DecimalCompat

BalanceType = TypeVar('BalanceType', bound='Balance')

ElementType = Callable[..., odf.element.Element]

LinkType = Union[str, Tuple[str, Optional[str]]]

RelatedType = TypeVar('RelatedType', bound='RelatedPostings')

RT = TypeVar('RT', bound=Sequence)

ST = TypeVar('ST')

T = TypeVar('T')

class Balance(Mapping[str, data.Amount]):

    """A collection of amounts mapped by currency

    Each key is a Beancount currency string, and each value represents the

    balance in that currency.

    """

    __slots__ = ('_currency_map', 'tolerance')

    TOLERANCE = Decimal('0.01')

    def __init__(self,

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

                 *,

                 _can_own: bool=False,

    ) -> None:

        self._postings: List[data.Posting]

        if _can_own and isinstance(source, list):

            self._postings = source

        else:

            self._postings = list(source)

    @classmethod

    def group_by_meta(cls: Type[RelatedType],

                      postings: Iterable[data.Posting],

                      key: MetaKey,

                      default: Optional[MetaValue]=None,

    ) -> Iterator[Tuple[Optional[MetaValue], RelatedType]]:

        """Relate postings by metadata value

        This method takes an iterable of postings and returns a mapping.

        The keys of the mapping are the values of post.meta.get(key, default).

        The values are RelatedPostings instances that contain all the postings

        that had that same metadata value.

        """

        mapping: DefaultDict[Optional[MetaValue], List[data.Posting]] = collections.defaultdict(list)

        for post in postings:

            mapping[post.meta.get(key, default)].append(post)

        for value, posts in mapping.items():

            yield value, cls(posts, _can_own=True)

    def __repr__(self) -> str:

        return f'<{type(self).__name__} {self._postings!r}>'

    @overload

    def __getitem__(self: RelatedType, index: int) -> data.Posting: ...

    @overload

    def __getitem__(self: RelatedType, s: slice) -> RelatedType: ...

    def __getitem__(self: RelatedType,

                    index: Union[int, slice],

    ) -> Union[data.Posting, RelatedType]:

        if isinstance(index, slice):

            return type(self)(self._postings[index], _can_own=True)

        else:

            return self._postings[index]

    def __len__(self) -> int:

        return len(self._postings)

        for post in self:

            try:

            except TypeError:

                pass

    def iter_with_balance(self) -> Iterator[Tuple[data.Posting, Balance]]:

        balance = MutableBalance()

        for post in self:

            balance += post.units

            yield post, balance

    def balance(self) -> Balance:

        for _, balance in self.iter_with_balance():

            pass

        try:

            return balance

        except NameError:

            return Balance()

    def balance_at_cost(self) -> Balance:

        balance = MutableBalance()

        for post in self:

            if post.cost is None:

                balance += post.units

            else:

                number = post.units.number * post.cost.number

                balance += data.Amount(number, post.cost.currency)

        return balance

    def meta_values(self,

                    key: MetaKey,

                    default: Optional[MetaValue]=None,

    ) -> Set[Optional[MetaValue]]:

        return {post.meta.get(key, default) for post in self}

class BaseSpreadsheet(Generic[RT, ST], metaclass=abc.ABCMeta):

    """Abstract base class to help write spreadsheets

    This class provides the very core logic to write an arbitrary set of data

    rows to arbitrary output. It calls hooks when it starts writing the

    spreadsheet, starts a new "section" of rows, ends a section, and ends the

    spreadsheet.

    RT is the type of the input data rows. ST is the type of the section

    identifier that you create from each row. If you don't want to use the

    section logic at all, set ST to None and define section_key to return None.

    """

    @abc.abstractmethod

    def section_key(self, row: RT) -> ST:

        """Return the section a row belongs to

        testutil.Posting('Income:Donations', -2, metakey='2'),

    ])

    assert related.meta_values('key', '') == {'1', ''}

def test_meta_values_all_match():

    related = core.RelatedPostings([

        testutil.Posting('Income:Donations', -1, key='1'),

        testutil.Posting('Income:Donations', -2, key='2'),

    ])

    assert related.meta_values('key') == {'1', '2'}

def test_meta_values_all_match_one_value():

    related = core.RelatedPostings([

        testutil.Posting('Income:Donations', -1, key='1'),

        testutil.Posting('Income:Donations', -2, key='1'),

    ])

    assert related.meta_values('key') == {'1'}

def test_meta_values_all_match_default_given():

    related = core.RelatedPostings([

        testutil.Posting('Income:Donations', -1, key='1'),

        testutil.Posting('Income:Donations', -2, key='2'),

    ])

    assert related.meta_values('key', '') == {'1', '2'}

def test_meta_values_many_types():

    expected = {

        datetime.date(2020, 4, 1),

        Decimal(42),

        testutil.Amount(5),

        'rt:42',

    }

    related = core.RelatedPostings(

        testutil.Posting('Income:Donations', -index, key=value)

        for index, value in enumerate(expected)

    )

    assert related.meta_values('key') == expected

@pytest.mark.parametrize('count', range(3))

def test_all_meta_links_zero(count):

    postings = (

        testutil.Posting('Income:Donations', -n, testkey=str(n))

        for n in range(count)

    )

    related = core.RelatedPostings(

        post._replace(meta=data.Metadata(post.meta))

        for post in postings

    )

def test_all_meta_links_singletons():

    postings = (

        testutil.Posting('Income:Donations', -10, statement=value)

        for value in itertools.chain(

            testutil.NON_LINK_METADATA_STRINGS,

            testutil.LINK_METADATA_STRINGS,

            testutil.NON_STRING_METADATA_VALUES,

        ))

    related = core.RelatedPostings(

        post._replace(meta=data.Metadata(post.meta))

        for post in postings

    )

def test_all_meta_links_multiples():

    postings = (

        testutil.Posting('Income:Donations', -10, approval=' '.join(value))

        for value in itertools.permutations(testutil.LINK_METADATA_STRINGS, 2)

    )

    related = core.RelatedPostings(

        post._replace(meta=data.Metadata(post.meta))

        for post in postings

    )

def test_group_by_meta_zero():

    assert not list(core.RelatedPostings.group_by_meta([], 'metacurrency'))

def test_group_by_meta_one(credit_card_cycle):

    posting = next(post for post in data.Posting.from_entries(credit_card_cycle)

                   if post.account.is_credit_card())

    actual = core.RelatedPostings.group_by_meta([posting], 'metacurrency')

    assert set(key for key, _ in actual) == {'USD'}

def test_group_by_meta_many(two_accruals_three_payments):

    postings = [post for post in data.Posting.from_entries(two_accruals_three_payments)

                if post.account == 'Assets:Receivable:Accounts']

    actual = dict(core.RelatedPostings.group_by_meta(postings, 'metacurrency'))

    assert set(actual) == {'USD', 'EUR'}

    for key, group in actual.items():

        assert 2 <= len(group) <= 3

        assert group.balance().is_zero()

def test_group_by_meta_many_single_posts(two_accruals_three_payments):

    postings = [post for post in data.Posting.from_entries(two_accruals_three_payments)

                if post.account == 'Assets:Receivable:Accounts']

    actual = dict(core.RelatedPostings.group_by_meta(postings, 'metanumber'))

    assert set(actual) == {post.units.number for post in postings}

    assert len(actual) == len(postings)