Changeset - 8b8bdc022508
[Not reviewed]
0 5 0
Brett Smith - 4 years ago 2020-06-17 02:41:13
brettcsmith@brettcsmith.org
reports: Add BaseODS.column_style() method.

Use this to provide more consistent column styles throughout the reports.
5 files changed with 102 insertions and 39 deletions:
0 comments (0 inline, 0 general)
conservancy_beancount/reports/accrual.py
Show inline comments
...
 
@@ -216,194 +216,199 @@ class AccrualPostings(core.RelatedPostings):
 
        invoice_ok = isinstance(self.invoice, str) and '/' in self.invoice
 
        if account_ok and entity_ok and invoice_ok:
 
            # type ignore for <https://github.com/python/mypy/issues/6670>
 
            yield (self.invoice, self)  # type:ignore[misc]
 
            return
 
        groups = collections.defaultdict(list)
 
        for post in self:
 
            post_invoice = self.invoice if invoice_ok else (
 
                post.meta.get('invoice') or 'BlankInvoice'
 
            )
 
            post_entity = self.entity if entity_ok else (
 
                post.meta.get('entity') or 'BlankEntity'
 
            )
 
            groups[f'{post.account} {post_invoice} {post_entity}'].append(post)
 
        type_self = type(self)
 
        for group_key, posts in groups.items():
 
            yield group_key, type_self(posts, _can_own=True)
 

	
 
    def is_paid(self, default: Optional[bool]=None) -> Optional[bool]:
 
        if self.accrual_type is None:
 
            return default
 
        else:
 
            return self.end_balance.le_zero()
 

	
 
    def is_zero(self, default: Optional[bool]=None) -> Optional[bool]:
 
        if self.accrual_type is None:
 
            return default
 
        else:
 
            return self.end_balance.is_zero()
 

	
 
    def since_last_nonzero(self) -> 'AccrualPostings':
 
        for index, (post, balance) in enumerate(self.iter_with_balance()):
 
            if balance.is_zero():
 
                start_index = index
 
        try:
 
            empty = start_index == index
 
        except NameError:
 
            empty = True
 
        return self if empty else self[start_index + 1:]
 

	
 
    @property
 
    def rt_id(self) -> Union[str, None, Sentinel]:
 
        return self._single_item(self.first_meta_links('rt-id', None))
 

	
 

	
 
class BaseReport:
 
    def __init__(self, out_file: TextIO) -> None:
 
        self.out_file = out_file
 
        self.logger = logger.getChild(type(self).__name__)
 

	
 
    def _report(self, posts: AccrualPostings, index: int) -> Iterable[str]:
 
        raise NotImplementedError("BaseReport._report")
 

	
 
    def run(self, groups: PostGroups) -> None:
 
        for index, invoice in enumerate(groups):
 
            for line in self._report(groups[invoice], index):
 
                print(line, file=self.out_file)
 

	
 

	
 
class AgingODS(core.BaseODS[AccrualPostings, Optional[data.Account]]):
 
    DOC_COLUMNS = [
 
        'rt-id',
 
        'invoice',
 
        'approval',
 
        'contract',
 
        'purchase-order',
 
    ]
 
    COLUMNS = [
 
        'Date',
 
        data.Metadata.human_name('entity'),
 
        'Invoice Amount',
 
        'Booked Amount',
 
        data.Metadata.human_name('project'),
 
        *(data.Metadata.human_name(key) for key in DOC_COLUMNS),
 
    ]
 
    COL_COUNT = len(COLUMNS)
 

	
 
    def __init__(self,
 
                 rt_wrapper: rtutil.RT,
 
                 date: datetime.date,
 
                 logger: logging.Logger,
 
    ) -> None:
 
        super().__init__(rt_wrapper)
 
        self.date = date
 
        self.logger = logger
 

	
 
    def section_key(self, row: AccrualPostings) -> Optional[data.Account]:
 
        if isinstance(row.account, str):
 
            return row.account
 
        else:
 
            return None
 

	
 
    def start_spreadsheet(self) -> None:
 
        for accrual_type in AccrualAccount:
 
            self.use_sheet(accrual_type.name.title())
 
            for index in range(self.COL_COUNT):
 
                stylename = self.style_col1_25 if index else ''
 
                self.sheet.addElement(odf.table.TableColumn(stylename=stylename))
 
                if index == 0:
 
                    style: Union[str, odf.style.Style] = ''
 
                elif index < 6:
 
                    style = self.column_style(1.2)
 
                else:
 
                    style = self.column_style(1.5)
 
                self.sheet.addElement(odf.table.TableColumn(stylename=style))
 
            self.add_row(*(
 
                self.string_cell(name, stylename=self.style_bold)
 
                for name in self.COLUMNS
 
            ))
 
            self.lock_first_row()
 

	
 
    def start_section(self, key: Optional[data.Account]) -> None:
 
        if key is None:
 
            return
 
        self.age_thresholds = list(AccrualAccount.by_account(key).value.aging_thresholds)
 
        self.age_balances = [core.MutableBalance() for _ in self.age_thresholds]
 
        accrual_date = self.date - datetime.timedelta(days=self.age_thresholds[-1])
 
        acct_parts = key.slice_parts()
 
        self.use_sheet(acct_parts[1])
 
        self.add_row()
 
        self.add_row(self.string_cell(
 
            f"{' '.join(acct_parts[2:])} {acct_parts[1]} Aging Report"
 
            f" Accrued by {accrual_date.isoformat()} Unpaid by {self.date.isoformat()}",
 
            stylename=self.merge_styles(self.style_bold, self.style_centertext),
 
            numbercolumnsspanned=self.COL_COUNT,
 
        ))
 
        self.add_row()
 

	
 
    def end_section(self, key: Optional[data.Account]) -> None:
 
        if key is None:
 
            return
 
        total_balance = core.MutableBalance()
 
        text_style = self.merge_styles(self.style_bold, self.style_endtext)
 
        text_span = 4
 
        last_age_text: Optional[str] = None
 
        self.add_row()
 
        for threshold, balance in zip(self.age_thresholds, self.age_balances):
 
            years, days = divmod(threshold, 365)
 
            years_text = f"{years} {'Year' if years == 1 else 'Years'}"
 
            days_text = f"{days} Days"
 
            if years and days:
 
                age_text = f"{years_text} {days_text}"
 
            elif years:
 
                age_text = years_text
 
            else:
 
                age_text = days_text
 
            if last_age_text is None:
 
                age_range = f"Over {age_text}"
 
            else:
 
                age_range = f"{age_text}–{last_age_text}"
 
            self.add_row(
 
                self.string_cell(
 
                    f"Total Aged {age_range}: ",
 
                    stylename=text_style,
 
                    numbercolumnsspanned=text_span,
 
                ),
 
                *(odf.table.TableCell() for _ in range(1, text_span)),
 
                self.balance_cell(balance),
 
            )
 
            last_age_text = age_text
 
            total_balance += balance
 
        self.add_row(
 
            self.string_cell(
 
                "Total Unpaid: ",
 
                stylename=text_style,
 
                numbercolumnsspanned=text_span,
 
            ),
 
            *(odf.table.TableCell() for _ in range(1, text_span)),
 
            self.balance_cell(total_balance),
 
        )
 

	
 
    def write_row(self, row: AccrualPostings) -> None:
 
        age = (self.date - row[0].meta.date).days
 
        if row.end_balance.ge_zero():
 
            for index, threshold in enumerate(self.age_thresholds):
 
                if age >= threshold:
 
                    self.age_balances[index] += row.end_balance
 
                    break
 
            else:
 
                return
 
        raw_balance = row.balance()
 
        if row.accrual_type is not None:
 
            raw_balance = row.accrual_type.normalize_amount(raw_balance)
 
        if raw_balance == row.end_balance:
 
            amount_cell = odf.table.TableCell()
 
        else:
 
            amount_cell = self.balance_cell(raw_balance)
 
        projects = {post.meta.get('project') or None for post in row}
 
        projects.discard(None)
 
        self.add_row(
 
            self.date_cell(row[0].meta.date),
 
            self.multiline_cell(row.entities()),
 
            amount_cell,
 
            self.balance_cell(row.end_balance),
 
            self.multiline_cell(sorted(projects)),
 
            *(self.meta_links_cell(row.all_meta_links(key))
 
              for key in self.DOC_COLUMNS),
 
        )
 

	
 

	
 
