devops_admin
/
odex25_standard
mirror of https://github.com/hydracp9/odex25_standard
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #5912 from expsa/change_token_flow 

[IMP] payment_hyperpay_tokenization: change the flow of CIT to be don…
This commit is contained in:
abdurrahman-saber 2025-12-31 19:02:48 +04:00 committed by GitHub
parent 04dbf6f55d 40c880bfdc
commit c4fd014af7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 91 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
odex25_donation/payment_hyperpay_tokenization/controllers/main.py
View File

@ -1,10 +1,12 @@
import requests
import re
import logging
from odoo import _
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
from odoo.addons.payment.models.payment_acquirer import PaymentToken
from datetime import datetime
_logger = logging.getLogger(__name__)
@ -36,16 +38,16 @@ class HyperPayTokenization(Controller):
acquirer = request.env['payment.acquirer'].sudo().search([('id', '=', acquirer_id)])
payment_icons = acquirer.payment_icon_ids.mapped('name')
data_brands = "VISA"
base_url = request.httprequest.host_url
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,
}
tx = self._create_validation_transaction(acquirer)
payload = self._get_hyperpay_token_payload(acquirer, data_brands, data)
payload.update(self._get_hyperpay_token_custom_parameters(data))
payload.update({
"merchantTransactionId": tx.reference,
})
if acquirer.state == 'test':
domain = TEST_URL
@ -68,7 +70,7 @@ class HyperPayTokenization(Controller):
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_url = f'{base_url}hyperpay/tokens/result?transaction_id={tx.id}'
return {
'state': True,
@ -79,20 +81,76 @@ class HyperPayTokenization(Controller):
'return_url': return_url,
}
def _get_hyperpay_token_custom_parameters(self, data):
def _get_hyperpay_token_payload(self, acquirer, data_brands, data):
reference_id = request.session.get('hyperpay_token_reference_id')
reference_model = request.session.get('hyperpay_token_reference_model')
partner_id = request.env.user.partner_id
currency = partner_id.currency_id
lang_code = str(request.env['res.lang'].sudo().search([('code', '=', request.env.user.lang)]).iso_code or '').upper()
amount = self._get_validation_amount(currency)
return {
"entityId": acquirer.hyperpay_merchant_id,
'createRegistration': True,
"amount": '%.2f' % amount,
"currency": currency.name,
'paymentBrand': data_brands,
"paymentType": "DB",
'standingInstruction.mode': 'INITIAL',
'standingInstruction.source': 'CIT',
'standingInstruction.recurringType': 'STANDING_ORDER',
'customParameters[paymentFrequency]': 'OTHER',
'standingInstruction.type': 'UNSCHEDULED',
"billing.street1": partner_id.street or 'Riyadh',
"billing.street2": partner_id.street2 or '',
"billing.city": partner_id.city or 'Riyadh',
"billing.state": partner_id.state_id.name or 'Riyadh',
"billing.postcode": partner_id.zip or '',
"billing.country": partner_id.country_id.code or 'SA',
"customer.givenName": partner_id.name,
"customer.surname": '',
"customer.email": partner_id.email,
"customer.mobile": partner_id.mobile or partner_id.phone or '',
"customer.phone": partner_id.phone or partner_id.mobile or '',
'customer.ip': request.httprequest.environ["REMOTE_ADDR"],
'customer.language': lang_code,
"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,
}
def _create_validation_transaction(self, acquirer_id):
partner_id = request.env.user.partner_id
currency = partner_id.currency_id
amount = self._get_validation_amount(currency)
reference = "VALIDATION-%s" % (datetime.now().strftime('%y%m%d_%H%M%S'))
tx = request.env['payment.transaction'].sudo().create({
'amount': amount,
'acquirer_id': acquirer_id.id,
'type': 'validation',
'currency_id': currency.id,
'reference': reference,
'partner_id': partner_id.id,
'partner_country_id': partner_id.country_id.id,
'state_message': _('This Transaction was automatically processed & refunded in order to validate a new credit card.'),
})
return tx
def _get_validation_amount(self, currency):
if PaymentToken.VALIDATION_AMOUNTS.get(currency.name):
return PaymentToken.VALIDATION_AMOUNTS.get(currency.name)
else:
# If we don't find the user's currency, then we set the currency to EUR and the amount to 1€50.
currency = self.env['res.currency'].search([('name', '=', 'EUR')])
return 1.5
@route('/hyperpay/tokens/result', type='http', auth='public', website=True)
def token_return(self, **post):
try:
_logger.info('Hyperpay Token Return Post: %s' % post)
acquirer_id = request.env['payment.acquirer'].sudo().search([('id', '=', int(post.get('acquirer_id', 0)))])
transaction_id = request.env['payment.transaction'].sudo().search([('id', '=', int(post.get('transaction_id', 0)))])
acquirer_id = transaction_id.acquirer_id
if acquirer_id.state == 'test':
domain = TEST_URL
@ -119,26 +177,38 @@ class HyperPayTokenization(Controller):
card = resp.get('card', {})
if not card:
return {'state': False, 'message': 'Card data not found'}
self._post_process_token_return(resp)
self._post_process_token_return(transaction_id, 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))
def _post_process_token_return(self, transaction_id, data):
data.update({'tx_id': transaction_id.id})
request.env['payment.transaction'].sudo().form_feedback(data, "hyperpay")
if not transaction_id.state == 'done':
_logger.error('Hyperpay Token Return Transaction not done: %s' % transaction_id.id)
return False
try:
transaction_id.hyperpay_s2s_do_refund()
except Exception as er:
_logger.error('Hyperpay Token Return Transaction refund failed: %s' % er)
acquirer_id = transaction_id.acquirer_id
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_id': acquirer_id.id,
'acquirer_ref': data.get('id', ''),
'hyperpay_payment_brand': data.get('paymentBrand'),
'initial_transaction_id': data.get('resultDetails.CardholderInitiatedTransactionID') or data.get('CardholderInitiatedTransactionID', ''),
'verified': True,
}
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:

