NPO-Accounting/conservancy_beancount Changeset - a27d68beab7b

Changeset - a27d68beab7b

Parent rev.

Child rev.

[Not reviewed]

0 4 0

Brett Smith - 4 years ago 2020-07-27 15:50:11
brettcsmith@brettcsmith.org

cliutil: Add diff_year function.

4 files changed with 33 insertions and 26 deletions:

conservancy_beancount/cliutil.py

conservancy_beancount/reports/fund.py

conservancy_beancount/reports/ledger.py

tests/test_cliutil.py

0 comments (0 inline, 0 general)

conservancy_beancount/cliutil.py

➞

Show inline comments

 """cliutil - Utilities for CLI tools"""
 PKGNAME = 'conservancy_beancount'
 LICENSE = """
 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 argparse
 import datetime
 import enum
 import io
 import logging
 import operator
 import os
 import pkg_resources
 import re
 import signal
 import sys
 import traceback
 import types
 from pathlib import Path
 from . import data
 from . import filters
 from . import rtutil
 from typing import (
     cast,
     Any,
     BinaryIO,
     Callable,
     IO,
     Iterable,
     NamedTuple,
     NoReturn,
     Optional,
     Sequence,
     TextIO,
     Type,
     Union,
+)
 from .beancount_types import (
     MetaKey,
+)
 OutputFile = Union[int, IO]
 STDSTREAM_PATH = Path('-')
 VERSION = pkg_resources.require(PKGNAME)[0].version
 class ExceptHook:
     def __init__(self,
                  logger: Optional[logging.Logger]=None,
                  default_exitcode: int=3,
     ) -> None:
         if logger is None:
             logger = logging.getLogger()
         self.logger = logger
         self.default_exitcode = default_exitcode
     def __call__(self,
                  exc_type: Type[BaseException],
                  exc_value: BaseException,
                  exc_tb: types.TracebackType,
     ) -> NoReturn:
         exitcode = self.default_exitcode
         if isinstance(exc_value, KeyboardInterrupt):
             signal.signal(signal.SIGINT, signal.SIG_DFL)
             os.kill(0, signal.SIGINT)
             signal.pause()
         elif isinstance(exc_value, OSError):
             exitcode += 1
             msg = "I/O error: {e.filename}: {e.strerror}".format(e=exc_value)
         else:
             parts = [type(exc_value).__name__, *exc_value.args]
             msg = "internal " + ": ".join(parts)
         self.logger.critical(msg)
         self.logger.debug(
             ''.join(traceback.format_exception(exc_type, exc_value, exc_tb)),
+        )
         raise SystemExit(exitcode)
 class InfoAction(argparse.Action):
     def __call__(self,
                  parser: argparse.ArgumentParser,
                  namespace: argparse.Namespace,
                  values: Union[Sequence[Any], str, None]=None,
                  option_string: Optional[str]=None,
     ) -> NoReturn:
         if isinstance(self.const, str):
             info = self.const
             exitcode = 0
         else:
             info, exitcode = self.const
         print(info)
         raise SystemExit(exitcode)
 class LogLevel(enum.IntEnum):
     DEBUG = logging.DEBUG
     INFO = logging.INFO
     WARNING = logging.WARNING
     ERROR = logging.ERROR
     CRITICAL = logging.CRITICAL
     WARN = WARNING
     ERR = ERROR
     CRIT = CRITICAL
     @classmethod
     def from_arg(cls, arg: str) -> int:
         try:
             return cls[arg.upper()].value
         except KeyError:
             raise ValueError(f"unknown loglevel {arg!r}") from None
     @classmethod
     def choices(cls) -> Iterable[str]:
         for level in sorted(cls, key=operator.attrgetter('value')):
             yield level.name.lower()
 class ReturnFlag(enum.IntFlag):
     """Common return codes for tools
     Tools should combine these flags to report different errors, and then use
     ReturnFlag.returncode(flags) to report their final exit status code.
     Values 1, 2, 4, and 8 should be reserved for this class to be shared across
     all tools. Flags 16, 32, and 64 are available for tools to report their own
     specific errors.
     """
     LOAD_ERRORS = 1
     NOTHING_TO_REPORT = 2
     _RESERVED4 = 4
     _RESERVED8 = 8
     @classmethod
     def returncode(cls, flags: int) -> int:
         return 0 if flags == 0 else 16 + flags
 class SearchTerm(NamedTuple):
     """NamedTuple representing a user's metadata filter
     SearchTerm knows how to parse and store posting metadata filters provided
     by the user in `key=value` format. Reporting tools can use this to filter
     postings that match the user's criteria, to report on subsets of the books.
     Typical usage looks like::
       argument_parser.add_argument(
         'search_terms',
         type=SearchTerm.arg_parser(),
         …,
+      )
       args = argument_parser.parse_args(…)
       for query in args.search_terms:
         postings = query.filter_postings(postings)
     """
     meta_key: MetaKey
     pattern: str
     @classmethod
     def arg_parser(cls,
                    default_key: Optional[str]=None,
                    ticket_default_key: Optional[str]=None,
     ) -> Callable[[str], 'SearchTerm']:
         """Build a SearchTerm parser
         This method returns a function that can parse strings in ``key=value``
         format and return a corresponding SearchTerm.
         If you specify a default key, then strings that just specify a ``value``
         will be parsed as if they said ``default_key=value``. Otherwise,
         parsing strings without a metadata key will raise a ValueError.
         If you specify a default key ticket links, then values in the format
         ``number``, ``rt:number``, or ``rt://ticket/number`` will be parsed as
         if they said ``ticket_default_key=value``.
         """
         if ticket_default_key is None:
             ticket_default_key = default_key
         def parse_search_term(arg: str) -> 'SearchTerm':
             key: Optional[str] = None
             if re.match(r'^[a-z][-\w]*=', arg):
                 key, _, raw_link = arg.partition('=')
             else:
                 raw_link = arg
             rt_ids = rtutil.RT.parse(raw_link)
             if rt_ids is None:
                 rt_ids = rtutil.RT.parse('rt:' + raw_link)
             if rt_ids is None:
                 if key is None:
                     key = default_key
                 pattern = r'(?:^|\s){}(?:\s|$)'.format(re.escape(raw_link))
             else:
                 ticket_id, attachment_id = rt_ids
                 if key is None:
                     if attachment_id is None:
                         key = ticket_default_key
                     else:
                         key = default_key
                 pattern = rtutil.RT.metadata_regexp(
                     ticket_id,
                     attachment_id,
                     first_link_only=key == 'rt-id' and attachment_id is None,
+                )
             if key is None:
                 raise ValueError(f"invalid search term {arg!r}: no metadata key")
             return cls(key, pattern)
         return parse_search_term
     def filter_postings(self, postings: Iterable[data.Posting]) -> Iterable[data.Posting]:
         return filters.filter_meta_match(
             postings, self.meta_key, re.compile(self.pattern),
+        )
 def add_loglevel_argument(parser: argparse.ArgumentParser,
                           default: LogLevel=LogLevel.INFO) -> argparse.Action:
     return parser.add_argument(
         '--loglevel',
         metavar='LEVEL',
         default=default.value,
         type=LogLevel.from_arg,
         help="Show logs at this level and above."
         f" Specify one of {', '.join(LogLevel.choices())}."
         f" Default {default.name.lower()}.",
+    )
 def add_version_argument(parser: argparse.ArgumentParser) -> argparse.Action:
     progname = parser.prog or sys.argv[0]
     return parser.add_argument(
         '--version', '--copyright', '--license',
         action=InfoAction,
         nargs=0,
         const=f"{progname} version {VERSION}\n{LICENSE}",
         help="Show program version and license information",
+    )
 def date_arg(arg: str) -> datetime.date:
     return datetime.datetime.strptime(arg, '%Y-%m-%d').date()
 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 year_or_date_arg(arg: str) -> Union[int, datetime.date]:
     """Get either a date or a year (int) from an argument string
     This is a useful argument type for arguments that will be passed into
     Books loader methods which can accept either a fiscal year or a full date.
     """
     try:
         year = int(arg, 10)
     except ValueError:
         ok = False
     else:
         ok = datetime.MINYEAR <= year <= datetime.MAXYEAR
     if ok:
         return year
     else:
         return date_arg(arg)
 def make_entry_point(mod_name: str, prog_name: str=sys.argv[0]) -> Callable[[], int]:
     """Create an entry_point function for a tool
     The returned function is suitable for use as an entry_point in setup.py.
     It sets up the root logger and excepthook, then calls the module's main
     function.
     """
     def entry_point():  # type:ignore
         prog_mod = sys.modules[mod_name]
         setup_logger()
         prog_mod.logger = logging.getLogger(prog_name)
         sys.excepthook = ExceptHook(prog_mod.logger)
         return prog_mod.main()
     return entry_point
 def setup_logger(logger: Union[str, logging.Logger]='',
                  stream: TextIO=sys.stderr,
                  fmt: str='%(name)s: %(levelname)s: %(message)s',
 ) -> logging.Logger:
     """Set up a logger with a StreamHandler with the given format"""
     if isinstance(logger, str):
         logger = logging.getLogger(logger)
     formatter = logging.Formatter(fmt)
     handler = logging.StreamHandler(stream)
     handler.setFormatter(formatter)
     logger.addHandler(handler)
     return logger
 def set_loglevel(logger: logging.Logger, loglevel: int=logging.INFO) -> None:
     """Set the loglevel for a tool or module
     If the given logger is not under a hierarchy, this function sets the
     loglevel for the root logger, along with some specific levels for libraries
     used by reporting tools. Otherwise, it's the same as
     ``logger.setLevel(loglevel)``.
     """
     if '.' not in logger.name:
         logger = logging.getLogger()
         if loglevel <= logging.DEBUG:
             # At the debug level, the rt module logs the full body of every
             # request and response. That's too much.
             logging.getLogger('rt.rt').setLevel(logging.INFO)
     logger.setLevel(loglevel)
 def bytes_output(path: Optional[Path]=None,
                  default: OutputFile=sys.stdout,
                  mode: str='w',
 ) -> BinaryIO:
     """Get a file-like object suitable for binary output
     If ``path`` is ``None`` or ``-``, returns a file-like object backed by
     ``default``. If ``default`` is a file descriptor or text IO object, this
     method returns a file-like object that writes to the same place.
     Otherwise, returns ``path.open(mode)``.
     """
     mode = f'{mode}b'
     if path is None or path == STDSTREAM_PATH:
         if isinstance(default, int):
             retval = open(default, mode)
         elif isinstance(default, TextIO):
             retval = default.buffer
         else:
             retval = default
     else:
         retval = path.open(mode)
     return cast(BinaryIO, retval)
 def text_output(path: Optional[Path]=None,
                 default: OutputFile=sys.stdout,
                 mode: str='w',
                 encoding: Optional[str]=None,
 ) -> TextIO:
     """Get a file-like object suitable for text output
     If ``path`` is ``None`` or ``-``, returns a file-like object backed by
     ``default``. If ``default`` is a file descriptor or binary IO object, this
     method returns a file-like object that writes to the same place.
     Otherwise, returns ``path.open(mode)``.
     """
     if path is None or path == STDSTREAM_PATH:
         if isinstance(default, int):
             retval = open(default, mode, encoding=encoding)
         elif isinstance(default, BinaryIO):
             retval = io.TextIOWrapper(default, encoding=encoding)
         else:
             retval = default
     else:
         retval = path.open(mode, encoding=encoding)
     return cast(TextIO, retval)

conservancy_beancount/reports/fund.py

➞

Show inline comments

 """funds.py - Funds report from Beancount
 This tool produces a report of account balances for restricted funds,
 including income, expenses, and realized gain/loss over a given time period.
 Restricted funds are designated with the "project" metadata.
 Specify the date range you want to report with the ``--begin`` and ``--end``
 options.
 Select the accounts you want to report with the ``--account`` option. You can
 specify this option multiple times. The report will include at least one sheet
 for each account you specify. Subaccounts will be reported on that sheet as
 well.
 Select the postings you want to report by passing metadata search terms in
 ``name=value`` format.
 Run ``ledger-report --help`` for abbreviations and other options.
 Examples
 --------
 Generate a report of all restricted funds in ODS format over the last year::
       fund-report
 Query a specific restricted fund and get a quick balance on the terminal::
       fund-report <PROJECT>
 """
 # 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 argparse
 import collections
 import datetime
 import enum
 import locale
 import logging
 import sys
 from typing import (
     Dict,
     Iterable,
     Iterator,
     Mapping,
     Optional,
     Sequence,
     TextIO,
     Tuple,
     Union,
+)
 from ..beancount_types import (
     MetaValue,
+)
 from decimal import Decimal
 from pathlib import Path
 import odf.table  # type:ignore[import]
 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
 AccountsMap = Mapping[data.Account, core.PeriodPostings]
 FundPosts = Tuple[MetaValue, AccountsMap]
 EQUITY_ACCOUNTS = ['Equity', 'Income', 'Expenses']
 INFO_ACCOUNTS = [
     'Assets:Receivable',
     'Assets:Prepaid',
     'Liabilities:UnearnedIncome',
     'Liabilities:Payable',
+]
 PROGNAME = 'fund-report'
 UNRESTRICTED_FUND = 'Conservancy'
 logger = logging.getLogger('conservancy_beancount.reports.fund')
 class ODSReport(core.BaseODS[FundPosts, None]):
     def __init__(self, start_date: datetime.date, stop_date: datetime.date) -> None:
         super().__init__()
         self.start_date = start_date
         self.stop_date = stop_date
     def section_key(self, row: FundPosts) -> None:
         return None
     def start_spreadsheet(self, *, expanded: bool=True) -> None:
         headers = [["Fund"], ["Balance as of", self.start_date.isoformat()]]
         if expanded:
             sheet_name = "With Breakdowns"
             headers += [["Income"], ["Expenses"], ["Equity"]]
         else:
             sheet_name = "Fund Report"
             headers += [["Additions"], ["Releases from", "Restrictions"]]
         headers.append(["Balance as of", self.stop_date.isoformat()])
         if expanded:
             headers += [
                 ["Of which", "Receivable"],
                 ["Of which", "Prepaid Expenses"],
                 ["Of which", "Payable"],
                 ["Of which", "Unearned Income"],
+            ]
         self.use_sheet(sheet_name)
         for header in headers:
             first_line = header[0]
             if first_line == 'Fund':
                 width = 2.0
             elif first_line == 'Balance as of':
                 width = 1.5
             elif first_line == 'Of which':
                 width = 1.3
             else:
                 width = 1.2
             col_style = self.column_style(width)
             self.sheet.addElement(odf.table.TableColumn(stylename=col_style))
         center_bold = self.merge_styles(self.style_centertext, self.style_bold)
         row = self.add_row(*(
             self.multiline_cell(header, stylename=center_bold)
             for header in headers
         ))
         row.firstChild.setAttribute(
             'stylename', self.merge_styles(self.style_endtext, self.style_bold),
+        )
         self.lock_first_row()
         self.lock_first_column()
         self.add_row()
         self.add_row(self.string_cell(
             f"Fund Report From {self.start_date.isoformat()} To {self.stop_date.isoformat()}",
             stylename=center_bold,
             numbercolumnsspanned=6,
         ))
         self.add_row()
     def end_spreadsheet(self) -> None:
         start_sheet = self.sheet
         self.set_open_sheet(self.sheet)
         self.start_spreadsheet(expanded=False)
         bal_indexes = [0, 1, 2, 4]
         totals = [core.MutableBalance() for _ in bal_indexes]
         threshold = Decimal('.5')
         for fund, balances in self.balances.items():
             balances = [balances[index] for index in bal_indexes]
             if (not all(bal.clean_copy(threshold).le_zero() for bal in balances)
                 and fund != UNRESTRICTED_FUND):
                 self.write_balances(fund, balances)
                 for total, bal in zip(totals, balances):
                     total += bal
         self.write_balances('', totals, self.merge_styles(
             self.border_style(core.Border.TOP, '.75pt'),
             self.border_style(core.Border.BOTTOM, '1.5pt', 'double'),
         ))
         self.document.spreadsheet.childNodes.reverse()
         self.sheet = start_sheet
     def _row_balances(self, accounts_map: AccountsMap) -> Iterator[core.Balance]:
         acct_order = ['Income', 'Expenses', 'Equity']
         key_order = [core.OPENING_BALANCE_NAME, *acct_order, core.ENDING_BALANCE_NAME]
         balances: Dict[str, core.Balance] = {key: core.MutableBalance() for key in key_order}
         for acct_s, balance in core.account_balances(accounts_map, acct_order):
             if acct_s in balances:
                 balances[acct_s] = balance
             else:
                 acct_root, _, _ = acct_s.partition(':')
                 balances[acct_root] += balance
         for key in key_order:
             if key == 'Expenses':
                 yield balances[key]
             else:
                 yield -balances[key]
         for info_key in INFO_ACCOUNTS:
             for _, balance in core.account_balances(accounts_map, [info_key]):
                 pass
             yield core.normalize_amount_func(info_key)(balance)
     def write_balances(self,
                        fund: str,
                        balances: Iterable[core.Balance],
                        style: Union[None, str, odf.style.Style]=None,
     ) -> odf.table.TableRow:
         return self.add_row(
             self.string_cell(fund, stylename=self.style_endtext),
             *(self.balance_cell(bal, stylename=style) for bal in balances),
+        )
     def write_row(self, row: FundPosts) -> None:
         fund, accounts_map = row
         self.balances[fund] = list(self._row_balances(accounts_map))
         if fund != UNRESTRICTED_FUND:
             self.write_balances(fund, self.balances[fund])
     def write(self, rows: Iterable[FundPosts]) -> None:
         self.balances: Dict[str, Sequence[core.Balance]] = collections.OrderedDict()
         super().write(rows)
         try:
             unrestricted = self.balances[UNRESTRICTED_FUND]
         except KeyError:
             pass
         else:
             self.add_row()
             self.write_balances("Unrestricted", unrestricted)
 class TextReport:
     def __init__(self,
                  start_date: datetime.date,
                  stop_date: datetime.date,
                  out_file: TextIO) -> None:
         self.start_date = start_date
         self.stop_date = stop_date
         self.out_file = out_file
     def _account_balances(self,
                           fund: str,
                           account_map: AccountsMap,
     ) -> Iterator[Tuple[str, Sequence[str]]]:
         total_fmt = f'{fund} balance as of {{}}'
         for acct_s, balance in core.account_balances(account_map, EQUITY_ACCOUNTS):
             if acct_s is core.OPENING_BALANCE_NAME:
                 acct_s = total_fmt.format(self.start_date.isoformat())
             elif acct_s is core.ENDING_BALANCE_NAME:
                 acct_s = total_fmt.format(self.stop_date.isoformat())
             yield acct_s, (-balance).format(None, sep='\0').split('\0')
         for _, account in core.sort_and_filter_accounts(account_map, INFO_ACCOUNTS):
             balance = account_map[account].stop_bal
             if not balance.is_zero():
                 balance = core.normalize_amount_func(account)(balance)
                 yield account, balance.format(None, sep='\0').split('\0')
     def write(self, rows: Iterable[FundPosts]) -> None:
         output = [
             line
             for fund, account_map in rows
             for line in self._account_balances(fund, account_map)
+        ]
         acct_width = max(len(acct_s) for acct_s, _ in output) + 2
         bal_width = max(len(s) for _, bal_s in output for s in bal_s)
         bal_width = max(bal_width, 8)
         line_fmt = f'{{:>{acct_width}}}  {{:>{bal_width}}}'
         print(line_fmt.replace('{:>', '{:^').format("ACCOUNT", "BALANCE"),
               file=self.out_file)
         fund_start = f' balance as of {self.start_date.isoformat()}'
         for acct_s, bal_seq in output:
             if acct_s.endswith(fund_start):
                 print(line_fmt.format('―' * acct_width, '―' * bal_width),
                       file=self.out_file)
             bal_iter = iter(bal_seq)
             print(line_fmt.format(acct_s, next(bal_iter)), file=self.out_file)
             for bal_s in bal_iter:
                 print(line_fmt.format('', bal_s), file=self.out_file)
 class ReportType(enum.Enum):
     TEXT = TextReport
     ODS = ODSReport
     TXT = TEXT
     SPREADSHEET = ODS
     @classmethod
     def from_arg(cls, s: str) -> 'ReportType':
         try:
             return cls[s.upper()]
         except KeyError:
             raise ValueError(f"no report type matches {s!r}") from None
 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.
 """)
     parser.add_argument(
         '--report-type', '-t',
         metavar='TYPE',
         type=ReportType.from_arg,
         help="""Type of report to generate. `text` gives a plain two-column text
 report listing accounts and balances over the period, and is the default when
 you search for a specific project/fund. `ods` produces a higher-level
 spreadsheet, meant to provide an overview of all funds, and is the default when
 you don't specify a project/fund.
 """)
     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 text reports, and a generated filename for ODS
 reports.
 """)
     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.report_type is None:
         if any(term.meta_key == 'project' for term in args.search_terms):
             args.report_type = ReportType.TEXT
         else:
             args.report_type = ReportType.ODS
     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()
     if args.stop_date is None:
         if args.start_date is None:
             args.stop_date = datetime.date.today()
         else:
             args.stop_date = diff_year(args.start_date, 1)
+            args.stop_date = cliutil.diff_year(args.start_date, 1)
     if args.start_date is None:
         args.start_date = diff_year(args.stop_date, -1)
+        args.start_date = cliutil.diff_year(args.stop_date, -1)
     returncode = 0
     books_loader = config.books_loader()
     if books_loader is None:
         entries, load_errors, _ = books.Loader.load_none(config.config_file_path())
     else:
         entries, load_errors, _ = books_loader.load_fy_range(args.start_date, args.stop_date)
     for error in load_errors:
         bc_printer.print_error(error, file=stderr)
         returncode |= cliutil.ReturnFlag.LOAD_ERRORS
     postings = (
         post
         for post in data.Posting.from_entries(entries)
         if post.meta.date < args.stop_date
+    )
     for search_term in args.search_terms:
         postings = search_term.filter_postings(postings)
     fund_postings = {
         key: related
         for key, related in core.RelatedPostings.group_by_meta(postings, 'project')
         if isinstance(key, str)
+    }
     period_cls = core.PeriodPostings.with_start_date(args.start_date)
     fund_map = collections.OrderedDict(
         (fund, dict(period_cls.group_by_account(fund_postings[fund])))
         for fund in sorted(fund_postings, key=lambda s: locale.strxfrm(s.casefold()))
+    )
     if not fund_map:
         logger.warning("no matching postings found to report")
         returncode |= cliutil.ReturnFlag.NOTHING_TO_REPORT
     elif args.report_type is ReportType.TEXT:
         out_file = cliutil.text_output(args.output_file, stdout)
         report = TextReport(args.start_date, args.stop_date, out_file)
         report.write(fund_map.items())
     else:
         ods_report = ODSReport(args.start_date, args.stop_date)
         ods_report.write(fund_map.items())
         if args.output_file is None:
             out_dir_path = config.repository_path() or Path()
             args.output_file = out_dir_path / 'FundReport_{}_{}.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)
         ods_report.save_file(ods_file)
     return cliutil.ReturnFlag.returncode(returncode)
 entry_point = cliutil.make_entry_point(__name__, PROGNAME)
 if __name__ == '__main__':
     exit(entry_point())

conservancy_beancount/reports/ledger.py

➞

Show inline comments

 """ledger.py - General ledger report from Beancount
 This tool produces a spreadsheet that shows postings in Beancount, organized
 by account.
 Specify the date range you want to report with the ``--begin`` and ``--end``
 options.
 Select the accounts you want to report with the ``--account`` option. You can
 specify this option multiple times. The report will include at least one sheet
 for each account you specify. Subaccounts will be reported on that sheet as
 well.
 Select the postings you want to report by passing metadata search terms in
 ``name=value`` format.
 Run ``ledger-report --help`` for abbreviations and other options.
 Examples
 --------
 Report all activity related to a given project::
     ledger-report project=NAME
 Get all Assets postings for a given month to help with reconciliation::
     ledger-report -a Assets -b 2018-05-01 -e 2018-06-01
 """
 # 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 argparse
 import collections
 import datetime
 import enum
 import itertools
 import operator
 import logging
 import sys
 from typing import (
     Any,
     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 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-allocation']),
         ('Equity', ['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']),
     ])
     # 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:
                     self.accounts.update(subaccounts)
                     self._require_sheet(acct_spec)
                 else:
                     account_roots_map = collections.defaultdict(list)
                     for account in data.Account.iter_accounts_by_classification(acct_spec):
                         self.accounts.add(account)
                         account_roots_map[account.root_part()].append(account)
                     if not account_roots_map:
                         raise ValueError("unknown account name or classification", acct_spec)
                     for root_part, accounts in account_roots_map.items():
                         start_count = min(account.count_parts() for account in accounts)
                         for count in range(start_count, 1, -1):
                             target = accounts[0].root_part(count)
                             if all(acct.root_part(count) == target for acct in accounts):
                                 self._require_sheet(target)
                                 break
                         else:
                             self._require_sheet(root_part)
     def _require_sheet(self, new_sheet: str) -> None:
         for index, sheet in enumerate(self.required_sheet_names):
             if new_sheet == sheet:
                 break
             elif new_sheet.startswith(sheet):
                 self.required_sheet_names.insert(index, new_sheet)
                 break
         else:
             self.required_sheet_names.append(new_sheet)
     def init_styles(self) -> None:
         super().init_styles()
         self.amount_column = self.column_style(1.2)
         self.default_column = self.column_style(1.5)
         self.column_styles: Mapping[str, Union[str, odf.style.Style]] = {
             'Account': self.column_style(2),  # for TransactionODS
             'Date': '',
             'Description': self.column_style(2),
             'Original Amount': self.amount_column,
             'Booked Amount': self.amount_column,
             data.Metadata.human_name('project'): self.amount_column,
             data.Metadata.human_name('rt-id'): self.amount_column,
+        }
     @classmethod
     def _group_tally(
             cls,
             tally_by_account: PostTally,
             key: Callable[[data.Account], Optional[str]],
     ) -> Dict[str, PostTally]:
         retval: Dict[str, PostTally] = collections.defaultdict(list)
         for count, account in tally_by_account:
             item_key = key(account)
             if item_key is not None:
                 retval[item_key].append((count, account))
         return retval
     @classmethod
     def _split_sheet(
             cls,
             tally_by_account: PostTally,
             sheet_size: int,
             sheet_name: str,
     ) -> Iterator[str]:
         total = 0
         for index, (count, account) in enumerate(tally_by_account):
             total += count
             if total > sheet_size:
                 break
         else:
             # All the accounts fit in this sheet.
             yield sheet_name
             return
         if index == 0 and len(tally_by_account) == 1:
             # With one account, we can't split any further, so warn and stop.
             logger.warning(
                 "%s has %s rows, over size %s",
                 account, f'{count:,g}', f'{sheet_size:,g}',
+            )
             yield sheet_name
             return
         group_func = operator.methodcaller('root_part', sheet_name.count(':') + 2)
         maybe_split = cls._group_tally(tally_by_account[:index], group_func)
         must_split = cls._group_tally(tally_by_account[index:], group_func)
         for subkey, must_split_tally in sorted(must_split.items()):
             split_names = cls._split_sheet(
                 maybe_split.get(subkey, []) + must_split_tally, sheet_size, subkey,
+            )
             # We must be willing to split out at least as many sheets as there
             # are accounts that didn't fit. Do that first.
             yield from itertools.islice(split_names, len(must_split_tally))
             # After that, we can be in one of two cases:
             # 1. There is no next sheet. All the accounts, including the
             #    maybe_splits and must_splits, fit on planned subsheets.
             #    Update state to note we don't need a sheet for them anymore.
             # 2. The next sheet is named `subkey`, and is planned to include
             #    all of our maybe_split accounts. However, we don't need to
             #    yield that sheet name, because those accounts already fit in
             #    the sheet we're planning, and it would be a needless split.
             next_sheet_name = next(split_names, None)
             if next_sheet_name is None:
                 maybe_split.pop(subkey, None)
             else:
                 assert next_sheet_name == subkey
                 assert not any(split_names)
         if maybe_split:
             yield sheet_name
     @classmethod
     def plan_sheets(
             cls,
             tally_by_account: Mapping[data.Account, int],
             base_sheets: Sequence[str],
             sheet_size: int,
     ) -> Sequence[str]:
         sorted_tally: PostTally = [
             (count, account)
             for account, count in tally_by_account.items()
+        ]
         sorted_tally.sort()
         split_tally = cls._group_tally(
             sorted_tally,
             operator.methodcaller('is_under', *base_sheets),
+        )
         return [
             sheet_name
             for key in base_sheets
             for sheet_name in cls._split_sheet(split_tally[key], sheet_size, key)
+        ]
     # The write() override does its own sectioning by account, so this method
     # goes unused.
     def section_key(self, row: data.Posting) -> None:
         return None
     def metadata_columns_for(self, sheet_name: str) -> Sequence[str]:
         columns_key = data.Account(sheet_name).is_under(*self.ACCOUNT_COLUMNS)
         # columns_key must not be None because ACCOUNT_COLUMNS has an entry
         # for all five root accounts.
         assert columns_key is not None
         return self.ACCOUNT_COLUMNS[columns_key]
     def start_sheet(self, sheet_name: str) -> None:
         self.use_sheet(sheet_name.replace(':', ' '))
         self.metadata_columns = self.metadata_columns_for(sheet_name)
         self.sheet_columns: Sequence[str] = [
             *self.CORE_COLUMNS,
             *(data.Metadata.human_name(meta_key) for meta_key in self.metadata_columns),
+        ]
         for col_name in self.sheet_columns:
             self.sheet.addElement(odf.table.TableColumn(
                 stylename=self.column_styles.get(col_name, self.default_column),
             ))
         self.add_row(*(
             self.string_cell(col_name, stylename=self.style_bold)
             for col_name in self.sheet_columns
         ))
         self.lock_first_row()
     def _write_total_row(self,
                          date: datetime.date,
                          description: str,
                          balance: core.Balance,
     ) -> None:
         cells: List[odf.table.TableCell] = []
         for column in self.CORE_COLUMNS:
             if column == 'Date':
                 cell = self.date_cell(date, stylename=self.merge_styles(
                     self.style_bold, self.style_date,
                 ))
             elif column == 'Description':
                 cell = self.string_cell(description, stylename=self.style_bold)
             elif column == 'Booked Amount':
                 cell = self.balance_cell(self.norm_func(balance), stylename=self.style_bold)
             else:
                 cell = odf.table.TableCell()
             cells.append(cell)
         self.add_row(*cells)
     def _report_section_balance(self, key: data.Account, date_key: str) -> None:
         related = self.account_groups[key]
         if date_key == 'start':
             if not key.keeps_balance():
                 return
             date = self.date_range.start
             balance = related.start_bal
             description = "Opening Balance"
         else:
             date = self.date_range.stop
             if key.keeps_balance():
                 balance = related.stop_bal
                 description = "Ending Balance"
             else:
                 balance = related.period_bal
                 description = "Period Total"
         self._write_total_row(date, description, balance)
     def write_header(self, key: data.Account) -> None:
         self.add_row()
         self.add_row(
             odf.table.TableCell(),
             self.string_cell(
                 f"{key} {self.report_name}"
                 f" From {self.date_range.start.isoformat()}"
                 f" To {self.date_range.stop.isoformat()}",
                 stylename=self.style_bold,
                 numbercolumnsspanned=len(self.sheet_columns) - 1,
             ),
+        )
     def write_balance_sheet(self) -> None:
         self.use_sheet("Balance")
         self.sheet.addElement(odf.table.TableColumn(stylename=self.column_style(3)))
         self.sheet.addElement(odf.table.TableColumn(stylename=self.column_style(1.5)))
         self.add_row(
             self.string_cell("Account", stylename=self.style_bold),
             self.string_cell("Balance", stylename=self.style_bold),
+        )
         self.lock_first_row()
         self.add_row()
         self.add_row(self.string_cell(
             f"Ledger From {self.date_range.start.isoformat()}"
             f" To {self.date_range.stop.isoformat()}",
             stylename=self.merge_styles(self.style_centertext, self.style_bold),
             numbercolumnsspanned=2,
         ))
         self.add_row()
         for account, balance in core.account_balances(self.account_groups):
             if account is core.OPENING_BALANCE_NAME:
                 text = f"Balance as of {self.date_range.start.isoformat()}"
                 style = self.merge_styles(self.style_bold, self.style_endtext)
             elif account is core.ENDING_BALANCE_NAME:
                 text = f"Balance as of {self.date_range.stop.isoformat()}"
                 style = self.merge_styles(self.style_bold, self.style_endtext)
             else:
                 text = account
                 style = self.style_endtext
             self.add_row(
                 self.string_cell(text, stylename=style),
                 self.balance_cell(-balance, stylename=style),
+            )
     def _account_tally(self, account: data.Account) -> int:
         return len(self.account_groups[account])
     def write_entries(self, account: data.Account, rows: Iterable[data.Posting]) -> None:
         for row in rows:
             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 ''),
                 self.string_cell(row.meta.txn.narration),
                 amount_cell,
                 self.currency_cell(self.norm_func(row.at_cost())),
                 *(self.meta_links_cell(row.meta.report_links(key))
                   if key in data.LINK_METADATA
                   else self.string_cell(row.meta.get(key, ''))
                   for key in self.metadata_columns),
+            )
     def write(self, rows: Iterable[data.Posting]) -> None:
         related_cls = core.PeriodPostings.with_start_date(self.date_range.start)
         self.account_groups = dict(related_cls.group_by_account(
             post for post in rows if post.meta.date < self.date_range.stop
         ))
         for empty_acct in self.accounts.difference(self.account_groups):
             self.account_groups[empty_acct] = related_cls()
         self.write_balance_sheet()
         tally_by_account_iter = (
             (account, self._account_tally(account))
             for account in self.accounts
+        )
         tally_by_account = {
             # 3 for the rows generated by start_section+end_section
             account: count + 3
             for account, count in tally_by_account_iter
+        }
         sheet_names = self.plan_sheets(
             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,
         ):
             if not account.is_open_on_date(self.date_range.start):
                 continue
             while using_sheet_index < sheet_index:
                 using_sheet_index += 1
                 self.start_sheet(sheet_names[using_sheet_index])
             self.norm_func = core.normalize_amount_func(account)
             postings = self.account_groups[account]
             if postings:
                 totals_set = self.totals_with_entries
             else:
                 totals_set = self.totals_without_entries
             want_totals = account.is_under(*totals_set) is not None
             if postings or want_totals:
                 self.write_header(account)
             if want_totals:
                 self._report_section_balance(account, 'start')
             self.write_entries(account, postings)
             if want_totals:
                 self._report_section_balance(account, 'stop')
         for index in range(using_sheet_index + 1, len(sheet_names)):
             self.start_sheet(sheet_names[index])
 class TransactionFilter(enum.IntFlag):
     ZERO = 1
     CREDIT = 2
     DEBIT = 4
     ALL = ZERO | CREDIT | DEBIT
     @classmethod
     def from_arg(cls, s: str) -> 'TransactionFilter':
         try:
             return cls[s.upper()]
         except KeyError:
             raise ValueError(f"unknown transaction filter {s!r}")
     @classmethod
     def post_flag(cls, post: data.Posting) -> int:
         norm_func = core.normalize_amount_func(post.account)
         number = norm_func(post.units.number)
         if not number:
             return cls.ZERO
         elif number > 0:
             return cls.CREDIT
         else:
             return cls.DEBIT
 class TransactionODS(LedgerODS):
     CORE_COLUMNS: Sequence[str] = [
         'Date',
         'Description',
         'Account',
         data.Metadata.human_name('entity'),
         'Original Amount',
         'Booked Amount',
+    ]
     METADATA_COLUMNS: Sequence[str] = [
         'project',
         'rt-id',
         'receipt',
         'check',
         'invoice',
         'contract',
         'approval',
         'paypal-id',
         'check-number',
         'bank-statement',
+    ]
     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,
                  txn_filter: int=TransactionFilter.ALL,
     ) -> None:
         super().__init__(
             start_date,
             stop_date,
             accounts,
             rt_wrapper,
             sheet_size,
             totals_with_entries,
             totals_without_entries,
+        )
         self.txn_filter = txn_filter
         if self.txn_filter == TransactionFilter.CREDIT:
             self.report_name = "Receipts"
         elif self.txn_filter == TransactionFilter.DEBIT:
             self.report_name = "Disbursements"
         else:
             self.report_name = "Transactions"
     def _wanted_txns(self, postings: Iterable[data.Posting]) -> Iterator[Transaction]:
         last_txn: Optional[Transaction] = None
         for post in postings:
             txn = post.meta.txn
             if (txn is not last_txn
                 and TransactionFilter.post_flag(post) & self.txn_filter):
                 yield txn
                 last_txn = txn
     def metadata_columns_for(self, sheet_name: str) -> Sequence[str]:
         return self.METADATA_COLUMNS
     def write_balance_sheet(self) -> None:
         return
     def _report_section_balance(self, key: data.Account, date_key: str) -> None:
         if self.txn_filter == TransactionFilter.ALL:
             super()._report_section_balance(key, date_key)
         elif date_key == 'stop':
             balance = core.Balance(
                 post.at_cost()
                 for txn in self._wanted_txns(self.account_groups[key])
                 for post in data.Posting.from_txn(txn)
                 if post.account == key
+            )
             self._write_total_row(self.date_range.stop, "Period Activity", balance)
     def _account_tally(self, account: data.Account) -> int:
         return sum(len(txn.postings)
                    for txn in self._wanted_txns(self.account_groups[account]))
     def write_entries(self, account: data.Account, rows: Iterable[data.Posting]) -> None:
         for txn in self._wanted_txns(rows):
             post_list = list(data.Posting.from_txn(txn))
             post_list.sort(key=lambda post: (
 if post.account == account else 1,
                 -abs(post.at_cost().number),
             ))
             postings = iter(post_list)
             post1 = next(postings)
             if post1.cost is None:
                 amount_cell = odf.table.TableCell()
             else:
                 amount_cell = self.currency_cell(self.norm_func(post1.units))
             self.add_row(
                 self.date_cell(txn.date),
                 self.string_cell(txn.narration),
                 self.string_cell(post1.account),
                 self.string_cell(post1.meta.get('entity') or ''),
                 amount_cell,
                 self.currency_cell(self.norm_func(post1.at_cost())),
                 *(self.meta_links_cell(post1.meta.report_links(key))
                   if key in data.LINK_METADATA
                   else self.string_cell(post1.meta.get(key, ''))
                   for key in self.metadata_columns),
+            )
             for post in postings:
                 meta_cells: List[odf.table.TableCell] = []
                 for meta_key in self.metadata_columns:
                     try:
                         dup = post.meta[meta_key] is txn.meta[meta_key]
                     except KeyError:
                         dup = False
                     if dup:
                         meta_cell = odf.table.TableCell()
                     elif meta_key in data.LINK_METADATA:
                         meta_cell = self.meta_links_cell(post.meta.report_links(meta_key))
                     else:
                         meta_cell = self.string_cell(post.meta.get(meta_key, ''))
                     meta_cells.append(meta_cell)
                 if post.cost is None:
                     amount_cell = odf.table.TableCell()
                 else:
                     amount_cell = self.currency_cell(self.norm_func(post.units))
                 self.add_row(
                     odf.table.TableCell(),
                     odf.table.TableCell(),
                     self.string_cell(post.account),
                     self.string_cell(post.meta.get('entity') or ''),
                     amount_cell,
                     self.currency_cell(self.norm_func(post.at_cost())),
                     *meta_cells,
+                )
 class CashReportAction(argparse.Action):
     def __call__(self,
                  parser: argparse.ArgumentParser,
                  namespace: argparse.Namespace,
                  values: Union[Sequence[Any], str, None]=None,
                  option_string: Optional[str]=None,
     ) -> None:
         namespace.txn_filter = self.const
         if namespace.accounts is None:
             namespace.accounts = []
         namespace.accounts.append('Assets:PayPal')
         namespace.accounts.append('Cash')
         if namespace.stop_date is None:
             namespace.stop_date = datetime.date.today()
 def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(prog=PROGNAME)
     cliutil.add_version_argument(parser)
     parser.add_argument(
         '--disbursements',
         action=CashReportAction,
         const=TransactionFilter.DEBIT,
         nargs=0,
         help="""Shortcut to set all the necessary options to generate a cash
 disbursements report.
 """)
     parser.add_argument(
         '--receipts',
         action=CashReportAction,
         const=TransactionFilter.CREDIT,
         nargs=0,
         help="""Shortcut to set all the necessary options to generate a cash
 receipts report.
 """)
     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(
         '--transactions', '-t',
         dest='txn_filter',
         metavar='TYPE',
         type=TransactionFilter.from_arg,
         help="""Report whole transactions rather than individual postings.
 The type argument selects which type of transactions to report. Choices are
 credit, debit, or all.
 """)
     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(
         '--show-totals', '-S',
         metavar='ACCOUNT',
         action='append',
         help="""When entries for this account appear in the report, include
 account balance(s) as well. You can specify this option multiple times. Pass in
 a part of the account hierarchy. The default is all accounts.
 """)
     parser.add_argument(
         '--add-totals', '-T',
         metavar='ACCOUNT',
         action='append',
         help="""When an account could be included in the report but does not
 have any entries in the date range, include a header and account balance(s) for
 it. You can specify this option multiple times. Pass in a part of the account
 hierarchy. 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',
         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 args.add_totals is None and args.search_terms:
         args.add_totals = []
     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)
     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)
