from typing import Any, Dict, Optional from src.core.providers.base import BasePaymentProvider, ProviderConfig, IframeConfig, TokenDetails, ChargeResult from src.core.providers.registry import ProviderRegistry @ProviderRegistry.register class PayrixProvider(BasePaymentProvider): """ Payrix (Worldpay) payment provider implementation. Falls back to global settings.PAYRIX_* env vars when config.credentials is empty (backward compatibility for merchants onboarded before multi-provider support). """ @property def slug(self) -> str: return "payrix" def _get_auth_headers(self, config: ProviderConfig) -> Dict[str, str]: """Build auth headers from config credentials or fall back to global settings.""" headers = { "Accept": "application/json", "Content-Type": "application/json", } creds = config.credentials if creds.get("api_key") and creds.get("api_secret"): import base64 token = base64.b64encode(f"{creds['api_key']}:{creds['api_secret']}".encode()).decode() headers["Authorization"] = f"Basic {token}" elif creds.get("auth_token"): headers["Authorization"] = f"Bearer {creds['auth_token']}" else: # Fall back to global settings from src.core.payrix.config import get_auth_headers return get_auth_headers() return headers def _get_base_url(self, config: ProviderConfig) -> str: from src.core.config import settings return config.config_data.get("base_url") or settings.PAYRIX_BASE_URL or "" def get_iframe_config(self, config: ProviderConfig, amount: float = 0.0, currency: str = "USD") -> IframeConfig: creds = config.credentials cfg_data = config.config_data from src.core.config import settings base_url = self._get_base_url(config) is_sandbox = "test" in base_url # NOTE: api_secret MUST NOT be included in extra — IframeConfig is serialised # and returned to the browser, which would expose the secret to end users. return IframeConfig( sdk_url="https://test-api.payrix.com/payFieldsScript" if is_sandbox else "https://api.payrix.com/payFieldsScript", api_key=creds.get("api_key") or settings.PAYRIX_API_KEY or "", merchant_id=cfg_data.get("provider_merchant_id", ""), mode="test" if is_sandbox else "live", extra={}, ) async def get_token_details(self, config: ProviderConfig, token: str) -> TokenDetails: import httpx base_url = self._get_base_url(config) auth_headers = self._get_auth_headers(config) async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.get( f"{base_url}/tokens/{token}", headers=auth_headers, ) resp.raise_for_status() body = resp.json() return self._normalize_token_response(body, token) async def create_payment_method( self, config: ProviderConfig, token: str, customer_context: Dict[str, Any] ) -> TokenDetails: # For Payrix, the token from the iframe is already the payment method reference. # We fetch the token details to normalize the response. return await self.get_token_details(config, token) async def submit_charge( self, config: ProviderConfig, amount: float, currency: str, payment_method_token: str, payment_method_type: str, capture: bool, idempotency_key: Optional[str], metadata: Optional[Dict[str, Any]], ) -> ChargeResult: import httpx base_url = self._get_base_url(config) auth_headers = self._get_auth_headers(config) type_map = {"card": 1, "ach": 7} txn_type = type_map.get(payment_method_type, 1) payload: Dict[str, Any] = { "token": payment_method_token, "amount": int(amount * 100), # Payrix expects cents as integer "currency": currency, "capture": capture, "type": txn_type, } if metadata: payload["description"] = metadata.get("description", "") request_headers = dict(auth_headers) if idempotency_key: request_headers["Idempotency-Key"] = idempotency_key async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.post( f"{base_url}/txns", headers=request_headers, json=payload, ) body = resp.json() if resp.status_code >= 400: from src.core.exceptions import PayrixApiError raise PayrixApiError( status_code=resp.status_code, payrix_error_body=body, request_id=resp.headers.get("x-request-id", "") or None, ) return self._normalize_charge_response(body, amount, currency) async def refund_charge( self, config: ProviderConfig, provider_transaction_id: str, amount: float, reason: Optional[str], ) -> ChargeResult: raise NotImplementedError("Refund via Payrix not yet implemented. See the Refunds PRD.") def verify_webhook( self, config: ProviderConfig, headers: Dict[str, str], raw_body: bytes ) -> bool: from src.core.config import settings # Use per-merchant webhook secret from credentials, fall back to global setting webhook_secret = config.credentials.get("webhook_secret") or settings.PAYRIX_WEBHOOK_HEADER_VALUE if not webhook_secret: return False header_name = ( config.config_data.get("webhook_header_name") or settings.PAYRIX_WEBHOOK_HEADER_NAME or "x-payrix-webhook-secret" ) received_secret = headers.get(header_name, "") or headers.get(header_name.lower(), "") return received_secret == webhook_secret def parse_webhook_event(self, payload: Dict[str, Any]) -> Dict[str, Any]: from src.core.payrix.status_mapper import map_payrix_payload_to_transaction_status, transaction_status_enum_to_string # Extract transaction data from Payrix webhook payload data = payload.get("data", payload) if isinstance(data, list): data = data[0] if data else {} txn_id = ( data.get("id") or data.get("txnId") or data.get("transaction_id") or payload.get("id") or "" ) event_id = ( payload.get("id") or payload.get("event_id") or payload.get("eventId") or "" ) status_enum = map_payrix_payload_to_transaction_status(payload) status = transaction_status_enum_to_string(status_enum) amount_raw = data.get("amount") or data.get("total") or 0 # Payrix always returns amounts as integer cents; divide by 100 unconditionally. amount = float(amount_raw) / 100 if isinstance(amount_raw, int) else float(amount_raw) currency = data.get("currency", "USD") return { "event_id": event_id, "transaction_id": txn_id, "status": status, "amount": amount, "currency": currency, "provider_slug": self.slug, "raw": payload, } def _normalize_token_response(self, body: Dict[str, Any], token: str) -> TokenDetails: data = body.get("data", body) if isinstance(data, list): data = data[0] if data else {} expiry = data.get("expiry") exp_month = data.get("expMonth") exp_year = data.get("expYear") if isinstance(expiry, dict): exp_month = exp_month or expiry.get("month") exp_year = exp_year or expiry.get("year") return TokenDetails( provider_payment_method_id=data.get("id") or token, provider=self.slug, brand=data.get("brand") or data.get("cardType"), last4=data.get("last4") or data.get("cardLast4"), exp_month=exp_month, exp_year=exp_year, account_type=data.get("accountType"), bank_name=data.get("bankName"), reference_token=data.get("token") or token, ) def _normalize_charge_response(self, body: Dict[str, Any], amount: float, currency: str) -> ChargeResult: from src.core.payrix.status_mapper import map_payrix_payload_to_transaction_status, transaction_status_enum_to_string data = body.get("data", body) if isinstance(data, list): data = data[0] if data else {} txn_id = data.get("id") or data.get("txnId") or "" status_enum = map_payrix_payload_to_transaction_status(body) status = transaction_status_enum_to_string(status_enum) return ChargeResult( transaction_id=txn_id, status=status, amount=amount, currency=currency, raw_response=body, )