14fe9ab716
Rental orders no longer show the "Authorizer Required?" question or the Authorizer field. The sale type is automatically set to 'Rentals' when creating or confirming a rental order. Validation logic also skips authorizer checks for rental sale type. Made-with: Cursor
544 lines
20 KiB
Python
544 lines
20 KiB
Python
# 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,
|
|
},
|
|
}
|