"""Ledger account contract model.""" from sqlalchemy import and_ from sqlalchemy import Column from sqlalchemy import DateTime from sqlalchemy import ForeignKey from sqlalchemy import func from sqlalchemy import Integer from sqlalchemy import Numeric from sqlalchemy import select from sqlalchemy import String from sqlalchemy.orm import aliased from sqlalchemy.sql import literal_column from sqlalchemy.sql import table from moneyhub.connectors.mysql import db from moneyhub.models.abacus_event import AbacusEvent from moneyhub.models.abacus_state import AbacusState from moneyhub.models.account_payment_term import AccountPaymentTerm from moneyhub.models.contract import Contract from moneyhub.models.mysql_base import BaseModel from moneyhub.models.reference_signing_entity import ReferenceSigningEntity from moneyhub.models.statement_period_payment_entity import StatementPeriodPaymentEntity class LedgerAccountContract(BaseModel): """Ledger account contract model.""" __tablename__ = 'ledger_account_contract' ledger_account_contract_id = Column(Integer, primary_key=True) abacus_event_id = Column(Integer, ForeignKey(AbacusEvent.abacus_event_id)) account_id = Column(Integer, nullable=False) contract_id = Column(Integer, ForeignKey(Contract.contract_id)) currency_code = Column(String(3), nullable=True) currency_amount = Column(Numeric(20, 2), nullable=True) previous_balance = Column(Numeric(20, 2), nullable=True) current_balance = Column(Numeric(20, 2), nullable=True) note = Column(String(255), nullable=True) created_by = Column(String(255), nullable=True) created_at = Column(DateTime, nullable=True) last_modified_by = Column(String(255), nullable=True) last_modified = Column(DateTime, nullable=True) @classmethod def get_contracts_by_statement_period( cls, statement_period_id: int ) -> list: """Get contracts that have ledger events for a given statement period. Args: statement_period_id (int): Statement period to get contracts for. Returns: list: Contracts with associated signing entity. """ query = select( LedgerAccountContract.account_id, LedgerAccountContract.contract_id, AbacusEvent.statement_period_id, ReferenceSigningEntity.company_code, ReferenceSigningEntity.tax_entity_company_code ) \ .distinct() \ .select_from(LedgerAccountContract) \ .join( AbacusEvent, LedgerAccountContract.abacus_event_id == AbacusEvent.abacus_event_id) \ .join( Contract, LedgerAccountContract.contract_id == Contract.contract_id) \ .join( ReferenceSigningEntity, Contract.reference_signing_entity_id == ReferenceSigningEntity.reference_signing_entity_id) \ .join( AccountPaymentTerm, AccountPaymentTerm.account_id == LedgerAccountContract.account_id) \ .join( StatementPeriodPaymentEntity, StatementPeriodPaymentEntity.reference_payment_entity_id == AccountPaymentTerm.payment_entity_id)\ .filter( AbacusEvent.statement_period_id == statement_period_id, StatementPeriodPaymentEntity.statement_period_id == statement_period_id, StatementPeriodPaymentEntity.is_visible_to_customer == 1) \ .order_by(LedgerAccountContract.ledger_account_contract_id.asc()) return db.session.execute(query).fetchall() @classmethod def get_events_for_account_and_contract( cls, account_id: int, contract_id: int | None, visible_periods: list[int] | None = None, first_only: bool = False ) -> list: """Get ledger events for an account. Args: account_id (int): Account to get events for. visible_periods (list): list of visible statement periods to filter by contract_id (int): Contract to filter by. first_only (Bool): optional boolean parameter to return only the first record Returns: list: Ledger account events. """ filters = [LedgerAccountContract.account_id == account_id] if visible_periods: filters.append(AbacusEvent.statement_period_id.in_(visible_periods)) if contract_id: filters.append(LedgerAccountContract.contract_id == contract_id) query = select( LedgerAccountContract.ledger_account_contract_id, LedgerAccountContract.account_id, LedgerAccountContract.contract_id, LedgerAccountContract.currency_code, LedgerAccountContract.currency_amount, LedgerAccountContract.previous_balance, LedgerAccountContract.current_balance, AbacusEvent.statement_period_id, AbacusEvent.event_name, ) \ .select_from(LedgerAccountContract) \ .join( AbacusEvent, LedgerAccountContract.abacus_event_id == AbacusEvent.abacus_event_id) \ .filter(*filters) \ .order_by(LedgerAccountContract.ledger_account_contract_id.asc()) if first_only: return db.session.execute(query).first() return db.session.execute(query).fetchall() @classmethod def get_latest_balances_for_account(cls, account_id, statement_period_id) -> list: """Get most recent balances for all contracts in an account before (or at) a given period. Args: account_id (int): Account to get events for. statement_period_id (int): Statement period id to filter by. Returns: list: Payable balances for each contract. """ filters = [ LedgerAccountContract.account_id == account_id, AbacusEvent.statement_period_id <= statement_period_id] # Sub-query for getting the ID of latest ledger entry for each contract latest_id_query = select( func.max(LedgerAccountContract.ledger_account_contract_id).label('latest_id'), LedgerAccountContract.contract_id ) \ .select_from(LedgerAccountContract) \ .join( AbacusEvent, AbacusEvent.abacus_event_id == LedgerAccountContract.abacus_event_id) \ .filter(*filters) \ .group_by(LedgerAccountContract.contract_id) \ .subquery('latest_id_query') # Get the rest of the data based on those ledger entries query = select( LedgerAccountContract.ledger_account_contract_id, LedgerAccountContract.account_id, LedgerAccountContract.contract_id, LedgerAccountContract.current_balance ) \ .select_from(LedgerAccountContract, latest_id_query) \ .filter( LedgerAccountContract.ledger_account_contract_id == latest_id_query.c.latest_id, LedgerAccountContract.contract_id == latest_id_query.c.contract_id) return db.session.execute(query).fetchall() @classmethod def get_payments_by_account_and_statement_periods( cls, account_id: int, contract_id: int | None, statement_period_ids: list[int] ) -> list: """Get payments for specified account and statement period. Args: account_id (int): The id of account contract_id (int): Optional contract id to filter by statement_period_ids (list): The ids of the statement periods Returns: list: list of payments """ filters = [ LedgerAccountContract.account_id == account_id, AbacusEvent.statement_period_id.in_(statement_period_ids), AbacusEvent.event_name == 'send_payments', AbacusEvent.target_type == 'payment_group_payment' ] if contract_id: filters.append(LedgerAccountContract.contract_id == contract_id) WithholdingTaxLedgerAccountContract = aliased(LedgerAccountContract) WithholdingTaxAbacusEvent = aliased(AbacusEvent) query = select([ LedgerAccountContract.ledger_account_contract_id.label('ledger_account_id'), LedgerAccountContract.account_id, LedgerAccountContract.contract_id, LedgerAccountContract.currency_code, LedgerAccountContract.currency_amount, LedgerAccountContract.created_at, AbacusEvent.statement_period_id, AbacusEvent.event_name, AbacusState.action_status, WithholdingTaxLedgerAccountContract.ledger_account_contract_id.label('withholding_tax_ledger_account_id'), # noqa: E501 WithholdingTaxLedgerAccountContract.currency_code.label('withholding_tax_currency_code'), # noqa: E501 WithholdingTaxLedgerAccountContract.currency_amount.label('withholding_tax_currency_amount'), # noqa: E501 WithholdingTaxLedgerAccountContract.created_at.label('withholding_tax_created_at') ]) \ .select_from(LedgerAccountContract) \ .join( AbacusEvent, AbacusEvent.abacus_event_id == LedgerAccountContract.abacus_event_id) \ .outerjoin( table('payment_group_payment').alias('pgp'), literal_column('pgp.payment_group_payment_id') == AbacusEvent.target_id) \ .outerjoin( table('payment_group_payment_account').alias('pgpa'), and_( literal_column('pgpa.account_id') == LedgerAccountContract.account_id, literal_column('pgpa.payment_group_payment_id') == literal_column('pgp.payment_group_payment_id') # noqa: E501 )) \ .outerjoin( AbacusState, and_( AbacusState.parent_table_id == literal_column('pgpa.payment_group_payment_account_id'), # noqa: E501 AbacusState.parent_table_name == 'payment_group_payment_account' )) \ .outerjoin( WithholdingTaxAbacusEvent, and_( WithholdingTaxAbacusEvent.target_type == AbacusEvent.target_type, WithholdingTaxAbacusEvent.target_id == AbacusEvent.target_id, WithholdingTaxAbacusEvent.event_name == 'tax_withholding' )) \ .outerjoin( WithholdingTaxLedgerAccountContract, and_( WithholdingTaxLedgerAccountContract.abacus_event_id == WithholdingTaxAbacusEvent.abacus_event_id, # noqa: E501 WithholdingTaxLedgerAccountContract.account_id == LedgerAccountContract.account_id # noqa: E501 )) \ .where(*filters) return db.session.execute(query).fetchall() @classmethod def get_credit_payments_for_account_statement_periods( cls, account_id: int, contract_id: int | None, statement_period_ids: list[int] ) -> list: """Get credit payments for specified account and statement period. Args: account_id (int): The id of account contract_id (int): Optional contract id to filter by statement_period_ids (list): The ids of the statement periods Returns: list: list of payments """ filters = [ LedgerAccountContract.account_id == account_id, AbacusEvent.statement_period_id.in_(statement_period_ids), AbacusEvent.event_name == 'payment_returned', AbacusEvent.target_type == 'payment_group_payment_account' ] if contract_id: filters.append(LedgerAccountContract.contract_id == contract_id) query = select([ LedgerAccountContract.ledger_account_contract_id.label('ledger_account_id'), LedgerAccountContract.account_id, LedgerAccountContract.contract_id, LedgerAccountContract.currency_code, LedgerAccountContract.currency_amount, LedgerAccountContract.created_at, AbacusEvent.statement_period_id, AbacusEvent.event_name ]) \ .select_from(LedgerAccountContract) \ .join( AbacusEvent, AbacusEvent.abacus_event_id == LedgerAccountContract.abacus_event_id) \ .where(*filters) return db.session.execute(query).fetchall() @classmethod def get_reserves_released_by_account_and_statement_periods( cls, account_id: int, contract_id: int | None, statement_period_ids: list[int] ) -> list: """Get reserves released for specified account and statement period. Args: account_id (int): The id of account contract_id (int): Optional contract id to filter by statement_period_ids (list): The id of statement periods Returns: list: list of reserves released """ filters = [ LedgerAccountContract.account_id == account_id, AbacusEvent.statement_period_id.in_(statement_period_ids), AbacusEvent.target_type == 'statement_period', AbacusEvent.event_name == 'release_reserves' ] if contract_id: filters.append(LedgerAccountContract.contract_id == contract_id) query = select([ LedgerAccountContract.ledger_account_contract_id, LedgerAccountContract.account_id, LedgerAccountContract.currency_code, LedgerAccountContract.currency_amount, AbacusEvent.statement_period_id ]) \ .select_from(LedgerAccountContract) \ .join( AbacusEvent, AbacusEvent.abacus_event_id == LedgerAccountContract.abacus_event_id) \ .where(*filters) return db.session.execute(query).fetchall() @classmethod def get_reserves_taken_by_account_and_statement_periods( cls, account_id: int, contract_id: int | None, statement_period_ids: list[int] ) -> list: """Get reserves taken for specified account and statement period. Args: account_id (int): The id of account contract_id (int): Optional contract id to filter by statement_period_ids (list): The id of statement periods Returns: list: list of reserves taken """ filters = [ LedgerAccountContract.account_id == account_id, AbacusEvent.statement_period_id.in_(statement_period_ids), AbacusEvent.target_type == 'accounting_run', AbacusEvent.event_name == 'take_reserves' ] if contract_id: filters.append(LedgerAccountContract.contract_id == contract_id) query = select([ LedgerAccountContract.ledger_account_contract_id, LedgerAccountContract.account_id, LedgerAccountContract.currency_code, LedgerAccountContract.currency_amount, AbacusEvent.statement_period_id ]) \ .select_from(LedgerAccountContract) \ .join( AbacusEvent, AbacusEvent.abacus_event_id == LedgerAccountContract.abacus_event_id) \ .where(*filters) return db.session.execute(query).fetchall() @classmethod def get_vat_summaries( cls, account_id: int, contract_id: int | None, statement_period_ids: list[int], ) -> list: """Get VAT summaries for an account/statement periods. Args: account_id (int): Account to get entries for contract_id (int): Optional contract to filter by statement_period_ids (list): Statement periods to get entries Returns: list: VAT summary entries """ filters = [ cls.account_id == account_id, AbacusEvent.statement_period_id.in_(statement_period_ids), AbacusEvent.event_name == 'commit_vat_summary' ] if contract_id: filters.append(cls.contract_id == contract_id) fields = [ cls.account_id, cls.currency_code, cls.currency_amount, AbacusEvent.statement_period_id ] return cls.query.with_entities(*fields)\ .join( AbacusEvent, AbacusEvent.abacus_event_id == cls.abacus_event_id) \ .filter(*filters) \ .all() @classmethod def get_by_id_and_account(cls, account_id: int, contract_id: int): """Get contract for specified account and contract id. Args: account_id (int): The id of account contract_id (int): The id of contract Returns: list: list of account contract records """ filters = [cls.contract_id == contract_id, cls.account_id == account_id] return cls.query.filter(*filters).first()