class AgingReport(BaseReport):
conservancy_beancount/reports/core.py
Show inline comments
...
 
@@ -370,305 +370,343 @@ class RelatedPostings(Sequence[data.Posting]):
 
        return Balance(post.at_cost() for post in self)
 

	
 
    def balance_at_cost_by_date(self, date: datetime.date) -> Balance:
 
        for index, post in enumerate(self):
 
            if post.meta.date >= date:
 
                break
 
        else:
 
            index += 1
 
        return Balance(post.at_cost() for post in self._postings[:index])
 

	
 
    def meta_values(self,
 
                    key: MetaKey,
 
                    default: Optional[MetaValue]=None,
 
    ) -> Set[Optional[MetaValue]]:
 
        return {post.meta.get(key, default) for post in self}
 

	
 

	
 
class BaseSpreadsheet(Generic[RT, ST], metaclass=abc.ABCMeta):
 
    """Abstract base class to help write spreadsheets
 

	
 
    This class provides the very core logic to write an arbitrary set of data
 
    rows to arbitrary output. It calls hooks when it starts writing the
 
    spreadsheet, starts a new "section" of rows, ends a section, and ends the
 
    spreadsheet.
 

	
 
    RT is the type of the input data rows. ST is the type of the section
 
    identifier that you create from each row. If you don't want to use the
 
    section logic at all, set ST to None and define section_key to return None.
 
    """
 

	
 
    @abc.abstractmethod
 
    def section_key(self, row: RT) -> ST:
 
        """Return the section a row belongs to
 

	
 
        Given a data row, this method should return some identifier for the
 
        "section" the row belongs to. The write method uses this to
 
        determine when to call start_section and end_section.
 

	
 
        If your spreadsheet doesn't need sections, define this to return None.
 
        """
 
        ...
 

	
 
    @abc.abstractmethod
 
    def write_row(self, row: RT) -> None:
 
        """Write a data row to the output spreadsheet
 

	
 
        This method is called once for each data row in the input.
 
        """
 
        ...
 

	
 
    # The next four methods are all called by the write method when the name
 
    # says. You may override them to output headers or sums, record
 
    # state, etc. The default implementations are all noops.
 

	
 
    def start_spreadsheet(self) -> None:
 
        pass
 

	
 
    def start_section(self, key: ST) -> None:
 
        pass
 

	
 
    def end_section(self, key: ST) -> None:
 
        pass
 

	
 
    def end_spreadsheet(self) -> None:
 
        pass
 

	
 
    def write(self, rows: Iterable[RT]) -> None:
 
        prev_section: Optional[ST] = None
 
        self.start_spreadsheet()
 
        for row in rows:
 
            section = self.section_key(row)
 
            if section != prev_section:
 
                if prev_section is not None:
 
                    self.end_section(prev_section)
 
                self.start_section(section)
 
                prev_section = section
 
            self.write_row(row)
 
        try:
 
            should_end = section is not None
 
        except NameError:
 
            should_end = False
 
        if should_end:
 
            self.end_section(section)
 
        self.end_spreadsheet()
 

	
 

	
 
