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

import collections

import datetime

import enum

import itertools

import operator

import logging

import sys

from typing import (

    Callable,

    Dict,

    Iterable,

    Iterator,

    List,

    Mapping,

    Optional,

    Sequence,

    TextIO,

    Tuple,

    Union,

)

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

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

    parser = argparse.ArgumentParser(prog=PROGNAME)

    cliutil.add_version_argument(parser)

    parser.add_argument(

        '--begin', '--start', '-b',

        dest='start_date',

        metavar='DATE',

        type=cliutil.date_arg,

        help="""Date to start reporting entries, inclusive, in YYYY-MM-DD format.

The default is one year ago.

""")

    parser.add_argument(

        '--end', '--stop', '-e',

        dest='stop_date',

        metavar='DATE',

        type=cliutil.date_arg,

        help="""Date to stop reporting entries, exclusive, in YYYY-MM-DD format.

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

        metavar='ACCOUNT',

        action='append',

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

""")

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

        metavar='PATH',

        type=Path,

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

The default is `LedgerReport_<StartDate>_<StopDate>.ods`.

""")

    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 any(term.meta_key == 'project' for term in args.search_terms):

                'Income',

                'Expenses',

                'Assets:Receivable',

                'Assets:Prepaid',

                'Liabilities:UnearnedIncome',

            ]

        else:

    return args

def diff_year(date: datetime.date, diff: int) -> datetime.date:

    new_year = date.year + diff

    try:

        return date.replace(year=new_year)

    except ValueError:

        # The original date is Feb 29, which doesn't exist in the new year.

        if diff < 0:

            return datetime.date(new_year, 2, 28)

        else:

            return datetime.date(new_year, 3, 1)

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

         stdout: TextIO=sys.stdout,

         stderr: TextIO=sys.stderr,

         config: Optional[configmod.Config]=None,

) -> int:

    args = parse_arguments(arglist)

    cliutil.set_loglevel(logger, args.loglevel)

    if config is None:

        config = configmod.Config()

        config.load_file()

    today = datetime.date.today()

    if args.start_date is None:

        args.start_date = diff_year(today, -1)

        if args.stop_date is None:

            args.stop_date = today + datetime.timedelta(days=30)

    elif args.stop_date is None:

        args.stop_date = diff_year(args.start_date, 1)

    returncode = 0

    books_loader = config.books_loader()

    if books_loader is None:

    else:

    for error in load_errors:

        bc_printer.print_error(error, file=stderr)

        returncode |= ReturnFlag.LOAD_ERRORS

    for search_term in args.search_terms:

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

    report = LedgerODS(

        args.start_date,

        args.stop_date,

        rt_wrapper,

        args.sheet_size,

    )

    report.write(postings)

    if not report.account_groups:

        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(

            args.start_date.isoformat(), args.stop_date.isoformat(),

        )

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

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

    report.save_file(ods_file)

    return 0 if returncode == 0 else 16 + returncode

entry_point = cliutil.make_entry_point(__name__, PROGNAME)

if __name__ == '__main__':

    exit(entry_point())

2018-01-01 open Equity:OpeningBalance

2018-01-01 open Assets:Checking

2018-01-01 open Assets:Receivable:Accounts

2018-01-01 open Expenses:Other

2018-01-01 open Income:Other

2018-01-01 open Liabilities:CreditCard

2018-01-01 open Liabilities:Payable:Accounts

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

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

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

import copy

import datetime

import io

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

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

    'Liabilities',

]

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:

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

])

def test_account_names_report(ledger_entries, sheet_names):

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

    report = ledger.LedgerODS(START_DATE, STOP_DATE, sheet_names=sheet_names)

    report.write(iter(postings))

    for key, expected in ExpectedPostings.group_by_account(postings):

        should_find = key.startswith(sheet_names)

        try:

            expected.check_report(report.document, START_DATE, STOP_DATE)

        except NotFound:

            assert not should_find

        else:

            assert should_find

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

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

    ])

    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(ledger_entries)

    for _, expected in ExpectedPostings.group_by_account(postings):

        expected.check_report(ods, START_DATE, STOP_DATE)

@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):

    postings = data.Posting.from_entries(ledger_entries)

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

    for _, expected in ExpectedPostings.group_by_account(related):

        expected.check_report(ods, start_date, stop_date)

def test_main_no_postings(caplog):

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

import beancount.loader as bc_loader

import beancount.parser.options as bc_options

import odf.element

import odf.opendocument

import odf.table

from decimal import Decimal

from pathlib import Path

from typing import Any, Optional, NamedTuple

from conservancy_beancount import books, data, rtutil

EXTREME_FUTURE_DATE = datetime.date(datetime.MAXYEAR, 12, 30)

FUTURE_DATE = datetime.date.today() + datetime.timedelta(days=365 * 99)

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

FY_MID_DATE = datetime.date(2020, 9, 1)

PAST_DATE = datetime.date(2000, 1, 1)

TESTS_DIR = Path(__file__).parent

# This function is a teardown fixture, but different test files use

# it with different scopes. Typical usage looks like:

#   clean_account_meta = pytest.fixture([options])(testutil.clean_account_meta)

def clean_account_meta():

def _ods_cell_value_type(cell):

    assert cell.tagName == 'table:table-cell'

    return cell.getAttribute('valuetype')

def _ods_cell_value(cell):

    value_type = cell.getAttribute('valuetype')

    if value_type == 'currency' or value_type == 'float':

        return Decimal(cell.getAttribute('value'))

    elif value_type == 'date':

        return datetime.datetime.strptime(

            cell.getAttribute('datevalue'), '%Y-%m-%d',

        ).date()

    else:

        return cell.getAttribute('value')

def _ods_elem_text(elem):

    if isinstance(elem, odf.element.Text):

        return elem.data

    else:

        return '\0'.join(_ods_elem_text(child) for child in elem.childNodes)

odf.element.Element.value_type = property(_ods_cell_value_type)

odf.element.Element.value = property(_ods_cell_value)