diff --git a/src/accounting/report/reports/unapplied.py b/src/accounting/report/reports/unapplied.py index 4913dcf..d24ec25 100644 --- a/src/accounting/report/reports/unapplied.py +++ b/src/accounting/report/reports/unapplied.py @@ -20,21 +20,26 @@ from datetime import date from decimal import Decimal +import sqlalchemy as sa from flask import render_template, Response +from sqlalchemy.orm import selectinload +from accounting import db from accounting.locale import gettext -from accounting.models import Currency, Account, JournalEntryLineItem +from accounting.models import Currency, Account, JournalEntry, \ + JournalEntryLineItem from accounting.report.period import Period, PeriodChooser from accounting.report.utils.base_page_params import BasePageParams from accounting.report.utils.base_report import BaseReport from accounting.report.utils.csv_export import BaseCSVRow, csv_download, \ period_spec -from accounting.report.utils.offset_matcher import OffsetMatcher from accounting.report.utils.option_link import OptionLink from accounting.report.utils.report_chooser import ReportChooser from accounting.report.utils.report_type import ReportType from accounting.report.utils.unapplied import get_accounts_with_unapplied from accounting.report.utils.urls import unapplied_url +from accounting.utils.cast import be +from accounting.utils.offset_alias import offset_alias from accounting.utils.pagination import Pagination @@ -179,12 +184,64 @@ class UnappliedOriginalLineItems(BaseReport): """The account.""" self.__period: Period = period """The period.""" - offset_matcher: OffsetMatcher \ - = OffsetMatcher(self.__currency, self.__account, self.__period) self.__line_items: list[JournalEntryLineItem] \ - = offset_matcher.unapplied + = self.__query_line_items() """The line items.""" + def __query_line_items(self) -> list[JournalEntryLineItem]: + """Queries and returns the line items. + + :return: The line items. + """ + net_balances: dict[int, Decimal | None] = self.__get_net_balances() + line_items: list[JournalEntryLineItem] = JournalEntryLineItem.query \ + .join(Account).join(JournalEntry) \ + .filter(JournalEntryLineItem.id.in_(net_balances)) \ + .order_by(JournalEntry.date, JournalEntry.no, + JournalEntryLineItem.is_debit, JournalEntryLineItem.no) \ + .options(selectinload(JournalEntryLineItem.currency), + selectinload(JournalEntryLineItem.journal_entry)).all() + for line_item in line_items: + line_item.net_balance = line_item.amount \ + if net_balances[line_item.id] is None \ + else net_balances[line_item.id] + return line_items + + def __get_net_balances(self) -> dict[int, Decimal | None]: + """Returns the net balances of the unapplied line items of the account. + + :return: The net balances of the unapplied line items of the account. + """ + offset: sa.Alias = offset_alias() + net_balance: sa.Label \ + = (JournalEntryLineItem.amount + + sa.func.sum(sa.case( + (be(offset.c.is_debit == JournalEntryLineItem.is_debit), + offset.c.amount), + else_=-offset.c.amount))).label("net_balance") + conditions: list[sa.BinaryExpression] \ + = [be(Account.id == self.__account.id), + be(JournalEntryLineItem.currency_code == self.__currency.code), + sa.or_(sa.and_(Account.base_code.startswith("2"), + sa.not_(JournalEntryLineItem.is_debit)), + sa.and_(Account.base_code.startswith("1"), + JournalEntryLineItem.is_debit))] + if self.__period.start is not None: + conditions.append(JournalEntry.date >= self.__period.start) + if self.__period.end is not None: + conditions.append(JournalEntry.date <= self.__period.end) + select_net_balances: sa.Select \ + = sa.select(JournalEntryLineItem.id, net_balance) \ + .join(JournalEntry).join(Account) \ + .join(offset, be(JournalEntryLineItem.id + == offset.c.original_line_item_id), + isouter=True) \ + .filter(*conditions) \ + .group_by(JournalEntryLineItem.id) \ + .having(sa.or_(sa.func.count(offset.c.id) == 0, net_balance != 0)) + return {x.id: x.net_balance + for x in db.session.execute(select_net_balances).all()} + def csv(self) -> Response: """Returns the report as CSV for download.