+        args.start_date = cliutil.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)
+        args.stop_date = cliutil.diff_year(args.start_date, 1)
     returncode = 0
     books_loader = config.books_loader()
     if books_loader is None:
         entries, load_errors, options = books.Loader.load_none(config.config_file_path())
     else:
         entries, load_errors, options = books_loader.load_fy_range(args.start_date, args.stop_date)
     for error in load_errors:
         bc_printer.print_error(error, file=stderr)
         returncode |= cliutil.ReturnFlag.LOAD_ERRORS
     data.Account.load_from_books(entries, options)
     postings = data.Posting.from_entries(entries)
     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")
     try:
         if args.txn_filter is None:
             report = LedgerODS(
                 args.start_date,
                 args.stop_date,
                 args.accounts,
                 rt_wrapper,
                 args.sheet_size,
                 args.show_totals,
                 args.add_totals,
+            )
         else:
             report = TransactionODS(
                 args.start_date,
                 args.stop_date,
                 args.accounts,
                 rt_wrapper,
                 args.sheet_size,
                 args.show_totals,
                 args.add_totals,
                 args.txn_filter,
+            )
     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 |= cliutil.ReturnFlag.NOTHING_TO_REPORT
     if args.output_file is None:
         out_dir_path = config.repository_path() or Path()
         args.output_file = out_dir_path / '{}Report_{}_{}.ods'.format(
             report.report_name,
             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 cliutil.ReturnFlag.returncode(returncode)
 entry_point = cliutil.make_entry_point(__name__, PROGNAME)
 if __name__ == '__main__':
     exit(entry_point())

tests/test_cliutil.py

➞

Show inline comments

 """Test CLI utilities"""
 # 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 argparse
 import datetime
 import errno
 import io
 import inspect
 import logging
 import os
 import re
 import sys
 import traceback
 import pytest
 from pathlib import Path
 from conservancy_beancount import cliutil
 FILE_NAMES = ['-foobar', '-foo.bin']
 STREAM_PATHS = [None, Path('-')]
 class MockTraceback:
     def __init__(self, stack=None, index=0):
         if stack is None:
             stack = inspect.stack(context=False)
         self._stack = stack
         self._index = index
         frame_record = self._stack[self._index]
         self.tb_frame = frame_record.frame
         self.tb_lineno = frame_record.lineno
     @property
     def tb_next(self):
         try:
             return type(self)(self._stack, self._index + 1)
         except IndexError:
             return None
 @pytest.fixture(scope='module')
 def argparser():
     parser = argparse.ArgumentParser(prog='test_cliutil')
     cliutil.add_loglevel_argument(parser)
     cliutil.add_version_argument(parser)
     return parser
 @pytest.mark.parametrize('path_name', FILE_NAMES)
 def test_bytes_output_path(path_name, tmp_path):
     path = tmp_path / path_name
     stream = io.BytesIO()
     actual = cliutil.bytes_output(path, stream)
     assert actual is not stream
     assert str(actual.name) == str(path)
     assert 'w' in actual.mode
     assert 'b' in actual.mode
 @pytest.mark.parametrize('path', STREAM_PATHS)
 def test_bytes_output_stream(path):
     stream = io.BytesIO()
     actual = cliutil.bytes_output(path, stream)
     assert actual is stream
 @pytest.mark.parametrize('year,month,day', [
     (2000, 1, 1),
     (2016, 2, 29),
     (2020, 12, 31),
 ])
 def test_date_arg_valid(year, month, day):
     expected = datetime.date(year, month, day)
     assert cliutil.date_arg(expected.isoformat()) == expected
 @pytest.mark.parametrize('arg', [
     '2000',
     '20-02-12',
     '2019-02-29',
     'two thousand',
 ])
 def test_date_arg_invalid(arg):
     with pytest.raises(ValueError):
         cliutil.date_arg(arg)
 @pytest.mark.parametrize('year', [
 ,
 ,
 ,
 ])
 def test_year_or_date_arg_year(year):
     assert cliutil.year_or_date_arg(str(year)) == year
 @pytest.mark.parametrize('year,month,day', [
     (2000, 1, 1),
     (2016, 2, 29),
     (2020, 12, 31),
 ])
 def test_year_or_date_arg_date(year, month, day):
     expected = datetime.date(year, month, day)
     assert cliutil.year_or_date_arg(expected.isoformat()) == expected
 @pytest.mark.parametrize('arg', [
     '-1',
     str(sys.maxsize),
     'MMDVIII',
     '2019-02-29',
 ])
 def test_year_or_date_arg_invalid(arg):
     with pytest.raises(ValueError):
         cliutil.year_or_date_arg(arg)
 @pytest.mark.parametrize('func_name', [
     'bytes_output',
     'text_output',
 ])
 def test_default_output(func_name):
     actual = getattr(cliutil, func_name)()
     assert actual.fileno() == sys.stdout.fileno()
 @pytest.mark.parametrize('path_name', FILE_NAMES)
 def test_text_output_path(path_name, tmp_path):
     path = tmp_path / path_name
     stream = io.StringIO()
     actual = cliutil.text_output(path, stream)
     assert actual is not stream
     assert str(actual.name) == str(path)
     assert 'w' in actual.mode
 @pytest.mark.parametrize('path', STREAM_PATHS)
 def test_text_output_stream(path):
     stream = io.StringIO()
     actual = cliutil.text_output(path, stream)
     assert actual is stream
 @pytest.mark.parametrize('errnum', [
     errno.EACCES,
     errno.EPERM,
     errno.ENOENT,
 ])
 def test_excepthook_oserror(errnum, caplog):
     error = OSError(errnum, os.strerror(errnum), 'TestFilename')
     with pytest.raises(SystemExit) as exc_check:
         cliutil.ExceptHook()(type(error), error, None)
     assert exc_check.value.args[0] == 4
     assert caplog.records
     for log in caplog.records:
         assert log.levelname == 'CRITICAL'
         assert log.message == f"I/O error: {error.filename}: {error.strerror}"
 @pytest.mark.parametrize('exc_type', [
     AttributeError,
     RuntimeError,
     ValueError,
 ])
 def test_excepthook_bug(exc_type, caplog):
     error = exc_type("test message")
     with pytest.raises(SystemExit) as exc_check:
         cliutil.ExceptHook()(exc_type, error, None)
     assert exc_check.value.args[0] == 3
     assert caplog.records
     for log in caplog.records:
         assert log.levelname == 'CRITICAL'
         assert log.message == f"internal {exc_type.__name__}: {error.args[0]}"
 def test_excepthook_traceback(caplog):
     error = KeyError('test')
     args = (type(error), error, MockTraceback())
     caplog.set_level(logging.DEBUG)
     with pytest.raises(SystemExit) as exc_check:
         cliutil.ExceptHook()(*args)
     assert caplog.records
     assert caplog.records[-1].message == ''.join(traceback.format_exception(*args))
 @pytest.mark.parametrize('arg,expected', [
     ('debug', logging.DEBUG),
     ('info', logging.INFO),
     ('warning', logging.WARNING),
     ('warn', logging.WARNING),
     ('error', logging.ERROR),
     ('err', logging.ERROR),
     ('critical', logging.CRITICAL),
     ('crit', logging.CRITICAL),
 ])
 def test_loglevel_argument(argparser, arg, expected):
     for method in ['lower', 'title', 'upper']:
         args = argparser.parse_args(['--loglevel', getattr(arg, method)()])
         assert args.loglevel is expected
 def test_setup_logger():
     stream = io.StringIO()
     logger = cliutil.setup_logger(
         'test_cliutil', stream, '%(name)s %(levelname)s: %(message)s',
+    )
     logger.critical("test crit")
     assert stream.getvalue() == "test_cliutil CRITICAL: test crit\n"
 @pytest.mark.parametrize('arg', [
     '--license',
     '--version',
     '--copyright',
 ])
 def test_version_argument(argparser, capsys, arg):
     with pytest.raises(SystemExit) as exc_check:
         args = argparser.parse_args(['--version'])
     assert exc_check.value.args[0] == 0
     stdout, _ = capsys.readouterr()
     lines = iter(stdout.splitlines())
     assert re.match(r'^test_cliutil version \d+\.\d+\.\d+', next(lines, "<EOF>"))
 @pytest.mark.parametrize('date,diff,expected', [
     (datetime.date(2010, 2, 28), 0, datetime.date(2010, 2, 28)),
     (datetime.date(2010, 2, 28), 1, datetime.date(2011, 2, 28)),
     (datetime.date(2010, 2, 28), 2, datetime.date(2012, 2, 28)),
     (datetime.date(2010, 2, 28), -1, datetime.date(2009, 2, 28)),
     (datetime.date(2010, 2, 28), -2, datetime.date(2008, 2, 28)),
     (datetime.date(2012, 2, 29), 2, datetime.date(2014, 3, 1)),
     (datetime.date(2012, 2, 29), 4, datetime.date(2016, 2, 29)),
     (datetime.date(2012, 2, 29), -2, datetime.date(2010, 2, 28)),
     (datetime.date(2012, 2, 29), -4, datetime.date(2008, 2, 29)),
     (datetime.date(2010, 3, 1), 1, datetime.date(2011, 3, 1)),
     (datetime.date(2010, 3, 1), 2, datetime.date(2012, 3, 1)),
     (datetime.date(2010, 3, 1), -1, datetime.date(2009, 3, 1)),
     (datetime.date(2010, 3, 1), -2, datetime.date(2008, 3, 1)),
 ])
 def test_diff_year(date, diff, expected):
     assert cliutil.diff_year(date, diff) == expected

0 comments (0 inline, 0 general)