metadata to match. A single ticket number is a shortcut for

`rt-id=rt:NUMBER`. Any other link, including an RT attachment link in

`TIK/ATT` format, is a shortcut for `invoice=LINK`.

""")

    args = parser.parse_args(arglist)

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

        args.report_type = ReportType.AGING

    return args

def main(arglist: Optional[Sequence[str]]=None,

         stdout: TextIO=sys.stdout,

         stderr: TextIO=sys.stderr,

         config: Optional[configmod.Config]=None,

) -> int:

    if cliutil.is_main_script(PROGNAME):

        global logger

        logger = logging.getLogger(PROGNAME)

        sys.excepthook = cliutil.ExceptHook(logger)

    args = parse_arguments(arglist)

    cliutil.setup_logger(logger, args.loglevel, stderr)

    if config is None:

        config = configmod.Config()

        config.load_file()

    returncode = 0

    books_loader = config.books_loader()

    if books_loader is None:

        entries, load_errors, _ = books.Loader.load_none(config.config_file_path())

    elif args.report_type is ReportType.AGING:

        entries, load_errors, _ = books_loader.load_all()

    else:

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

    filters.remove_opening_balance_txn(entries)

    for error in load_errors:

        bc_printer.print_error(error, file=stderr)

        returncode |= ReturnFlag.LOAD_ERRORS

    postings = list(filter_search(

        data.Posting.from_entries(entries), args.search_terms,

    ))

    if not postings:

        logger.warning("no matching entries found to report")

        returncode |= ReturnFlag.NOTHING_TO_REPORT

    groups: PostGroups

    if args.report_type is None or args.report_type is ReportType.OUTGOING:

        groups = dict(AccrualPostings.group_by_first_meta_link(postings, 'rt-id'))

        if (args.report_type is None

            and len(groups) == 1

        ):

            args.report_type = ReportType.OUTGOING

    if args.report_type is not ReportType.OUTGOING:

        groups = {

            key: group

            for _, source in AccrualPostings.group_by_first_meta_link(postings, 'invoice')

            for key, group in source.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

    del postings

    report: Optional[BaseReport] = None

    output_path: Optional[Path] = None

    if args.report_type is ReportType.AGING:

        rt_wrapper = config.rt_wrapper()

        if rt_wrapper is None:

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

        else:

            now = datetime.datetime.now()

            if args.output_file is None:

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

                logger.info("Writing report to %s", args.output_file)

            out_bin = cliutil.bytes_output(args.output_file, stdout)

            report = AgingReport(rt_wrapper, out_bin)

    elif args.report_type is ReportType.OUTGOING:

        rt_wrapper = config.rt_wrapper()

        if rt_wrapper is None:

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

        else:

            out_file = cliutil.text_output(args.output_file, stdout)

            report = OutgoingReport(rt_wrapper, out_file)

    else:

        out_file = cliutil.text_output(args.output_file, stdout)

        report = BalanceReport(out_file)

    if report is None:

        returncode |= ReturnFlag.REPORT_ERRORS

    else:

        report.run(groups)

    return 0 if returncode == 0 else 16 + returncode

if __name__ == '__main__':

    exit(main())

#!/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

        # 1.4.1 crashes when trying to save some documents.

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

        '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.plugin',

        'conservancy_beancount.reports',

    ],

    entry_points={

        'console_scripts': [

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

        ],

    },

)

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

  rt-id: "rt:510"

  invoice: "rt:510/5100"

  contract: "rt:510/4000"

  project: "Conservancy"

  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"

  project: "Conservancy"

  Income:Donations  -1500.00 USD

  Assets:Receivable:Accounts  1500.00 USD

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

  rt-id: "rt:505"

  invoice: "rt:505/5050"

  project: "Conservancy"

  Assets:Receivable:Accounts  -2,750.00 USD

  Assets:Checking  2,750.00 USD

  receipt: "DonorAWire.pdf"

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

  rt-id: "rt:510"

  invoice: "rt:510/5100"

  project: "Conservancy"

  Liabilities:Payable:Accounts  200.00 USD

  contract: "rt:510/4000"

  Assets:Checking  -200.00 USD

  receipt: "rt:510/5105"

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

  rt-id: "rt:510"

  invoice: "rt:510/6100"

  contract: "rt:510/4000"

  project: "Conservancy"

  Expenses:Services:Legal  220.00 USD

  Liabilities:Payable:Accounts  -220.00 USD

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

  rt-id: "rt:510"

  invoice: "rt:510/6100"

  contract: "rt:510/4000"

  project: "Conservancy"

  Expenses:FilingFees  60.00 USD

  Liabilities:Payable:Accounts  -60.00 USD

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

  ; Multiple rt-ids are used to test proper handling for both

  ; searching and generating the outgoing report.

  rt-id: "rt:520 rt:525"

  invoice: "rt:520/5200"

  contract: "rt:520/5220"

  project: "Conservancy"

  Liabilities:Payable:Accounts  -1,000 EUR {1.100 USD}

  Expenses:FilingFees  1,000 EUR {1.100 USD}

2010-09-15 * "GrantCo" "2010Q3 grant"

  rt-id: "rt:470"

  invoice: "rt:470/4700"

  project: "Development Grant"

  Assets:Receivable:Accounts  6000 USD

  Income:Donations           -6000 USD

from decimal import Decimal

from typing import NamedTuple, Optional, Sequence

from beancount.core import data as bc_data

from beancount import loader as bc_loader

from conservancy_beancount import cliutil

from conservancy_beancount import data

from conservancy_beancount import rtutil

from conservancy_beancount.reports import accrual

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

]

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]

    project: Sequence[str]

    @classmethod

    def make_simple(cls, date, entity, at_cost, invoice,

                    rt_id=None, orig_amount=None, project='Conservancy'):

        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)

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

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

    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

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

        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-25', 'Vendor', 200, 'FIXME'),

    AgingRow.make_simple('2010-04-30', 'Vendor', 220, 'rt:310/3120'),

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

    AgingRow.make_simple('2010-06-15', 'GrantCo', 11500, 'rt:470/4700',

                         project='Development Grant'),

]

class RTClient(testutil.RTClient):

    TICKET_DATA = {

        '40': [

            ('400', 'invoice feb.csv', 'text/csv', '40.4k'),

        ],

        '44': [

            ('440', 'invoice feb.csv', 'text/csv', '40.4k'),

        ],

        '310': [

            ('3100', 'VendorContract.pdf', 'application/pdf', '1.7m'),

            ('3120', 'VendorInvoiceB.pdf', 'application/pdf', '1.8m'),

        ],

        '490': [],

        '505': [],

        '510': [

            ('4000', 'contract.pdf', 'application/pdf', '1.4m'),

            ('5100', 'invoice april.pdf', 'application/pdf', '1.5m'),

            ('5105', 'payment.png', 'image/png', '51.5k'),

            ('6100', 'invoice may.pdf', 'application/pdf', '1.6m'),

        ],

        '515': [],

        '520': [],

    }

@pytest.fixture

def accrual_postings():

    return data.Posting.from_entries(copy.deepcopy(ACCRUAL_TXNS))

def accruals_by_meta(postings, value, key='invoice', wrap_type=iter):

    return wrap_type(

        post for post in postings

        if post.meta.get(key) == value

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

    )

def find_row_by_text(row_source, want_text):

    for row in row_source:

        assert not re.match(r'\brt:4\d\b', line)

@pytest.mark.parametrize('arglist,expect_invoice', [

    (['40'], 'rt:40/400'),

    (['44/440'], 'rt:44/440'),

])

def test_output_payments_when_only_match(arglist, expect_invoice):

    retcode, output, errors = run_main(arglist)

    assert not errors.getvalue()

    assert retcode == 0

    check_output(output, [

        rf'^{re.escape(expect_invoice)}:$',

        r' outstanding since ',

    ])

@pytest.mark.parametrize('arglist,expect_amount', [

    (['310'], 420),

    (['310/3120'], 220),

    (['entity=Vendor'], 420),

])

def test_main_outgoing_report(arglist, expect_amount):

    retcode, output, errors = run_main(arglist)

    assert not errors.getvalue()

    assert retcode == 0

    rt_url = RTClient.DEFAULT_URL[:-9]

    rt_id_url = re.escape(f'<{rt_url}Ticket/Display.html?id=310>')

    contract_url = re.escape(f'<{rt_url}Ticket/Attachment/3120/3120/VendorContract.pdf>')

    check_output(output, [

        r'^REQUESTOR: Mx\. 310 <mx310@example\.org>$',

        rf'^TOTAL TO PAY: \${expect_amount}\.00$',

        r'^\s*2010-04-30\s',

        r'^\s+Expenses:Travel\s+220 USD$',

    ])

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

    ['-t', 'balance'],

    ['515/5150'],

    ['entity=MatchingProgram'],

])

def test_main_balance_report(arglist):

    retcode, output, errors = run_main(arglist)

    assert not errors.getvalue()

    assert retcode == 0

    check_output(output, [

        r'\brt://ticket/515/attachments/5150:$',

        r'^\s+1,500\.00 USD outstanding since 2010-05-15$',

    ])

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

    [],

    ['-t', 'aging', 'entity=Lawyer'],

])

def test_main_aging_report(tmp_path, arglist):

    if arglist:

        recv_rows = [row for row in AGING_AR if 'Lawyer' in row.entity]

        pay_rows = [row for row in AGING_AP if 'Lawyer' in row.entity]

    else:

        recv_rows = AGING_AR

        pay_rows = AGING_AP

    output_path = tmp_path / 'AgingReport.ods'

    arglist.insert(0, f'--output-file={output_path}')

    retcode, output, errors = run_main(arglist)

    assert not errors.getvalue()

    assert retcode == 0

    assert not output.getvalue()

    with output_path.open('rb') as ods_file:

        check_aging_ods(ods_file, None, recv_rows, pay_rows)

def test_main_no_books():

    check_main_fails([], testutil.TestConfig(), 1 | 8, [

        r':[01]: +no books to load in configuration\b',

    ])

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

    ['499'],

    ['505/99999'],

    ['entity=NonExistent'],

])

def test_main_no_matches(arglist):

    check_main_fails(arglist, None, 8, [

        r': WARNING: no matching entries found to report$',

    ])

def test_main_no_rt():

    config = testutil.TestConfig(

        books_path=testutil.test_path('books/accruals.beancount'),

    )

    check_main_fails(['-t', 'out'], config, 4, [

        r': ERROR: unable to generate outgoing report: RT client is required\b',

    ])