class BaseODS(BaseSpreadsheet[RT, ST], metaclass=abc.ABCMeta):
 
    """Abstract base class to help write OpenDocument spreadsheets
 

	
 
    This class provides the very core logic to write an arbitrary set of data
 
    rows to an OpenDocument spreadsheet. It provides helper methods for
 
    building sheets, rows, and cells.
 

	
 
    See also the BaseSpreadsheet base class for additional documentation about
 
    methods you must and can define, the definition of RT and ST, etc.
 
    """
 
    # Defined in the XSL spec, "Definitions of Units of Measure"
 
    MEASUREMENT_UNITS = frozenset([
 
        'cm',
 
        'em',
 
        'in',
 
        'mm',
 
        'pc',
 
        'pt',
 
        'px',
 
    ])
 
    MEASUREMENT_RE = re.compile(
 
        r'([-+]?(?:\d+\.?|\.\d+|\d+\.\d+))({})'.format('|'.join(MEASUREMENT_UNITS)),
 
        re.ASCII,
 
    )
 

	
 
    def __init__(self, rt_wrapper: Optional[rtutil.RT]=None) -> None:
 
        self.rt_wrapper = rt_wrapper
 
        self.locale = babel.core.Locale.default('LC_MONETARY')
 
        self.currency_fmt_key = 'accounting'
 
        self._name_counter = itertools.count(1)
 
        self._currency_style_cache: MutableMapping[str, odf.style.Style] = {}
 
        self.document = odf.opendocument.OpenDocumentSpreadsheet()
 
        self.init_settings()
 
        self.init_styles()
 
        self.sheet = self.use_sheet("Report")
 

	
 
    ### Low-level document tree manipulation
 
    # The *intent* is that you only need to use these if you're adding new
 
    # methods to manipulate document settings or styles.
 

	
 
    def copy_element(self, elem: odf.element.Element) -> odf.element.Element:
 
        qattrs = dict(self.iter_qattributes(elem))
 
        retval = odf.element.Element(qname=elem.qname, qattributes=qattrs)
 
        try:
 
            orig_name = retval.getAttribute('name')
 
        except ValueError:
 
            orig_name = None
 
        if orig_name is not None:
 
            retval.setAttribute('name', f'{orig_name}{next(self._name_counter)}')
 
        return retval
 

	
 
    def ensure_child(self,
 
                     parent: odf.element.Element,
 
                     child_type: ElementType,
 
                     **kwargs: Any,
 
    ) -> odf.element.Element:
 
        new_child = child_type(**kwargs)
 
        found_child = self.find_child(parent, new_child)
 
        if found_child is None:
 
            parent.addElement(new_child)
 
            return parent.lastChild
 
        else:
 
            return found_child
 

	
 
    def ensure_config_map_entry(self,
 
                                root: odf.element.Element,
 
                                map_name: str,
 
                                entry_name: str,
 
    ) -> odf.element.Element:
 
        """Return a ``ConfigItemMapEntry`` under ``root``
 

	
 
        This method ensures there's a ``ConfigItemMapNamed`` named ``map_name``
 
        under ``root``, and a ``ConfigItemMapEntry`` named ``entry_name`` under
 
        that. Return the ``ConfigItemMapEntry`` element.
 
        """
 
        config_map = self.ensure_child(root, odf.config.ConfigItemMapNamed, name=map_name)
 
        return self.ensure_child(config_map, odf.config.ConfigItemMapEntry, name=entry_name)
 

	
 
    def find_child(self,
 
                   parent: odf.element.Element,
 
                   child: odf.element.Element,
 
    ) -> Optional[odf.element.Element]:
 
        attrs = {k: v for k, v in self.iter_attributes(child)}
 
        if not attrs:
 
            return None
 
        for elem in parent.childNodes:
 
            if (elem.qname == child.qname
 
                and all(elem.getAttribute(k) == v for k, v in attrs.items())):
 
                return elem
 
        return None
 

	
 
    def iter_attributes(self, elem: odf.element.Element) -> Iterator[Tuple[str, str]]:
 
        for (_, key), value in self.iter_qattributes(elem):
 
            yield key.lower().replace('-', ''), value
 

	
 
    def iter_qattributes(self, elem: odf.element.Element) -> Iterator[Tuple[Tuple[str, str], str]]:
 
        if elem.attributes:
 
            yield from elem.attributes.items()
 

	
 
    def replace_child(self,
 
                     parent: odf.element.Element,
 
                     child_type: ElementType,
 
                     **kwargs: Any,
 
    ) -> odf.element.Element:
 
        new_child = child_type(**kwargs)
 
        found_child = self.find_child(parent, new_child)
 
        parent.insertBefore(new_child, found_child)
 
        if found_child is not None:
 
            parent.removeChild(found_child)
 
        return new_child
 

	
 
    def set_config(self,
 
                   root: odf.element.Element,
 
                   name: str,
 
                   value: Union[bool, int, str],
 
                   config_type: Optional[str]=None,
 
    ) -> None:
 
        """Ensure ``root`` has a ``ConfigItem`` with the given name, type, and value"""
 
        value_s = str(value)
 
        if isinstance(value, bool):
 
            value_s = str(value).lower()
 
            default_type = 'boolean'
 
        elif isinstance(value, str):
 
            default_type = 'string'
 
        if config_type is None:
 
            try:
 
                config_type = default_type
 
            except NameError:
 
                raise ValueError(
 
                    f"need config_type for {type(value).__name__} value",
 
                ) from None
 
        item = self.replace_child(
 
            root, odf.config.ConfigItem, name=name, type=config_type,
 
        )
 
        item.addText(value_s)
 

	
 
    ### Styles
 

	
 
    def column_style(self, width: Union[float, str], **attrs: Any) -> odf.style.Style:
 
        if not isinstance(width, str) or (width and not width[-1].isalpha()):
 
            width = f'{width}in'
 
        match = self.MEASUREMENT_RE.fullmatch(width)
 
        if match is None:
 
            raise ValueError(f"invalid width {width!r}")
 
        width_float = float(match.group(1))
 
        if width_float <= 0:
 
            # Per the OpenDocument spec, column-width is a positiveLength.
 
            raise ValueError(f"width {width!r} must be positive")
 
        width = '{:.3g}{}'.format(width_float, match.group(2))
 
        retval = self.ensure_child(
 
            self.document.automaticstyles,
 
            odf.style.Style,
 
            name=f'col_{width.replace(".", "_")}'
 
        )
 
        retval.setAttribute('family', 'table-column')
 
        if retval.firstChild is None:
 
            retval.addElement(odf.style.TableColumnProperties(
 
                columnwidth=width, **attrs
 
            ))
 
        return retval
 

	
 
    def _build_currency_style(
 
            self,
 
            root: odf.element.Element,
 
            locale: babel.core.Locale,
 
            code: str,
 
            fmt_index: int,
 
            properties: Optional[odf.style.TextProperties]=None,
 
            *,
 
            fmt_key: Optional[str]=None,
 
            volatile: bool=False,
 
            minintegerdigits: int=1,
 
    ) -> odf.element.Element:
 
        if fmt_key is None:
 
            fmt_key = self.currency_fmt_key
 
        pattern = locale.currency_formats[fmt_key]
 
        fmts = pattern.pattern.split(';')
 
        try:
 
            fmt = fmts[fmt_index]
 
        except IndexError:
 
            fmt = fmts[0]
 
            grouping = pattern.grouping[0]
 
        else:
 
            grouping = pattern.grouping[fmt_index]
 
        zero_s = babel.numbers.format_currency(0, code, '##0.0', locale)
 
        try:
 
            decimal_index = zero_s.rindex('.') + 1
 
        except ValueError:
 
            decimalplaces = 0
 
        else:
 
            decimalplaces = len(zero_s) - decimal_index
 
        style = self.replace_child(
 
            root,
 
            odf.number.CurrencyStyle,
 
            name=f'{code}{next(self._name_counter)}',
 
        )
 
        style.setAttribute('volatile', 'true' if volatile else 'false')
 
        if properties is not None:
 
            style.addElement(properties)
 
        for part in re.split(r"(¤+|[#0,.]+|'[^']+')", fmt):
 
            if not part:
 
                pass
 
            elif not part.strip('#0,.'):
 
                style.addElement(odf.number.Number(
 
                    decimalplaces=str(decimalplaces),
 
                    grouping='true' if grouping else 'false',
 
                    minintegerdigits=str(minintegerdigits),
 
                ))
 
            elif part == '¤':
 
                style.addElement(odf.number.CurrencySymbol(
 
                    country=locale.territory,
 
                    language=locale.language,
 
                    text=babel.numbers.get_currency_symbol(code, locale),
 
                ))
 
            elif part == '¤¤':
 
                style.addElement(odf.number.Text(text=code))
 
            else:
 
                style.addElement(odf.number.Text(text=part.strip("'")))
 
        return style
 

	
 
    def currency_style(
 
            self,
 
            code: str,
 
            locale: Optional[babel.core.Locale]=None,
 
            negative_properties: Optional[odf.style.TextProperties]=None,
 
            positive_properties: Optional[odf.style.TextProperties]=None,
 
            root: odf.element.Element=None,
 
    ) -> odf.style.Style:
 
        """Create and return a spreadsheet style to format currency data
 

	
 
        Given a currency code and a locale, this method will create all the
 
        styles necessary to format the currency according to the locale's
 
        rules, including rendering of decimal points and negative values.
 

	
 
        You may optionally pass in TextProperties to use for negative and
 
        positive amounts, respectively. If you don't, negative values will
 
        automatically be rendered in red (text color #f00).
 

	
 
        Results are cached. If you repeatedly call this method with the same
 
        arguments, you'll keep getting the same style returned, which will
 
        only be added to the document once.
 
        """
 
        if locale is None:
 
            locale = self.locale
 
        if negative_properties is None:
 
            negative_properties = odf.style.TextProperties(color='#ff0000')
 
        if root is None:
 
            root = self.document.styles
 
        cache_parts = [str(id(root)), code, str(locale)]
 
        for key, value in self.iter_attributes(negative_properties):
 
            cache_parts.append(f'{key}={value}')
 
        if positive_properties is not None:
 
            cache_parts.append('')
 
            for key, value in self.iter_attributes(positive_properties):
 
                cache_parts.append(f'{key}={value}')
 
        cache_key = '\0'.join(cache_parts)
 
        try:
