Callable,

    Dict,

    Iterable,

    Iterator,

    List,

    Mapping,

    Optional,

    Sequence,

    Set,

    TextIO,

    Tuple,

    Union,

)

from ..beancount_types import (

    Transaction,

)

from pathlib import Path

import odf.table  # type:ignore[import]

from beancount.core import data as bc_data

from beancount.parser import printer as bc_printer

from . import core

from . import rewrite

from .. import books

from .. import cliutil

from .. import config as configmod

from .. import data

from .. import ranges

from .. import rtutil

PostTally = List[Tuple[int, data.Account]]

PROGNAME = 'ledger-report'

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

class LedgerODS(core.BaseODS[data.Posting, None]):

    CORE_COLUMNS: Sequence[str] = [

        'Date',

        data.Metadata.human_name('entity'),

        'Description',

        'Original Amount',

        'Booked Amount',

    ]

    ACCOUNT_COLUMNS: Dict[str, Sequence[str]] = collections.OrderedDict([

        ('Income', ['project', 'rt-id', 'receipt', 'income-type', 'memo']),

        ('Expenses', ['project', 'rt-id', 'receipt', 'approval', 'expense-type']),

        ('Equity', ['project', 'rt-id']),

        ('Assets:Receivable', ['project', 'rt-id', 'invoice', 'approval', 'contract', 'purchase-order']),

        ('Liabilities:Payable', ['project', 'rt-id', 'invoice', 'approval', 'contract', 'purchase-order']),

        ('Assets:PayPal', ['rt-id', 'paypal-id', 'receipt', 'approval']),

        ('Assets', ['rt-id', 'receipt', 'approval', 'bank-statement']),

        ('Liabilities', ['rt-id', 'receipt', 'approval', 'bank-statement']),

    ])

    CLASSIFICATION_COLUMN = "Account Classification"

    # Excel 2003 was limited to 65,536 rows per worksheet.

    # While we can probably count on all our users supporting more modern

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

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

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

    ) -> None:

        if sheet_size is None:

            sheet_size = self.SHEET_SIZE

        if totals_with_entries is None:

            totals_with_entries = [s for s in self.ACCOUNT_COLUMNS if ':' not in s]

        if totals_without_entries is None:

            totals_without_entries = totals_with_entries

        super().__init__(rt_wrapper)

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

        self.sheet_size = sheet_size

        self.totals_with_entries = totals_with_entries

        self.totals_without_entries = totals_without_entries

        self.report_name = "Ledger"

        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:

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

        'GitPython>=2.0',  # Debian:python3-git

        # 1.4.1 crashes when trying to save some documents.

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

        'pdfminer.six>=20200101',

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

        'conservancy_beancount.pdfforms.extract',

        'conservancy_beancount.plugin',

        'conservancy_beancount.reports',

        'conservancy_beancount.tools',

    ],

    entry_points={

        'console_scripts': [

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

            'assemble-audit-reports = conservancy_beancount.tools.audit_report:entry_point',

            'balance-sheet-report = conservancy_beancount.reports.balance_sheet:entry_point',

            'budget-report = conservancy_beancount.reports.budget:entry_point',

            'bean-sort = conservancy_beancount.tools.sort_entries:entry_point',

            'extract-odf-links = conservancy_beancount.tools.extract_odf_links:entry_point',

            'fund-report = conservancy_beancount.reports.fund:entry_point',

            'ledger-report = conservancy_beancount.reports.ledger:entry_point',

            'opening-balances = conservancy_beancount.tools.opening_balances:entry_point',

            'pdfform-extract = conservancy_beancount.pdfforms.extract:entry_point',

            'pdfform-extract-irs990scheduleA = conservancy_beancount.pdfforms.extract.irs990scheduleA:entry_point',

            'pdfform-fill = conservancy_beancount.pdfforms.fill:entry_point',

            'split-ods-links = conservancy_beancount.tools.split_ods_links:entry_point',

        ],

2018-01-01 open Equity:OpeningBalance

2018-01-01 open Assets:Checking

  classification: "Cash"

2018-01-01 open Assets:PayPal

  classification: "Cash"

2018-01-01 open Assets:Prepaid

  classification: "Prepaid expenses"

2018-01-01 open Assets:Receivable:Accounts

  classification: "Accounts receivable"

2018-01-01 open Expenses:Other

  classification: "Other expenses"

2018-01-01 open Income:Other

  classification: "Other income"

2018-01-01 open Liabilities:CreditCard

  classification: "Accounts payable"

2018-01-01 open Liabilities:Payable:Accounts

  classification: "Accounts payable"

2018-01-01 open Liabilities:UnearnedIncome

  classification: "Unearned income"

2018-02-28 * "Opening balance"

  Equity:OpeningBalance  -10,000 USD

  Assets:Checking         10,000 USD

2018-06-06 * "Accrued expense"

  project: "eighteen"

  Liabilities:Payable:Accounts  -60 USD

  Expenses:Other                 60 USD

2018-09-09 * "Paid expense"

  Liabilities:Payable:Accounts  60 USD

  project: "eighteen"

  Assets:Checking              -60 USD

2018-12-12 * "Accrued income"

  project: "eighteen"

  Assets:Receivable:Accounts  120 USD

  Income:Other               -120 USD

2019-03-03 * "Paid income"

  Assets:Receivable:Accounts  -120 USD

  project: "eighteen"

  Assets:Checking              120 USD

2019-06-06 * "Credit card expense"

  Liabilities:CreditCard  -65 USD

  Expenses:Other           65 USD

  project: "nineteen"

2019-09-09 * "Credit card paid"

  Liabilities:CreditCard  65 USD

  Assets:Checking        -65 USD

"""test_reports_ledger.py - Unit tests for general ledger report"""

# Copyright © 2020  Brett Smith

# License: AGPLv3-or-later WITH Beancount-Plugin-Additional-Permission-1.0

#

# Full copyright and licensing details can be found at toplevel file

# LICENSE.txt in the repository.

import collections

import contextlib

import copy

import datetime

import io

import itertools

import re

import pytest

from . import testutil

import odf.table

import odf.text

from beancount.core import data as bc_data

from beancount import loader as bc_loader

from conservancy_beancount import data

from conservancy_beancount.reports import core

from conservancy_beancount.reports import ledger

clean_account_meta = contextlib.contextmanager(testutil.clean_account_meta)

Acct = data.Account

_ledger_load = bc_loader.load_file(testutil.test_path('books/ledger.beancount'))

DEFAULT_REPORT_SHEETS = [

    'Balance',

    'Income',

    'Expenses',

    'Equity',

    'Assets:Receivable',

    'Liabilities:Payable',

    'Assets:PayPal',

    'Assets',

    'Liabilities',

]

    'Assets:Prepaid',

    'Liabilities:UnearnedIncome',

    'Liabilities:Payable',

]

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)