36
odex25_donation/payment_hyperpay_tokenization/models/payment.py
View File

@ -45,45 +45,20 @@ class HyperPayTransaction(models.Model):
return self._hyperpay_s2s_validate_transaction(data)
def _hyperpay_get_s2s_transaction_payload(self, data):
partner_id = self.env.user.partner_id
base_url = request.httprequest.host_url
lang_code = str(self.env['res.lang'].sudo().search([('code', '=', self.env.user.lang)]).iso_code or '').upper()
payload = {
"entityId": self.acquirer_id.hyperpay_s2s_entity_id,
"amount": '%.2f' % self.amount,
"currency": self.currency_id.name,
'paymentBrand': self.payment_token_id.hyperpay_payment_brand,
"paymentType": "DB",
"createRegistration": True,
'standingInstruction.mode': 'INITIAL',
'standingInstruction.source': 'CIT',
'standingInstruction.mode': 'REPEATED',
'standingInstruction.source': 'MIT',
'standingInstruction.recurringType': 'STANDING_ORDER',
'customParameters[paymentFrequency]': 'OTHER',
'standingInstruction.initialTransactionId': self.payment_token_id.hyperpay_initial_transaction_id,
'standingInstruction.type': 'UNSCHEDULED',
'shopperResultUrl': f'{base_url}hyperpay/tokens/result?transaction_id={self.id}',
'notificationUrl': f'{base_url}hyperpay/tokens/result?transaction_id={self.id}',
"billing.street1": partner_id.street or 'Riyadh',
"billing.street2": partner_id.street2 or '',
"billing.city": partner_id.city or 'Riyadh',
"billing.state": partner_id.state_id.name or 'Riyadh',
"billing.postcode": partner_id.zip or '',
"billing.country": partner_id.country_id.code or 'SA',
"customer.givenName": partner_id.name,
"customer.surname": '',
"customer.email": partner_id.email,
"customer.mobile": partner_id.mobile or partner_id.phone or '',
"customer.phone": partner_id.phone or partner_id.mobile or '',
'customer.ip': request.httprequest.environ["REMOTE_ADDR"],
'customer.language': lang_code,
}
if self.type != 'validation':
payload.update({
'standingInstruction.mode': 'REPEATED',
'standingInstruction.source': 'MIT',
'standingInstruction.initialTransactionId': self.payment_token_id.hyperpay_initial_transaction_id
})
payload.pop('customParameters[paymentFrequency]', None)
payload.pop('createRegistration', None)
return payload
def hyperpay_s2s_do_refund(self, **kwargs):
@ -95,8 +70,6 @@ class HyperPayTransaction(models.Model):
domain = LIVE_URL
url = f"{domain}/v1/payments/{self.acquirer_reference}"
base_url = request.httprequest.host_url
payload = {
'entityId': self.acquirer_id.hyperpay_s2s_entity_id,
@ -104,13 +77,14 @@ class HyperPayTransaction(models.Model):
'paymentType': 'RF',
'amount': '%.2f' % self.amount,
'currency': self.currency_id.name,
'shopperResultUrl': f'{base_url}hyperpay/tokens/result',
}
headers = {
"Authorization": f"Bearer {self.acquirer_id.hyperpay_authorization}"
}
_logger.info('Hyperpay S2S Refund Payload: %s' % payload)
response = requests.post(url, data=payload, headers=headers)
data = response.json()
_logger.info('Hyperpay S2S Refund Response JSON: %s' % data)
return self._hyperpay_s2s_validate_transaction(data)
def _hyperpay_s2s_validate_transaction(self, data):