...
 
@@ -782,206 +820,192 @@ class BaseODS(BaseSpreadsheet[RT, ST], metaclass=abc.ABCMeta):
 
        if sheet is None:
 
            sheet = self.sheet
 
        config_map = self.ensure_config_map_entry(
 
            self.view, 'Tables', sheet.getAttribute('name'),
 
        )
 
        self.set_config(config_map, 'PositionBottom', 1, 'int')
 
        self.set_config(config_map, 'VerticalSplitMode', 2, 'short')
 
        self.set_config(config_map, 'VerticalSplitPosition', 1, 'short')
 

	
 
    def use_sheet(self, name: str) -> odf.table.Table:
 
        """Switch the active sheet ``self.sheet`` to the one with the given name
 

	
 
        If there is no sheet with the given name, create it and append it to
 
        the spreadsheet first.
 

	
 
        If the current active sheet is empty when this method is called, it
 
        will be removed from the spreadsheet.
 
        """
 
        try:
 
            empty_sheet = not self.sheet.hasChildNodes()
 
        except AttributeError:
 
            empty_sheet = False
 
        if empty_sheet:
 
            self.document.spreadsheet.removeChild(self.sheet)
 
        self.sheet = self.ensure_child(
 
            self.document.spreadsheet, odf.table.Table, name=name,
 
        )
 
        return self.sheet
 

	
 
    ### Initialization hooks
 

	
 
    def init_settings(self) -> None:
 
        """Hook called to initialize settings
 

	
 
        This method is called by __init__ to populate
 
        ``self.document.settings``. This implementation creates the barest
 
        skeleton structure necessary to support other methods, in particular
 
        ``lock_first_row``.
 
        """
 
        view_settings = self.ensure_child(
 
            self.document.settings, odf.config.ConfigItemSet, name='ooo:view-settings',
 
        )
 
        views = self.ensure_child(
 
            view_settings, odf.config.ConfigItemMapIndexed, name='Views',
 
        )
 
        self.view = self.ensure_child(views, odf.config.ConfigItemMapEntry)
 
        self.set_config(self.view, 'ViewId', 'view1')
 

	
 
    def init_styles(self) -> None:
 
        """Hook called to initialize settings
 

	
 
        This method is called by __init__ to populate
 
        ``self.document.styles``. This implementation creates basic building
 
        block cell styles often used in financial reports.
 
        """
 
        styles = self.document.styles
 
        self.style_bold = self.ensure_child(
 
            styles, odf.style.Style, name='Bold', family='table-cell',
 
        )
 
        self.ensure_child(
 
            self.style_bold, odf.style.TextProperties, fontweight='bold',
 
        )
 
        self.style_dividerline = self.ensure_child(
 
            styles, odf.style.Style, name='DividerLine', family='table-cell',
 
        )
 
        self.ensure_child(
 
            self.style_dividerline,
 
            odf.style.TableCellProperties,
 
            borderbottom='1pt solid #0000ff',
 
        )
 

	
 
        date_style = self.replace_child(styles, odf.number.DateStyle, name='ISODate')
 
        date_style.addElement(odf.number.Year(style='long'))
 
        date_style.addElement(odf.number.Text(text='-'))
 
        date_style.addElement(odf.number.Month(style='long'))
 
        date_style.addElement(odf.number.Text(text='-'))
 
        date_style.addElement(odf.number.Day(style='long'))
 
        self.style_date = self.ensure_child(
 
            styles,
 
            odf.style.Style,
 
            name=f'{date_style.getAttribute("name")}Cell',
 
            family='table-cell',
 
            datastylename=date_style,
 
        )
 

	
 
        self.style_starttext: odf.style.Style
 
        self.style_centertext: odf.style.Style
 
        self.style_endtext: odf.style.Style
 
        for textalign in ['start', 'center', 'end']:
 
            aligned_style = self.replace_child(
 
                styles, odf.style.Style, name=f'{textalign.title()}Text',
 
            )
 
            aligned_style.setAttribute('family', 'table-cell')
 
            aligned_style.addElement(odf.style.ParagraphProperties(textalign=textalign))
 
            setattr(self, f'style_{textalign}text', aligned_style)
 

	
 
        self.style_col1: odf.style.Style
 
        self.style_col1_25: odf.style.Style
 
        self.style_col1_5: odf.style.Style
 
        self.style_col1_75: odf.style.Style
 
        self.style_col2: odf.style.Style
 
        for width in ['1', '1.25', '1.5', '1.75', '2']:
 
            width_name = width.replace('.', '_')
 
            column_style = self.replace_child(
 
                self.document.automaticstyles, odf.style.Style, name=f'col_{width_name}',
 
            )
 
            column_style.setAttribute('family', 'table-column')
 
            column_style.addElement(odf.style.TableColumnProperties(columnwidth=f'{width}in'))
 
            setattr(self, f'style_col{width_name}', column_style)
 

	
 
    ### Rows and cells
 

	
 
    def add_row(self, *cells: odf.table.TableCell, **attrs: Any) -> odf.table.TableRow:
 
        row = odf.table.TableRow(**attrs)
 
        for cell in cells:
 
            row.addElement(cell)
 
        self.sheet.addElement(row)
 
        return row
 

	
 
    def balance_cell(self, balance: Balance, **attrs: Any) -> odf.table.TableCell:
 
        if balance.is_zero():
 
            return self.float_cell(0, **attrs)
 
        elif len(balance) == 1:
 
            amount = next(iter(balance.values()))
 
            attrs['stylename'] = self.merge_styles(
 
                attrs.get('stylename'), self.currency_style(amount.currency),
 
            )
 
            return self.currency_cell(amount, **attrs)
 
        else:
 
            lines = [babel.numbers.format_currency(
 
                number, currency, locale=self.locale, format_type=self.currency_fmt_key,
 
            ) for number, currency in balance.values()]
 
            attrs['stylename'] = self.merge_styles(
 
                attrs.get('stylename'), self.style_endtext,
 
            )
 
            return self.multiline_cell(lines, **attrs)
 

	
 
    def currency_cell(self, amount: data.Amount, **attrs: Any) -> odf.table.TableCell:
 
        if 'stylename' not in attrs:
 
            attrs['stylename'] = self.currency_style(amount.currency)
 
        number, currency = amount
 
        cell = odf.table.TableCell(valuetype='currency', value=number, **attrs)
 
        cell.addElement(odf.text.P(text=babel.numbers.format_currency(
 
            number, currency, locale=self.locale, format_type=self.currency_fmt_key,
 
        )))
 
        return cell
 

	
 
    def date_cell(self, date: datetime.date, **attrs: Any) -> odf.table.TableCell:
 
        attrs.setdefault('stylename', self.style_date)
 
        cell = odf.table.TableCell(valuetype='date', datevalue=date, **attrs)
 
        cell.addElement(odf.text.P(text=date.isoformat()))
 
        return cell
 

	
 
    def float_cell(self, value: Union[int, float, Decimal], **attrs: Any) -> odf.table.TableCell:
 
        cell = odf.table.TableCell(valuetype='float', value=value, **attrs)
 
        cell.addElement(odf.text.P(text=str(value)))
 
        return cell
 

	
 
    def _meta_link_pairs(self, links: Iterable[Optional[str]]) -> Iterator[Tuple[str, str]]:
 
        for href in links:
 
            if href is None:
 
                continue
 
            elif self.rt_wrapper is not None:
 
                rt_ids = self.rt_wrapper.parse(href)
 
                rt_href = rt_ids and self.rt_wrapper.url(*rt_ids)
 
            else:
 
                rt_ids = None
 
                rt_href = None
 
            if rt_ids is None or rt_href is None:
 
                # '..' pops the ODS filename off the link path. In other words,
 
                # make the link relative to the directory the ODS is in.
 
                href_path = Path('..', href)
 
                href = str(href_path)
 
                text = href_path.name
 
            else:
 
                rt_path = urlparse.urlparse(rt_href).path
 
                if rt_path.endswith('/Ticket/Display.html'):
 
                    text = rtutil.RT.unparse(*rt_ids)
 
                else:
 
                    text = urlparse.unquote(Path(rt_path).name)
 
                href = rt_href
 
            yield (href, text)
 

	
 
    def meta_links_cell(self, links: Iterable[Optional[str]], **attrs: Any) -> odf.table.TableCell:
 
        return self.multilink_cell(self._meta_link_pairs(links), **attrs)
 

	
 
    def multiline_cell(self, lines: Iterable[Any], **attrs: Any) -> odf.table.TableCell:
 
        cell = odf.table.TableCell(valuetype='string', **attrs)
 
        for line in lines:
 
            cell.addElement(odf.text.P(text=str(line)))
 
        return cell
 

	
 
    def multilink_cell(self, links: Iterable[LinkType], **attrs: Any) -> odf.table.TableCell:
 
        cell = odf.table.TableCell(valuetype='string', **attrs)
 
        for link in links:
 
            if isinstance(link, tuple):
 
                href, text = link
 
            else:
 
                href = link
 
                text = None
 
            cell.addElement(odf.text.P())
 
            cell.lastChild.addElement(odf.text.A(
 
                type='simple', href=href, text=text or href,
 
            ))
 
        return cell
 

	
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 (
 
    Callable,
 
    Dict,
 
    Iterable,
 
    Iterator,
 
    List,
 
    Mapping,
 
    Optional,
 
    Sequence,
 
    TextIO,
 
    Tuple,
 
    Union,
 
)
 

	
 
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
 
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, data.Account]):
 
    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']),
 
        ('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']),
 
    ])
 
    COLUMN_STYLES: Mapping[str, str] = {
 
        'Date': '',
 
        'Description': 'col_1_75',
 
        data.Metadata.human_name('paypal-id'): 'col_1_5',
 
    }
 
    # 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 = 65000
 

	
 
    def __init__(self,
 
                 start_date: datetime.date,
 
                 stop_date: datetime.date,
 
                 sheet_names: Optional[Sequence[str]]=None,
 
                 rt_wrapper: Optional[rtutil.RT]=None,
 
                 sheet_size: Optional[int]=None,
 
    ) -> None:
 
        if sheet_names is None:
 
            sheet_names = list(self.ACCOUNT_COLUMNS)
 
        if sheet_size is None:
 
            sheet_size = self.SHEET_SIZE
 
        super().__init__(rt_wrapper)
 
        self.date_range = ranges.DateRange(start_date, stop_date)
 
        self.required_sheet_names = sheet_names
 
        self.sheet_size = sheet_size
 

	
 
    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]] = {
 
            '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)
 
        ]
 

	
 
    @staticmethod
 
    def _sort_and_filter_accounts(
 
            accounts: Iterable[data.Account],
 
            order: Sequence[str],
 
    ) -> Iterator[Tuple[int, data.Account]]:
 
        index_map = {s: ii for ii, s in enumerate(order)}
 
        retval: Mapping[int, List[data.Account]] = collections.defaultdict(list)
 
        for account in accounts:
 
            acct_key = account.is_under(*order)
 
            if acct_key is not None:
 
                retval[index_map[acct_key]].append(account)
 
        for key in sorted(retval):
 
            acct_list = retval[key]
 
            acct_list.sort()
 
            for account in acct_list:
 
                yield key, account
 

	
 
    def section_key(self, row: data.Posting) -> data.Account:
 
        return row.account
 

	
 
    def start_sheet(self, sheet_name: str) -> None:
 
        self.use_sheet(sheet_name.replace(':', ' '))
 
        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
 
        self.metadata_columns = self.ACCOUNT_COLUMNS[columns_key]
 
        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, 'col_1_25'),
 
                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 _report_section_balance(self, key: data.Account, date_key: str) -> None:
 
        uses_opening = key.is_under('Assets', 'Equity', 'Liabilities')
 
        if date_key == 'start':
 
            if not uses_opening:
 
                return
 
            date = self.date_range.start
 
            description = "Opening Balance"
 
        else:
 
            date = self.date_range.stop
 
            description = "Ending Balance" if uses_opening else "Period Total"
 
        balance = self.norm_func(
 
            self.account_groups[key].balance_at_cost_by_date(date)
 
        )
 
        self.add_row(
 
            self.date_cell(date, stylename=self.merge_styles(
 
                self.style_bold, self.style_date,
 
            )),
 
            odf.table.TableCell(),
 
            self.string_cell(description, stylename=self.style_bold),
 
            odf.table.TableCell(),
 
            self.balance_cell(balance, stylename=self.style_bold),
 
        )
 

	
 
    def start_section(self, key: data.Account) -> None:
 
        self.add_row()
 
        self.add_row(
 
            odf.table.TableCell(),
 
            self.string_cell(
 
                f"{key} Ledger"
 
                f" From {self.date_range.start.isoformat()}"
 
                f" To {self.date_range.stop.isoformat()}",
 
                stylename=self.style_bold,
 
                numbercolumnsspanned=len(self.sheet_columns) - 1,
 
            ),
 
        )
 
        self.norm_func = core.normalize_amount_func(key)
 
        self._report_section_balance(key, 'start')
 

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

	
 
    def write_row(self, row: data.Posting) -> None:
 
        if row.meta.date not in self.date_range:
 
            return
 
        elif 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 _combined_balance_row(self,
 
                              date: datetime.date,
 
                              balance_accounts: Sequence[str],
 
    ) -> None:
 
        balance = -sum((
 
            related.balance_at_cost_by_date(date)
 
            for account, related in self.account_groups.items()
 
            if account.is_under(*balance_accounts)
 
        ), core.MutableBalance())
 
        self.add_row(
 
            self.string_cell(
 
                f"Balance as of {date.isoformat()}",
 
                stylename=self.merge_styles(self.style_bold, self.style_endtext),
 
            ),
 
            self.balance_cell(balance, stylename=self.style_bold),
 
        )
 

	
 
    def write_balance_sheet(self) -> None:
 
        balance_accounts = ['Equity', 'Income', 'Expenses']
 
        # FIXME: This is a hack to exclude non-project Equity accounts from
 
        # project reports.
 
        if balance_accounts[0] not in self.required_sheet_names:
 
            balance_accounts[0] = 'Equity:Funds'
 
        self.use_sheet("Balance")
 
        column_style = self.replace_child(
 
            self.document.automaticstyles, odf.style.Style, name='col_3',
 
        )
 
        column_style.setAttribute('family', 'table-column')
 
        column_style.addElement(odf.style.TableColumnProperties(columnwidth='3in'))
 
        for _ in range(2):
 
            self.sheet.addElement(odf.table.TableColumn(stylename=column_style))
 
        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()
 
        self._combined_balance_row(self.date_range.start, balance_accounts)
 
        for _, account in self._sort_and_filter_accounts(
 
                self.account_groups, balance_accounts,
 
        ):
 
            related = self.account_groups[account]
 
            # start_bal - stop_bal == -(stop_bal - start_bal)
 
            balance = related.balance_at_cost_by_date(self.date_range.start)
 
            balance -= related.balance_at_cost_by_date(self.date_range.stop)
 
            if not balance.is_zero():
 
                self.add_row(
 
                    self.string_cell(account, stylename=self.style_endtext),
 
                    self.balance_cell(balance),
 
                )
 
        self._combined_balance_row(self.date_range.stop, balance_accounts)
 

	
 
    def write(self, rows: Iterable[data.Posting]) -> None:
 
        self.account_groups = dict(core.RelatedPostings.group_by_account(rows))
 
        self.write_balance_sheet()
 
        tally_by_account_iter = (
 
            (account, sum(1 for post in related if post.meta.date in self.date_range))
 
            for account, related in self.account_groups.items()
 
        )
 
        tally_by_account = {
 
            account: count
 
            for account, count in tally_by_account_iter
 
            if count
 
        }
 
        sheet_names = self.plan_sheets(
 
            tally_by_account, self.required_sheet_names, self.sheet_size,
 
        )
 
        using_sheet_index = -1
 
        for sheet_index, account in self._sort_and_filter_accounts(
 
                tally_by_account, sheet_names,
 
        ):
 
            while using_sheet_index < sheet_index:
 
                using_sheet_index += 1
 
                self.start_sheet(sheet_names[using_sheet_index])
 
            super().write(self.account_groups[account])
 
        for index in range(using_sheet_index + 1, len(sheet_names)):
 
            self.start_sheet(sheet_names[index])
 

	
 

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

	
 

	
 
