Changeset - 461536453835
[Not reviewed]
0 8 0
Brett Smith - 4 years ago 2020-07-27 14:54:04
brettcsmith@brettcsmith.org
cliutil: Add ReturnFlag.

Take this opportunity to re-standardize flag values now that it's clear
what's most common.
8 files changed with 37 insertions and 38 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/cliutil.py
Show inline comments
...
 
@@ -87,96 +87,116 @@ class ExceptHook:
 
            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
conservancy_beancount/reports/accrual.py
Show inline comments
...
 
@@ -590,204 +590,197 @@ class OutgoingReport(BaseReport):
 
                last_txn = txn
 
                txn = self.rt_wrapper.txn_with_urls(txn, '{}')
 
                # Suppress payment-method metadata from the report.
 
                txn.meta.pop('payment-method', None)
 
                for txn_post in txn.postings:
 
                    if txn_post.meta:
 
                        txn_post.meta.pop('payment-method', None)
 
                yield bc_printer.format_entry(txn)
 

	
 
        cf_targets = {
 
            'payment-amount': payment_amount,
 
            'payment-method': (self._get_payment_method(posts, ticket_id)
 
                               or ticket.get('CF.{payment-method}')),
 
            'payment-to': payment_to,
 
        }
 

	
 
        cf_updates = {
 
            f'CF_{key}': value
 
            for key, value in cf_targets.items()
 
            if ticket.get(f'CF.{{{key}}}') != value
 
        }
 
        if cf_updates:
 
            try:
 
                ok = self.rt_client.edit_ticket(ticket_id, **cf_updates)
 
            except rt.RtError:
 
                self.logger.debug("RT exception on edit_ticket", exc_info=True)
 
                ok = False
 
            if not ok:
 
                self.logger.warning("failed to set custom fields for rt:%s", ticket_id)
 

	
 

	
 
class ReportType(enum.Enum):
 
    AGING = AgingReport
 
    BALANCE = BalanceReport
 
    OUTGOING = OutgoingReport
 
    AGE = AGING
 
    BAL = BALANCE
 
    OUT = OUTGOING
 
    OUTGOINGS = OUTGOING
 

	
 
    @classmethod
 
    def by_name(cls, name: str) -> 'ReportType':
 
        try:
 
            return cls[name.upper()]
 
        except KeyError:
 
            raise ValueError(f"unknown report type {name!r}") from None
 

	
 

	
 
class ReturnFlag(enum.IntFlag):
 
    LOAD_ERRORS = 1
 
    # 2 was used in the past, it can probably be reclaimed.
 
    REPORT_ERRORS = 4
 
    NOTHING_TO_REPORT = 8
 

	
 

	
 
def filter_search(postings: Iterable[data.Posting],
 
                  search_terms: Iterable[cliutil.SearchTerm],
 
) -> Iterable[data.Posting]:
 
    accounts = tuple(AccrualAccount.account_names())
 
    postings = (post for post in postings if post.account.is_under(*accounts))
 
    for query in search_terms:
 
        postings = query.filter_postings(postings)
 
    return postings
 

	
 
def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
 
    parser = argparse.ArgumentParser(prog=PROGNAME)
 
    cliutil.add_version_argument(parser)
 
    parser.add_argument(
 
        '--report-type', '-t',
 
        metavar='NAME',
 
        type=ReportType.by_name,
 
        help="""The type of report to generate, one of `aging`, `balance`, or
 
`outgoing`. If not specified, the default is `aging` when no search terms are
 
given, `outgoing` for search terms that return a single outstanding payable,
 
and `balance` any other time.
 
""")
 
    parser.add_argument(
 
        '--since',
 
        metavar='YEAR',
 
        type=int,
 
        default=0,
 
        help="""How far back to search the books for related transactions.
 
You can either specify a fiscal year, or a negative offset from the current
 
fiscal year, to start loading entries from. The default is to load the current,
 
unaudited books.
 
""")
 
    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 the balance and outgoing reports, and a generated
 
filename for other reports.
 
""")
 
    cliutil.add_loglevel_argument(parser)
 
    parser.add_argument(
 
        'search_terms',
 
        metavar='FILTER',
 
        type=cliutil.SearchTerm.arg_parser('invoice', 'rt-id'),
 
        nargs=argparse.ZERO_OR_MORE,
 
        help="""Report on accruals 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 link, including an RT attachment link in
 
