admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

changes

Browse Source
This commit is contained in:
gsinghpal
2026-02-25 09:40:41 -05:00
parent 0e1aebe60b
commit e71bc503f9
69 changed files with 7537 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
fusion_poynt/models/payment_provider.py
View File

@@ -53,6 +53,13 @@ class PaymentProvider(models.Model):
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(
@@ -121,10 +128,16 @@ class PaymentProvider(models.Model):
},
headers={
'Content-Type': 'application/x-www-form-urlencoded',
'Api-Version': const.API_VERSION,
'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:
@@ -210,24 +223,32 @@ class PaymentProvider(models.Model):
_("Poynt authentication expired. Please retry.")
)
if response.status_code == 204:
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)",
"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=error_msg)
code=response.status_code, msg=dev_msg or error_msg)
)
return result
@@ -252,6 +273,7 @@ class PaymentProvider(models.Model):
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',
@@ -283,7 +305,11 @@ class PaymentProvider(models.Model):
# === ACTION METHODS === #
def action_poynt_test_connection(self):
"""Test the connection to Poynt by fetching business info.
"""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
@@ -291,11 +317,25 @@ class PaymentProvider(models.Model):
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",
"Connection successful. Business: %(name)s (ID: %(bid)s)",
name=business_name,
bid=self.poynt_business_id,
)
notification_type = 'success'
except (ValidationError, UserError) as e:
@@ -312,30 +352,71 @@ class PaymentProvider(models.Model):
},
}
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:
store_id = self.poynt_store_id
if store_id:
endpoint = f'stores/{store_id}/storeDevices'
else:
endpoint = 'storeDevices'
result = self._poynt_make_request('GET', 'stores')
stores = result if isinstance(result, list) else result.get('stores', [])
result = self._poynt_make_request('GET', endpoint)
devices = result if isinstance(result, list) else result.get('storeDevices', [])
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
for device in devices:
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),
@@ -347,7 +428,7 @@ class PaymentProvider(models.Model):
'serial_number': device.get('serialNumber', ''),
'provider_id': self.id,
'status': 'online' if device.get('status') == 'ACTIVATED' else 'offline',
'store_id_poynt': device.get('storeId', ''),
'store_id_poynt': store_id,
}
if existing:
@@ -357,15 +438,53 @@ class PaymentProvider(models.Model):
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,
)
notification_type = 'success'
return self._poynt_notification(message, 'success')
except (ValidationError, UserError) as e:
message = _("Failed to fetch terminals: %(error)s", error=str(e))
notification_type = 'danger'
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',