# formats (Excel 2007 supports over 1 million rows per worksheet),

    # keeping the default limit conservative seems good to avoid running into

    # other limits (like the number of hyperlinks per worksheet), plus just

    # better for human organization and readability.

    SHEET_SIZE = 65500

    def __init__(self,

                 start_date: datetime.date,

                 stop_date: datetime.date,

                 accounts: Optional[Sequence[str]]=None,

                 rt_wrapper: Optional[rtutil.RT]=None,

                 sheet_size: Optional[int]=None,

    ) -> None:

        if sheet_size is None:

            sheet_size = self.SHEET_SIZE

        super().__init__(rt_wrapper)

        self.date_range = ranges.DateRange(start_date, stop_date)

        self.sheet_size = sheet_size

        if accounts is None:

            self.accounts = set(data.Account.iter_accounts())

            self.required_sheet_names = list(self.ACCOUNT_COLUMNS)

        else:

            self.accounts = set()

            self.required_sheet_names = []

            for acct_spec in accounts:

                subaccounts = frozenset(data.Account.iter_accounts_by_hierarchy(acct_spec))

                if subaccounts:

                    self.accounts.update(subaccounts)

                    self._require_sheet(acct_spec)

                balance = related.period_bal

                description = "Period Total"

        self.add_row(

            self.date_cell(date, stylename=self.merge_styles(

                self.style_bold, self.style_date,

            )),

            odf.table.TableCell(),

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

            odf.table.TableCell(),

            self.balance_cell(self.norm_func(balance), stylename=self.style_bold),

        )

        self.add_row()

        self.add_row(

            odf.table.TableCell(),

            self.string_cell(

                f"{key} Ledger"

                f" From {self.date_range.start.isoformat()}"

                f" To {self.date_range.stop.isoformat()}",

                stylename=self.style_bold,

                numbercolumnsspanned=len(self.sheet_columns) - 1,

            ),

        )

        self.norm_func = core.normalize_amount_func(key)

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

        self._report_section_balance(key, 'stop')

    def write_row(self, row: data.Posting) -> None:

        if row.cost is None:

            amount_cell = odf.table.TableCell()

        else:

            amount_cell = self.currency_cell(self.norm_func(row.units))

        self.add_row(

            self.date_cell(row.meta.date),

            self.string_cell(row.meta.get('entity') or ''),

            tally_by_account, self.required_sheet_names, self.sheet_size,

        )

        using_sheet_index = -1

        for sheet_index, account in core.sort_and_filter_accounts(

                tally_by_account, sheet_names,

        ):

            while using_sheet_index < sheet_index:

                using_sheet_index += 1

                self.start_sheet(sheet_names[using_sheet_index])

            postings = self.account_groups[account]

            if postings:

                super().write(postings)

                self.end_section(account)

        for index in range(using_sheet_index + 1, len(sheet_names)):

            self.start_sheet(sheet_names[index])

class ReturnFlag(enum.IntFlag):

    LOAD_ERRORS = 1

    NOTHING_TO_REPORT = 8

def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:

    parser = argparse.ArgumentParser(prog=PROGNAME)

The default is a year after the start date, or 30 days from today if the start

date was also not specified.

""")

    parser.add_argument(

        '--account', '-a',

        dest='accounts',

        metavar='ACCOUNT',

        action='append',

        help="""Show this account in the report. You can specify this option

multiple times. You can specify a part of the account hierarchy, or an account

classification from metadata. If not specified, the default set adapts to your

search criteria.

""")

    parser.add_argument(

        '--sheet-size', '--size',

        metavar='SIZE',

        type=int,

        default=LedgerODS.SHEET_SIZE,

        help="""Try to limit sheets to this many rows. The report will

automatically create new sheets to make this happen. When that's not possible,

it will issue a warning.

""")

    parser.add_argument(

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

    cliutil.add_loglevel_argument(parser)

    parser.add_argument(

        'search_terms',

        metavar='FILTER',

        type=cliutil.SearchTerm.arg_parser('project', 'rt-id'),

        nargs=argparse.ZERO_OR_MORE,

        help="""Report on postings that match this criteria. The format is

NAME=TERM. TERM is a link or word that must exist in a posting's NAME

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

`rt-id=rt:NUMBER`. Any other word is a shortcut for `project=TERM`.