REPORT_KWARGS = [

    {'report_class': ledger.LedgerODS},

    *({'report_class': ledger.TransactionODS, 'txn_filter': flags}

      for flags in ledger.TransactionFilter),

]

@pytest.fixture

def ledger_entries():

    return copy.deepcopy(_ledger_load[0])

def iter_accounts(entries):

    for entry in entries:

        if isinstance(entry, bc_data.Open):

            yield entry.account

class NotFound(Exception): pass

class NoSheet(NotFound): pass

class NoHeader(NotFound): pass

class ExpectedPostings(core.RelatedPostings):

    @classmethod

    def find_section(cls, ods, account):

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

            sheet_account = sheet.getAttribute('name').replace(' ', ':')

            if sheet_account and account.is_under(sheet_account):

                break

        else:

            raise NoSheet(account)

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

        for row in rows:

            cells = row.childNodes

            if (len(cells) >= 3

                and cells[1].text == account

                and not cells[0].text):

                break

        else:

            raise NoHeader(account)

                )

@pytest.mark.parametrize('report_kwargs', iter(REPORT_KWARGS))

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

    (),

    ('Assets', 'Liabilities'),

    ('Income', 'Expenses'),

    ('Assets', 'Liabilities', 'Income', 'Expenses'),

])

def test_report_filter_totals(ledger_entries, tot_accts, report_kwargs):

    txn_filter = report_kwargs.get('txn_filter')

    postings, report = build_report(ledger_entries, START_DATE, STOP_DATE,

                                    totals_with_entries=tot_accts,

                                    totals_without_entries=tot_accts,

                                    **report_kwargs)

    expected = dict(ExpectedPostings.group_by_account(postings))

    for account in iter_accounts(ledger_entries):

        expect_totals = account.startswith(tot_accts)

        if account in expected and expected[account][-1].meta.date >= START_DATE:

            if txn_filter is None:

                expected[account].check_report(

                    report.document, START_DATE, STOP_DATE, expect_totals=expect_totals,

                )

            else:

                expected[account].check_txn_report(

                    report.document, txn_filter,

                    START_DATE, STOP_DATE, expect_totals=expect_totals,

                )

        elif expect_totals:

            ExpectedPostings.check_in_report(

                report.document, account, START_DATE, STOP_DATE, txn_filter,

            )

        else:

            ExpectedPostings.check_not_in_report(report.document, account)

@pytest.mark.parametrize('report_kwargs', iter(REPORT_KWARGS))

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

    ('Income', 'Expenses'),

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

])

def test_account_names_report(ledger_entries, accounts, report_kwargs):

    txn_filter = report_kwargs.get('txn_filter')

    postings, report = build_report(ledger_entries, START_DATE, STOP_DATE,

                                    accounts, **report_kwargs)

    expected = dict(ExpectedPostings.group_by_account(postings))

    for account in iter_accounts(ledger_entries):

        if not account.startswith(accounts):

            ExpectedPostings.check_not_in_report(report.document, account)

        elif txn_filter is None:

            expected[account].check_report(report.document, START_DATE, STOP_DATE)

        else:

            expected[account].check_txn_report(

                report.document, txn_filter, START_DATE, STOP_DATE,

            )

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)

    output.seek(0)

    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[:]

    postings = data.Posting.from_entries(iter(ledger_entries))

    expected = dict(ExpectedPostings.group_by_account(postings))

    for account in iter_accounts(ledger_entries):

        try:

            expected[account].check_report(ods, START_DATE, STOP_DATE)

        except KeyError:

            ExpectedPostings.check_in_report(ods, account)

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