Files
Odoo-Modules/fusion_poynt/controllers/main.py
gsinghpal 2563208f53 changes
2026-03-26 15:16:51 -04:00

638 lines
24 KiB
Python

# Part of Odoo. See LICENSE file for full copyright and licensing details.
import hashlib
import hmac
import json
import logging
import pprint
from werkzeug.exceptions import Forbidden
from odoo import http
from odoo.exceptions import ValidationError
from odoo.http import request
from odoo.tools import mute_logger
from odoo.addons.fusion_poynt import const
from odoo.addons.fusion_poynt import utils as poynt_utils
_logger = logging.getLogger(__name__)
def _detect_card_brand(card_number):
"""Detect the card brand from the card number using BIN prefixes."""
num = (card_number or '').replace(' ', '')
if len(num) < 2:
return 'other'
if num[:2] in ('34', '37'):
return 'amex'
if num[0] == '4':
return 'visa'
prefix2 = int(num[:2])
if 51 <= prefix2 <= 55:
return 'mastercard'
if len(num) >= 4:
prefix4 = int(num[:4])
if 2221 <= prefix4 <= 2720:
return 'mastercard'
return 'other'
class PoyntController(http.Controller):
_return_url = '/payment/poynt/return'
_webhook_url = '/payment/poynt/webhook'
_terminal_callback_url = '/payment/poynt/terminal/callback'
_oauth_callback_url = '/payment/poynt/oauth/callback'
# === RETURN ROUTE === #
@http.route(_return_url, type='http', methods=['GET'], auth='public')
def poynt_return(self, **data):
"""Process the return from a Poynt payment flow.
The customer is redirected here after completing (or abandoning) a payment.
We look up the transaction by reference and fetch the latest status from Poynt.
"""
tx_sudo = request.env['payment.transaction'].sudo()._search_by_reference(
'poynt', data,
)
if tx_sudo and tx_sudo.poynt_transaction_id:
try:
txn_data = tx_sudo.provider_id._poynt_make_request(
'GET',
f'transactions/{tx_sudo.poynt_transaction_id}',
)
payment_data = {
'reference': tx_sudo.reference,
'poynt_transaction_id': txn_data.get('id'),
'poynt_order_id': tx_sudo.poynt_order_id,
'poynt_status': txn_data.get('status', ''),
'funding_source': txn_data.get('fundingSource', {}),
}
tx_sudo._process('poynt', payment_data)
except ValidationError:
_logger.error(
"Failed to fetch Poynt transaction %s on return.",
tx_sudo.poynt_transaction_id,
)
with mute_logger('werkzeug'):
return request.redirect('/payment/status')
# === WEBHOOK ROUTE === #
@http.route(_webhook_url, type='http', methods=['POST'], auth='public', csrf=False)
def poynt_webhook(self):
"""Process webhook notifications from Poynt.
Poynt sends cloud hook events for transaction and order status changes.
We verify the payload, match it to an Odoo transaction, and update accordingly.
:return: An empty JSON response to acknowledge the notification.
:rtype: Response
"""
try:
raw_body = request.httprequest.data.decode('utf-8')
event = json.loads(raw_body)
except (ValueError, UnicodeDecodeError):
_logger.warning("Received invalid JSON from Poynt webhook")
return request.make_json_response({'status': 'error'}, status=400)
_logger.info(
"Poynt webhook notification received:\n%s",
pprint.pformat(event),
)
try:
event_type = event.get('eventType', event.get('type', ''))
resource = event.get('resource', {})
business_id = event.get('businessId', '')
if event_type not in const.HANDLED_WEBHOOK_EVENTS:
_logger.info("Ignoring unhandled Poynt event type: %s", event_type)
return request.make_json_response({'status': 'ignored'})
self._verify_webhook_signature(event, business_id)
if event_type.startswith('TRANSACTION_'):
self._handle_transaction_webhook(event_type, resource, business_id)
elif event_type.startswith('ORDER_'):
self._handle_order_webhook(event_type, resource, business_id)
except ValidationError:
_logger.exception("Unable to process Poynt webhook; acknowledging to avoid retries")
except Forbidden:
_logger.warning("Poynt webhook signature verification failed")
return request.make_json_response({'status': 'forbidden'}, status=403)
return request.make_json_response({'status': 'ok'})
def _handle_transaction_webhook(self, event_type, resource, business_id):
"""Process a transaction-related webhook event.
:param str event_type: The Poynt event type.
:param dict resource: The Poynt resource data from the webhook.
:param str business_id: The Poynt business ID.
"""
transaction_id = resource.get('id', '')
if not transaction_id:
_logger.warning("Transaction webhook missing transaction ID")
return
provider_sudo = request.env['payment.provider'].sudo().search([
('code', '=', 'poynt'),
('poynt_business_id', '=', business_id),
], limit=1)
if not provider_sudo:
_logger.warning("No Poynt provider found for business %s", business_id)
return
try:
txn_data = provider_sudo._poynt_make_request(
'GET', f'transactions/{transaction_id}',
)
except ValidationError:
_logger.error("Failed to fetch transaction %s from Poynt", transaction_id)
return
reference = txn_data.get('notes', '')
status = txn_data.get('status', '')
payment_data = {
'reference': reference,
'poynt_transaction_id': transaction_id,
'poynt_status': status,
'funding_source': txn_data.get('fundingSource', {}),
}
tx_sudo = request.env['payment.transaction'].sudo()._search_by_reference(
'poynt', payment_data,
)
if not tx_sudo:
_logger.warning(
"No matching transaction for Poynt txn %s (ref: %s)",
transaction_id, reference,
)
return
if event_type == 'TRANSACTION_REFUNDED':
action = txn_data.get('action', '')
if action == 'REFUND':
parent_id = txn_data.get('parentId', '')
source_tx = request.env['payment.transaction'].sudo().search([
('provider_reference', '=', parent_id),
('provider_code', '=', 'poynt'),
], limit=1)
if source_tx:
refund_amount = poynt_utils.parse_poynt_amount(
txn_data.get('amounts', {}).get('transactionAmount', 0),
source_tx.currency_id,
)
existing_refund = source_tx.child_transaction_ids.filtered(
lambda t: t.provider_reference == transaction_id
)
if not existing_refund:
refund_tx = source_tx._create_child_transaction(
refund_amount, is_refund=True,
)
payment_data['reference'] = refund_tx.reference
refund_tx._process('poynt', payment_data)
return
tx_sudo._process('poynt', payment_data)
def _handle_order_webhook(self, event_type, resource, business_id):
"""Process an order-related webhook event.
:param str event_type: The Poynt event type.
:param dict resource: The Poynt resource data from the webhook.
:param str business_id: The Poynt business ID.
"""
order_id = resource.get('id', '')
if not order_id:
return
tx_sudo = request.env['payment.transaction'].sudo().search([
('poynt_order_id', '=', order_id),
('provider_code', '=', 'poynt'),
], limit=1)
if not tx_sudo:
_logger.info("No Odoo transaction found for Poynt order %s", order_id)
return
if event_type == 'ORDER_CANCELLED' and tx_sudo.state not in ('done', 'cancel', 'error'):
tx_sudo._set_canceled()
def _verify_webhook_signature(self, event, business_id):
"""Verify the webhook notification signature.
:param dict event: The webhook event data.
:param str business_id: The Poynt business ID.
:raises Forbidden: If signature verification fails.
"""
provider_sudo = request.env['payment.provider'].sudo().search([
('code', '=', 'poynt'),
('poynt_business_id', '=', business_id),
], limit=1)
if not provider_sudo or not provider_sudo.poynt_webhook_secret:
_logger.info("No webhook secret configured; skipping signature verification")
return
signature = request.httprequest.headers.get('X-Poynt-Webhook-Signature', '')
if not signature:
_logger.warning("Webhook missing X-Poynt-Webhook-Signature header")
return
raw_body = request.httprequest.data
expected_signature = hmac.new(
provider_sudo.poynt_webhook_secret.encode('utf-8'),
raw_body,
hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(signature, expected_signature):
_logger.warning("Poynt webhook signature mismatch")
raise Forbidden()
# === TERMINAL CALLBACK ROUTE === #
@http.route(
_terminal_callback_url, type='http', methods=['POST'],
auth='public', csrf=False,
)
def poynt_terminal_callback(self, **data):
"""Handle callback from a Poynt terminal after a payment completes.
The terminal sends transaction results here after the customer
taps/inserts their card at the physical device.
:return: A JSON acknowledgement.
:rtype: Response
"""
try:
raw_body = request.httprequest.data.decode('utf-8')
event = json.loads(raw_body)
except (ValueError, UnicodeDecodeError):
return request.make_json_response({'status': 'error'}, status=400)
_logger.info(
"Poynt terminal callback received:\n%s",
pprint.pformat(event),
)
reference = event.get('referenceId', event.get('data', {}).get('referenceId', ''))
transaction_id = event.get('transactionId', event.get('data', {}).get('transactionId', ''))
if not reference and not transaction_id:
_logger.warning("Terminal callback missing reference and transaction ID")
return request.make_json_response({'status': 'error'}, status=400)
payment_data = {
'reference': reference,
'poynt_transaction_id': transaction_id,
}
tx_sudo = request.env['payment.transaction'].sudo()._search_by_reference(
'poynt', payment_data,
)
if tx_sudo and transaction_id:
try:
txn_data = tx_sudo.provider_id._poynt_make_request(
'GET', f'transactions/{transaction_id}',
)
payment_data.update({
'poynt_status': txn_data.get('status', ''),
'funding_source': txn_data.get('fundingSource', {}),
'poynt_order_id': tx_sudo.poynt_order_id,
})
tx_sudo._process('poynt', payment_data)
except ValidationError:
_logger.error("Failed to process terminal callback for txn %s", transaction_id)
return request.make_json_response({'status': 'ok'})
# === OAUTH CALLBACK ROUTE === #
@http.route(_oauth_callback_url, type='http', methods=['GET'], auth='user')
def poynt_oauth_callback(self, **data):
"""Handle the OAuth2 authorization callback from Poynt.
After a merchant authorizes the application on poynt.net, they are
redirected here with an authorization code (JWT) and business ID.
:return: Redirect to the payment provider form.
:rtype: Response
"""
code = data.get('code', '')
status = data.get('status', '')
context = data.get('context', '')
business_id = data.get('businessId', '')
if status != 'AUTHORIZED':
_logger.warning("Poynt OAuth callback with status: %s", status)
return request.redirect('/odoo/settings')
if code:
try:
import jwt as pyjwt
decoded = pyjwt.decode(code, options={"verify_signature": False})
business_id = decoded.get('poynt.biz', business_id)
except Exception:
_logger.warning("Failed to decode Poynt OAuth JWT")
if business_id and context:
try:
provider_id = int(context)
provider = request.env['payment.provider'].browse(provider_id)
if provider.exists() and provider.code == 'poynt':
provider.sudo().write({
'poynt_business_id': business_id,
})
_logger.info(
"Poynt OAuth: linked business %s to provider %s",
business_id, provider_id,
)
except (ValueError, TypeError):
_logger.warning("Invalid provider context in Poynt OAuth callback: %s", context)
return request.redirect('/odoo/settings')
# === SURCHARGE HELPER === #
def _apply_portal_surcharge(self, tx_sudo, card_type):
"""Apply credit card surcharge to the linked invoice if enabled.
Detects the card brand from the number if card_type is not provided,
adds a surcharge line to the invoice, and updates the transaction
amount to include the fee.
:param tx_sudo: The sudo payment.transaction record.
:param str card_type: The card brand (visa, mastercard, amex, other).
:return: The surcharge fee amount, or 0 if not applied.
:rtype: float
"""
ICP = request.env['ir.config_parameter'].sudo()
if ICP.get_param('fusion_poynt.surcharge_enabled', 'False') != 'True':
return 0.0
if not card_type:
card_type = 'other'
rate_key = {
'visa': 'fusion_poynt.surcharge_visa_rate',
'mastercard': 'fusion_poynt.surcharge_mastercard_rate',
'amex': 'fusion_poynt.surcharge_amex_rate',
'debit': 'fusion_poynt.surcharge_debit_rate',
}.get(card_type, 'fusion_poynt.surcharge_other_rate')
rate = float(ICP.get_param(rate_key, '0') or 0)
if rate <= 0:
return 0.0
invoices = tx_sudo.invoice_ids
if not invoices:
base_amount = tx_sudo.amount
else:
base_amount = sum(invoices.mapped('amount_residual'))
fee_amount = round(base_amount * rate / 100.0, 2)
if fee_amount <= 0:
return 0.0
product_id = int(ICP.get_param('fusion_poynt.surcharge_product_id', '0') or 0)
product = request.env['product.product'].sudo().browse(product_id).exists()
if not product:
product = request.env.ref(
'fusion_poynt.product_cc_processing_fee', raise_if_not_found=False,
)
if not product:
_logger.warning("Surcharge product not configured; skipping surcharge")
return 0.0
for invoice in invoices.sudo():
was_posted = invoice.state == 'posted'
if was_posted:
invoice.button_draft()
description = "Credit Card Processing Fee (%.2f%% surcharge)" % rate
invoice.write({
'invoice_line_ids': [(0, 0, {
'product_id': product.id,
'name': description,
'quantity': 1,
'price_unit': fee_amount,
'tax_ids': [(5, 0, 0)],
})],
})
if was_posted:
invoice.action_post()
new_amount = tx_sudo.amount + fee_amount
tx_sudo.write({'amount': new_amount})
return fee_amount
# === JSON-RPC ROUTES (called from frontend JS) === #
@http.route('/payment/poynt/terminals', type='jsonrpc', auth='public')
def poynt_get_terminals(self, provider_id=None, **kwargs):
"""Return available Poynt terminals for the given provider.
:param int provider_id: The payment provider ID.
:return: List of terminal dicts with id, name, status.
:rtype: list
"""
if not provider_id:
return []
terminals = request.env['poynt.terminal'].sudo().search([
('provider_id', '=', int(provider_id)),
('active', '=', True),
])
return [{
'id': t.id,
'name': t.name,
'status': t.status,
'device_id': t.device_id,
} for t in terminals]
@http.route('/payment/poynt/process_card', type='jsonrpc', auth='public')
def poynt_process_card(self, reference=None, poynt_order_id=None,
card_number=None, exp_month=None, exp_year=None,
cvv=None, cardholder_name=None, card_type=None,
billing_address=None, billing_city=None,
billing_state=None, billing_zip=None,
billing_country=None, **kwargs):
"""Process a card payment through Poynt Cloud API.
The frontend sends card details which are passed to Poynt for
authorization. Card data is NOT stored in Odoo.
:return: Dict with success status or error message.
:rtype: dict
"""
if not reference:
return {'error': 'Missing payment reference.'}
tx_sudo = request.env['payment.transaction'].sudo().search([
('reference', '=', reference),
('provider_code', '=', 'poynt'),
], limit=1)
if not tx_sudo:
return {'error': 'Transaction not found.'}
try:
if not card_type and card_number:
card_type = _detect_card_brand(card_number)
surcharge_fee = self._apply_portal_surcharge(tx_sudo, card_type)
funding_source = {
'type': 'CREDIT_DEBIT',
'card': {
'number': card_number,
'expirationMonth': int(exp_month),
'expirationYear': int(exp_year),
'cardHolderFullName': cardholder_name or '',
},
'verificationData': {
'cvData': cvv,
'cardHolderBillingAddress': {
'line1': billing_address or '',
'city': billing_city or '',
'territory': billing_state or '',
'postalCode': billing_zip or '',
'countryCode': billing_country or '',
},
},
'entryDetails': {
'customerPresenceStatus': 'ECOMMERCE',
'entryMode': 'KEYED',
},
}
provider = tx_sudo.provider_id.sudo()
action = 'AUTHORIZE' if provider.capture_manually else 'SALE'
minor_amount = poynt_utils.format_poynt_amount(
tx_sudo.amount, tx_sudo.currency_id,
)
context = {
'source': 'WEB',
'sourceApp': 'odoo.fusion_poynt',
'transactionInstruction': 'ONLINE_AUTH_REQUIRED',
}
if provider.poynt_business_id:
context['businessId'] = provider.poynt_business_id
if provider.poynt_store_id:
context['storeId'] = provider.poynt_store_id
txn_payload = {
'action': action,
'fundingSourceType': 'CREDIT_DEBIT',
'amounts': {
'transactionAmount': minor_amount,
'orderAmount': minor_amount,
'tipAmount': 0,
'cashbackAmount': 0,
'currency': tx_sudo.currency_id.name,
},
'fundingSource': funding_source,
'context': context,
'notes': reference,
}
if poynt_order_id:
txn_payload['references'] = [{
'id': poynt_order_id,
'type': 'POYNT_ORDER',
}]
result = provider._poynt_make_request(
'POST', 'transactions', payload=txn_payload,
)
transaction_id = result.get('id', '')
status = result.get('status', '')
tx_sudo.write({
'poynt_transaction_id': transaction_id,
'provider_reference': transaction_id,
})
payment_data = {
'reference': reference,
'poynt_transaction_id': transaction_id,
'poynt_order_id': poynt_order_id,
'poynt_status': status,
'funding_source': result.get('fundingSource', {}),
}
tx_sudo._process('poynt', payment_data)
return {'success': True, 'status': status}
except ValidationError as e:
return {'error': str(e)}
except Exception as e:
_logger.error("Card payment processing failed: %s", e)
return {'error': 'Payment processing failed. Please try again.'}
@http.route('/payment/poynt/send_to_terminal', type='jsonrpc', auth='public')
def poynt_send_to_terminal(self, reference=None, terminal_id=None,
poynt_order_id=None, card_type=None, **kwargs):
"""Send a payment request to a Poynt terminal device.
:return: Dict with success status or error message.
:rtype: dict
"""
if not reference or not terminal_id:
return {'error': 'Missing reference or terminal ID.'}
tx_sudo = request.env['payment.transaction'].sudo().search([
('reference', '=', reference),
('provider_code', '=', 'poynt'),
], limit=1)
if not tx_sudo:
return {'error': 'Transaction not found.'}
terminal = request.env['poynt.terminal'].sudo().browse(int(terminal_id))
if not terminal.exists():
return {'error': 'Terminal not found.'}
try:
surcharge_fee = self._apply_portal_surcharge(tx_sudo, card_type or 'other')
result = terminal.action_send_payment_to_terminal(
amount=tx_sudo.amount,
currency=tx_sudo.currency_id,
reference=reference,
order_id=poynt_order_id,
)
return {'success': True, 'message_id': result.get('id', '')}
except (ValidationError, Exception) as e:
return {'error': str(e)}
@http.route('/payment/poynt/terminal_status', type='jsonrpc', auth='public')
def poynt_terminal_status(self, reference=None, terminal_id=None, **kwargs):
"""Poll the status of a terminal payment.
:return: Dict with current payment status.
:rtype: dict
"""
if not reference:
return {'status': 'error', 'message': 'Missing reference.'}
terminal = request.env['poynt.terminal'].sudo().browse(int(terminal_id or 0))
if not terminal.exists():
return {'status': 'error', 'message': 'Terminal not found.'}
return terminal.action_check_terminal_payment_status(reference)