138 lines
5.0 KiB
Python
138 lines
5.0 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
|
|
import pprint
|
|
|
|
|
|
_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: %s' % payload)
|
|
response = requests.post(url, data=payload, headers=headers)
|
|
_logger.info('Hyperpay S2S Transaction Response Text: %s' % response.text)
|
|
data = response.json()
|
|
_logger.info('Hyperpay S2S Transaction Response JSON: %s' % data)
|
|
return self._hyperpay_s2s_validate_transaction(data)
|
|
|
|
def _hyperpay_get_s2s_transaction_payload(self, data):
|
|
base_url = self.acquirer_id.get_base_url()
|
|
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",
|
|
'standingInstruction.mode': 'REPEATED',
|
|
'standingInstruction.source': 'MIT',
|
|
'standingInstruction.recurringType': 'STANDING_ORDER',
|
|
'standingInstruction.initialTransactionId': self.payment_token_id.hyperpay_initial_transaction_id,
|
|
'standingInstruction.type': 'UNSCHEDULED',
|
|
'shopperResultUrl': f'{base_url}hyperpay/tokens/result?transaction_id={self.id}',
|
|
}
|
|
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}"
|
|
|
|
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,
|
|
}
|
|
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):
|
|
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 self.hyperpay_response_payload:
|
|
res.update({
|
|
'hyperpay_response_payload': f"{self.hyperpay_response_payload} \n===============\n {pprint.pformat(data)}"
|
|
})
|
|
else:
|
|
res.update({
|
|
'hyperpay_response_payload': pprint.pformat(data)
|
|
})
|
|
|
|
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
|
|
})
|