def parse_arguments(arglist: Optional[Sequence[str]]=None) -> argparse.Namespace:
 
    parser = argparse.ArgumentParser(prog=PROGNAME)
 
    cliutil.add_version_argument(parser)
 
    parser.add_argument(
 
        '--begin', '--start', '-b',
 
        dest='start_date',
 
        metavar='DATE',
 
        type=cliutil.date_arg,
 
        help="""Date to start reporting entries, inclusive, in YYYY-MM-DD format.
 
The default is one year ago.
 
""")
 
    parser.add_argument(
 
        '--end', '--stop', '-e',
 
        dest='stop_date',
 
        metavar='DATE',
 
        type=cliutil.date_arg,
 
        help="""Date to stop reporting entries, exclusive, in YYYY-MM-DD format.
 
The default is a year after the start date, or 30 days from today if the start
 
date was also not specified.
 
""")
 
    parser.add_argument(
 
        '--account', '-a',
 
        dest='sheet_names',
 
        metavar='ACCOUNT',
 
        action='append',
 
        help="""Show this account in the report. You can specify this option
 
multiple times. If not specified, the default set adapts to your search
 
criteria.
 
""")
 
    parser.add_argument(
 
        '--sheet-size', '--size',
 
        metavar='SIZE',
 
        type=int,
 
        default=LedgerODS.SHEET_SIZE,
 
        help="""Try to limit sheets to this many rows. The report will
 
automatically create new sheets to make this happen. When that's not possible,
setup.py
Show inline comments
 
#!/usr/bin/env python3
 

	
 
from setuptools import setup
 

	
 
setup(
 
    name='conservancy_beancount',
 
    description="Plugin, library, and reports for reading Conservancy's books",
 
    version='1.2.3',
 
    version='1.2.4',
 
    author='Software Freedom Conservancy',
 
    author_email='info@sfconservancy.org',
 
    license='GNU AGPLv3+',
 

	
 
    install_requires=[
 
        'babel>=2.6',  # Debian:python3-babel
 
        'beancount>=2.2',  # Debian:beancount
 
        # 1.4.1 crashes when trying to save some documents.
 
        'odfpy>=1.4.0,!=1.4.1',  # Debian:python3-odf
 
        'PyYAML>=3.0',  # Debian:python3-yaml
 
        'regex',  # Debian:python3-regex
 
        'rt>=2.0',
 
    ],
 
    setup_requires=[
 
        'pytest-mypy',
 
        'pytest-runner',  # Debian:python3-pytest-runner
 
    ],
 
    tests_require=[
 
        'mypy>=0.770',  # Debian:python3-mypy
 
        'pytest',  # Debian:python3-pytest
 
    ],
 

	
 
    packages=[
 
        'conservancy_beancount',
 
        'conservancy_beancount.plugin',
 
        'conservancy_beancount.reports',
 
    ],
 
    entry_points={
 
        'console_scripts': [
 
            'accrual-report = conservancy_beancount.reports.accrual:entry_point',
 
            'ledger-report = conservancy_beancount.reports.ledger:entry_point',
 
        ],
 
    },
 
)
tests/test_reports_spreadsheet.py
Show inline comments
...
 
@@ -130,246 +130,276 @@ def check_currency_style(curr_style):
 
    assert len(child_names) > 1
 

	
 
def test_spreadsheet(spreadsheet):
 
    rows = [(ch, ii) for ii, ch in enumerate('aabbcc', 1)]
 
    spreadsheet.write(iter(rows))
 
    assert spreadsheet.written_rows == rows
 
    assert spreadsheet.ended_sections == spreadsheet.started_sections
 
    assert spreadsheet.started_sections == list('abc')
 
    assert spreadsheet.start_call == []
 
    assert spreadsheet.end_call == spreadsheet.ended_sections
 

	
 
def test_empty_spreadsheet(spreadsheet):
 
    empty_list = []
 
    spreadsheet.write(iter(empty_list))
 
    assert spreadsheet.start_call == empty_list
 
    assert spreadsheet.end_call == empty_list
 
    assert spreadsheet.started_sections == empty_list
 
    assert spreadsheet.ended_sections == empty_list
 
    assert spreadsheet.written_rows == empty_list
 

	
 
def test_one_section_spreadsheet(spreadsheet):
 
    rows = [('A', n) for n in range(1, 4)]
 
    spreadsheet.write(iter(rows))
 
    assert spreadsheet.written_rows == rows
 
    assert spreadsheet.ended_sections == spreadsheet.started_sections
 
    assert spreadsheet.started_sections == list('A')
 
    assert spreadsheet.start_call == []
 
    assert spreadsheet.end_call == spreadsheet.ended_sections
 

	
 
def test_ods_writer(ods_writer):
 
    rows = [(ch, ii) for ii, ch in enumerate('aabbcc', 1)]
 
    ods_writer.write(iter(rows))
 
    sheets = ods_writer.document.getElementsByType(odf.table.Table)
 
    assert len(sheets) == 1
 
    for exp_row, act_row in zip(rows, testutil.ODSCell.from_sheet(sheets[0])):
 
        expected1, expected2 = exp_row
 
        actual1, actual2 = act_row
 
        assert actual1.value_type == 'string'
 
        assert actual1.text == expected1
 
        assert actual2.value_type == 'float'
 
        assert actual2.value == expected2
 
        assert actual2.text == str(expected2)
 

	
 
@pytest.mark.parametrize('save_type', ['file', 'path'])
 
def test_ods_writer_save(tmp_path, save_type):
 
    rows = list(zip('ABC', 'abc'))
 
    ods_writer = ODSTester()
 
    ods_writer.write(iter(rows))
 
    if save_type == 'file':
 
        ods_output = io.BytesIO()
 
        ods_writer.save_file(ods_output)
 
        ods_output.seek(0)
 
    else:
 
        ods_output = tmp_path / 'SavePathTest.ods'
 
        ods_writer.save_path(ods_output)
 
    for exp_row, act_row in zip(rows, testutil.ODSCell.from_ods_file(ods_output)):
 
        assert len(exp_row) == len(act_row)
 
        for expected, actual in zip(exp_row, act_row):
 
            assert actual.value_type == 'string'
 
            assert actual.value is None
 
            assert actual.text == expected
 

	
 
def test_ods_writer_use_sheet(ods_writer):
 
    names = ['One', 'Two']
 
    for name in names:
 
        ods_writer.use_sheet(name)
 
        ods_writer.write([(name,)])
 
    ods_writer.use_sheet('End')
 
    sheets = ods_writer.document.getElementsByType(odf.table.Table)
 
    assert len(sheets) == len(names) + 1
 
    for name, sheet in zip(names, sheets):
 
        texts = [cell.text for row in testutil.ODSCell.from_sheet(sheet)
 
                 for cell in row]
 
        assert texts == [name]
 

	
 
def test_ods_writer_use_sheet_returns_to_prior_sheets(ods_writer):
 
    names = ['One', 'Two']
 
    sheets = []
 
    for name in names:
 
        sheets.append(ods_writer.use_sheet(name))
 
        ods_writer.write([(name,)])
 
    for name, expected in zip(names, sheets):
 
        actual = ods_writer.use_sheet(name)
 
        assert actual is expected
 
        texts = [cell.text for row in testutil.ODSCell.from_sheet(actual)
 
                 for cell in row]
 
        assert texts == [name]
 

	
 
def test_ods_writer_use_sheet_discards_unused_sheets(ods_writer):
 
    ods_writer.use_sheet('Three')
 
    ods_writer.use_sheet('Two')
 
    ods_writer.use_sheet('One')
 
    sheets = ods_writer.document.getElementsByType(odf.table.Table)
 
    assert len(sheets) == 1
 
    assert sheets[0].getAttribute('name') == 'One'
 

	
 
@pytest.mark.parametrize('width,expect_name', [
 
    ('.750', 'col_0_75in'),
 
    (2, 'col_2in'),
 
    ('2.2in', 'col_2_2in'),
 
    (3.5, 'col_3_5in'),
 
    ('4cm', 'col_4cm'),
 
])
 
def test_ods_column_style(ods_writer, width, expect_name):
 
    style = ods_writer.column_style(width)
 
    assert style.getAttribute('name') == expect_name
 
    assert style.getAttribute('family') == 'table-column'
 
    curr_style = get_child(
 
        ods_writer.document.automaticstyles,
 
        odf.style.Style,
 
        name=expect_name,
 
    )
 
    assert get_child(
 
        curr_style,
 
        odf.style.TableColumnProperties,
 
        columnwidth=expect_name[4:].replace('_', '.'),
 
    )
 

	
 
def test_ods_column_style_caches(ods_writer):
 
    int_width = ods_writer.column_style('1in')
 
    float_width = ods_writer.column_style('1.00in')
 
    assert int_width is float_width
 

	
 
@pytest.mark.parametrize('width', [
 
    '1mi',
 
    '0in',
 
    '-1cm',
 
    'in',
 
    '.cm',
 
])
 
def test_ods_column_style_invalid_width(ods_writer, width):
 
    with pytest.raises(ValueError):
 
        ods_writer.column_style(width)
 

	
 
@pytest.mark.parametrize('currency_code', [
 
    'USD',
 
    'EUR',
 
    'BRL',
 
])
 
def test_ods_currency_style(ods_writer, currency_code):
 
    style = ods_writer.currency_style(currency_code)
 
    assert style.getAttribute('family') == 'table-cell'
 
    curr_style = get_child(
 
        ods_writer.document.styles,
 
        odf.number.CurrencyStyle,
 
        name=style.getAttribute('datastylename'),
 
    )
 
    check_currency_style(curr_style)
 
    mappings = get_children(curr_style, odf.style.Map)
 
    assert mappings
 
    for mapping in mappings:
 
        check_currency_style(get_child(
 
            ods_writer.document.styles,
 
            odf.number.CurrencyStyle,
 
            name=mapping.getAttribute('applystylename'),
 
        ))
 

	
 
def test_ods_currency_style_caches(ods_writer):
 
    expected = ods_writer.currency_style('USD')
 
    _ = ods_writer.currency_style('EUR')
 
    actual = ods_writer.currency_style('USD')
 
    assert actual is expected
 

	
 
def test_ods_currency_style_cache_considers_properties(ods_writer):
 
    bold_text = odf.style.TextProperties(fontweight='bold')
 
    plain = ods_writer.currency_style('USD')
 
    bold = ods_writer.currency_style('USD', positive_properties=bold_text)
 
    assert plain is not bold
 
    assert plain.getAttribute('name') != bold.getAttribute('name')
 
    assert plain.getAttribute('datastylename') != bold.getAttribute('datastylename')
 

	
 
@pytest.mark.parametrize('attr_name,child_type,checked_attr', [
 
    ('style_col1', odf.style.TableColumnProperties, 'columnwidth'),
 
    ('style_col1_25', odf.style.TableColumnProperties, 'columnwidth'),
 
    ('style_col1_5', odf.style.TableColumnProperties, 'columnwidth'),
 
    ('style_col1_75', odf.style.TableColumnProperties, 'columnwidth'),
 
    ('style_col2', odf.style.TableColumnProperties, 'columnwidth'),
 
    ('style_bold', odf.style.TextProperties, 'fontweight'),
 
    ('style_centertext', odf.style.ParagraphProperties, 'textalign'),
 
    ('style_dividerline', odf.style.TableCellProperties, 'borderbottom'),
 
    ('style_endtext', odf.style.ParagraphProperties, 'textalign'),
 
    ('style_starttext', odf.style.ParagraphProperties, 'textalign'),
 
])
 
def test_ods_writer_style(ods_writer, attr_name, child_type, checked_attr):
 
    if child_type is odf.style.TableColumnProperties:
 
        root = ods_writer.document.automaticstyles
 
    else:
 
        root = ods_writer.document.styles
 
    root = ods_writer.document.styles
 
    style = getattr(ods_writer, attr_name)
 
    actual = get_child(root, odf.style.Style, name=style.getAttribute('name'))
 
    assert actual is style
 
    child = get_child(actual, child_type)
 
    assert child.getAttribute(checked_attr)
 

	
 
def test_ods_writer_merge_styles(ods_writer):
 
    style = ods_writer.merge_styles(ods_writer.style_bold, ods_writer.style_dividerline)
 
    actual = get_child(
 
        ods_writer.document.styles,
 
        odf.style.Style,
 
        name=style.getAttribute('name'),
 
    )
 
    assert actual is style
 
    assert actual.getAttribute('family') == 'table-cell'
 
    text_props = get_child(actual, odf.style.TextProperties)
 
    assert text_props.getAttribute('fontweight') == 'bold'
 
    cell_props = get_child(actual, odf.style.TableCellProperties)
 
    assert cell_props.getAttribute('borderbottom')
 

	
 
def test_ods_writer_merge_styles_with_children_and_attributes(ods_writer):
 
    jpy_style = ods_writer.currency_style('JPY')
 
    style = ods_writer.merge_styles(ods_writer.style_bold, jpy_style)
 
    actual = get_child(
 
        ods_writer.document.styles,
 
        odf.style.Style,
 
        name=style.getAttribute('name'),
 
    )
 
    assert actual is style
 
    assert actual.getAttribute('family') == 'table-cell'
 
    assert actual.getAttribute('datastylename') == jpy_style.getAttribute('datastylename')
 
    text_props = get_child(actual, odf.style.TextProperties)
 
    assert text_props.getAttribute('fontweight') == 'bold'
 

	
 
def test_ods_writer_merge_styles_caches(ods_writer):
 
    sources = [ods_writer.style_bold, ods_writer.style_dividerline]
 
    style1 = ods_writer.merge_styles(*sources)
 
    style2 = ods_writer.merge_styles(*reversed(sources))
 
    assert style1 is style2
 
    assert get_child(
 
        ods_writer.document.styles,
 
        odf.style.Style,
 
        name=style1.getAttribute('name'),
 
    )
 

	
 
def test_ods_writer_layer_merge_styles(ods_writer):
 
    usd_style = ods_writer.currency_style('USD')
 
    layer1 = ods_writer.merge_styles(ods_writer.style_bold, ods_writer.style_dividerline)
 
    layer2 = ods_writer.merge_styles(layer1, usd_style)
 
    style_name = layer2.getAttribute('name')
 
    assert style_name.count('Merge_') == 1
 
    actual = get_child(
 
        ods_writer.document.styles,
 
        odf.style.Style,
 
        name=style_name,
 
    )
 
    assert actual is layer2
 
    assert actual.getAttribute('family') == 'table-cell'
 
    assert actual.getAttribute('datastylename') == usd_style.getAttribute('datastylename')
 
    text_props = get_child(actual, odf.style.TextProperties)
 
    assert text_props.getAttribute('fontweight') == 'bold'
 
    cell_props = get_child(actual, odf.style.TableCellProperties)
 
    assert cell_props.getAttribute('borderbottom')
 

	
 
def test_ods_writer_merge_one_style(ods_writer):
 
    actual = ods_writer.merge_styles(None, ods_writer.style_bold)
 
    assert actual is ods_writer.style_bold
 

	
 
def test_ods_writer_merge_no_styles(ods_writer):
 
    assert ods_writer.merge_styles() is None
 

	
 
def test_ods_writer_merge_nonexistent_style(ods_writer):
 
    name = 'Non Existent Style'
 
    with pytest.raises(ValueError, match=repr(name)):
 
        ods_writer.merge_styles(ods_writer.style_bold, name)
 

	
 
def test_ods_writer_merge_conflicting_styles(ods_writer):
 
    sources = [ods_writer.currency_style(code) for code in ['USD', 'EUR']]
 
    with pytest.raises(ValueError, match='conflicting datastylename'):
 
        ods_writer.merge_styles(*sources)
 

	
 
def test_ods_writer_date_style(ods_writer):
 
    data_style_name = ods_writer.style_date.getAttribute('datastylename')
 
    actual = get_child(
 
        ods_writer.document.styles,
 
        odf.style.Style,
 
        family='table-cell',
 
        datastylename=data_style_name,
 
    )
 
    assert actual is ods_writer.style_date
 
    data_style = get_child(
 
        ods_writer.document.styles,
 
        odf.number.DateStyle,
 
        name=data_style_name,
 
    )
 
    assert len(data_style.childNodes) == 5
0 comments (0 inline, 0 general)