"""Parser for EarningsTransfer records from ows-royalties. Fetches earnings transfer configurations via GET /earnings-transfers/ and maps each record to a TransferRecord for downstream calculation. API record schema (royalty_accounting.earnings_transfer table): - earnings_transfer_id: int - from_contract_id: mediumint unsigned - to_contract_id: mediumint unsigned - transfer_type: enum('cross_recoup','reclass','override','transfer','nr_transfer') - rate_type: enum('percent','flat_rate') - transfer_amount: decimal(20,12) — percent stored as 0–1; flat_rate as dollar amount - input: enum('net_revenue','gross_revenue','closing_balance') - negative: tinyint(1) — if True, allow transfers from a negative balance - active: tinyint(1) — if False, record is skipped - comment: str | None """ from __future__ import annotations from typing import Any from lambdacommon.common_config import logger from owsclient import OwsClient from src.connectors.ows_royalties import get_earnings_transfers from src.connectors.snowflake.connection import SnowflakeConnection from src.connectors.snowflake.query import fetch_contract_data from src.enums import RateType, TransferType from src.errors import InputValidationError from src.types import ContractData, ContractRef, TransferRecord _DB_TRANSFER_TYPE_MAP: dict[str, TransferType] = { 'cross_recoup': TransferType.CROSS_RECOUPMENT, 'reclass': TransferType.RECLASS, 'override': TransferType.OVERRIDE, 'transfer': TransferType.TRANSFER, 'nr_transfer': TransferType.TRANSFER, } _VALID_INPUTS: frozenset[str] = frozenset( {'closing_balance', 'net_revenue', 'gross_revenue'} ) def _map_transfer_type(db_type: str) -> TransferType: try: return _DB_TRANSFER_TYPE_MAP[db_type] except KeyError: raise InputValidationError(f'Unknown transfer_type: {db_type!r}') def _map_rate_type(db_rate_type: str) -> RateType: try: return RateType(db_rate_type) except ValueError: raise InputValidationError(f'Unknown rate_type: {db_rate_type!r}') def _validate_input(db_input: str) -> str: if db_input not in _VALID_INPUTS: raise InputValidationError(f'Unknown input: {db_input!r}') return db_input def api_record_to_transfer_record(r: dict[str, Any]) -> TransferRecord: """Map a single OWS API record to a TransferRecord. Raises: InputValidationError: If required fields are missing or invalid. """ transfer_type = _map_transfer_type(r['transfer_type']) rate_type = _map_rate_type(r['rate_type']) transfer_amount = float(r['transfer_amount']) comment = r.get('comment') or '' _id = r.get('earnings_transfer_id', '') if rate_type == RateType.PERCENT and not (0 <= transfer_amount <= 1.0): raise InputValidationError( f'earnings_transfer_id={_id}: ' f'percent transfer_amount must be between 0.0 and 1.0; got {transfer_amount}' ) if rate_type == RateType.FLAT_RATE and transfer_amount <= 0: raise InputValidationError( f'earnings_transfer_id={_id}: ' f'flat_rate transfer_amount must be > 0; got {transfer_amount}' ) return TransferRecord( earnings_transfer_id=int(r['earnings_transfer_id']), transfer_type=transfer_type, rate_type=rate_type, transfer_amount=transfer_amount, input=_validate_input(r.get('input', 'closing_balance')), negative=bool(r.get('negative', False)), use_static_balance=bool(r.get('use_static_balance', False)), from_contract=ContractRef(contract_id=int(r['from_contract_id'])), to_contract=ContractRef(contract_id=int(r['to_contract_id'])), from_comment=comment, to_comment=comment, description=comment, status='active', ) def _enrich_record( record: TransferRecord, contract_data: dict[int, ContractData] ) -> TransferRecord: """Annotate a TransferRecord with contract metadata and balances from Snowflake for both FROM and TO contracts.""" updates: dict[str, object] = {} from_cid = record.from_contract.contract_id if from_cid is not None and from_cid in contract_data: data = contract_data[from_cid] updates['closing_balance'] = data.closing_balance updates['net_revenue'] = data.net_revenue updates['gross_revenue'] = data.gross_revenue if not record.currency: updates['currency'] = data.currency updates['from_contract'] = record.from_contract.model_copy( update={ 'account_name': data.account_name or record.from_contract.account_name, 'account_id': data.account_id, 'contract_name': data.contract_name or record.from_contract.contract_name, } ) to_cid = record.to_contract.contract_id if to_cid is not None and to_cid in contract_data: to_data = contract_data[to_cid] updates['to_closing_balance'] = to_data.closing_balance updates['to_net_revenue'] = to_data.net_revenue updates['to_gross_revenue'] = to_data.gross_revenue updates['to_contract'] = record.to_contract.model_copy( update={ 'account_name': to_data.account_name or record.to_contract.account_name, 'account_id': to_data.account_id, 'contract_name': to_data.contract_name or record.to_contract.contract_name, } ) return record.model_copy(update=updates) if updates else record _INPUT_BALANCE_FIELD: dict[str, str] = { 'closing_balance': 'closing_balance', 'net_revenue': 'net_revenue', 'gross_revenue': 'gross_revenue', } def _validate_enriched(record: TransferRecord) -> TransferRecord: """Flag records missing required Snowflake data so they surface as error rows.""" from_cid = record.from_contract.contract_id if ( record.closing_balance is None and record.net_revenue is None and record.gross_revenue is None ): return record.model_copy( update={ 'error': f'Contract {from_cid} not found in Snowflake', 'error_contract_id': from_cid, } ) balance_field = _INPUT_BALANCE_FIELD[record.input] balance_value = getattr(record, balance_field) if balance_value is None: return record.model_copy( update={ 'error': f'Contract {from_cid} has no {record.input} in Snowflake', 'error_contract_id': from_cid, } ) return record def parse_earnings_transfers( ows_client: OwsClient, conn: SnowflakeConnection ) -> list[TransferRecord]: """Fetch and parse all earnings transfers from ows-royalties, enriched with Snowflake data. Args: ows_client: Configured OwsClient instance. conn: Active SnowflakeConnection for fetching contract metadata and balances. Returns: List of TransferRecord objects with contract metadata and balances populated. Raises: InputValidationError: If the API call fails or returns invalid data. """ raw_records: list[dict[str, Any]] = get_earnings_transfers(ows_client) sample = [ { 'id': r.get('earnings_transfer_id'), 'active': r.get('active'), 'active_type': type(r.get('active')).__name__, } for r in raw_records[:5] ] logger.info( 'Raw records from ows-royalties: count=%s sample_active_values=%s', len(raw_records), sample, ) active_records = [r for r in raw_records if r.get('active') in (1, True, '1')] filtered_out = [ { 'id': r.get('earnings_transfer_id'), 'active': r.get('active'), 'active_type': type(r.get('active')).__name__, } for r in raw_records if r.get('active') not in (1, True, '1') ][:5] logger.info( 'After active filter: before=%s after=%s filtered_out=%s', len(raw_records), len(active_records), filtered_out, ) # Cross-recoupment is not yet supported via the OWS API format. # Skip those records until the new EarningsTransfer structure is ready. cr_skipped = [r for r in active_records if r.get('transfer_type') == 'cross_recoup'] if cr_skipped: logger.info( 'Skipping cross-recoupment records (not yet supported): count=%s', len(cr_skipped), ) active_records = [ r for r in active_records if r.get('transfer_type') != 'cross_recoup' ] skipped = len(raw_records) - len(active_records) logger.info( 'Parsing earnings transfers: total=%s active=%s skipped=%s', len(raw_records), len(active_records), skipped, ) records: list[TransferRecord] = [] errors: list[str] = [] for r in active_records: try: records.append(api_record_to_transfer_record(r)) except (InputValidationError, KeyError, TypeError, ValueError) as exc: errors.append(str(exc)) logger.warning( 'Skipping invalid earnings transfer record', extra={'error': str(exc), 'record': r}, ) if errors: raise InputValidationError( f'{len(errors)} invalid record(s) in ows-royalties response:\n' + '\n'.join(f' - {e}' for e in errors) ) from_ids = [ r.from_contract.contract_id for r in records if r.from_contract.contract_id ] to_ids = [r.to_contract.contract_id for r in records if r.to_contract.contract_id] all_ids = list({*from_ids, *to_ids}) contract_data = fetch_contract_data(conn, all_ids) logger.info( 'Fetched contract data from Snowflake', extra={'contract_count': len(contract_data)}, ) enriched = [_enrich_record(r, contract_data) for r in records] return [_validate_enriched(r) for r in enriched]