`TIK/ATT` format, is a shortcut for `invoice=LINK`.
 
""")
 
    args = parser.parse_args(arglist)
 
    if args.report_type is None and not any(
 
            term.meta_key == 'invoice' or term.meta_key == 'rt-id'
 
            for term in args.search_terms
 
    ):
 
        args.report_type = ReportType.AGING
 
    return args
 

	
 
def main(arglist: Optional[Sequence[str]]=None,
 
         stdout: TextIO=sys.stdout,
 
         stderr: TextIO=sys.stderr,
 
         config: Optional[configmod.Config]=None,
 
) -> int:
 
    args = parse_arguments(arglist)
 
    cliutil.set_loglevel(logger, args.loglevel)
 
    if config is None:
 
        config = configmod.Config()
 
        config.load_file()
 

	
 
    returncode = 0
 
    books_loader = config.books_loader()
 
    if books_loader is None:
 
        entries, load_errors, _ = books.Loader.load_none(config.config_file_path())
 
    else:
 
        load_since = None if args.report_type == ReportType.AGING else args.since
 
        entries, load_errors, _ = books_loader.load_all(load_since)
 
    filters.remove_opening_balance_txn(entries)
 
    for error in load_errors:
 
        bc_printer.print_error(error, file=stderr)
 
        returncode |= ReturnFlag.LOAD_ERRORS
 
        returncode |= cliutil.ReturnFlag.LOAD_ERRORS
 

	
 
    postings = list(filter_search(
 
        data.Posting.from_entries(entries), args.search_terms,
 
    ))
 
    if not postings:
 
        logger.warning("no matching entries found to report")
 
        returncode |= ReturnFlag.NOTHING_TO_REPORT
 
        returncode |= cliutil.ReturnFlag.NOTHING_TO_REPORT
 
    # groups is a mapping of metadata value strings to AccrualPostings.
 
    # The keys are basically arbitrary, the report classes don't rely on them,
 
    # but they do help symbolize what's being grouped.
 
    # For the outgoing approval report, groups maps rt-id link strings to
 
    # associated accruals.
 
    # For all other reports, groups comes from AccrualReport.make_consistent().
 
    groups: PostGroups
 
    if args.report_type is None or args.report_type is ReportType.OUTGOING:
 
        groups = dict(AccrualPostings.group_by_first_meta_link(postings, 'rt-id'))
 
        if args.report_type is None and len(groups) == 1:
 
            key = next(iter(groups))
 
            group = groups[key]
 
            account = group[0].account
 
            if (AccrualAccount.by_account(account) is AccrualAccount.PAYABLE
 
                and all(post.account == account for post in group)
 
                and not group.balance().ge_zero()
 
                and key):  # Make sure we have a usable rt-id
 
                args.report_type = ReportType.OUTGOING
 
    if args.report_type is not ReportType.OUTGOING:
 
        groups = dict(AccrualPostings.make_consistent(postings))
 
    if args.report_type is not ReportType.AGING:
 
        groups = {
 
            key: posts for key, posts in groups.items() if not posts.is_paid()
 
        } or groups
 
    del postings
 

	
 
    report: Optional[BaseReport] = None
 
    output_path: Optional[Path] = None
 
    if args.report_type is ReportType.AGING:
 
        rt_wrapper = config.rt_wrapper()
 
        if rt_wrapper is None:
 
            logger.error("unable to generate aging report: RT client is required")
 
        else:
 
            now = datetime.datetime.now()
 
            if args.output_file is None:
 
                out_dir_path = config.repository_path() or Path()
 
                args.output_file = out_dir_path / now.strftime('AgingReport_%Y-%m-%d_%H:%M.ods')
 
                logger.info("Writing report to %s", args.output_file)
 
            out_bin = cliutil.bytes_output(args.output_file, stdout)
 
            report = AgingReport(rt_wrapper, out_bin)
 
    elif args.report_type is ReportType.OUTGOING:
 
        rt_wrapper = config.rt_wrapper()
 
        if rt_wrapper is None:
 
            logger.error("unable to generate outgoing report: RT client is required")
 
        else:
 
            out_file = cliutil.text_output(args.output_file, stdout)
 
            report = OutgoingReport(rt_wrapper, out_file)
 
    else:
 
        out_file = cliutil.text_output(args.output_file, stdout)
 
        report = BalanceReport(out_file)
 

	
 
    if report is None:
 
        returncode |= ReturnFlag.REPORT_ERRORS
 
        returncode |= 16
 
    else:
 
        report.run(groups)
 
    return 0 if returncode == 0 else 16 + returncode
 
    return cliutil.ReturnFlag.returncode(returncode)
 

	
 
entry_point = cliutil.make_entry_point(__name__, PROGNAME)
 

	
 
if __name__ == '__main__':
 
    exit(entry_point())
conservancy_beancount/reports/fund.py
Show inline comments
...
 
@@ -238,190 +238,185 @@ class TextReport:
 
            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
 

	
 

	
 
class ReturnFlag(enum.IntFlag):
 
    LOAD_ERRORS = 1
 
    NOTHING_TO_REPORT = 8
 

	
 

	
 
def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
 
    parser = argparse.ArgumentParser(prog=PROGNAME)
 
    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)
 
    if args.start_date is None:
 
        args.start_date = 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 |= ReturnFlag.LOAD_ERRORS
 
        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 |= ReturnFlag.NOTHING_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 0 if returncode == 0 else 16 + returncode
 
    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
...
 
@@ -570,101 +570,96 @@ class TransactionODS(LedgerODS):
 
            ))
 
            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 ReturnFlag(enum.IntFlag):
 
    LOAD_ERRORS = 1
 
    NOTHING_TO_REPORT = 8
 

	
 

	
 
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,
...
 
@@ -745,104 +740,104 @@ metadata to match. A single ticket number is a shortcut for
 
                '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)
 
        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:
 
        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 |= ReturnFlag.LOAD_ERRORS
 
        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 |= ReturnFlag.NOTHING_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 0 if returncode == 0 else 16 + returncode
 
    return cliutil.ReturnFlag.returncode(returncode)
 

	
 
entry_point = cliutil.make_entry_point(__name__, PROGNAME)
 

	
 
if __name__ == '__main__':
 
    exit(entry_point())
conservancy_beancount/tools/opening_balances.py
Show inline comments
...
 
@@ -97,163 +97,159 @@ def quantize_amount(
 
) -> data.Amount:
 
    return amount._replace(number=amount.number.quantize(exp, rounding=rounding))
 

	
 
class AccountWithFund(NamedTuple):
 
    account: data.Account
 
    fund: Optional[MetaValue]
 

	
 
    def sortkey(self) -> Hashable:
 
        account, fund = self
 
        return (
 
            0 if fund is None else 1,
 
            locale.strxfrm(account),
 
            locale.strxfrm(str(fund).casefold()),
 
        )
 

	
 

	
 
class Posting(data.Posting):
 
    @staticmethod
 
    def _position_sortkey(position: Position) -> str:
 
        units, cost = position
 
        if cost is None:
 
            # Beancount type-declares that position.cost must be a Cost, but
 
            # in practice that's not true. Call get_position(post) on any
 
            # post without a cost and see what it returns. Hence the ignore.
 
            return units.currency  # type:ignore[unreachable]
 
        else:
 
            return f'{units.currency} {cost.currency} {cost.date.isoformat()}'
 

	
 
    @classmethod
 
    def build_opening(
 
            cls,
 
            key: AccountWithFund,
 
            meta_key: MetaKey,
 
            inventory: Inventory,
 
    ) -> Iterator[bc_data.Posting]:
 
        account, project = key
 
        if project is None:
 
            meta: Optional[Dict[MetaKey, MetaValue]] = None
 
        else:
 
            meta = {meta_key: project}
 
        for units, cost in sorted(inventory, key=cls._position_sortkey):
 
            if cost is None:
 
                units = quantize_amount(units)
 
            yield bc_data.Posting(
 
                account, units, cost, None, None, copy.copy(meta),
 
            )
 

	
 

	
 
class ReturnFlag(enum.IntFlag):
 
    LOAD_ERRORS = 1
 

	
 

	
 
def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
 
    parser = argparse.ArgumentParser(prog=PROGNAME)
 
    cliutil.add_version_argument(parser)
 
    cliutil.add_loglevel_argument(parser)
 
    parser.add_argument(
 
        '--fund-metadata-key', '-m',
 
        metavar='KEY',
 
        dest='meta_key',
 
        default='project',
 
        help="""Name of the fund metadata key. Default %(default)s.
 
