"""Build GraphQL inputs from parsed CSV rows. Handles required-field validation, business defaults, and entity map lookups. Returns BuildResult (BuildSuccess | BuildSkip) with explicit tracking of any defaults applied. """ import logging from collections.abc import Callable from datetime import datetime from schemas import ( AbacusContractInput, AbacusContractLifecycleInput, AbacusContractLifecycleScheduleInput, AbacusContractWithLifecyclesInput, BuildResult, BuildSkip, BuildSuccess, ContractRow, MissingFieldPolicy, PeriodType, ReferenceSigningEntity, RenewalType, RunController, ) logger = logging.getLogger(__name__) class ContractInputBuilder: """Converts parsed ContractRows into GraphQL input models. Accepts lists of domain models and builds internal lookup maps. """ def __init__( self, signing_entities: list[ReferenceSigningEntity], run_controllers: list[RunController], clock: Callable[[], datetime] = datetime.now, ): self._signing_entity_map: dict[str, ReferenceSigningEntity] = { e.legal_name.lower(): e for e in signing_entities } self._run_controller_map: dict[str, RunController] = { rc.run_controller_name.lower(): rc for rc in run_controllers if rc.run_controller_name } self._clock = clock def build( self, row: ContractRow, policy: MissingFieldPolicy = MissingFieldPolicy.DEFAULT, ) -> BuildResult: """Convert a ContractRow into a GraphQL input. Returns BuildSuccess with the input and any defaults applied, or BuildSkip with the reason the row was skipped. """ defaults: list[str] = [] # --- Required fields --- if not row.account_id: logger.warning('Skipping row: Account ID is required') return BuildSkip('Account ID is required') if not row.contract_name: logger.warning( f'Skipping row for Account {row.account_id}: Contract Name is required' ) return BuildSkip('Contract Name is required') if row.contract_type is None: logger.warning( f"Skipping '{row.contract_name}': Invalid or missing contract type" ) return BuildSkip('Invalid or missing contract type') # --- Entity lookups --- signing_entity_id = self._resolve_signing_entity( row.contract_name, row.signing_entity ) if signing_entity_id is None and row.signing_entity: if policy == MissingFieldPolicy.SKIP: return BuildSkip( f"Signing entity '{row.signing_entity}' not found in map" ) run_controller_id = self._resolve_run_controller( row.contract_name, row.run_controller ) # --- Lifecycle --- period_start = self._resolve_period_start( row.contract_name, row.current_period_start, policy, defaults ) if period_start is None: return BuildSkip('Current Period Start Date is required') # --- Schedule --- schedule = self._build_schedule(row, period_start, defaults) # --- Assemble GraphQL input --- gql_input = AbacusContractWithLifecyclesInput( contract=AbacusContractInput( account_id=row.account_id, contract_name=row.contract_name, contract_type=row.contract_type, execution_date=row.execution_date, is_excluded_from_accounting_run=( row.is_excluded if row.is_excluded is not None else False ), is_primary_contract=( row.is_primary if row.is_primary is not None else True ), reference_signing_entity_id=signing_entity_id, run_controller_id=run_controller_id, ), lifecycle=AbacusContractLifecycleInput( lifecycle_term_start=period_start, ), lifecycle_schedules=[schedule], ) return BuildSuccess(input=gql_input, defaults_applied=tuple(defaults)) # ------------------------------------------------------------------ # Private resolution helpers # ------------------------------------------------------------------ def _resolve_signing_entity( self, contract_name: str, signing_entity: str | None ) -> int | None: if not signing_entity: return None entity = self._signing_entity_map.get(signing_entity.lower()) if entity is None: logger.warning( f"'{contract_name}': Signing entity '{signing_entity}' not found" ) return None return entity.reference_signing_entity_id def _resolve_run_controller( self, contract_name: str, run_controller: str | None ) -> int | None: if not run_controller: return None rc = self._run_controller_map.get(run_controller.lower()) if rc is None: logger.warning( f"'{contract_name}': Run controller '{run_controller}' not found" ) return None return rc.run_controller_id def _resolve_period_start( self, contract_name: str, period_start: str | None, policy: MissingFieldPolicy, defaults: list[str], ) -> str | None: if period_start: return period_start logger.warning( f"'{contract_name}': Current Period Start Date is required for lifecycle" ) if policy == MissingFieldPolicy.SKIP: return None today = self._clock().strftime('%Y-%m-%d') logger.warning(f"'{contract_name}': Using today's date as lifecycle_term_start") defaults.append(f'lifecycle_term_start={today}') return today @staticmethod def _build_schedule( row: ContractRow, period_start: str, defaults: list[str] ) -> AbacusContractLifecycleScheduleInput: renewal_type = row.renewal_rules if renewal_type is None: logger.warning( f"'{row.contract_name}': Invalid or missing " "renewal rule, defaulting to 'continuously_active'" ) renewal_type = RenewalType.CONTINUOUSLY_ACTIVE defaults.append('renewal_type=continuously_active') termination_interval = row.termination_interval termination_type = row.termination_type if not termination_interval or termination_type is None: logger.warning( f"'{row.contract_name}': Termination notice " 'period is required, using default (30 days)' ) termination_interval = 30 termination_type = PeriodType.DAY defaults.append('termination_notice=30 days') renewal_offset_interval = None renewal_offset_type = None if renewal_type == RenewalType.RENEW_PERIODICALLY: renewal_offset_interval = row.renew_after_interval renewal_offset_type = row.renew_after_type if not renewal_offset_interval or renewal_offset_type is None: logger.warning( f"'{row.contract_name}': Renew periodically requires renewal offset" ) renewal_offset_interval = None renewal_offset_type = None collection_interval = row.collection_interval collection_type = row.collection_type if not collection_interval or collection_type is None: collection_interval = None collection_type = None return AbacusContractLifecycleScheduleInput( renewal_type=renewal_type, schedule_end=row.current_period_end, termination_notice_detail_interval=termination_interval, termination_notice_detail_type=termination_type, renewal_offset_detail_interval=renewal_offset_interval, renewal_offset_detail_type=renewal_offset_type, collection_period_detail_interval=collection_interval, collection_period_detail_type=collection_type, contract_lifecycle=AbacusContractLifecycleInput( lifecycle_term_start=period_start, ), )