"""Payment Batch Refresh Processor.""" from collections import defaultdict from decimal import Decimal from typing import Dict, List, Optional, Sequence from config import app_logger as logger from src.connectors.ows_payment import ( bulk_create_contract_payable_details, bulk_update_worksheet_contract_balance_after_tax, delete_payable_details_wht_vat_corrections, ) from src.constants import ( CORRECTION_DETAIL_GROUPS, DECIMAL_ZERO, NO_REFRESH_ENTRIES_MSG, PayableDetailTypes, REFRESH_BATCH_MSG, REFRESH_CLOSING_BALANCE_NOT_FOUND_ERR, REFRESH_DRIFT_MSG, REFRESH_ENTRIES_MSG, TableNames, TaxCorrectionTypes, ) from src.models import ( ContractCloseBalance, ContractPayableDetails, Event, PayableBalanceAfterTaxBulkUpdate, PayableBalanceAfterTaxEntry, TaxCorrection, TaxCorrectionVAT, ) from src.processors.base.worksheet_calculator import WorksheetCalculator from src.processors.models import WorksheetPayableCalculatorDetail from src.utils import ( fetch_all_contract_closing_balance_entries_bulk, fetch_all_payable_details_entries, fetch_all_pending_tax_corrections, fetch_all_pending_tax_corrections_vat, ) class PaymentBatchRefreshProcessor: """Refresh wht/vat for a batch of existing worksheet payable after tax entries. Pending tax corrections are the source of truth. For each entry whose stored amounts or correction payable details no longer match the currently pending corrections, the correction details are rebuilt and the after-tax amounts re-set to the pending sums. """ def __init__( self, abacus_event: Event, worksheet_after_tax_batch: List[PayableBalanceAfterTaxEntry], append_vat_corrections: Optional[bool] = True, ) -> None: """Init.""" self._abacus_event = abacus_event self._statement_period_id = abacus_event.statement_period_id self._worksheet_after_tax_batch = worksheet_after_tax_batch self._append_vat_corrections = append_vat_corrections self._contracts_by_period: Dict[int, List[int]] = {} # populated during prep self._pending_wht_by_contract: Dict[int, List[TaxCorrection]] = defaultdict( list ) self._pending_vat_by_contract: Dict[int, List[TaxCorrectionVAT]] = defaultdict( list ) self._detail_sums_by_after_tax: Dict[int, Dict[int, Decimal]] = {} def _get_closing_balance_related_data(self) -> List[ContractCloseBalance]: """Fetch closing balances for the batch using their known closing balance IDs.""" closing_balance_ids = [ entry.worksheet_account_contract_closing_balance_id for entry in self._worksheet_after_tax_batch ] return fetch_all_contract_closing_balance_entries_bulk(closing_balance_ids) def _get_contract_ids(self) -> List[int]: """Return contract ids for the batch.""" return [entry.contract_id for entry in self._worksheet_after_tax_batch] def _get_after_tax_ids(self) -> List[int]: """Return worksheet payable after tax ids for the batch.""" return [ entry.worksheet_account_contract_payable_after_tax_id for entry in self._worksheet_after_tax_batch ] def _set_contracts_by_period(self) -> None: """Populate self._contracts_by_period: a mapping of statement_period_id -> [contract_ids]. Each worksheet after-tax entry in the batch is linked to a closing balance, which carries the statement_period_id the contract was processed under. Payoneer payments can span two periods — the current event period for normal contracts and the previous period for "missed" contracts that had no closing balance in the current period. Steps: 1. Fetch closing balances by their known PKs from the batch entries. 2. Build a contract_id -> statement_period_id lookup from those closing balances. 3. Group the batch's contract_ids by their resolved statement_period_id. The resulting self._contracts_by_period is used by _set_pending_corrections to fetch WHT/VAT corrections for each contract from the correct statement period. """ closing_balances = self._get_closing_balance_related_data() # Step 2: map each contract to its statement period via the closing balance statement_period_by_contract: Dict[int, int] = { closing_balance.contract_id: closing_balance.statement_period_id for closing_balance in closing_balances } # Step 3: group batch contract_ids by their resolved statement period contracts_by_statement_period: Dict[int, List[int]] = defaultdict(list) for contract_id in self._get_contract_ids(): statement_period_id = statement_period_by_contract.get(contract_id) if statement_period_id is None: raise ValueError( REFRESH_CLOSING_BALANCE_NOT_FOUND_ERR.format(contract_id) ) contracts_by_statement_period[statement_period_id].append(contract_id) self._contracts_by_period = contracts_by_statement_period def _set_pending_corrections(self) -> None: """Fetch pending wht/vat corrections and group them by contract id. VAT processing is controlled at the processor level via self._append_vat_corrections (check payments pass False to skip VAT). """ for statement_period_id, contract_ids in self._contracts_by_period.items(): for wht_correction in fetch_all_pending_tax_corrections( TaxCorrectionTypes.wht, contract_ids, statement_period_id ): self._pending_wht_by_contract[wht_correction.contract_id].append( wht_correction ) if self._append_vat_corrections: for statement_period_id, contract_ids in self._contracts_by_period.items(): for vat_correction in fetch_all_pending_tax_corrections_vat( contract_ids, statement_period_id ): self._pending_vat_by_contract[vat_correction.contract_id].append( vat_correction ) def _set_current_detail_sums(self) -> None: """Fetch current correction payable details and sum by after-tax id and type.""" details = fetch_all_payable_details_entries( self._statement_period_id, self._get_after_tax_ids(), CORRECTION_DETAIL_GROUPS, ) for detail in details: # Response can have payable_detail_type_id in [2,3,4,5], only 4 and 5 are corrections, skip the rest if detail.payable_detail_type_id not in [ PayableDetailTypes.withholding_tax_correction, PayableDetailTypes.vat_correction, ]: continue per_type = self._detail_sums_by_after_tax.setdefault( detail.worksheet_account_contract_payable_after_tax_id, {} ) per_type[detail.payable_detail_type_id] = ( per_type.get(detail.payable_detail_type_id, DECIMAL_ZERO) + detail.amount_payable ) @staticmethod def _sum_amounts( corrections: Sequence[TaxCorrection] | Sequence[TaxCorrectionVAT], ) -> Decimal: """Sum the amounts of a list of corrections.""" total = DECIMAL_ZERO for correction in corrections: total += correction.amount return total def _needs_refresh(self, entry: PayableBalanceAfterTaxEntry) -> bool: """Return True if the entry's wht or vat is out of sync with pending corrections.""" detail_sums = self._detail_sums_by_after_tax.get( entry.worksheet_account_contract_payable_after_tax_id, {} ) current_wht = entry.tax_withholding_amount or DECIMAL_ZERO details_wht_sum = detail_sums.get( PayableDetailTypes.withholding_tax_correction, DECIMAL_ZERO ) pending_wht_sum = self._sum_amounts( self._pending_wht_by_contract.get(entry.contract_id, []) ) needs_wht = not (current_wht == details_wht_sum == pending_wht_sum) current_vat = entry.vat_amount or DECIMAL_ZERO details_vat_sum = detail_sums.get( PayableDetailTypes.vat_correction, DECIMAL_ZERO ) pending_vat_sum = self._sum_amounts( self._pending_vat_by_contract.get(entry.contract_id, []) ) needs_vat = not (current_vat == details_vat_sum == pending_vat_sum) if needs_wht: logger.info( REFRESH_DRIFT_MSG, TaxCorrectionTypes.wht, entry.worksheet_account_contract_payable_after_tax_id, current_wht, details_wht_sum, pending_wht_sum, ) if needs_vat: logger.info( REFRESH_DRIFT_MSG, TaxCorrectionTypes.vat, entry.worksheet_account_contract_payable_after_tax_id, current_vat, details_vat_sum, pending_vat_sum, ) return needs_wht or needs_vat def _build_refreshed_calculator( self, entry: PayableBalanceAfterTaxEntry ) -> WorksheetCalculator: """Rebuild a worksheet calculator from the entry and its pending corrections.""" calculator = WorksheetCalculator.from_payable_balance_after_tax_entry(entry) for wht_correction in self._pending_wht_by_contract.get(entry.contract_id, []): calculator.append_wht_item( WorksheetPayableCalculatorDetail( amount_payable=wht_correction.amount, target_table=TableNames.worksheet_tax_correction, target_id=wht_correction.worksheet_tax_correction_id, payable_detail_type_id=wht_correction.payable_detail_type_id, ) ) for vat_correction in self._pending_vat_by_contract.get(entry.contract_id, []): calculator.append_vat_item( WorksheetPayableCalculatorDetail( amount_payable=vat_correction.amount, target_table=TableNames.worksheet_tax_correction_vat, target_id=vat_correction.worksheet_tax_correction_vat_id, payable_detail_type_id=vat_correction.payable_detail_type_id, ) ) return calculator def process(self) -> None: """Validate the batch and refresh entries whose corrections drifted.""" if not self._worksheet_after_tax_batch: return logger.info(REFRESH_BATCH_MSG.format(len(self._worksheet_after_tax_batch))) self._set_contracts_by_period() self._set_pending_corrections() self._set_current_detail_sums() soft_delete_ids: List[int] = [] bulk_updates: List[PayableBalanceAfterTaxBulkUpdate] = [] reinsert_details: List[ContractPayableDetails] = [] for entry in self._worksheet_after_tax_batch: if not self._needs_refresh(entry): continue after_tax_id = entry.worksheet_account_contract_payable_after_tax_id calculator = self._build_refreshed_calculator(entry) soft_delete_ids.append(after_tax_id) bulk_updates.append( PayableBalanceAfterTaxBulkUpdate( worksheet_account_contract_payable_after_tax_id=after_tax_id, tax_withholding_amount=calculator.tax_withholding_amount, vat_amount=calculator.vat_amount, payable_amount_post_tax=calculator.payable_amount_post_tax, ) ) reinsert_details.extend(calculator.get_payable_detail_items(after_tax_id)) if not soft_delete_ids: logger.info(NO_REFRESH_ENTRIES_MSG) return logger.info( REFRESH_ENTRIES_MSG.format( len(soft_delete_ids), len(self._worksheet_after_tax_batch) ) ) delete_payable_details_wht_vat_corrections(soft_delete_ids) bulk_update_worksheet_contract_balance_after_tax(bulk_updates) bulk_create_contract_payable_details( self._abacus_event.abacus_event_id, self._statement_period_id, reinsert_details, )