"""Account Statement Period Logic.""" from decimal import Decimal from sqlalchemy.engine.row import Row from moneyhub import models from moneyhub.constants.constants import OrderDirection from moneyhub.constants.features import MONEYHUB_STATEMENTS_VAT_LEDGER from moneyhub.models.ledger_account_contract import LedgerAccountContract from moneyhub.models.workstation_summary import WorkstationSummary from moneyhub.schemas.account_statement_period import AccountPaymentDetailStatusSchema from moneyhub.schemas.account_statement_period import AccountStatementPeriodBalanceSchema from moneyhub.schemas.account_statement_period import AccountStatementPeriodDetailSchema from moneyhub.schemas.account_statement_period import AccountStatementPeriodPaymentDetailSchema from moneyhub.schemas.account_statement_period import AccountStatementPeriodVatDetailSchema from moneyhub.schemas.account_statements import PaymentAllocationDetailSchema from moneyhub.utils.features import is_feature_enabled from moneyhub.utils.request import create_paginated_response def get_account_statement_periods_by_account( account_id: int, contract_id: int | None, limit: int = 50, offset: int = 0, order_dir: OrderDirection = OrderDirection.ASC ) -> dict: """Get ledger info per statement period for a specified account. Args: account_id (int): The id of an account contract_id (int | None): Contract to filter by limit (int): Number of records to return (pagination) offset (int): Number of records to skip (pagination) order_dir (OrderDirection): Order direction for statement IDs Returns: dict: objet containing list items and pagination details """ active_periods, total_periods = models.AccountStatementPeriods.get_by_account_id( account_id, contract_id, limit, offset, order_dir) if not active_periods: return create_paginated_response(items=[], total_records=0) statement_period_ids = [s.statement_period_id for s in active_periods] ledger_summary = models.LedgerSummary.get_for_account( account_id, contract_id, statement_period_ids, order_dir) workstation_summary = models.WorkstationSummary.get_revenue_for_account( account_id, contract_id, statement_period_ids, order_dir) def _add_workstation_entry(entry: Row): entries[entry.statement_period_id] = AccountStatementPeriodDetailSchema( account_id=account_id, contract_id=contract_id, statement_period_id=entry.statement_period_id, currency_code=entry.currency, total_gross_revenue_amount=entry.gross_revenue, total_net_revenue_amount=entry.net_revenue, distribution_fee=entry.fee, mechanical_deduction_total=entry.mechanicals, mechanical_deduction_admin_fee_total=entry.mechanical_fees, ) def _add_ledger_entry(entry: Row): if (entry.statement_period_id in entries and entry.total_gross_revenue_amount == 0 and entry.total_net_revenue_amount == 0 and entry.distribution_fee == 0): return # assume this is empty and defer to the Workstation entry entries[entry.statement_period_id] = AccountStatementPeriodDetailSchema( account_id=account_id, contract_id=contract_id, statement_period_id=entry.statement_period_id, currency_code=entry.currency_code, total_gross_revenue_amount=entry.total_gross_revenue_amount, total_net_revenue_amount=entry.total_net_revenue_amount, distribution_fee=entry.distribution_fee, mechanical_deduction_total=entry.mechanical_deduction_total, mechanical_deduction_admin_fee_total=entry.mechanical_deduction_admin_fee_total, ) entries = {} list(map(_add_workstation_entry, workstation_summary)) list(map(_add_ledger_entry, ledger_summary)) records_list = entries.values() return create_paginated_response(records_list, total_periods) def get_payments_by_account_and_statement_periods( account_id: int, contract_id: int | None, statement_period_ids: list[int] ) -> list[AccountStatementPeriodPaymentDetailSchema]: """GET payments info for a specified account and statement period. Args: account_id (int): The id of an account contract_id (int): Contract to filter by statement_period_ids (list): The ids of the statement periods Returns: list: list of payment details """ payments = models.CombinedPayments.get_payments_by_account_and_statement_periods( account_id, statement_period_ids, contract_id ) def _create_payment_entry(entry: Row): entries[entry.unique_key] = AccountStatementPeriodPaymentDetailSchema( account_id=entry.account_id, contract_id=entry.contract_id, ledger_account_id=entry.unique_key, statement_period_id=entry.statement_period_id, action_status=entry.action_status, currency_code=entry.currency_code, currency_amount=entry.currency_amount, created_at=entry.created_at, event_name=entry.event_name, withholding_tax_ledger_account_id=entry.withholding_tax_ledger_account_id, withholding_tax_currency_code=entry.withholding_tax_currency_code, withholding_tax_currency_amount=entry.withholding_tax_currency_amount, withholding_tax_created_at=entry.withholding_tax_created_at ) entries = {} list(map(_create_payment_entry, payments)) return list(entries.values()) def get_payments_details_by_account_and_statement_periods( account_id: int, contract_id: int | None, statement_period_ids: list[int] ) -> list[AccountPaymentDetailStatusSchema]: """GET payment details info for a specified account and statement period. Args: account_id (int): The id of an account contract_id (int): Contract to filter by statement_period_ids (list): The ids of the statement periods Returns: list: list of payment details """ if contract_id: payment_details = models.WorksheetAccountContractClosingBalance.get_payment_details( account_id, statement_period_ids, contract_id ) return payment_details return models.PaymentGroupPaymentAccount.get_payment_details( account_id, statement_period_ids ) def get_payment_allocation_details( account_id: int, contract_id: int | None, statement_period_ids: list[int] ) -> list[PaymentAllocationDetailSchema]: """GET payment allocation flowthrough details for a specified contract and statement period. Args: account_id (int): The id of an account contract_id (int): The contract id to filter by statement_period_ids (list): The ids of the statement periods Returns: list: list of payment allocation details filtered by flowthrough """ contract_ids = [contract_id] if not contract_id: contract_ids = models.AccountContract.get_by_account(account_id) return models.PaymentAllocation.get_flowthrough_details( contract_ids, statement_period_ids) def get_vat_by_account_and_statement_periods( account_id: int, contract_id: int | None, statement_period_ids: list[int] ) -> list[AccountStatementPeriodVatDetailSchema]: """Get VAT info for a specified account and statement period. Args: account_id (int): The id of an account contract_id (int): Contract to filter by statement_period_ids (list): List of statement period IDs to filter by Returns: list: List of VAT entries """ run_results = {} summary_results = {} # Ledger VAT summary entries get_vat_from_ledger = is_feature_enabled(MONEYHUB_STATEMENTS_VAT_LEDGER) if get_vat_from_ledger: ledger_entries = models.LedgerAccountContract.get_vat_summaries( account_id, contract_id, statement_period_ids) summary_results = {} for entry in ledger_entries: if entry.statement_period_id not in summary_results: summary_results[entry.statement_period_id] = { 'statement_period_id': entry.statement_period_id, 'account_id': account_id, 'payee_currency_code': entry.currency_code, 'currency_code': entry.currency_code, 'gross_revenue': None, 'net_revenue': None, 'distribution_fee': None, 'gross_vat_rate': None, 'distribution_vat_rate': None, 'gross_vat': None, 'distribution_vat': None, 'adjusted_net_revenue': None, 'base_amount_payee_currency': None, 'net_amount_payee_currency': None, 'vat_amount_payee_currency': Decimal(0), } summary_results[entry.statement_period_id]['vat_amount_payee_currency'] += entry.currency_amount # noqa: E501 else: summary_entries = models.LedgerVatSummary.get_visible_by_account_and_statement_periods( account_id, contract_id, statement_period_ids) aggregate_fields = [ 'base_amount_payee_currency', 'net_amount_payee_currency', 'vat_amount_payee_currency', ] for entry in summary_entries: if entry.statement_period_id not in summary_results: item = { 'statement_period_id': entry.statement_period_id, 'account_id': account_id, 'payee_currency_code': entry.payee_currency_code, 'currency_code': entry.payee_currency_code, 'gross_revenue': None, 'net_revenue': None, 'distribution_fee': None, 'gross_vat_rate': None, 'distribution_vat_rate': None, 'gross_vat': None, 'distribution_vat': None, 'adjusted_net_revenue': None, } for field in aggregate_fields: item[field] = Decimal(0) summary_results[entry.statement_period_id] = item for field in aggregate_fields: summary_results[entry.statement_period_id][field] += (getattr(entry, field) or 0) # Accounting run VAT entries (used in lambda when generating old distribution fee invoices) run_entries = models.LedgerAccountingRunVat.get_committed_for_account( account_id, contract_id, statement_period_ids) if len(run_entries) != 0: aggregate_fields = [ 'gross_revenue', 'net_revenue', 'distribution_fee', 'gross_vat', 'distribution_vat', 'adjusted_net_revenue', ] for entry in run_entries: entry = entry._asdict() statement_period_id = entry['statement_period_id'] if statement_period_id not in run_results: run_results[statement_period_id] = { 'statement_period_id': statement_period_id, 'account_id': account_id, 'currency_code': entry['currency_code'], 'gross_revenue': None, 'net_revenue': None, 'distribution_fee': None, 'gross_vat_rate': entry['gross_vat_rate'], 'distribution_vat_rate': entry['distribution_vat_rate'], 'gross_vat': None, 'distribution_vat': None, 'adjusted_net_revenue': None, } for field in aggregate_fields: if field in entry and entry[field] is not None: if run_results[statement_period_id][field] is None: run_results[statement_period_id][field] = Decimal(0) run_results[statement_period_id][field] += entry[field] # Fill in the new fields for key, item in run_results.items(): run_results[key]['payee_currency_code'] = item['currency_code'] run_results[key]['base_amount_payee_currency'] = item['gross_revenue'] run_results[key]['net_amount_payee_currency'] = item['adjusted_net_revenue'] run_results[key]['vat_amount_payee_currency'] = ( item['gross_vat'] or item['distribution_vat']) results = summary_results | run_results return [ AccountStatementPeriodVatDetailSchema(**entry) for entry in results.values() ] def get_balance_by_account_statement_periods( # noqa: C901 account_id: int, statement_period_ids: list[int], contract_id: int | None ) -> list[AccountStatementPeriodBalanceSchema] | None: """Get balance info for a specified account and statement periods. Args: account_id (int): The id of an account statement_period_ids (list[int]): Statement periods to filter by contract_id (int): Contract to filter by Returns: list[AccountStatementPeriodBalanceSchema]: account statement period details """ ledger_events = models.LedgerAccountContract.get_events_for_account_and_contract( account_id, contract_id) visible_periods = models.StatementPeriodPaymentEntity.get_visible_statement_period_ids( account_id) all_contract_ids = models.AccountContract.get_by_account(account_id) def _init_entries(statement_period_id: int): entries[statement_period_id] = { 'statement_period_id': statement_period_id, 'currency_code': None, 'opening_balance': None, 'ledger_amount': Decimal(0), 'closing_balance': None, } def _update_entry_with_ledger_event(event: LedgerAccountContract): event_statement_period = event.statement_period_id if event_statement_period not in visible_periods: return if event_statement_period not in entries: _init_entries(event_statement_period) if entries[event_statement_period]['opening_balance'] is None: entries[event_statement_period]['opening_balance'] = event.previous_balance entries[event_statement_period]['closing_balance'] = event.current_balance entries[event_statement_period]['ledger_amount'] += event.currency_amount if entries[event_statement_period]['currency_code'] is None: entries[event_statement_period]['currency_code'] = event.currency_code def _set_payable_balance(statement_period_id: int): current_balances = models.LedgerAccountContract.get_latest_balances_for_account( account_id, statement_period_id ) total_positive_current_balances = sum( item.current_balance for item in current_balances if item.current_balance > 0 ) flowthrough_balances = models.LedgerContractFlowthrough.get_latest_balances_for_contract( all_contract_ids, [statement_period_id] ) total_flowthrough_balances = sum(balance.amount for balance in flowthrough_balances) entries[statement_period_id]['payable_balance'] = (total_positive_current_balances + total_flowthrough_balances) running = { 'latest_balance': Decimal(0), 'latest_currency_code': None } def _update_latest_balances(statement_period_id: int): if entries[statement_period_id]['opening_balance'] is None: entries[statement_period_id]['opening_balance'] = running['latest_balance'] else: running['latest_balance'] = entries[statement_period_id]['opening_balance'] if entries[statement_period_id]['closing_balance'] is None: entries[statement_period_id]['closing_balance'] = running['latest_balance'] else: running['latest_balance'] = entries[statement_period_id]['closing_balance'] if entries[statement_period_id]['currency_code'] is None: entries[statement_period_id]['currency_code'] = running['latest_currency_code'] else: running['latest_currency_code'] = entries[statement_period_id]['currency_code'] legacy_balances = models.WorkstationSummary.get_balances_by_account_id( account_id, statement_period_ids, contract_id) if len(ledger_events) == 0 and len(legacy_balances) == 0: return None def _add_legacy_balances(balance: WorkstationSummary): balance_statement_period_id = balance.statement_period_id if (balance_statement_period_id in entries and entries[balance_statement_period_id]['opening_balance'] is not None): return if balance_statement_period_id in entries: entries[balance_statement_period_id] = { 'statement_period_id': balance.statement_period_id, 'currency_code': balance.currency, 'opening_balance': balance.opening_balance, 'ledger_amount': None, 'closing_balance': balance.closing_balance, } entries = {} list(map(_init_entries, statement_period_ids)) list(map(_update_entry_with_ledger_event, ledger_events)) list(map(_add_legacy_balances, legacy_balances)) list(map(_set_payable_balance, statement_period_ids)) sorted_keys = sorted(entries.keys()) list(map(_update_latest_balances, sorted_keys)) return [ AccountStatementPeriodBalanceSchema(**entry) for entry in entries.values() if entry['statement_period_id'] in statement_period_ids ] def get_flowthrough_payable_balances( contract_ids: list[int], statement_period_ids: list[int] ) -> list: """Get flowthrough payable balances for a contract and statement periods. Args: contract_ids (list[int]): ID of the contract to fetch balances for statement_period_ids (list): IDs of the statement periods Returns: list: list of flowthrough balances """ return models.LedgerContractFlowthrough.get_latest_balances_for_contract( contract_ids, statement_period_ids)