1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
# coding: utf-8
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
import requests
import werkzeug
from odoo import fields, models, api, _
from odoo.exceptions import ValidationError, UserError, AccessError
_logger = logging.getLogger(__name__)
TIMEOUT = 10
class PosPaymentMethod(models.Model):
_inherit = 'pos.payment.method'
def _get_payment_terminal_selection(self):
return super()._get_payment_terminal_selection() + [('stripe', 'Stripe')]
# Stripe
stripe_serial_number = fields.Char(help='[Serial number of the stripe terminal], for example: WSC513105011295', copy=False)
@api.model
def _load_pos_data_fields(self, config_id):
params = super()._load_pos_data_fields(config_id)
params += ['stripe_serial_number']
return params
@api.constrains('stripe_serial_number')
def _check_stripe_serial_number(self):
for payment_method in self:
if not payment_method.stripe_serial_number:
continue
existing_payment_method = self.search([('id', '!=', payment_method.id),
('stripe_serial_number', '=', payment_method.stripe_serial_number)],
limit=1)
if existing_payment_method:
raise ValidationError(_('Terminal %(terminal)s is already used on payment method %(payment_method)s.',
terminal=payment_method.stripe_serial_number, payment_method=existing_payment_method.display_name))
def _get_stripe_payment_provider(self):
stripe_payment_provider = self.env['payment.provider'].search([
('code', '=', 'stripe'),
('company_id', '=', self.env.company.id)
], limit=1)
if not stripe_payment_provider:
raise UserError(_("Stripe payment provider for company %s is missing", self.env.company.name))
return stripe_payment_provider
@api.model
def _get_stripe_secret_key(self):
stripe_secret_key = self._get_stripe_payment_provider().stripe_secret_key
if not stripe_secret_key:
raise ValidationError(_('Complete the Stripe onboarding for company %s.', self.env.company.name))
return stripe_secret_key
@api.model
def stripe_connection_token(self):
if not self.env.user.has_group('point_of_sale.group_pos_user'):
raise AccessError(_("Do not have access to fetch token from Stripe"))
endpoint = 'https://api.stripe.com/v1/terminal/connection_tokens'
try:
resp = requests.post(endpoint, auth=(self.sudo()._get_stripe_secret_key(), ''), timeout=TIMEOUT)
except requests.exceptions.RequestException:
_logger.exception("Failed to call stripe_connection_token endpoint")
raise UserError(_("There are some issues between us and Stripe, try again later."))
return resp.json()
def _stripe_calculate_amount(self, amount):
currency = self.journal_id.currency_id or self.company_id.currency_id
return round(amount/currency.rounding)
def stripe_payment_intent(self, amount):
if not self.env.user.has_group('point_of_sale.group_pos_user'):
raise AccessError(_("Do not have access to fetch token from Stripe"))
# For Terminal payments, the 'payment_method_types' parameter must include
# at least 'card_present' and the 'capture_method' must be set to 'manual'.
endpoint = 'https://api.stripe.com/v1/payment_intents'
currency = self.journal_id.currency_id or self.company_id.currency_id
params = [
("currency", currency.name),
("amount", self._stripe_calculate_amount(amount)),
("payment_method_types[]", "card_present"),
("capture_method", "manual"),
]
if currency.name == 'AUD' and self.company_id.country_code == 'AU':
# See https://stripe.com/docs/terminal/payments/regional?integration-country=AU
# This parameter overrides "capture_method": "manual" above.
params.append(("payment_method_options[card_present][capture_method]", "manual_preferred"))
elif currency.name == 'CAD' and self.company_id.country_code == 'CA':
params.append(("payment_method_types[]", "interac_present"))
try:
data = werkzeug.urls.url_encode(params)
resp = requests.post(endpoint, data=data, auth=(self.sudo()._get_stripe_secret_key(), ''), timeout=TIMEOUT)
except requests.exceptions.RequestException:
_logger.exception("Failed to call stripe_payment_intent endpoint")
raise UserError(_("There are some issues between us and Stripe, try again later."))
return resp.json()
@api.model
def stripe_capture_payment(self, paymentIntentId, amount=None):
"""Captures the payment identified by paymentIntentId.
:param paymentIntentId: the id of the payment to capture
:param amount: without this parameter the entire authorized
amount is captured. Specifying a larger amount allows
overcapturing to support tips.
"""
if not self.env.user.has_group('point_of_sale.group_pos_user'):
raise AccessError(_("Do not have access to fetch token from Stripe"))
endpoint = ('payment_intents/%s/capture') % (werkzeug.urls.url_quote(paymentIntentId))
data = None
if amount is not None:
data = {
"amount_to_capture": self._stripe_calculate_amount(amount),
}
return self.sudo()._get_stripe_payment_provider()._stripe_make_request(endpoint, data)
def action_stripe_key(self):
res_id = self._get_stripe_payment_provider().id
# Redirect
return {
'name': _('Stripe'),
'res_model': 'payment.provider',
'type': 'ir.actions.act_window',
'view_mode': 'form',
'res_id': res_id,
}
|
