154 lines
6.1 KiB
Python
154 lines
6.1 KiB
Python
from odoo import models, fields
|
|
from odoo.http import request
|
|
from odoo.addons.payment.models.payment_acquirer import PaymentToken
|
|
import requests
|
|
import re
|
|
import dateutil
|
|
import pytz
|
|
import logging
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
TEST_URL = "https://eu-test.oppwa.com"
|
|
LIVE_URL = "https://eu-prod.oppwa.com"
|
|
|
|
|
|
|
|
class AcquirerHyperPay(models.Model):
|
|
_inherit = 'payment.acquirer'
|
|
|
|
hyperpay_s2s_entity_id = fields.Char('Server2Server Entity Id', groups='base.group_user')
|
|
|
|
class HyperPayTransaction(models.Model):
|
|
_inherit = 'payment.transaction'
|
|
|
|
def hyperpay_s2s_do_transaction(self, **kwargs):
|
|
self.ensure_one()
|
|
|
|
if self.acquirer_id.state == 'test':
|
|
domain = TEST_URL
|
|
else:
|
|
domain = LIVE_URL
|
|
|
|
url = f"{domain}/v1/registrations/{self.payment_token_id.acquirer_ref}/payments"
|
|
|
|
payload = self._hyperpay_get_s2s_transaction_payload(kwargs)
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {self.acquirer_id.hyperpay_authorization}"
|
|
}
|
|
_logger.info('Hyperpay S2S Transaction Payload:', payload)
|
|
response = requests.post(url, data=payload, headers=headers)
|
|
_logger.info('Hyperpay S2S Transaction Response Text:', response.text)
|
|
data = response.json()
|
|
_logger.info('Hyperpay S2S Transaction Response JSON:', data)
|
|
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.recurringType': 'STANDING_ORDER',
|
|
'customParameters[paymentFrequency]': 'OTHER',
|
|
'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):
|
|
self.ensure_one()
|
|
|
|
if self.acquirer_id.state == 'test':
|
|
domain = TEST_URL
|
|
else:
|
|
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,
|
|
'paymentBrand': self.payment_token_id.hyperpay_payment_brand,
|
|
'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}"
|
|
}
|
|
response = requests.post(url, data=payload, headers=headers)
|
|
data = response.json()
|
|
return self._hyperpay_s2s_validate_transaction(data)
|
|
|
|
def _hyperpay_s2s_validate_transaction(self, data):
|
|
success_re = r"^(000\.000\.|000\.100\.1|000\.[36]|000\.400\.1[12]0|000\.400\.0[^3]|000\.400\.100)"
|
|
pending_re = r"^(000\.200|800\.400\.5|100\.400\.500)"
|
|
|
|
result = data.get('result')
|
|
result_code = result.get('code')
|
|
res = {
|
|
'acquirer_reference': data.get('id'),
|
|
'state_message': f"{result.get('description', '')}\n{','.join([str(s) for s in data.get('parameterErrors', [])])}"
|
|
}
|
|
|
|
if re.match(success_re, result_code):
|
|
date_validate = dateutil.parser.parse(data.get('timestamp')).astimezone(pytz.utc).replace(tzinfo=None)
|
|
res.update(date=date_validate)
|
|
if self.type == 'validation' and not self.payment_token_id.verified:
|
|
self.payment_token_id.write({
|
|
'verified': True,
|
|
'hyperpay_initial_transaction_id': data.get('CardholderInitiatedTransactionID', '')
|
|
})
|
|
self._set_transaction_done()
|
|
elif re.match(pending_re, result_code):
|
|
self.write({'html_3ds': data.get('id', ' ')})
|
|
self._set_transaction_pending()
|
|
else:
|
|
self._set_transaction_error(result.get('description', ''))
|
|
return self.write(res)
|
|
|
|
|
|
class HyperPayToken(models.Model):
|
|
_inherit = 'payment.token'
|
|
|
|
hyperpay_payment_brand = fields.Char('Payment Brand')
|
|
hyperpay_initial_transaction_id = fields.Char()
|
|
|
|
|
|
PaymentToken.VALIDATION_AMOUNTS.update({
|
|
'SAR': 5.00
|
|
})
|