# Part of Odoo. See LICENSE file for full copyright and licensing details. import json import logging import time import requests from odoo import _, api, fields, models from odoo.exceptions import UserError, ValidationError from odoo.addons.fusion_poynt import const from odoo.addons.fusion_poynt import utils as poynt_utils _logger = logging.getLogger(__name__) class PaymentProvider(models.Model): _inherit = 'payment.provider' code = fields.Selection( selection_add=[('poynt', "Poynt")], ondelete={'poynt': 'set default'}, ) poynt_application_id = fields.Char( string="Application ID", help="The Poynt application ID (urn:aid:...) from your developer portal.", required_if_provider='poynt', copy=False, ) poynt_private_key = fields.Text( string="Private Key (PEM)", help="The RSA private key in PEM format, downloaded from the Poynt developer portal. " "Used to sign JWT tokens for OAuth2 authentication.", required_if_provider='poynt', copy=False, groups='base.group_system', ) poynt_business_id = fields.Char( string="Business ID", help="The merchant's Poynt business UUID.", required_if_provider='poynt', copy=False, ) poynt_store_id = fields.Char( string="Store ID", help="The Poynt store UUID for this location.", copy=False, ) poynt_webhook_secret = fields.Char( string="Webhook Secret", help="Secret key used to verify webhook notifications from Poynt.", copy=False, groups='base.group_system', ) poynt_default_terminal_id = fields.Many2one( 'poynt.terminal', string="Default Terminal", help="The default Poynt terminal used for in-store payment collection. " "Staff can override this per transaction.", domain="[('provider_id', '=', id), ('active', '=', True)]", ) # Cached access token fields (not visible in UI) _poynt_access_token = fields.Char( string="Access Token", copy=False, groups='base.group_system', ) _poynt_token_expiry = fields.Integer( string="Token Expiry Timestamp", copy=False, groups='base.group_system', ) # === COMPUTE METHODS === # def _compute_feature_support_fields(self): """Override of `payment` to enable additional features.""" super()._compute_feature_support_fields() self.filtered(lambda p: p.code == 'poynt').update({ 'support_manual_capture': 'full_only', 'support_refund': 'partial', 'support_tokenization': True, }) # === CRUD METHODS === # def _get_default_payment_method_codes(self): """Override of `payment` to return the default payment method codes.""" self.ensure_one() if self.code != 'poynt': return super()._get_default_payment_method_codes() return const.DEFAULT_PAYMENT_METHOD_CODES # === BUSINESS METHODS - AUTHENTICATION === # def _poynt_get_access_token(self): """Obtain an OAuth2 access token from Poynt using JWT bearer grant. Caches the token and only refreshes when it's about to expire. :return: The access token string. :rtype: str :raises ValidationError: If authentication fails. """ self.ensure_one() now = int(time.time()) if self._poynt_access_token and self._poynt_token_expiry and now < self._poynt_token_expiry - 30: return self._poynt_access_token jwt_assertion = poynt_utils.create_self_signed_jwt( self.poynt_application_id, self.poynt_private_key, ) token_url = f"{const.API_BASE_URL}{const.TOKEN_ENDPOINT}" if self.state == 'test': token_url = f"{const.API_BASE_URL_TEST}{const.TOKEN_ENDPOINT}" try: response = requests.post( token_url, data={ 'grantType': 'urn:ietf:params:oauth:grant-type:jwt-bearer', 'assertion': jwt_assertion, }, headers={ 'Content-Type': 'application/x-www-form-urlencoded', 'api-version': const.API_VERSION, 'Accept': 'application/json', }, timeout=30, ) if response.status_code >= 400: _logger.error( "Poynt token request failed (HTTP %s): %s", response.status_code, response.text[:1000], ) response.raise_for_status() token_data = response.json() except requests.exceptions.RequestException as e: _logger.error("Poynt OAuth2 token request failed: %s", e) raise ValidationError( _("Failed to authenticate with Poynt. Please check your credentials. Error: %s", e) ) access_token = token_data.get('accessToken') expires_in = token_data.get('expiresIn', 900) if not access_token: raise ValidationError( _("Poynt authentication returned no access token. " "Please verify your Application ID and Private Key.") ) self.sudo().write({ '_poynt_access_token': access_token, '_poynt_token_expiry': now + expires_in, }) return access_token # === BUSINESS METHODS - API REQUESTS === # def _poynt_make_request(self, method, endpoint, payload=None, params=None, business_scoped=True, store_scoped=False): """Make an authenticated API request to the Poynt REST API. :param str method: HTTP method (GET, POST, PUT, PATCH, DELETE). :param str endpoint: The API endpoint path. :param dict payload: The JSON request body (optional). :param dict params: The query parameters (optional). :param bool business_scoped: Whether to scope the URL to the business. :param bool store_scoped: Whether to scope the URL to the store. :return: The parsed JSON response. :rtype: dict :raises ValidationError: If the API request fails. """ self.ensure_one() access_token = self._poynt_get_access_token() is_test = self.state == 'test' business_id = self.poynt_business_id if business_scoped else None store_id = self.poynt_store_id if store_scoped and self.poynt_store_id else None url = poynt_utils.build_api_url( endpoint, business_id=business_id, store_id=store_id, is_test=is_test, ) request_id = poynt_utils.generate_request_id() headers = poynt_utils.build_api_headers(access_token, request_id=request_id) _logger.info( "Poynt API %s request to %s (request_id=%s)", method, url, request_id, ) try: response = requests.request( method, url, json=payload, params=params, headers=headers, timeout=60, ) except requests.exceptions.RequestException as e: _logger.error("Poynt API request failed: %s", e) raise ValidationError(_("Communication with Poynt failed: %s", e)) if response.status_code == 401: self.sudo().write({ '_poynt_access_token': False, '_poynt_token_expiry': 0, }) raise ValidationError( _("Poynt authentication expired. Please retry.") ) if response.status_code in (202, 204): return {} try: result = response.json() except ValueError: if response.status_code < 400: return {} _logger.error("Poynt returned non-JSON response: %s", response.text[:500]) raise ValidationError(_("Poynt returned an invalid response.")) if response.status_code >= 400: error_msg = result.get('message', result.get('developerMessage', 'Unknown error')) dev_msg = result.get('developerMessage', '') _logger.error( "Poynt API error %s: %s (request_id=%s)\n" " URL: %s %s\n Payload: %s\n Response: %s\n Developer: %s", response.status_code, error_msg, request_id, method, url, json.dumps(payload)[:2000] if payload else 'None', response.text[:2000], dev_msg, ) raise ValidationError( _("Poynt API error (%(code)s): %(msg)s", code=response.status_code, msg=dev_msg or error_msg) ) return result # === BUSINESS METHODS - TOKENIZE / CHARGE === # def _poynt_tokenize_nonce(self, nonce): """Exchange a Poynt Collect nonce for a long-lived payment token JWT. :param str nonce: The one-time nonce from Poynt Collect JS. :return: The tokenize response containing card details, cardId, paymentToken (JWT), and AVS/CVV verification results. :rtype: dict :raises ValidationError: If the tokenize call fails. """ self.ensure_one() return self._poynt_make_request( 'POST', 'cards/tokenize', payload={'nonce': nonce}, ) def _poynt_charge_token(self, payment_jwt, amount, currency, action='SALE', reference=''): """Charge a stored payment token JWT via the tokenize/charge endpoint. :param str payment_jwt: The payment token JWT from _poynt_tokenize_nonce. :param float amount: The charge amount in major currency units. :param recordset currency: The currency record. :param str action: SALE or AUTHORIZE (default SALE). :param str reference: Optional reference note for the transaction. :return: The transaction result dict from Poynt. :rtype: dict :raises ValidationError: If the charge fails. """ self.ensure_one() payload = poynt_utils.build_token_charge_payload( action=action, amount=amount, currency=currency, payment_jwt=payment_jwt, business_id=self.poynt_business_id, store_id=self.poynt_store_id or '', reference=reference, ) return self._poynt_make_request( 'POST', 'cards/tokenize/charge', payload=payload, ) # === BUSINESS METHODS - INLINE FORM === # def _poynt_get_inline_form_values(self, amount, currency, partner_id, is_validation, payment_method_sudo=None, **kwargs): """Return a serialized JSON of values needed to render the inline payment form. :param float amount: The payment amount. :param recordset currency: The currency of the transaction. :param int partner_id: The partner ID. :param bool is_validation: Whether this is a validation (tokenization) operation. :param recordset payment_method_sudo: The sudoed payment method record. :return: The JSON-serialized inline form values. :rtype: str """ self.ensure_one() partner = self.env['res.partner'].browse(partner_id).exists() minor_amount = poynt_utils.format_poynt_amount(amount, currency) if amount else 0 inline_form_values = { 'provider_id': self.id, 'business_id': self.poynt_business_id, 'application_id': self.poynt_application_id, 'currency_name': currency.name if currency else 'USD', 'minor_amount': minor_amount, 'capture_method': 'manual' if self.capture_manually else 'automatic', 'is_test': self.state == 'test', 'billing_details': { 'name': partner.name or '', 'email': partner.email or '', 'phone': partner.phone or '', 'address': { 'line1': partner.street or '', 'line2': partner.street2 or '', 'city': partner.city or '', 'state': partner.state_id.code or '', 'country': partner.country_id.code or '', 'postal_code': partner.zip or '', }, }, 'is_tokenization_required': ( self.allow_tokenization and self._is_tokenization_required(**kwargs) and payment_method_sudo and payment_method_sudo.support_tokenization ), } return json.dumps(inline_form_values) # === ACTION METHODS === # def action_poynt_test_connection(self): """Test the connection to Poynt by authenticating and fetching business info. If the Business ID appears to be a numeric MID rather than a UUID, the method attempts to decode the access token to find the real business UUID and auto-correct it. :return: A notification action with the result. :rtype: dict """ self.ensure_one() try: access_token = self._poynt_get_access_token() business_id = self.poynt_business_id is_uuid = business_id and '-' in business_id and len(business_id) > 30 if not is_uuid and business_id: resolved_biz_id = self._poynt_resolve_business_id(access_token) if resolved_biz_id: self.sudo().write({'poynt_business_id': resolved_biz_id}) _logger.info( "Auto-corrected Business ID from MID %s to UUID %s", business_id, resolved_biz_id, ) result = self._poynt_make_request('GET', '') business_name = result.get('legalName', result.get('doingBusinessAs', 'Unknown')) message = _( "Connection successful. Business: %(name)s (ID: %(bid)s)", name=business_name, bid=self.poynt_business_id, ) notification_type = 'success' except (ValidationError, UserError) as e: message = _("Connection failed: %(error)s", error=str(e)) notification_type = 'danger' return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'message': message, 'sticky': False, 'type': notification_type, }, } def _poynt_resolve_business_id(self, access_token): """Try to extract the real business UUID from the access token JWT. The Poynt access token contains a 'poynt.biz' claim with the merchant's business UUID when the token was obtained via merchant authorization. For app-level tokens, we fall back to the 'poynt.org' claim or attempt a direct API lookup. :param str access_token: The current access token. :return: The business UUID, or False if it cannot be resolved. :rtype: str or bool """ try: import jwt as pyjwt claims = pyjwt.decode(access_token, options={"verify_signature": False}) biz_id = claims.get('poynt.biz') or claims.get('poynt.org') if biz_id: return biz_id except Exception as e: _logger.warning("Could not decode access token to find business ID: %s", e) return False def action_poynt_fetch_terminals(self): """Fetch terminal devices from Poynt and create/update local records. Uses GET /businesses/{id}/stores which returns stores with their nested storeDevices arrays. The main business endpoint does not include stores in its response. :return: A notification action with the result. :rtype: dict """ self.ensure_one() try: result = self._poynt_make_request('GET', 'stores') stores = result if isinstance(result, list) else result.get('stores', []) all_devices = [] for store in stores: store_id = store.get('id', '') for device in store.get('storeDevices', []): device['_store_id'] = store_id device['_store_name'] = store.get('displayName', store.get('name', '')) all_devices.append(device) if not all_devices: return self._poynt_notification( _("No terminal devices found for this business."), 'warning' ) terminal_model = self.env['poynt.terminal'] created = 0 updated = 0 first_store_id = None for device in all_devices: device_id = device.get('deviceId', '') if not device_id: continue store_id = device.get('_store_id', '') if not first_store_id and store_id: first_store_id = store_id existing = terminal_model.search([ ('device_id', '=', device_id), ('provider_id', '=', self.id), ], limit=1) vals = { 'name': device.get('name', device_id), 'device_id': device_id, 'serial_number': device.get('serialNumber', ''), 'provider_id': self.id, 'status': 'online' if device.get('status') == 'ACTIVATED' else 'offline', 'store_id_poynt': store_id, } if existing: existing.write(vals) updated += 1 else: terminal_model.create(vals) created += 1 if first_store_id and not self.poynt_store_id: self.sudo().write({'poynt_store_id': first_store_id}) _logger.info("Auto-filled Store ID: %s", first_store_id) message = _( "Terminals synced: %(created)s created, %(updated)s updated.", created=created, updated=updated, ) return self._poynt_notification(message, 'success') except (ValidationError, UserError) as e: return self._poynt_notification( _("Failed to fetch terminals: %(error)s", error=str(e)), 'danger' ) def _poynt_fetch_receipt(self, transaction_id): """Fetch the rendered receipt from Poynt for a given transaction. Calls GET /businesses/{businessId}/transactions/{transactionId}/receipt which returns a TransactionReceipt with a ``data`` field containing the rendered receipt content (HTML or text). :param str transaction_id: The Poynt transaction UUID. :return: The receipt content string, or None on failure. :rtype: str | None """ self.ensure_one() if not transaction_id: return None try: result = self._poynt_make_request( 'GET', f'transactions/{transaction_id}/receipt', ) return result.get('data') or None except (ValidationError, Exception): _logger.debug( "Could not fetch Poynt receipt for transaction %s", transaction_id, ) return None def _poynt_notification(self, message, notification_type='info'): """Return a display_notification action. :param str message: The notification message. :param str notification_type: One of 'success', 'warning', 'danger', 'info'. :return: The notification action dict. :rtype: dict """ return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'message': message, 'sticky': False, 'type': notification_type, }, }