import requests import re import logging from odoo.http import route, request, Controller from odoo.addons.payment_hyperpay.data.payment_icon import payment_icon from odoo.addons.payment.controllers.portal import PaymentProcessing _logger = logging.getLogger(__name__) TEST_URL = "https://eu-test.oppwa.com" LIVE_URL = "https://eu-prod.oppwa.com" class HyperPayTokenization(Controller): @route('/hyperpay/tokens/checkout', type='json', auth='public', website=True, methods=['POST']) def token_checkout(self, **kwargs): if not kwargs.get('acquirer_id'): return {'state': False, 'message': 'Couldn\'t identify acquirer'} try: self._save_session_data(kwargs) checkout_data = self._register_payment_card(kwargs) return checkout_data except Exception as er: return { 'state': False, 'message': er } def _save_session_data(self, data): # save data in session as a form of naive saving return True def _register_payment_card(self, data): acquirer_id = int(data.get('acquirer_id')) acquirer = request.env['payment.acquirer'].sudo().search([('id', '=', acquirer_id)]) payment_icons = acquirer.payment_icon_ids.mapped('name') data_brands = "VISA" if (len(payment_icon) > 1): brands = [payment_icon[i.upper()] for i in payment_icons if i.upper() in payment_icon.keys()] data_brands = brands and " ".join(brands) or data_brands base_url = request.httprequest.host_url payload = { "entityId": acquirer.hyperpay_merchant_id, 'createRegistration': True, } payload.update(self._get_hyperpay_token_custom_parameters(data)) if acquirer.state == 'test': domain = TEST_URL else: domain = LIVE_URL url = f"{domain}/v1/checkouts" headers = { "Authorization": f"Bearer {acquirer.hyperpay_authorization}" } _logger.info('Hyperpay Token Checkout Payload:', payload) response_data = requests.post(url=url, data=payload, headers=headers) _logger.info('Hyperpay Token Checkout Response Text:', response_data.text) response = response_data.json() _logger.info('Hyperpay Token Checkout Response JSON:', response) result = response.get('result', {}) result_code = result.get('code') if result_code and not re.match(r"^(000\.000\.|000\.100\.1|000\.[36]|000\.400\.1[12]0|000\.400\.0[^3]|000\.400\.100|000\.200)", result_code): return {'state': False, 'message': result.get('description', '')} return_url = f'{base_url}hyperpay/tokens/result?acquirer_id={acquirer_id}' return { 'state': True, 'checkout_id': response.get('id'), 'service_domain': domain, 'base_url': base_url, 'data_brands': data_brands, 'return_url': return_url, } def _get_hyperpay_token_custom_parameters(self, data): reference_id = request.session.get('hyperpay_token_reference_id') reference_model = request.session.get('hyperpay_token_reference_model') return { "customParameters[SHOPPER_acquirer_id]": data.get('acquirer_id', 0), "customParameters[SHOPPER_hyperpay_token_reference_id]": reference_id, "customParameters[SHOPPER_hyperpay_token_reference_model]": reference_model, } @route('/hyperpay/tokens/result', type='http', auth='public', website=True) def token_return(self, **post): try: _logger.info('Hyperpay Token Return Post:', post) acquirer_id = request.env['payment.acquirer'].sudo().search([('id', '=', int(post.get('acquirer_id', 0)))]) if acquirer_id.state == 'test': domain = TEST_URL else: domain = LIVE_URL url = f"{domain}{post.get('resourcePath')}?entityId={acquirer_id.hyperpay_merchant_id}" headers = { "Authorization": f"Bearer {acquirer_id.hyperpay_authorization}" } _logger.info('Hyperpay Token Status Request:', url) response_data = requests.get(url=url, headers=headers) _logger.info('Hyperpay Token Status Response Text:', response_data.text) resp = response_data.json() _logger.info('Hyperpay Token Status Response JSON:', resp) result = resp.get('result', {}) result_code = result.get('code') if result_code and not re.match(r"^(000\.000\.|000\.100\.1|000\.[36]|000\.400\.1[12]0|000\.400\.0[^3]|000\.400\.100|000\.200)", result_code): # Handle failed cards logic here return {'state': False, 'message': result.get('description', ''), 'errors': resp.get('parameterErrors', [])} # create card record and activate recurring donation card = resp.get('card', {}) if not card: return {'state': False, 'message': 'Card data not found'} self._post_process_token_return(resp) except Exception as er: # request.env.cr.rollback() _logger.error(er) return request.redirect('/my/recurring_donation') def _post_process_token_return(self, data): acquirer_id = int(data.get('customParameters', {}).get('SHOPPER_acquirer_id', 0)) card = data.get('card', {}) card_vals = { 'name': f"{card.get('bin', '')}XXXXXXXXXXXX{card.get('last4Digits', '')}", 'partner_id': request.env.user.partner_id.id, 'acquirer_id': acquirer_id, 'acquirer_ref': data.get('id', ''), 'hyperpay_payment_brand': data.get('paymentBrand'), } token_id = request.env['payment.token'].sudo().create(card_vals) token_id.validate() reference_id = int(data.get('customParameters', {}).get('SHOPPER_hyperpay_token_reference_id', 0)) reference_model = data.get('customParameters', {}).get('SHOPPER_hyperpay_token_reference_model', '') if reference_id and reference_model and reference_model in request.env: record_id = request.env[reference_model].sudo().search([('id', '=', reference_id)]) if record_id and hasattr(record_id, '_post_process_card_tokenization'): record_id._post_process_card_tokenization(token_id) return token_id