""")

    args = parser.parse_args(arglist)

    if args.accounts is None:

        if any(term.meta_key == 'project' for term in args.search_terms):

            args.accounts = [

                'Income',

                'Expenses',

                'Assets:Receivable',

                'Assets:Prepaid',

                'Liabilities:UnearnedIncome',

                'Liabilities:Payable',

            ]

        else:

            args.accounts = list(LedgerODS.ACCOUNT_COLUMNS)

        postings = search_term.filter_postings(postings)

    rt_wrapper = config.rt_wrapper()

    if rt_wrapper is None:

        logger.warning("could not initialize RT client; spreadsheet links will be broken")

    try:

        report = LedgerODS(

            args.start_date,

            args.stop_date,

            args.accounts,

            rt_wrapper,

            args.sheet_size,

        )

    except ValueError as error:

        logger.error("%s: %r", *error.args)

        return 2

    report.write(postings)

    if not any(report.account_groups.values()):

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

        returncode |= ReturnFlag.NOTHING_TO_REPORT

    if args.output_file is None:

        out_dir_path = config.repository_path() or Path()

        args.output_file = out_dir_path / 'LedgerReport_{}_{}.ods'.format(

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

del PROJECT_REPORT_SHEETS[3]

OVERSIZE_RE = re.compile(

    r'^([A-Za-z0-9:]+) has ([0-9,]+) rows, over size ([0-9,]+)$'

)

START_DATE = datetime.date(2018, 3, 1)

MID_DATE = datetime.date(2019, 3, 1)

STOP_DATE = datetime.date(2020, 3, 1)

@pytest.fixture

def ledger_entries():

    return copy.deepcopy(_ledger_load[0])

class NotFound(Exception): pass

class NoSheet(NotFound): pass

class NoHeader(NotFound): pass

class ExpectedPostings(core.RelatedPostings):

    def slice_date_range(self, start_date, end_date):

        postings = enumerate(self)

        for start_index, post in postings:

            if start_date <= post.meta.date:

                break

        else:

            start_index += 1

        if end_date <= post.meta.date:

            end_index = start_index

        else:

            for end_index, post in postings:

                if end_date <= post.meta.date:

                    break

            else:

                end_index = None

        return (self[:start_index].balance_at_cost(),

                self[start_index:end_index])

        account = self[0].account

        norm_func = core.normalize_amount_func(account)

        open_bal, expect_posts = self.slice_date_range(start_date, end_date)

        open_bal = norm_func(open_bal)

        closing_bal = norm_func(expect_posts.balance_at_cost())

            opening_row = testutil.ODSCell.from_row(next(rows))

            assert opening_row[0].value == start_date

            assert opening_row[4].text == open_bal.format(None, empty='0', sep='\0')

            closing_bal += open_bal

        for expected in expect_posts:

            cells = iter(testutil.ODSCell.from_row(next(rows)))

            assert next(cells).value == expected.meta.date

            assert next(cells).text == (expected.meta.get('entity') or '')

            assert next(cells).text == (expected.meta.txn.narration or '')

            if expected.cost is None:

                assert not next(cells).text

                assert next(cells).value == norm_func(expected.units.number)

            else:

                assert next(cells).value == norm_func(expected.units.number)

                assert next(cells).value == norm_func(expected.at_cost().number)

def get_sheet_names(ods):

    return [sheet.getAttribute('name').replace(' ', ':')

            for sheet in ods.getElementsByType(odf.table.Table)]

def check_oversize_logs(caplog, accounts, sheet_size):

    actual = {}

    for log in caplog.records:

        match = OVERSIZE_RE.match(log.message)

        if match:

            assert int(match.group(3).replace(',', '')) == sheet_size

    check_oversize_logs(caplog, have, 100)

def test_plan_sheets_full_split_required(caplog):

    have = {

        Acct('Assets:Bank:Savings'): 98,

        Acct('Assets:Bank:Checking'): 96,

        Acct('Assets:Bank:Investment'): 94,

    }

    actual = ledger.LedgerODS.plan_sheets(have, ['Assets'], 100)

    assert actual == ['Assets:Bank:Checking', 'Assets:Bank:Savings', 'Assets']

    assert not caplog.records

@pytest.mark.parametrize('start_date,stop_date', [

    (START_DATE, STOP_DATE),

    (START_DATE, MID_DATE),

    (MID_DATE, STOP_DATE),

    (START_DATE.replace(month=6), START_DATE.replace(month=12)),

    (STOP_DATE, STOP_DATE.replace(month=12)),

])

def test_date_range_report(ledger_entries, start_date, stop_date):

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

    ('Income', 'Expenses'),

    ('Assets:Receivable', 'Liabilities:Payable'),

])

def test_account_names_report(ledger_entries, accounts):

        else:

def run_main(arglist, config=None):

    if config is None:

        config = testutil.TestConfig(

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

            rt_client=testutil.RTClient(),

        )

    arglist.insert(0, '--output-file=-')

    output = io.BytesIO()

    errors = io.StringIO()

    with clean_account_meta():

        retcode = ledger.main(arglist, output, errors, config)

    return retcode, output, errors

def test_main(ledger_entries):

    retcode, output, errors = run_main([

        '-b', START_DATE.isoformat(),

        '-e', STOP_DATE.isoformat(),

    ])

    output.seek(0)

    assert not errors.getvalue()

    assert retcode == 0

    ods = odf.opendocument.load(output)

    assert get_sheet_names(ods) == DEFAULT_REPORT_SHEETS[:]

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

    'Liabilities',

    'Accounts payable',

])

def test_main_account_limit(ledger_entries, acct_arg):

    retcode, output, errors = run_main([

        '-a', acct_arg,

        '-b', START_DATE.isoformat(),

        '-e', STOP_DATE.isoformat(),

    ])

    assert not errors.getvalue()

        try:

            expected.check_report(ods, START_DATE, STOP_DATE)

        except NotFound:

            assert not should_find, f"{account} not found in report"

        else:

            assert should_find, f"{account} in report but should be excluded"

@pytest.mark.parametrize('project,start_date,stop_date', [

    ('eighteen', START_DATE, MID_DATE.replace(day=30)),

    ('nineteen', MID_DATE, STOP_DATE),

])

def test_main_project_report(ledger_entries, project, start_date, stop_date):

    for key, related in ExpectedPostings.group_by_meta(postings, 'project'):

        if key == project:

            break

    assert key == project

    retcode, output, errors = run_main([

        f'--begin={start_date.isoformat()}',

        f'--end={stop_date.isoformat()}',

        project,

    ])

    assert not errors.getvalue()

    assert retcode == 0

    ods = odf.opendocument.load(output)

    assert get_sheet_names(ods) == PROJECT_REPORT_SHEETS[:]

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

    'Assets:NoneSuchBank',

    'Funny money',

])

def test_main_invalid_account(caplog, arg):

    retcode, output, errors = run_main(['-a', arg])

    assert retcode == 2

    assert any(log.message.endswith(f': {arg!r}') for log in caplog.records)

def test_main_no_postings(caplog):

    retcode, output, errors = run_main(['NonexistentProject'])