# Part of Odoo. See LICENSE file for full copyright and licensing details. import json import logging import requests from odoo import _, api, fields, models from odoo.exceptions import UserError, ValidationError from odoo.addons.fusion_clover import const from odoo.addons.fusion_clover import utils as clover_utils _logger = logging.getLogger(__name__) class PaymentProvider(models.Model): _inherit = 'payment.provider' code = fields.Selection( selection_add=[('clover', "Clover")], ondelete={'clover': 'set default'}, ) clover_api_key = fields.Char( string="Ecommerce Private Token", help="The private token from Clover's Ecommerce API Tokens page. " "Used for online charges and refunds (scl.clover.com).", required_if_provider='clover', copy=False, groups='base.group_system', ) clover_merchant_id = fields.Char( string="Merchant ID", help="The Clover merchant ID for this business.", required_if_provider='clover', copy=False, ) clover_rest_api_token = fields.Char( string="REST API Token", help="The merchant's REST API token from the Clover dashboard " "(Setup > API Tokens). Used for Platform API (devices, orders) " "and terminal payments (Cloud Pay Display). This is different " "from the Ecommerce API token.", copy=False, groups='base.group_system', ) clover_app_id = fields.Char( string="App ID (Client ID)", help="The Clover App ID (client_id) from the developer dashboard. " "Used for OAuth2 merchant authorization flow.", copy=False, ) clover_app_secret = fields.Char( string="App Secret", help="The Clover App Secret (client_secret) from the developer dashboard.", copy=False, groups='base.group_system', ) clover_remote_app_id = fields.Char( string="Remote App ID (RAID)", help="The Remote App ID generated when the Clover dev app's " "App Type is set to Web > 'Is this an integration of an " "existing point of sale?' = Yes. Sent in the X-POS-Id " "header on every REST Pay Display call. Required for " "production terminal payments.", copy=False, ) clover_public_key = fields.Char( string="Public API Key (PAKMS)", help="The public token from Clover's Ecommerce API Tokens page. " "Used for client-side tokenization. Safe to expose in the browser.", copy=False, ) clover_oauth_access_token = fields.Char( string="OAuth Access Token", help="Long-lived merchant OAuth access_token, obtained via the " "Connect to Clover button. Used as the Bearer token for " "all Platform API and REST Pay Display calls. Different " "from the Ecommerce private token (clover_api_key).", copy=False, groups='base.group_system', ) clover_oauth_refresh_token = fields.Char( string="OAuth Refresh Token", help="Refresh token used to renew the access token before it " "expires.", copy=False, groups='base.group_system', ) clover_oauth_token_expiry = fields.Datetime( string="OAuth Token Expiry", help="When the current access_token expires. The refresh flow " "should run automatically before this time.", copy=False, ) clover_default_terminal_id = fields.Many2one( 'clover.terminal', string="Default Terminal", help="The default Clover terminal used for in-store payment collection. " "Staff can override this per transaction.", domain="[('provider_id', '=', id), ('active', '=', True)]", ) # === 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 == 'clover').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 != 'clover': return super()._get_default_payment_method_codes() return const.DEFAULT_PAYMENT_METHOD_CODES # === BUSINESS METHODS - OAUTH HELPERS === # def _clover_get_platform_token(self): """Pick the best Bearer token for Platform / REST Pay Display calls. Priority order: 1. Live OAuth access_token (preferred — refreshable, app-scoped, grants both Platform API and REST Pay Display). 2. Manual REST API token from Clover Dashboard > Setup > API Tokens (legacy / single-merchant fallback). 3. Ecommerce private token (last-resort — works only for the Platform GET /v3/merchants/{mId} test-connection call; will 401 on REST Pay Display). Will proactively refresh the OAuth access_token if it's expired (or within a 60-second grace window of expiry). """ self.ensure_one() if self.clover_oauth_access_token: self._clover_refresh_oauth_if_needed() return self.clover_oauth_access_token if self.clover_rest_api_token: return self.clover_rest_api_token return self.clover_api_key or '' def _clover_refresh_oauth_if_needed(self, grace_seconds=60): """Refresh the OAuth access_token if it's within ``grace_seconds`` of expiry. Silent no-op if no expiry / no refresh_token / refresh endpoint not available. Logs at WARNING on refresh failure but does not raise (the next API call will get 401 and may trigger retry-with-refresh from the request method).""" from datetime import datetime, timedelta self.ensure_one() if not self.clover_oauth_refresh_token: return False if self.clover_oauth_token_expiry: time_left = self.clover_oauth_token_expiry - datetime.utcnow() if time_left > timedelta(seconds=grace_seconds): return False # plenty of time return self._clover_refresh_oauth_token() def _clover_refresh_oauth_token(self): """Use the stored refresh_token to mint a new access_token via Clover's /oauth/v2/refresh. Stores the new token + new expiry on the provider record. Returns True on success, False on failure (caller decides whether to fall back or raise).""" from datetime import datetime, timedelta self.ensure_one() if not self.clover_oauth_refresh_token or not self.clover_app_id: return False is_test = self.state == 'test' url = const.OAUTH_REFRESH_URL_TEST if is_test else const.OAUTH_REFRESH_URL try: r = requests.post( url, json={ 'client_id': self.clover_app_id, 'refresh_token': self.clover_oauth_refresh_token, }, timeout=20, ) except requests.exceptions.RequestException as e: _logger.warning("Clover OAuth refresh network error: %s", e) return False if r.status_code != 200: _logger.warning( "Clover OAuth refresh failed (%s): %s", r.status_code, r.text[:300], ) return False try: data = r.json() except ValueError: _logger.warning("Clover OAuth refresh response is not JSON") return False new_token = data.get('access_token', '') if not new_token: return False vals = {'clover_oauth_access_token': new_token} new_refresh = data.get('refresh_token', '') if new_refresh: vals['clover_oauth_refresh_token'] = new_refresh expires = int(data.get('access_token_expiration', 0) or 0) if expires: if expires > 10 * 365 * 24 * 3600: # absolute timestamp vals['clover_oauth_token_expiry'] = datetime.utcfromtimestamp(expires) else: # duration in seconds vals['clover_oauth_token_expiry'] = ( datetime.utcnow() + timedelta(seconds=expires) ) self.sudo().write(vals) _logger.info("Clover OAuth access_token refreshed for provider %s", self.id) return True def _clover_dispatcher_url(self): """Return the Nexa OAuth dispatcher URL (where Clover redirects after the merchant authorises). Configurable per-deployment via ir.config_parameter ``fusion_clover.dispatcher_url``.""" return self.env['ir.config_parameter'].sudo().get_param( 'fusion_clover.dispatcher_url', 'https://oauth.nexasystems.ca/clover/callback', ) def _clover_dispatcher_secret(self): """Return the HMAC secret shared with the Nexa OAuth dispatcher. Stored in ``ir.config_parameter`` ``fusion_clover.dispatcher_secret`` so it can be rotated without a code deploy. Returns '' if not configured (in which case the OAuth flow falls back to a direct callback to this Odoo instance — useful for local dev).""" return self.env['ir.config_parameter'].sudo().get_param( 'fusion_clover.dispatcher_secret', '', ) def _clover_build_signed_state(self, customer_slug=''): """Build the HMAC-SHA256 signed state token sent to Clover's OAuth authorize endpoint. The Nexa dispatcher Worker verifies this signature and uses the embedded ``redirect_to`` to fan the OAuth callback back to THIS Odoo instance. Format: ``.`` """ import base64 import hashlib import hmac as _hmac import json as _json import secrets import time self.ensure_one() secret = self._clover_dispatcher_secret() if not secret: raise UserError(_( "The dispatcher HMAC secret is not configured. Set " "the system parameter " "'fusion_clover.dispatcher_secret' to the value " "stored in the Cloudflare Worker before initiating " "the Connect to Clover flow." )) base_url = self.get_base_url() payload = { 'redirect_to': f"{base_url}/payment/clover/oauth/callback", 'nonce': secrets.token_hex(16), 'iat': int(time.time()), 'customer': customer_slug or self.company_id.name or '', } payload_json = _json.dumps(payload, separators=(',', ':')).encode('utf-8') payload_b64u = base64.urlsafe_b64encode(payload_json).rstrip(b'=').decode('ascii') sig = _hmac.new( secret.encode('utf-8'), payload_b64u.encode('ascii'), hashlib.sha256, ).digest() sig_b64u = base64.urlsafe_b64encode(sig).rstrip(b'=').decode('ascii') return f"{payload_b64u}.{sig_b64u}" def action_clover_oauth_connect(self): """Kick off the OAuth flow: redirect the staff user to Clover's authorize endpoint, with our dispatcher URL as redirect_uri and a signed state encoding this Odoo's callback URL.""" from werkzeug.urls import url_encode self.ensure_one() if self.code != 'clover': raise UserError(_("This action is only available for Clover providers.")) if not self.clover_app_id: raise UserError(_( "App ID is required. Add it on the payment provider record " "under 'OAuth (Optional)' first." )) is_test = self.state == 'test' authorize_url = ( const.OAUTH_AUTHORIZE_URL_TEST if is_test else const.OAUTH_AUTHORIZE_URL ) state = self._clover_build_signed_state() # Per https://docs.clover.com/dev/docs/high-trust-app-auth-flow the # v2 authorize endpoint only takes client_id + redirect_uri. # `state` is documented elsewhere as supported for CSRF; we include # it. Notably ABSENT: `response_type` (legacy v1 only) and `scope` # (handled by app's Requested Permissions in the dev dashboard). params = { 'client_id': self.clover_app_id, 'redirect_uri': self._clover_dispatcher_url(), 'state': state, } return { 'type': 'ir.actions.act_url', 'url': f"{authorize_url}?{url_encode(params)}", 'target': 'self', } def _clover_exchange_oauth_code(self, code): """Exchange a one-time OAuth ``code`` for a long-lived access_token + refresh_token by POSTing to Clover's token endpoint with our App Secret. Stores the tokens on this provider record. Tries the v2 token endpoint first; if Clover responds with 401 "Failed to validate authentication code", it means the code was generated by the legacy partial OAuth flow (App Market Connect without an Alternate Launch Path) — in that case we fall back to the v1 token endpoint which is what generated the code. """ from datetime import datetime, timedelta self.ensure_one() if not self.clover_app_id or not self.clover_app_secret: raise ValidationError(_( "Cannot exchange OAuth code: App ID or App Secret is missing." )) is_test = self.state == 'test' v2_url = const.OAUTH_TOKEN_URL_TEST if is_test else const.OAUTH_TOKEN_URL # Legacy v1 token URL — same host, no /v2/. v1_url = v2_url.replace('/oauth/v2/token', '/oauth/token') payload = { 'client_id': self.clover_app_id, 'client_secret': self.clover_app_secret, 'code': code, } data = None last_error = '' for token_url in (v2_url, v1_url): try: # v1 endpoint historically used GET with query params; # v2 uses POST with JSON body. We send POST+JSON to v1 # too (Clover accepts both there in our experience). response = requests.post(token_url, json=payload, timeout=30) except requests.exceptions.RequestException as e: raise ValidationError(_("Could not reach Clover OAuth token endpoint: %s", e)) if response.status_code < 400: try: data = response.json() except ValueError: raise ValidationError(_("Clover OAuth response was not valid JSON.")) _logger.info("Clover OAuth code exchanged successfully via %s", token_url) break last_error = response.text[:500] _logger.warning( "Clover OAuth exchange via %s returned %s: %s — trying next endpoint", token_url, response.status_code, last_error, ) # Only try v1 fallback if v2 specifically said "Failed to # validate authentication code" — other 4xx errors won't # benefit from the fallback. if 'validate authentication code' not in last_error.lower(): break if data is None: raise ValidationError(_( "Clover rejected the OAuth code at both v2 and v1 token endpoints. " "Last response: %s", last_error, )) access_token = data.get('access_token', '') refresh_token = data.get('refresh_token', '') expires_in = int(data.get('access_token_expiration', 0) or 0) if not access_token: raise ValidationError(_("Clover OAuth response did not contain an access_token.")) vals = { 'clover_oauth_access_token': access_token, 'clover_oauth_refresh_token': refresh_token or False, } if expires_in: # Clover returns the expiry as a UNIX timestamp in seconds, not # a duration. Detect both shapes (a duration is < ~10 years). if expires_in > 10 * 365 * 24 * 3600: vals['clover_oauth_token_expiry'] = datetime.utcfromtimestamp(expires_in) else: vals['clover_oauth_token_expiry'] = ( datetime.utcnow() + timedelta(seconds=expires_in) ) self.sudo().write(vals) return True # === BUSINESS METHODS - API REQUESTS === # def _clover_make_ecom_request(self, method, endpoint, payload=None, params=None, _retry=True): """Make an authenticated API request to the Clover Ecommerce API. :param str method: HTTP method (GET, POST, PUT, DELETE). :param str endpoint: The API endpoint path (e.g., 'v1/charges'). :param dict payload: The JSON request body (optional). :param dict params: The query parameters (optional). :return: The parsed JSON response. :rtype: dict :raises ValidationError: If the API request fails. """ self.ensure_one() is_test = self.state == 'test' url = clover_utils.build_ecom_url(endpoint, is_test=is_test) idempotency_key = clover_utils.generate_idempotency_key() # Auth precedence: OAuth access_token (works for both Ecommerce # AND Platform APIs per Clover docs) > Ecommerce private token. # Routed through _clover_get_platform_token() so the OAuth token # is proactively refreshed if it's near expiry. if self.clover_oauth_access_token: ecom_token = self._clover_get_platform_token() else: ecom_token = self.clover_api_key or '' headers = clover_utils.build_ecom_headers( ecom_token, idempotency_key=idempotency_key, ) _logger.info( "Clover Ecom API %s request to %s (idempotency=%s)", method, url, idempotency_key, ) try: response = requests.request( method, url, json=payload, params=params, headers=headers, timeout=60, ) except requests.exceptions.RequestException as e: _logger.error("Clover Ecom API request failed: %s", e) raise ValidationError(_("Communication with Clover failed: %s", e)) if response.status_code in (202, 204): return {} try: result = response.json() except ValueError: if response.status_code < 400: return {} _logger.error("Clover returned non-JSON response: %s", response.text[:500]) raise ValidationError(_("Clover returned an invalid response.")) if response.status_code == 401 and _retry and self.clover_oauth_refresh_token: _logger.info("Clover Ecom 401, attempting token refresh + retry") if self._clover_refresh_oauth_token(): return self._clover_make_ecom_request( method, endpoint, payload=payload, params=params, _retry=False, ) if response.status_code >= 400: error = result.get('error', {}) error_msg = error.get('message', '') if isinstance(error, dict) else str(error) error_code = error.get('code', '') if isinstance(error, dict) else '' _logger.error( "Clover Ecom API error %s: %s (code=%s)\n" " URL: %s %s\n Payload: %s\n Response: %s", response.status_code, error_msg, error_code, method, url, json.dumps(payload)[:2000] if payload else 'None', response.text[:2000], ) raise ValidationError( _("Clover API error (%(code)s): %(msg)s", code=response.status_code, msg=error_msg or 'Unknown error') ) return result def _clover_make_platform_request(self, method, endpoint, payload=None, params=None, _retry=True): """Make an authenticated request to the Clover Platform API. :param str method: HTTP method. :param str endpoint: The API endpoint path. :param dict payload: The JSON request body (optional). :param dict params: The query parameters (optional). :return: The parsed JSON response. :rtype: dict :raises ValidationError: If the API request fails. """ self.ensure_one() is_test = self.state == 'test' url = clover_utils.build_platform_url( endpoint, merchant_id=self.clover_merchant_id, is_test=is_test, ) # Platform API auth precedence: OAuth > merchant REST API token > ecom key api_token = self._clover_get_platform_token() headers = clover_utils.build_ecom_headers(api_token) _logger.info("Clover Platform API %s request to %s", method, url) try: response = requests.request( method, url, json=payload, params=params, headers=headers, timeout=60, ) except requests.exceptions.RequestException as e: _logger.error("Clover Platform API request failed: %s", e) raise ValidationError(_("Communication with Clover failed: %s", e)) if response.status_code in (202, 204): return {} try: result = response.json() except ValueError: if response.status_code < 400: return {} raise ValidationError(_("Clover returned an invalid response.")) if response.status_code == 401 and _retry and self.clover_oauth_refresh_token: _logger.info("Clover Platform 401, attempting token refresh + retry") if self._clover_refresh_oauth_token(): return self._clover_make_platform_request( method, endpoint, payload=payload, params=params, _retry=False, ) if response.status_code >= 400: error_msg = result.get('message', result.get('error', 'Unknown error')) raise ValidationError( _("Clover API error (%(code)s): %(msg)s", code=response.status_code, msg=error_msg) ) return result # === BUSINESS METHODS - TOKENIZATION === # @staticmethod def _clover_detect_brand_from_pan(pan): """Detect the card brand from a digits-only PAN, returning one of Clover's accepted brand codes (VISA, MC, AMEX, DISCOVER, DINERS, JCB). Falls back to VISA if unrecognised — Clover will then decline at the charge step rather than the tokenize step, which is a slightly more useful error path.""" if not pan or len(pan) < 2: return 'VISA' if pan[:2] in ('34', '37'): return 'AMEX' if pan[0] == '4': return 'VISA' try: p2 = int(pan[:2]) p4 = int(pan[:4]) if len(pan) >= 4 else 0 except ValueError: return 'VISA' if 51 <= p2 <= 55: return 'MC' if 2221 <= p4 <= 2720: return 'MC' if pan[:4] == '6011' or pan[:2] == '65': return 'DISCOVER' if pan[:4] in ('3014', '3036', '3038') or pan[:2] in ('30', '36', '38'): return 'DINERS' if pan[:2] == '35': return 'JCB' return 'VISA' def _clover_tokenize_card(self, card_number, exp_month, exp_year, cvv, cardholder_name='', postal_code=''): """Server-side card tokenization via Clover's tokenization service. Used by the back-office wizard for staff-keyed card entry. Returns a ``clv_xxx`` token that can then be passed to ``/v1/charges`` as the ``source``. The raw PAN is sent ONCE to Clover's tokenization endpoint and never persisted in Odoo. Note: Westin Healthcare staff should normally use the terminal flow rather than manual card entry, which keeps even this brief in-memory PAN handling out of the picture. :param str card_number: Card number, digits only. :param int exp_month: Expiry month (1-12). :param int exp_year: Expiry year (4-digit). :param str cvv: Card verification value. :param str cardholder_name: Optional name on card. :param str postal_code: Optional billing postal code. :return: The Clover token string ``clv_xxx``. :rtype: str :raises ValidationError: If tokenization fails. """ self.ensure_one() is_test = self.state == 'test' base_url = const.TOKEN_BASE_URL_TEST if is_test else const.TOKEN_BASE_URL url = f"{base_url}/v1/tokens" if not self.clover_public_key: raise ValidationError(_( "No Clover Public API Key (PAKMS) configured. Add it on " "the payment provider record before staff can charge cards " "from the back-office wizard." )) # Clover's tokenization endpoint requires the lowercase `apikey` # header (NOT the `apiAccessKey` field name returned by /pakms). # https://docs.clover.com/dev/reference/create-card-token headers = { 'Content-Type': 'application/json', 'Accept': 'application/json', 'apikey': self.clover_public_key, } # Detect card brand from the BIN. Clover requires a real brand # name (VISA/MC/AMEX/DISCOVER/DINERS/JCB) — `CARD` is not valid. clean_pan = str(card_number).replace(' ', '').replace('-', '') brand = self._clover_detect_brand_from_pan(clean_pan) payload = { 'card': { 'number': clean_pan, 'exp_month': str(exp_month).zfill(2), 'exp_year': str(exp_year), 'cvv': str(cvv), 'brand': brand, }, } if cardholder_name: payload['card']['name'] = cardholder_name if postal_code: payload['card']['address_zip'] = postal_code try: response = requests.post(url, json=payload, headers=headers, timeout=30) except requests.exceptions.RequestException as e: raise ValidationError(_("Could not reach Clover tokenization service: %s", e)) try: result = response.json() except ValueError: raise ValidationError(_("Clover returned an invalid tokenization response.")) if response.status_code >= 400 or 'id' not in result: error = result.get('error', {}) error_msg = error.get('message', '') if isinstance(error, dict) else str(error) raise ValidationError( _("Clover tokenization failed (%(code)s): %(msg)s", code=response.status_code, msg=error_msg or 'Unknown error') ) token = result['id'] if not token.startswith('clv_'): raise ValidationError(_("Unexpected token format from Clover: %s", token)) return token # === BUSINESS METHODS - CHARGE / TOKENIZE === # def _clover_create_charge(self, source_token, amount, currency, capture=True, description='', ecomind='ecom', external_reference_id='', receipt_email='', metadata=None): """Create a charge via the Clover Ecommerce API. :param str source_token: The Clover card token. :param float amount: The charge amount in major currency units. :param recordset currency: The currency record. :param bool capture: Whether to capture immediately. :param str description: Optional charge description. :param str ecomind: 'ecom' or 'moto'. :param str external_reference_id: External reference. :param str receipt_email: Email for receipt. :param dict metadata: Optional metadata. :return: The charge response dict. :rtype: dict """ self.ensure_one() payload = clover_utils.build_charge_payload( amount=amount, currency=currency, source_token=source_token, capture=capture, description=description, ecomind=ecomind, external_reference_id=external_reference_id, receipt_email=receipt_email, metadata=metadata, ) return self._clover_make_ecom_request('POST', 'v1/charges', payload=payload) def _clover_capture_charge(self, charge_id, amount=None, currency=None): """Capture a previously authorized charge. :param str charge_id: The Clover charge ID. :param float amount: Optional capture amount (for partial captures). :param recordset currency: Optional currency record. :return: The capture response dict. :rtype: dict """ self.ensure_one() payload = {} if amount is not None and currency: payload['amount'] = clover_utils.format_clover_amount(amount, currency) return self._clover_make_ecom_request( 'POST', f'v1/charges/{charge_id}/capture', payload=payload, ) def _clover_create_refund(self, charge_id, amount=None, currency=None, reason=''): """Create a refund via the Clover Ecommerce API. :param str charge_id: The Clover charge ID to refund. :param float amount: Optional partial refund amount. :param recordset currency: Optional currency record. :param str reason: Optional reason. :return: The refund response dict. :rtype: dict """ self.ensure_one() payload = clover_utils.build_refund_payload( charge_id=charge_id, amount=amount, currency=currency, reason=reason, ) return self._clover_make_ecom_request('POST', 'v1/refunds', payload=payload) # === BUSINESS METHODS - NON-REFERENCED CREDIT === # def _clover_create_credit(self, amount, currency, description=''): """Issue a non-referenced credit (manual refund) via Clover Ecommerce API. This creates a credit without referencing an original charge. Useful when the original transaction is too old for a referenced refund. Note: merchants must have manual refunds enabled by Clover support. :param float amount: The credit amount in major currency units. :param recordset currency: The currency record. :param str description: Optional description. :return: The credit response dict. :rtype: dict """ self.ensure_one() minor_amount = clover_utils.format_clover_amount(amount, currency) payload = { 'amount': minor_amount, 'currency': currency.name.lower(), } if description: payload['description'] = description return self._clover_make_ecom_request('POST', 'v1/credits', payload=payload) # === BUSINESS METHODS - VERIFICATION === # def _clover_get_charge(self, charge_id): """Fetch a charge from the Clover Ecommerce API. :param str charge_id: The Clover charge ID. :return: The charge data dict. :rtype: dict """ self.ensure_one() return self._clover_make_ecom_request('GET', f'v1/charges/{charge_id}') def _clover_verify_charge_not_reversed(self, charge_id): """Check that a charge has not already been fully refunded or voided. :param str charge_id: The Clover charge ID. :return: The charge data dict. :rtype: dict :raises UserError: If the charge is already refunded. """ self.ensure_one() charge_data = self._clover_get_charge(charge_id) status = charge_data.get('status', '') if status == 'refunded': raise UserError(_( "This charge (%(charge_id)s) has already been fully refunded " "on Clover. A duplicate refund cannot be issued.", charge_id=charge_id, )) return charge_data # === BUSINESS METHODS - INLINE FORM === # def _clover_get_inline_form_values(self, amount, currency, partner_id, is_validation, payment_method_sudo=None, **kwargs): """Return serialized JSON of values needed for 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 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(int(partner_id)).exists() if partner_id else self.env['res.partner'] minor_amount = clover_utils.format_clover_amount(amount, currency) if amount else 0 # Map the Odoo language to a Clover-supported locale (Clover only # supports en-US, en-CA, fr-CA today). Anything else falls back to # en-US (Clover SDK default). partner_lang = (partner.lang or self.env.user.lang or 'en_US').replace('_', '-') clover_locale = '' if partner_lang.startswith('fr'): clover_locale = 'fr-CA' elif partner_lang in ('en-CA',): clover_locale = 'en-CA' inline_form_values = { 'provider_id': self.id, 'merchant_id': self.clover_merchant_id, 'public_key': self.clover_public_key or '', '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', 'locale': clover_locale, '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 ), } ICP = self.env['ir.config_parameter'].sudo() surcharge_enabled = ICP.get_param( 'fusion_clover.surcharge_enabled', 'False', ) == 'True' if surcharge_enabled: inline_form_values['surcharge'] = { 'enabled': True, 'visa': float(ICP.get_param('fusion_clover.surcharge_visa_rate', '0') or 0), 'mastercard': float(ICP.get_param('fusion_clover.surcharge_mastercard_rate', '0') or 0), 'amex': float(ICP.get_param('fusion_clover.surcharge_amex_rate', '0') or 0), 'debit': float(ICP.get_param('fusion_clover.surcharge_debit_rate', '0') or 0), 'other': float(ICP.get_param('fusion_clover.surcharge_other_rate', '0') or 0), } return json.dumps(inline_form_values) # === BUSINESS METHODS - TERMINAL (REST Pay Display Cloud API) === # def _clover_terminal_request(self, method, endpoint, serial_number=None, payload=None, params=None, _retry=True): """Make a request to the Clover REST Pay Display Cloud API. Sends commands to Clover terminals through Clover's cloud (Cloud Pay Display). :param str method: HTTP method (GET, POST). :param str endpoint: The API endpoint path (e.g., 'payments', 'device/ping'). :param str serial_number: The device serial number (X-Clover-Device-Id). :param dict payload: The JSON request body (optional). :param dict params: The query parameters (optional). :return: The parsed JSON response. :rtype: dict :raises ValidationError: If the API request fails. """ self.ensure_one() is_test = self.state == 'test' base_url = const.CONNECT_BASE_URL_TEST if is_test else const.CONNECT_BASE_URL url = f"{base_url}/{endpoint}" headers = { 'Content-Type': 'application/json', 'Accept': 'application/json', 'Authorization': f'Bearer {self._clover_get_platform_token()}', # X-POS-Id MUST be the Remote App ID (RAID) from the Clover # developer dashboard (App Type > Web > "is integration of an # existing POS" = Yes). Falls back to a static identifier in # sandbox where some merchants use the legacy free-form value. 'X-POS-Id': self.clover_remote_app_id or 'FusionCloverOdoo', } if serial_number: headers['X-Clover-Device-Id'] = serial_number idempotency_key = clover_utils.generate_idempotency_key() headers['Idempotency-Key'] = idempotency_key _logger.info( "Clover Terminal API %s request to %s (device=%s)", method, url, serial_number or 'none', ) try: response = requests.request( method, url, json=payload, params=params, headers=headers, timeout=120, ) except requests.exceptions.RequestException as e: _logger.error("Clover Terminal API request failed: %s", e) raise ValidationError(_("Communication with Clover terminal failed: %s", e)) if response.status_code in (202, 204): return {} try: result = response.json() except ValueError: if response.status_code < 400: return {} _logger.error("Clover Terminal returned non-JSON: %s", response.text[:500]) raise ValidationError(_("Clover terminal returned an invalid response.")) if response.status_code == 401 and _retry and self.clover_oauth_refresh_token: _logger.info("Clover Terminal 401, attempting token refresh + retry") if self._clover_refresh_oauth_token(): return self._clover_terminal_request( method, endpoint, serial_number=serial_number, payload=payload, params=params, _retry=False, ) if response.status_code >= 400: error_msg = result.get('message', result.get('error', 'Unknown error')) _logger.error( "Clover Terminal API error %s: %s\n URL: %s %s", response.status_code, error_msg, method, url, ) raise ValidationError( _("Clover terminal error (%(code)s): %(msg)s", code=response.status_code, msg=error_msg) ) return result def _clover_get_merchant_devices(self): """Fetch all devices provisioned to the merchant from the Platform API. :return: List of device dicts with id, serial, name, model. :rtype: list[dict] """ self.ensure_one() result = self._clover_make_platform_request('GET', 'devices') elements = result.get('elements', []) return [ { 'id': d.get('id', ''), 'serial': d.get('serial', ''), 'name': d.get('name', d.get('productName', 'Clover Device')), 'model': d.get('model', d.get('productName', '')), } for d in elements if d.get('serial') ] def action_sync_terminals(self): """Sync terminals from the Clover Platform API.""" self.ensure_one() if self.code != 'clover': return try: devices = self._clover_get_merchant_devices() except (ValidationError, UserError) as e: return self._clover_notification( _("Failed to fetch devices: %(error)s", error=str(e)), 'danger', ) if not devices: return self._clover_notification( _("No devices found for this merchant."), 'warning', ) Terminal = self.env['clover.terminal'].sudo() created = 0 updated = 0 for device in devices: serial = device['serial'] existing = Terminal.search([ ('serial_number', '=', serial), ('provider_id', '=', self.id), ], limit=1) if existing: # Only update metadata; don't overwrite user-set name vals = { 'device_id': device['id'], 'model_name': device['model'], 'clover_device_name': device['name'], } existing.write(vals) updated += 1 else: Terminal.create({ 'name': device['name'], 'clover_device_name': device['name'], 'serial_number': serial, 'device_id': device['id'], 'model_name': device['model'], 'provider_id': self.id, }) created += 1 return self._clover_notification( _("Sync complete: %(created)s created, %(updated)s updated.", created=created, updated=updated), 'success', ) # === ACTION METHODS === # def action_clover_test_connection(self): """Test the connection to Clover by fetching merchant info. :return: A notification action with the result. :rtype: dict """ self.ensure_one() try: result = self._clover_make_platform_request('GET', '') merchant_name = result.get('name', 'Unknown') message = _( "Connection successful. Merchant: %(name)s (ID: %(mid)s)", name=merchant_name, mid=self.clover_merchant_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 _clover_notification(self, message, notification_type='info'): """Return a display_notification action.""" return { 'type': 'ir.actions.client', 'tag': 'display_notification', 'params': { 'message': message, 'sticky': False, 'type': notification_type, }, }