accruals.

If you run it with no arguments, it will generate an aging report in ODS format

in the current directory.

Otherwise, the typical way to run it is to pass an RT ticket number or

invoice link as an argument, to report about accruals that match those

criteria::

accrual-report will automatically decide what kind of report to generate

from the search terms you provide and the results they return. If you pass

no search terms, it generates an aging report. If your search terms match a

single outstanding payable, it writes an outgoing approval report.

Otherwise, it writes a basic balance report. You can specify what report

type you want with the ``--report-type`` option::

    # Write an aging report for a specific project

    accrual-report --report-type aging project=ProjectName

import urllib.parse as urlparse

from pathlib import Path

    cast,

    BinaryIO,

    Entries,

    Errors,

import odf.style  # type:ignore[import]

import odf.table  # type:ignore[import]

STANDARD_PATH = Path('-')

    aging_thresholds: Sequence[int]

    # 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 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}")

        'end_balance',

        self.entity: Union[MetaValue, Sentinel]

            self.end_balance = self.balance_at_cost()

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

            return self.end_balance.le_zero()

            return self.end_balance.is_zero()

class AgingODS(core.BaseODS[AccrualPostings, Optional[data.Account]]):

    COLUMNS = [

        'Date',

        'Entity',

        'Invoice Amount',

        'Booked Amount',

        'Ticket',

        'Invoice',

    ]

    COL_COUNT = len(COLUMNS)

    def __init__(self,

                 rt_client: rt.Rt,

                 date: datetime.date,

                 logger: logging.Logger,

    ) -> None:

        super().__init__()

        self.rt_client = rt_client

        self.rt_wrapper = rtutil.RT(self.rt_client)

        self.date = date

        self.logger = logger

    def init_styles(self) -> None:

        super().init_styles()

        self.style_widecol = self.replace_child(

            self.document.automaticstyles,

            odf.style.Style,

            name='WideCol',

        )

        self.style_widecol.setAttribute('family', 'table-column')

        self.style_widecol.addElement(odf.style.TableColumnProperties(

            columnwidth='1.25in',

        ))

    def section_key(self, row: AccrualPostings) -> Optional[data.Account]:

        if isinstance(row.account, str):

            return row.account

        else:

            return None

    def start_spreadsheet(self) -> None:

        for accrual_type in AccrualAccount:

            self.use_sheet(accrual_type.name.title())

            for index in range(self.COL_COUNT):

                stylename = self.style_widecol if index else ''

                self.sheet.addElement(odf.table.TableColumn(stylename=stylename))

            self.add_row(*(

                self.string_cell(name, stylename=self.style_bold)

                for name in self.COLUMNS

            ))

            self.lock_first_row()

    def start_section(self, key: Optional[data.Account]) -> None:

        if key is None:

            return

        self.age_thresholds = list(AccrualAccount.by_account(key).value.aging_thresholds)

        self.age_balances = [core.MutableBalance() for _ in self.age_thresholds]

        accrual_date = self.date - datetime.timedelta(days=self.age_thresholds[-1])

        acct_parts = key.split(':')

        self.use_sheet(acct_parts[1])

        self.add_row()

        self.add_row(self.string_cell(

            f"{' '.join(acct_parts[2:])} {acct_parts[1]} Aging Report"

            f" Accrued by {accrual_date.isoformat()} Unpaid by {self.date.isoformat()}",

            stylename=self.merge_styles(self.style_bold, self.style_centertext),

            numbercolumnsspanned=self.COL_COUNT,

        ))

        self.add_row()

    def end_section(self, key: Optional[data.Account]) -> None:

        if key is None:

            return

        self.add_row()

        text_style = self.merge_styles(self.style_bold, self.style_endtext)

        text_span = self.COL_COUNT - 1

        for threshold, balance in zip(self.age_thresholds, self.age_balances):

            years, days = divmod(threshold, 365)

            years_text = f"{years} {'Year' if years == 1 else 'Years'}"

            days_text = f"{days} Days"

            if years and days:

                age_text = f"{years_text} {days_text}"

            elif years:

                age_text = years_text

            else:

                age_text = days_text

            self.add_row(

                self.string_cell(

                    f"Total Aged Over {age_text}: ",

                    stylename=text_style,

                    numbercolumnsspanned=text_span,

                ),

                *(odf.table.TableCell() for _ in range(1, text_span)),

                self.balance_cell(balance),

            )

    def _link_seq(self, row: AccrualPostings, key: MetaKey) -> Iterator[Tuple[str, str]]:

        for href in row.all_meta_links(key):

            text: Optional[str] = None

            rt_ids = self.rt_wrapper.parse(href)

            if rt_ids is not None:

                ticket_id, attachment_id = rt_ids

                if attachment_id is None:

                    text = f'RT#{ticket_id}'

                href = self.rt_wrapper.url(ticket_id, attachment_id) or href

            else:

                # '..' pops the ODS filename off the link path. In other words,

                # make the link relative to the directory the ODS is in.

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

            self.multiline_cell(row.entities),

            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,

            min(related.entities) if related.entities else '',

        ))

        self.ods.write(rows)

        self.ods.save_file(self.out_bin)

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

        balance_s = posts.end_balance.format(None)

        raw_balance = -posts.balance()

        if raw_balance != posts.end_balance:

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

    AGING = AgingReport

    AGE = AGING

        help="""The type of report to generate, one of `aging`, `balance`, or

`outgoing`. If not specified, the default is `aging` when no search terms are

given, `outgoing` for search terms that return a single outstanding payable,

and `balance` any other time.

""")

    parser.add_argument(

        '--output-file', '-O',

        metavar='PATH',

        type=Path,

        help="""Write the report to this file, or stdout when PATH is `-`.

The default is stdout for the balance and outgoing reports, and a generated

filename for other reports.

    if args.report_type is None and not args.search_terms:

        args.report_type = ReportType.AGING

def get_output_path(output_path: Optional[Path],

                    default_path: Path=STANDARD_PATH,

) -> Optional[Path]:

    if output_path is None:

        output_path = default_path

    if output_path == STANDARD_PATH:

        return None

    else:

        return output_path

def get_output_bin(path: Optional[Path], stdout: TextIO) -> BinaryIO:

    if path is None:

        return open(stdout.fileno(), 'wb')

    else:

        return path.open('wb')

def get_output_text(path: Optional[Path], stdout: TextIO) -> TextIO:

    if path is None:

        return stdout

    else:

        return path.open('w')

    if books_loader is None:

        entries: Entries = []

        load_errors: Errors = [Error(source, "no books to load in configuration", None)]

    elif args.report_type is ReportType.AGING:

        entries, load_errors, _ = books_loader.load_all()

    else:

        entries, load_errors, _ = books_loader.load_fy_range(args.since)

    returncode = 0

    groups = {

        key: posts

        for source_posts in groups.values()

        for key, posts in source_posts.make_consistent()

    }

    if args.report_type is not ReportType.AGING:

        groups = {

            key: posts for key, posts in groups.items() if not posts.is_paid()

        } or groups

    if args.report_type is None:

        args.report_type = ReportType.default_for(groups)

    output_path: Optional[Path] = None

    if args.report_type is ReportType.AGING:

        rt_client = config.rt_client()

        if rt_client is None:

            logger.error("unable to generate aging report: RT client is required")

        else:

            now = datetime.datetime.now()

            default_path = Path(now.strftime('AgingReport_%Y-%m-%d_%H:%M.ods'))

            output_path = get_output_path(args.output_file, default_path)

            out_bin = get_output_bin(output_path, stdout)

            report = AgingReport(rt_client, out_bin)

    elif args.report_type is ReportType.OUTGOING:

            output_path = get_output_path(args.output_file)

            out_file = get_output_text(output_path, stdout)

            report = OutgoingReport(rt_client, out_file)

        output_path = get_output_path(args.output_file)

        out_file = get_output_text(output_path, stdout)

        report = args.report_type.value(out_file)

        if args.output_file != output_path:

            logger.info("Report saved to %s", output_path)

    version='1.1.0',

2010-01-01 open Assets:Checking

2010-01-01 open Assets:Receivable:Accounts

2010-01-01 open Expenses:FilingFees

2010-01-01 open Expenses:Services:Legal

2010-01-01 open Expenses:Travel

2010-01-01 open Income:Donations

2010-01-01 open Liabilities:Payable:Accounts

2010-01-01 open Equity:Funds:Opening

2010-03-01 * "Opening balances"

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

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

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

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

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

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

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

2010-05-20 * "DonorA" "Donation made"

2010-05-25 * "Lawyer" "May payment"

2010-06-10 * "Lawyer" "May legal services"

2010-06-12 * "Lawyer" "Additional legal fees for May"

2010-06-18 * "EuroGov" "European legal fees"

import babel.numbers

import odf.opendocument

import odf.table

import odf.text

from decimal import Decimal

from typing import NamedTuple, Optional, Sequence

from beancount.core import data as bc_data

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)

        cells = iter(cells)

        assert next(cells).value == self.date

        assert next(cells).text == '\0'.join(self.entity)

        assert next(cells).text == '\0'.join(

            babel.numbers.format_currency(number, currency, format_type='accounting')

            for number, currency in self.amount or ()

        )

        usd_cell = next(cells)

        assert usd_cell.value_type == 'currency'

        assert usd_cell.value == self.at_cost.number

        for index, cell in enumerate(cells):

            links = cell.getElementsByType(odf.text.A)

            assert len(links) == len(cell.childNodes)

        assert index >= 1

AGING_AP = [

    AgingRow.make_simple('2010-03-06', 'EarlyBird', -125, 'rt:44/440'),

    AgingRow.make_simple('2010-03-30', 'EarlyBird', 75, 'rt:490/4900'),

    AgingRow.make_simple('2010-04-30', 'Vendor', 200, 'FIXME'),

    AgingRow.make_simple('2010-06-10', 'Lawyer', 280, 'rt:510/6100'),

    AgingRow.make_simple('2010-06-18', 'EuroGov', 1100, 'rt:520/5200',

                         orig_amount=[testutil.Amount(1000, 'EUR')]),

]

AGING_AR = [

    AgingRow.make_simple('2010-03-05', 'EarlyBird', -500, 'rt:40/400'),

    AgingRow.make_simple('2010-05-15', 'MatchingProgram', 1500,

                         'rt://ticket/515/attachments/5150'),

]

def find_row_by_text(row_source, want_text):

    for row in row_source:

        try:

            found_row = row.childNodes[0].text == want_text

        except IndexError:

            found_row = False

        if found_row:

            return row

    return None

def check_aging_sheet(sheet, aging_rows, date, accrue_date):

    if not aging_rows:

        return

    if isinstance(accrue_date, int):

        accrue_date = date + datetime.timedelta(days=accrue_date)

    rows = iter(sheet.getElementsByType(odf.table.TableRow))

    for row in rows:

        if "Aging Report" in row.text:

            break

    else:

        assert None, "Header row not found"

    assert f"Accrued by {accrue_date.isoformat()}" in row.text

    assert f"Unpaid by {date.isoformat()}" in row.text

    expect_rows = iter(aging_rows)

    row0 = find_row_by_text(rows, aging_rows[0].date.isoformat())

    next(expect_rows).check_row_match(row0)

    for actual, expected in zip(rows, expect_rows):

        expected.check_row_match(actual)

    for row in rows:

        if row.text.startswith("Total Aged Over "):

            break

    else:

        assert None, "Totals rows not found"

    actual_sum = Decimal(row.childNodes[-1].value)

    for row in rows:

        if row.text.startswith("Total Aged Over "):

            actual_sum += Decimal(row.childNodes[-1].value)

        else:

            break

    assert actual_sum == sum(

        row.at_cost.number

        for row in aging_rows

        if row.date <= accrue_date

        and row.at_cost.number > 0

    )

def check_aging_ods(ods_file,

                    date=None,

                    recv_rows=AGING_AR,

                    pay_rows=AGING_AP,

):

    if date is None:

        date = datetime.date.today()

    ods_file.seek(0)

    ods = odf.opendocument.load(ods_file)

    sheets = ods.spreadsheet.getElementsByType(odf.table.Table)

    assert len(sheets) == 2

    check_aging_sheet(sheets[0], recv_rows, date, -60)

    check_aging_sheet(sheets[1], pay_rows, date, -30)

    ('aging', accrual.AgingReport),

    ('age', accrual.AgingReport),

    ('rt:505/5050', "Zero balance outstanding since 2010-05-05"),

    ('rt:510/5100', "Zero balance outstanding since 2010-05-10"),

    ('rt:510/6100', "-280.00 USD outstanding since 2010-06-10"),

    ('rt://ticket/515/attachments/5150', "1,500.00 USD outstanding since 2010-05-15",),

        r'^\s*2010-06-10\s',

        r'^\s*2010-06-12\s',

def run_aging_report(postings, today=None):

    if today is None:

        today = datetime.date.today()

    postings = (

        post for post in postings

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

    )

    groups = {

        key: group

        for _, related in accrual.AccrualPostings.group_by_meta(postings, 'invoice')

        for key, group in related.make_consistent()

    }

    output = io.BytesIO()

    rt_client = RTClient()

    report = accrual.AgingReport(rt_client, output, today)

    report.run(groups)

    return output

def test_aging_report(accrual_postings):

    output = run_aging_report(accrual_postings)

    check_aging_ods(output)

@pytest.mark.parametrize('date,recv_end,pay_end', [

    # Both these dates are chosen for their off-by-one potential:

    # the first is exactly 30 days after the 2010-06-10 payable;

    # the second is exactly 60 days after the 2010-05-15 receivable.

    (datetime.date(2010, 7, 10), 1, 4),

    (datetime.date(2010, 7, 14), 2, 4),

])

def test_aging_report_date_cutoffs(accrual_postings, date, recv_end, pay_end):

    expect_recv = AGING_AR[:recv_end]

    expect_pay = AGING_AP[:pay_end]

    output = run_aging_report(accrual_postings, date)