"""Lambda process function module.""" from collections import defaultdict from decimal import Decimal from math import ceil from typing import Any, Dict, List, Optional import pydantic from config import app_logger as logger from src import constants from src.connectors.ows_account import get_account_payment_hold, get_eligible_accounts from src.connectors.ows_event import get_events_by_target_type from src.connectors.ows_payment import ( bulk_create_payment_accounts, bulk_create_worksheet_contract_balance_after_tax, bulk_update_payment_allocations_flowthrough, delete_balance_entries_after_tax, get_bulk_last_payments, get_bulk_payable_details, get_bulk_payment_allocations_flowthrough, get_contract_closing_balance_entries, get_last_payment, get_payable_balance_after_tax_entries, get_payment_group, get_payment_group_payment, get_payment_minimums, ) from src.connectors.ows_royalties import get_payment_entity from src.connectors.ows_state import get_abacus_states, update_abacus_state_by_id from src.features import is_refactoring_enabled from src.models import ( AbacusState, Account, AccountPayableContract, AccountPaymentDetails, AggregatedBalancesAfterTax, ContractCloseBalance, Event, LambdaResponse, PayableBalanceAfterTax, PayableDetails, PaymentAccount, PaymentAccountInstance, PaymentAllocationFlowthrough, PaymentAllocationFlowthroughUpdate, PaymentGroupPayment, PaymentMethodMinimum, ) from src.processors import Processor from src.utils import fetch_all_accounts def handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: """Lambda entry point.""" if is_refactoring_enabled(): return new_handler(event, context) else: # TODO: TAP-3493 remove old handler and all related code on FF teardown return old_handler(event, context) def new_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: """Lambda entry point.""" logger.info('Received event', extra={'event': event}) try: valid_event = Event.model_validate(event) except pydantic.ValidationError as exc: logger.error(f'Failed to parse event: {exc.errors()}') return LambdaResponse( status_code=400, status_description='INVALID_EVENT', ).model_dump(by_alias=True) payment_group_payment = _payment_group_payment(valid_event) if not payment_group_payment: return LambdaResponse( status_code=400, status_description=constants.NO_PAYMENT_GROUP_PAYMENT_FOUND.format( valid_event.target_id ), # noqa: E501 ).model_dump(by_alias=True) state = _get_generate_payment_state(valid_event) _update_state_status(state, constants.ACTION_STATUS_RUNNING) try: Processor().process(valid_event) except Exception as exc: _update_state_status(state, constants.ACTION_STATUS_ERROR) return LambdaResponse( status_code=500, status_description=f'Failed to process event: {exc}', ).model_dump(by_alias=True) _update_state_status(state, constants.ACTION_STATUS_COMPLETE) return LambdaResponse( status_code=200, status_description=constants.SUCCESS_MSG, ).model_dump(by_alias=True) def old_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: """Lambda entry point.""" logger.info('Received event', extra={'event': event}) try: valid_event = Event.model_validate(event) except pydantic.ValidationError as exc: logger.error(f'Failed to parse event: {exc.errors()}') return LambdaResponse( status_code=400, status_description='INVALID_EVENT', ).model_dump(by_alias=True) payment_group_payment = _payment_group_payment(valid_event) if not payment_group_payment: return LambdaResponse( status_code=400, status_description=constants.NO_PAYMENT_GROUP_PAYMENT_FOUND.format( valid_event.target_id ), # noqa: E501 ).model_dump(by_alias=True) state = _get_generate_payment_state(valid_event) _update_state_status(state, constants.ACTION_STATUS_RUNNING) payment_group_id = payment_group_payment.payment_group_id accounts = _get_payment_group_accounts(payment_group_id) if not accounts: logger.info(constants.NO_ELIGIBLE_ACCOUNTS.format(payment_group_id)) _update_state_status(state, constants.ACTION_STATUS_COMPLETE) return LambdaResponse( status_code=200, status_description=constants.NO_ELIGIBLE_ACCOUNTS.format(payment_group_id), ).model_dump(by_alias=True) response = _create_payments_for_precalculated_data(valid_event, state, accounts) return response def _get_payment_group_accounts(payment_group_id: int) -> List[Account]: payment_group = get_payment_group(payment_group_id) reference_payment_type_id = (payment_group.group_criteria or {}).get( 'reference_payment_type_id' ) # managed payment group for specific payment types if reference_payment_type_id: agreement_type_ids = (payment_group.group_criteria or {}).get( 'reference_agreement_types' ) logger.info( f'Fetching managed payment group accounts with reference_payment_type_id: {reference_payment_type_id},' f' agreement_type_ids: {agreement_type_ids}' ) accounts = [ account for account in fetch_all_accounts( reference_payment_type_id=reference_payment_type_id, agreement_type_ids=agreement_type_ids, ) if not ( (hold := get_account_payment_hold(account.account_id)) and hold.is_on_hold ) ] else: logger.info('Fetching regular payment group accounts') accounts = get_eligible_accounts(payment_group_id) return accounts def _create_payments_for_precalculated_data( valid_event: Event, state: AbacusState, eligible_accounts: List[Account] ) -> Dict[str, Any]: """Creates payment based on previously calculated data.""" payment_minimums = _get_payment_minimums() eligible_accounts_batched = _batch_accounts(eligible_accounts) payable_balance_after_tax_entries = _fetch_all_payable_balance_after_tax_entries( valid_event ) account_batch_mapping = _map_accounts_to_batch(eligible_accounts_batched) payable_balance_after_tax_entries_batches = ( _map_payable_balance_after_tax_entries_to_batch( account_batch_mapping, payable_balance_after_tax_entries ) ) payment_accounts_ids = [] for batch_id, accounts_batch in enumerate(eligible_accounts_batched): payable_balance_after_tax_entries_batch = ( payable_balance_after_tax_entries_batches.get(batch_id, []) ) account_balances_after_tax = _aggregate_worksheets_by_account( payable_balance_after_tax_entries_batch ) try: payment_accounts = _create_payment_accounts_from_agg_balances( valid_event, accounts_batch, payment_minimums, account_balances_after_tax, payable_balance_after_tax_entries_batch, ) except Exception as exc: logger.error(f'Failed to process batch: {exc}') _update_state_status(state, constants.ACTION_STATUS_ERROR) return LambdaResponse( status_code=500, status_description=f'Failed to process batch: {exc}', ).model_dump(by_alias=True) payment_accounts_ids.extend( [pa.payment_group_payment_account_id for pa in payment_accounts] if payment_accounts else [] ) try: _update_payment_allocations_flowthrough(payment_accounts_ids) except Exception as exc: logger.error(f'Failed to update flowthrough allocations: {exc}') _update_state_status(state, constants.ACTION_STATUS_ERROR) return LambdaResponse( status_code=500, status_description=f'Failed to update flowthrough allocations: {exc}', ).model_dump(by_alias=True) _update_state_status(state, constants.ACTION_STATUS_COMPLETE) return LambdaResponse( status_code=200, status_description=constants.SUCCESS_MSG, ).model_dump(by_alias=True) def _update_payment_allocations_flowthrough(payment_accounts_ids: List[int]) -> None: """Updates payment allocations for flowthrough payments.""" if not payment_accounts_ids: logger.info('No payment accounts flowthrough update') return payable_details = _fetch_all_payable_details_entries(payment_accounts_ids) if not payable_details: logger.info('No flowthrough details for update') return allocations = _fetch_all_allocations_flowthrough_entries( contract_ids=list({d.contract_id for d in payable_details}), payment_statuses=[ constants.PaymentAllocationStatuses.INIT, constants.PaymentAllocationStatuses.RETURNED, ], ) if not allocations: logger.info('No allocations for update') return update_payloads = [ PaymentAllocationFlowthroughUpdate( payment_allocation_id=a.payment_allocation_id, payment_status=constants.PaymentAllocationStatuses.ATTACHED_TO_PAYMENT, ledger_status=constants.PaymentAllocationLedgerStatuses.ATTACHED_TO_PAYMENT, ) for a in allocations ] logger.info(f'Updating allocations for {len(update_payloads)} items.') for i in range(0, len(update_payloads), constants.ACCOUNT_BATCH_SIZE): bulk_update_payment_allocations_flowthrough( update_payloads[i : i + constants.ACCOUNT_BATCH_SIZE] ) def _fetch_all_payable_details_entries( payment_accounts_ids: List[int], ) -> List[PayableDetails]: """Get all payable details.""" offset = 0 limit = constants.ACCOUNT_BATCH_SIZE entries: List[PayableDetails] = [] logger.info('Fetching all payable details...') while True: batch = get_bulk_payable_details( payment_accounts_ids, [constants.FLOWTHROUGH_PAYABLE_DETAIL_TYPE_ID], limit=limit, offset=offset, ) offset = offset + limit entries.extend(batch.items) logger.info('Fetched %d of %d entries.', offset, batch.total_count) if offset >= batch.total_count: break logger.info('Fetching all payable details completed.') return entries def _fetch_all_allocations_flowthrough_entries( payment_allocation_ids: List[int] | None = None, contract_ids: List[int] | None = None, payment_statuses: List[str] | None = None, ledger_statuses: List[str] | None = None, ) -> List[PaymentAllocationFlowthrough]: """Get all allocations flowthrough.""" offset = 0 limit = constants.ACCOUNT_BATCH_SIZE entries: List[PaymentAllocationFlowthrough] = [] logger.info('Fetching all allocation...') while True: batch = get_bulk_payment_allocations_flowthrough( payment_allocation_ids, contract_ids, payment_statuses, ledger_statuses, limit=limit, offset=offset, ) offset = offset + limit entries.extend(batch.items) logger.info('Fetched %d of %d entries.', offset, batch.total_count) if offset >= batch.total_count: break logger.info('Fetching all allocations completed.') return entries def _fetch_all_payable_balance_after_tax_entries( event: Event, ) -> List[PayableBalanceAfterTax]: """Get payable balance entries after tax for the event.""" offset = 0 limit = constants.ACCOUNT_BATCH_SIZE entries: List[PayableBalanceAfterTax] = [] calculate_payments_event = _get_related_calculate_payments_event(event) if not calculate_payments_event: return entries logger.info('Fetching all payable balance after tax entries...') while True: batch = get_payable_balance_after_tax_entries( calculate_payments_event.abacus_event_id, limit=limit, offset=offset ) offset = offset + limit entries.extend(batch.items) logger.info('Fetched %d of %d entries.', offset, batch.total_count) if offset >= batch.total_count: break logger.info('Fetching all payable balance after tax entries completed.') return entries def _get_related_calculate_payments_event(event: Event) -> Optional[Event]: """Get related calculate_payments event.""" events = get_events_by_target_type(event.target_type, event.target_id) calculate_event_name = constants.CALCULATE_PAYMENTS_ACTION_NAME return next( filter( lambda e: e.event_name == calculate_event_name, # type: ignore events, ), None, ) def _delete_balance_entries(event: Event) -> None: """Delete contract balance after tax entries.""" event_id = event.abacus_event_id logger.info(constants.DELETE_BALANCE_ENTRIES_AFTER_TAX.format(event_id)) return delete_balance_entries_after_tax(event_id) def _payment_group_payment(event: Event) -> Optional[PaymentGroupPayment]: """Get payment group payment detail.""" payment_group_payment_id = event.target_id logger.info(constants.PAYMENT_GROUP_MSG.format(payment_group_payment_id)) return get_payment_group_payment(payment_group_payment_id) or None def _batch_accounts(accounts: List[Account]) -> List[List[Account]]: """Split accounts into batches. Return a list of lists. Args: accounts (list): shallow list of distinct accounts Returns: a list of lists: [[account_1, account_2], [account_3]] """ batched_accounts = list() batch_size = int(constants.ACCOUNT_BATCH_SIZE) num_of_batches = ceil(len(accounts) / batch_size) for batch_num in range(num_of_batches): start = batch_num * batch_size end = start + batch_size batched_accounts.append(accounts[start:end]) logger.info(constants.ACCOUNTS_BATCHES_MSG.format(len(batched_accounts))) return batched_accounts def _map_accounts_to_batch( eligible_accounts_batched: List[List[Account]], ) -> Dict[int, int]: """Build account_id to batch_id mapping. Args: eligible_accounts_batched (list): list of lists of accounts Returns: dict: mapping of account_id to batch_id """ account_batch_mapping = { account.account_id: batch_id for batch_id, accounts_batch in enumerate(eligible_accounts_batched) for account in accounts_batch } return account_batch_mapping def _map_payable_balance_after_tax_entries_to_batch( account_batch_mapping: Dict[int, int], payable_balance_after_tax_entries: List[PayableBalanceAfterTax], ) -> Dict[int, List[PayableBalanceAfterTax]]: """Build batch_id to payable_balance_after_tax_entries mapping. Returns: dict: mapping of batch_id to list of PayableBalanceAfterTax entries """ payable_balance_after_tax_entries_batches = defaultdict(list) for entry in payable_balance_after_tax_entries: batch_id = account_batch_mapping.get(entry.account_id) if batch_id is not None: # not None to cover batch_id == 0 payable_balance_after_tax_entries_batches[batch_id].append(entry) return payable_balance_after_tax_entries_batches def _get_closing_balance_entries( event: Event, accounts: List[Account] ) -> List[ContractCloseBalance]: """Get contract close balance entities by accounts. It retrieves all closing balance entries for event.statement_period_id and if there are not all entries for this statement period it will try to get them from the previous period. """ logger.info(constants.CLOSE_BALANCE_MSG.format(len(accounts))) closing_balance_entries = _fetch_all_contract_closing_balance_entries( event.statement_period_id, accounts ) current_cbe_account_ids = set([cbe.account_id for cbe in closing_balance_entries]) accounts_missed = list( filter(lambda item: item.account_id not in current_cbe_account_ids, accounts) ) # noqa: E501 if accounts_missed: previous_statement_period = event.statement_period_id - 1 previous_balance_entries = _fetch_all_contract_closing_balance_entries( previous_statement_period, accounts_missed ) closing_balance_entries.extend(previous_balance_entries) return closing_balance_entries def _fetch_all_contract_closing_balance_entries( statement_period_id: int, accounts: List[Account] ) -> List[ContractCloseBalance]: """Get all contract closing balance entries.""" offset = 0 limit = constants.ACCOUNT_BATCH_SIZE closing_balance_entries: List[ContractCloseBalance] = [] while True: batch = get_contract_closing_balance_entries( statement_period_id, accounts, limit=limit, offset=offset ) offset = offset + limit closing_balance_entries.extend(batch.items) if offset >= batch.total_count: break return closing_balance_entries def _create_balance_entries_after_tax( event: Event, payment_entity_policy_country_mapping: Dict[int, str], eligible_accounts: List[Account], close_balance_entries: List[ContractCloseBalance], ) -> List[PayableBalanceAfterTax]: """Format and post contract balance after tax entries.""" formatted_balance_after_tax_entries = [ _format_payable_balance_after_tax_entry( payment_entity_policy_country_mapping, cbe, eligible_accounts ) for cbe in close_balance_entries # noqa: E501 ] logger.info( constants.POST_BALANCE_ENTRIES_AFTER_TAX.format( len(formatted_balance_after_tax_entries) ) ) # noqa: E501 bulk_create_worksheet_contract_balance_after_tax( event.abacus_event_id, event.statement_period_id, formatted_balance_after_tax_entries, ) return formatted_balance_after_tax_entries def _format_payable_balance_after_tax_entry( payment_entity_policy_country_mapping: Dict[int, str], close_balance_entry: ContractCloseBalance, eligible_accounts: List[Account], ) -> PayableBalanceAfterTax: """Format data to contract balance after tax entries.""" country_of_tax_policy = _get_country_of_tax_policy( close_balance_entry.reference_payment_entity_id, payment_entity_policy_country_mapping, ) country_of_tax_residence = _get_country_of_tax_residence( close_balance_entry, eligible_accounts ) amount = str(max(Decimal(close_balance_entry.amount), Decimal(0))) # don't change this return PayableBalanceAfterTax( worksheet_account_contract_closing_balance_id=close_balance_entry.worksheet_account_contract_closing_balance_id, # noqa: E501 contract_id=close_balance_entry.contract_id, account_id=close_balance_entry.account_id, payable_amount_pre_tax=amount, tax_withholding_amount=None, vat_amount=None, payable_amount_post_tax=amount, currency_code=close_balance_entry.currency_code, country_of_tax_policy=country_of_tax_policy, country_of_tax_residence=country_of_tax_residence, ) def _get_country_of_tax_policy( reference_payment_entity_id: int, payment_entity_policy_country_mapping: Dict[int, str], ) -> str: """Get country of tax policy for specified payment entity id.""" if reference_payment_entity_id not in payment_entity_policy_country_mapping: ref_payment_entity = get_payment_entity(reference_payment_entity_id) if ref_payment_entity: country_code = constants.PAYMENT_NAME_TO_COUNTRY_OF_TAX_POLICY_MAPPING[ ref_payment_entity.payment_entity_name ] # noqa: E501 payment_entity_id = ref_payment_entity.reference_payment_entity_id payment_entity_policy_country_mapping[payment_entity_id] = country_code return payment_entity_policy_country_mapping[reference_payment_entity_id] def _get_country_of_tax_residence( entry: ContractCloseBalance, eligible_accounts: List[Account] ) -> str: """Get country of tax residence for specified contract close balance entry.""" accounts = list( filter(lambda item: item.account_id == entry.account_id, eligible_accounts) ) # noqa: E501 return accounts[0].country_of_tax_residence # type: ignore def _get_payment_minimums() -> List[PaymentMethodMinimum]: """Get payment minimums.""" logger.info(constants.PAYMENT_MINIMUMS_MSG) payment_minimum_list = get_payment_minimums() return payment_minimum_list or list() def _calculate_account_balances_after_tax( balance_after_tax_entries: List[PayableBalanceAfterTax], ) -> Dict[int, Decimal]: """Calculate sum of contract balances per account.""" res: Dict[int, Decimal] = dict() for entry in balance_after_tax_entries: balance = res.get(entry.account_id, Decimal(0)) res[entry.account_id] = balance + Decimal(entry.payable_amount_post_tax) return res def _aggregate_worksheets_by_account( balance_after_tax_entries: List[PayableBalanceAfterTax], ) -> Dict[int, AggregatedBalancesAfterTax]: """Calculate sum of contract balances per account.""" res: Dict[int, AggregatedBalancesAfterTax] = dict() default = AggregatedBalancesAfterTax() for entry in balance_after_tax_entries: agg_obj = res.get(entry.account_id, default) payable_amount_pre_tax = Decimal(agg_obj.payable_amount_pre_tax) + Decimal( entry.payable_amount_pre_tax ) payable_amount_post_tax = agg_obj.payable_amount_post_tax + Decimal( entry.payable_amount_post_tax ) tax_withholding_amount: Decimal | None = agg_obj.tax_withholding_amount if entry.tax_withholding_amount is not None: if tax_withholding_amount is None: tax_withholding_amount = Decimal(0) tax_withholding_amount = tax_withholding_amount + Decimal( entry.tax_withholding_amount ) vat_amount: Decimal | None = agg_obj.vat_amount if entry.vat_amount is not None: if vat_amount is None: vat_amount = Decimal(0) vat_amount = vat_amount + Decimal(entry.vat_amount) upd_agg_obj = AggregatedBalancesAfterTax( payable_amount_pre_tax=payable_amount_pre_tax, tax_withholding_amount=tax_withholding_amount, vat_amount=vat_amount, payable_amount_post_tax=payable_amount_post_tax, ) res[entry.account_id] = upd_agg_obj return res def _get_generate_payment_state(event: Event) -> AbacusState: """Get generate_payment state using payment_group_payment_id.""" parent_table_name = 'payment_group_payment' logger.info(constants.GET_STATE_MSG.format(parent_table_name, event.target_id)) states = get_abacus_states(parent_table_name, event.target_id) action_name = constants.GENERATE_PAYMENTS_ACTION_NAME if not states: logger.error(constants.NO_STATE_MSG) raise ValueError( f'{parent_table_name} state not found for ID: {event.target_id}' ) # noqa: E501 generate_payments_state = next( (state for state in states if state.action_name == action_name), None ) if not generate_payments_state: logger.error(constants.NO_STATE_MSG.format(action_name)) raise ValueError( f'{parent_table_name} state not found for ID: {event.target_id}' ) # noqa: E501 return generate_payments_state def _update_state_status(state: AbacusState, status: str) -> None: """Update existing state with new state using abacus_state_id.""" abacus_state_id = state.abacus_state_id logger.info(constants.UPDATE_STATE_MSG.format(abacus_state_id)) update_abacus_state_by_id(abacus_state_id, dict(action_status=status)) def _get_last_account_payment_details(account: Account) -> AccountPaymentDetails: """Get last posted payment for each eligible account.""" account_payment_details = get_last_payment(account.account_id) if not account_payment_details: raise ValueError(f'Last posted payment not found for ID: {account.account_id}') return account_payment_details def _get_last_payments_map( accounts: List[Account], ) -> Dict[int, AccountPaymentDetails]: """Fetch and map last payments for all accounts.""" account_ids = [account.account_id for account in accounts] limit = constants.ACCOUNT_BATCH_SIZE last_payments_list = get_bulk_last_payments(account_ids, limit=limit) return { payment.account_id: payment for payment in last_payments_list if payment.account_id is not None } def _create_payment_accounts( event: Event, accounts: List[Account], payment_minimums: List[PaymentMethodMinimum], acc_balance_after_tax: Dict[int, Decimal], payable_balance_after_tax_entries: List[PayableBalanceAfterTax], ) -> None: """Post payment_accounts for eligible payments.""" last_payments_map = _get_last_payments_map(accounts) payment_accounts = [ _format_payment_account( account, acc_balance_after_tax, payable_balance_after_tax_entries, last_payments_map, ) for account in accounts # noqa: E501 if _check_balance_limit(account, payment_minimums, acc_balance_after_tax) ] if payment_accounts: logger.info(constants.CREATE_PAYMENT_ACCOUNT_MSG.format(len(payment_accounts))) bulk_create_payment_accounts(event.target_id, payment_accounts) def _create_payment_accounts_from_agg_balances( event: Event, accounts: List[Account], payment_minimums: List[PaymentMethodMinimum], agg_balance_after_tax: Dict[int, AggregatedBalancesAfterTax], payable_balance_after_tax_entries: List[PayableBalanceAfterTax], ) -> None | List[PaymentAccountInstance]: """Post payment_accounts for eligible payments.""" last_payments_map = _get_last_payments_map(accounts) payment_accounts = [ _format_payment_account_from_agg_balances( account, agg_balance_after_tax, payable_balance_after_tax_entries, last_payments_map, ) for account in accounts # noqa: E501 if _check_agg_balance_limit(account, payment_minimums, agg_balance_after_tax) ] if payment_accounts: logger.info(constants.CREATE_PAYMENT_ACCOUNT_MSG.format(len(payment_accounts))) return bulk_create_payment_accounts(event.target_id, payment_accounts) return None def _check_balance_limit( account: Account, payment_method_minimums: List[PaymentMethodMinimum], account_balances: Dict[int, Decimal], ) -> bool: """Check for payment more than minimum sum.""" payment_method_minimum = [ pmm for pmm in payment_method_minimums if pmm.currency_code == account.currency_code ][0] payment_method_minimum_dict = PaymentMethodMinimum.model_dump( payment_method_minimum ) # noqa: E501 payment_method_minimum_amount = payment_method_minimum_dict['check_amount'] return ( account.account_id in account_balances and Decimal(account.payment_minimum or 0) <= account_balances[account.account_id] and Decimal(payment_method_minimum_amount) <= account_balances[account.account_id] ) def _check_agg_balance_limit( account: Account, payment_method_minimums: List[PaymentMethodMinimum], account_agg_balances: Dict[int, AggregatedBalancesAfterTax], ) -> bool: """Check for payment more than minimum sum.""" payment_method_minimum = [ pmm for pmm in payment_method_minimums if pmm.currency_code == account.currency_code ][0] payment_method_minimum_dict = PaymentMethodMinimum.model_dump( payment_method_minimum ) # noqa: E501 payment_method_minimum_amount = payment_method_minimum_dict['check_amount'] if account.account_id in account_agg_balances: post_tax_amount = account_agg_balances[ account.account_id ].payable_amount_post_tax # noqa: E501 return ( Decimal(account.payment_minimum or 0) <= post_tax_amount and Decimal(payment_method_minimum_amount) <= post_tax_amount ) return False def _get_or_create_default_payment_details( account_id: int, last_payments_map: Dict[int, AccountPaymentDetails] ) -> AccountPaymentDetails: """Get payment details from map or create default for accounts without previous payments.""" acc_payment_details = last_payments_map.get(account_id) if not acc_payment_details: acc_payment_details = AccountPaymentDetails( account_id=account_id, current_statement_period_id=None, balance_after_tax=Decimal(0.0), ) return acc_payment_details def _format_payment_account_legacy( account: Account, account_balances: Dict[int, Decimal], payable_balance_after_tax_entries: List[PayableBalanceAfterTax], ) -> PaymentAccount: """Format data to payment_account entries (legacy individual fetch).""" account_balance = str(account_balances[account.account_id]) acc_payment_details = _get_last_account_payment_details(account) contracts_payable = _get_account_contracts_payable( account, payable_balance_after_tax_entries ) return PaymentAccount( contracts_payable=contracts_payable, currency_code=account.currency_code, current_balance=account_balance, last_payment=acc_payment_details.balance_after_tax or Decimal(0), tax_withholding=None, vat_amount=None, balance_after_tax=account_balance, account_id=account.account_id, payoneer_program_id=account.payoneer_program_id or 0, last_statement_period_id=acc_payment_details.current_statement_period_id, ) def _format_payment_account( account: Account, account_balances: Dict[int, Decimal], payable_balance_after_tax_entries: List[PayableBalanceAfterTax], last_payments_map: Dict[int, AccountPaymentDetails], ) -> PaymentAccount: """Format data to payment_account entries (optimized bulk fetch).""" account_balance = str(account_balances[account.account_id]) acc_payment_details = _get_or_create_default_payment_details( account.account_id, last_payments_map ) contracts_payable = _get_account_contracts_payable( account, payable_balance_after_tax_entries ) return PaymentAccount( contracts_payable=contracts_payable, currency_code=account.currency_code, current_balance=account_balance, last_payment=acc_payment_details.balance_after_tax or Decimal(0), tax_withholding=None, vat_amount=None, balance_after_tax=account_balance, account_id=account.account_id, payoneer_program_id=account.payoneer_program_id or 0, last_statement_period_id=acc_payment_details.current_statement_period_id, ) def _format_payment_account_from_agg_balances_legacy( account: Account, account_balances: Dict[int, AggregatedBalancesAfterTax], payable_balance_after_tax_entries: List[PayableBalanceAfterTax], ) -> PaymentAccount: """Format data to payment_account entries (legacy individual fetch).""" agg_account_balance = account_balances[account.account_id] acc_payment_details = _get_last_account_payment_details(account) contracts_payable = _get_account_contracts_payable( account, payable_balance_after_tax_entries ) return PaymentAccount( contracts_payable=contracts_payable, currency_code=account.currency_code, current_balance=agg_account_balance.payable_amount_pre_tax, last_payment=acc_payment_details.balance_after_tax or Decimal(0), tax_withholding=agg_account_balance.tax_withholding_amount, vat_amount=agg_account_balance.vat_amount, balance_after_tax=agg_account_balance.payable_amount_post_tax, account_id=account.account_id, payoneer_program_id=account.payoneer_program_id or 0, last_statement_period_id=acc_payment_details.current_statement_period_id, ) def _format_payment_account_from_agg_balances( account: Account, account_balances: Dict[int, AggregatedBalancesAfterTax], payable_balance_after_tax_entries: List[PayableBalanceAfterTax], last_payments_map: Dict[int, AccountPaymentDetails], ) -> PaymentAccount: """Format data to payment_account entries (optimized bulk fetch).""" agg_account_balance = account_balances[account.account_id] acc_payment_details = _get_or_create_default_payment_details( account.account_id, last_payments_map ) contracts_payable = _get_account_contracts_payable( account, payable_balance_after_tax_entries ) return PaymentAccount( contracts_payable=contracts_payable, currency_code=account.currency_code, current_balance=agg_account_balance.payable_amount_pre_tax, last_payment=acc_payment_details.balance_after_tax or Decimal(0), tax_withholding=agg_account_balance.tax_withholding_amount, vat_amount=agg_account_balance.vat_amount, balance_after_tax=agg_account_balance.payable_amount_post_tax, account_id=account.account_id, payoneer_program_id=account.payoneer_program_id or 0, last_statement_period_id=acc_payment_details.current_statement_period_id, ) def _is_valid_contract_payable(item: PayableBalanceAfterTax, account_id: int) -> bool: """Check that the payable entry is a valid contract payable item for the account.""" return item.account_id == account_id and Decimal(item.payable_amount_post_tax) > 0 def _get_account_contracts_payable( account: Account, payable_balance_after_tax_entries: List[PayableBalanceAfterTax] ) -> List[AccountPayableContract]: """Get account payable contracts from balance after tax entries.""" account_contract_balances = list( filter( lambda item: _is_valid_contract_payable(item, account.account_id), payable_balance_after_tax_entries, ) ) return [ AccountPayableContract( contract_id=acb.contract_id, currency_code=acb.currency_code, current_balance=acb.payable_amount_post_tax, ) for acb in account_contract_balances ]