""")
 
    parser.add_argument(
 
        '--unrestricted-fund', '-u',
 
        metavar='PROJECT',
 
        default='Conservancy',
 
        help="""Name of the unrestricted fund. Default %(default)s.
 
""")
 
    parser.add_argument(
 
        'as_of_date',
 
        metavar='YEAR_OR_DATE',
 
        type=cliutil.year_or_date_arg,
 
        nargs='?',
 
        help="""Date to generate opening balances for. You can provide just
 
a year to generate balances for the start of that fiscal year. Defaults to the
 
current fiscal year.
 
""")
 
    return parser.parse_args(arglist)
 

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

	
 
    fy = config.fiscal_year_begin()
 
    if args.as_of_date is None:
 
        args.as_of_date = fy.for_date()
 
    if isinstance(args.as_of_date, int):
 
        args.as_of_date = fy.first_date(args.as_of_date)
 

	
 
    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(0, args.as_of_date)
 
    for error in load_errors:
 
        bc_printer.print_error(error, file=stderr)
 
        returncode |= ReturnFlag.LOAD_ERRORS
 
        returncode |= cliutil.ReturnFlag.LOAD_ERRORS
 

	
 
    inventories: Mapping[AccountWithFund, Inventory] = collections.defaultdict(Inventory)
 
    for post in Posting.from_entries(entries):
 
        if post.meta.date >= args.as_of_date:
 
            continue
 
        account = post.account
 
        fund_acct_match = post.account.is_under(*FUND_ACCOUNTS)
 
        is_equity = account.root_part() in EQUITY_ACCOUNTS
 
        if fund_acct_match is None:
 
            project: MetaValue = None
 
        else:
 
            project = post.meta.get(args.meta_key)
 
            if project is None:
 
                bc_printer.print_error(Error(
 
                    post.meta, "no fund specified", post.meta.txn,
 
                ), file=stderr)
 
                project = args.unrestricted_fund
 
            if is_equity:
 
                if project == args.unrestricted_fund:
 
                    account = UNRESTRICTED_ACCOUNT
 
                else:
 
                    account = RESTRICTED_ACCOUNT
 
        inventory = inventories[AccountWithFund(account, project)]
 
        if is_equity:
 
            inventory.add_amount(post.at_cost())
 
        else:
 
            inventory.add_position(get_position(post))
 

	
 
    opening_date = args.as_of_date - datetime.timedelta(1)
 
    opening = bc_data.Transaction(  # type:ignore[operator]
 
        None,  # meta
 
        opening_date,
 
        '*',
 
        None,  # payee
 
        f"Opening balances for FY{fy.for_date(args.as_of_date)}",
 
        frozenset(),  # tags
 
        frozenset(),  # links
 
        [post
 
         for key in sorted(inventories, key=AccountWithFund.sortkey)
 
         for post in Posting.build_opening(key, args.meta_key, inventories[key])
 
        ])
 
    balance = Balance(get_cost(get_position(post))
 
                      for post in opening.postings)
 
    for amount in balance.clean_copy().values():
 
        opening.postings.append(bc_data.Posting(
 
            UNRESTRICTED_ACCOUNT, quantize_amount(-amount), None, None, None,
 
            {args.meta_key: args.unrestricted_fund},
 
        ))
 
    dcontext = bc_dcontext.DisplayContext()
 
    dcontext.set_commas(True)
 
    bc_printer.print_entry(opening, dcontext, file=stdout)
 
    return 0 if returncode == 0 else 16 + returncode
 
    return cliutil.ReturnFlag.returncode(returncode)
 

	
 
entry_point = cliutil.make_entry_point(__name__, PROGNAME)
 

	
 
if __name__ == '__main__':
 
    exit(entry_point())
tests/test_reports_accrual.py
Show inline comments
...
 
@@ -730,72 +730,72 @@ def test_main_outgoing_report(arglist, expect_amount):
 
    contract_url = re.escape(f'<{rt_url}Ticket/Attachment/3120/3120/VendorContract.pdf>')
 
    check_output(output, [
 
        r'^REQUESTOR: Mx\. 310 <mx310@example\.org>$',
 
        rf'^TOTAL TO PAY: \${expect_amount}\.00$',
 
        r'^\s*2010-04-30\s',
 
        r'^\s+Expenses:Travel\s+220 USD$',
 
    ])
 

	
 
@pytest.mark.parametrize('arglist', [
 
    ['-t', 'balance'],
 
    ['515/5150'],
 
])
 
def test_main_balance_report(arglist):
 
    retcode, output, errors = run_main(arglist)
 
    assert not errors.getvalue()
 
    assert retcode == 0
 
    check_output(output, [
 
        r'\brt://ticket/515/attachments/5150:$',
 
        r'^\s+1,500\.00 USD outstanding since 2010-05-15$',
 
    ])
 

	
 
def test_main_balance_report_because_no_rt_id():
 
    invoice = 'Invoices/2010StateRegistration.pdf'
 
    retcode, output, errors = run_main([invoice])
 
    assert not errors.getvalue()
 
    assert retcode == 0
 
    check_output(output, [
 
        rf'\b{re.escape(invoice)}:$',
 
        r'^\s+-50\.00 USD outstanding since 2010-06-20$',
 
    ])
 

	
 
@pytest.mark.parametrize('arglist', [
 
    [],
 
    ['entity=Lawyer'],
 
])
 
def test_main_aging_report(arglist):
 
    if arglist:
 
        recv_rows = [row for row in AGING_AR if 'Lawyer' in row.entity]
 
        pay_rows = [row for row in AGING_AP if 'Lawyer' in row.entity]
 
    else:
 
        recv_rows = AGING_AR
 
        pay_rows = AGING_AP
 
    retcode, output, errors = run_main(arglist, out_type=io.BytesIO)
 
    assert not errors.getvalue()
 
    assert retcode == 0
 
    check_aging_ods(output, datetime.date.today(), recv_rows, pay_rows)
 

	
 
def test_main_no_books():
 
    errors = check_main_fails([], testutil.TestConfig(), 1 | 8)
 
    errors = check_main_fails([], testutil.TestConfig(), 1 | 2)
 
    testutil.check_lines_match(iter(errors), [
 
        r':[01]: +no books to load in configuration\b',
 
    ])
 

	
 
@pytest.mark.parametrize('arglist', [
 
    ['499'],
 
    ['505/99999'],
 
    ['-t', 'balance', 'entity=NonExistent'],
 
])
 
def test_main_no_matches(arglist, caplog):
 
    check_main_fails(arglist, None, 8)
 
    check_main_fails(arglist, None, 2)
 
    testutil.check_logs_match(caplog, [
 
        ('WARNING', 'no matching entries found to report'),
 
    ])
 

	
 
def test_main_no_rt(caplog):
 
    config = testutil.TestConfig(
 
        books_path=testutil.test_path('books/accruals.beancount'),
 
    )
 
    check_main_fails(['-t', 'out'], config, 4)
 
    check_main_fails(['-t', 'out'], config, 16)
 
    testutil.check_logs_match(caplog, [
 
        ('ERROR', 'unable to generate outgoing report: RT client is required'),
 
    ])
tests/test_reports_fund.py
Show inline comments
...
 
@@ -235,50 +235,50 @@ def check_ods_report(ods, start_date, stop_date):
 
def run_main(out_type, arglist, config=None):
 
    if config is None:
 
        config = testutil.TestConfig(
 
            books_path=testutil.test_path('books/fund.beancount'),
 
        )
 
    arglist.insert(0, '--output-file=-')
 
    output = out_type()
 
    errors = io.StringIO()
 
    retcode = fund.main(arglist, output, errors, config)
 
    output.seek(0)
 
    return retcode, output, errors
 

	
 
@pytest.mark.parametrize('project,start_date,stop_date', [
 
    ('Conservancy', START_DATE, STOP_DATE),
 
    ('project=Conservancy', MID_DATE, STOP_DATE),
 
    ('Conservancy', START_DATE, MID_DATE),
 
    ('Alpha', START_DATE, STOP_DATE),
 
    ('project=Alpha', MID_DATE, STOP_DATE),
 
    ('Alpha', START_DATE, MID_DATE),
 
    ('Bravo', START_DATE, STOP_DATE),
 
    ('project=Bravo', MID_DATE, STOP_DATE),
 
    ('Bravo', START_DATE, MID_DATE),
 
    ('project=Charlie', START_DATE, STOP_DATE),
 
])
 
def test_text_report(project, start_date, stop_date):
 
    retcode, output, errors = run_main(io.StringIO, [
 
        '-b', start_date.isoformat(), '-e', stop_date.isoformat(), project,
 
    ])
 
    assert not errors.getvalue()
 
    assert retcode == 0
 
    check_text_report(output, project, start_date, stop_date)
 

	
 
@pytest.mark.parametrize('start_date,stop_date', [
 
    (START_DATE, STOP_DATE),
 
    (MID_DATE, STOP_DATE),
 
    (START_DATE, MID_DATE),
 
])
 
def test_ods_report(start_date, stop_date):
 
    retcode, output, errors = run_main(io.BytesIO, [
 
        '--begin', start_date.isoformat(), '--end', stop_date.isoformat(),
 
    ])
 
    assert not errors.getvalue()
 
    assert retcode == 0
 
    ods = odf.opendocument.load(output)
 
    check_ods_report(ods, start_date, stop_date)
 

	
 
def test_main_no_postings(caplog):
 
    retcode, output, errors = run_main(io.StringIO, ['NonexistentProject'])
 
    assert retcode == 24
 
    assert retcode == 18
 
    assert any(log.levelname == 'WARNING' for log in caplog.records)
tests/test_reports_ledger.py
Show inline comments
...
 
@@ -512,50 +512,50 @@ def test_main_project_report(ledger_entries, project, start_date, stop_date):
 
        project,
 
    ])
 
    assert not errors.getvalue()
 
    assert retcode == 0
 
    ods = odf.opendocument.load(output)
 
    assert get_sheet_names(ods) == PROJECT_REPORT_SHEETS[:]
 
    expected = dict(ExpectedPostings.group_by_account(related))
 
    for account in iter_accounts(ledger_entries):
 
        try:
 
            expected[account].check_report(ods, start_date, stop_date)
 
        except KeyError:
 
            ExpectedPostings.check_not_in_report(ods, account)
 

	
 
@pytest.mark.parametrize('flag', [
 
    '--disbursements',
 
    '--receipts',
 
])
 
def test_main_cash_report(ledger_entries, flag):
 
    if flag == '--receipts':
 
        txn_filter = ledger.TransactionFilter.CREDIT
 
    else:
 
        txn_filter = ledger.TransactionFilter.DEBIT
 
    retcode, output, errors = run_main([
 
        flag,
 
        '-b', START_DATE.isoformat(),
 
        '-e', STOP_DATE.isoformat(),
 
    ])
 
    assert not errors.getvalue()
 
    assert retcode == 0
 
    ods = odf.opendocument.load(output)
 
    postings = data.Posting.from_entries(ledger_entries)
 
    for account, expected in ExpectedPostings.group_by_account(postings):
 
        if account == 'Assets:Checking' or account == 'Assets:PayPal':
 
            expected.check_txn_report(ods, txn_filter, START_DATE, STOP_DATE)
 
        else:
 
            expected.check_not_in_report(ods)
 

	
 
@pytest.mark.parametrize('arg', [
 
    'Assets:NoneSuchBank',
 
    'Funny money',
 
])
 
def test_main_invalid_account(caplog, arg):
 
    retcode, output, errors = run_main(['-a', arg])
 
    assert retcode == 2
 
    assert any(log.message.endswith(f': {arg!r}') for log in caplog.records)
 

	
 
def test_main_no_postings(caplog):
 
    retcode, output, errors = run_main(['NonexistentProject'])
 
    assert retcode == 24
 
    assert retcode == 18
 
    assert any(log.levelname == 'WARNING' for log in caplog.records)
0 comments (0 inline, 0 general)