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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import logging
import json
from odoo import api, fields, models, exceptions, _
from odoo.addons.iap.tools import iap_tools
# TDE FIXME: check those errors at iap level ?
from requests.exceptions import ConnectionError, HTTPError
_logger = logging.getLogger(__name__)
DEFAULT_ENDPOINT = 'https://partner-autocomplete.odoo.com'
class ResPartner(models.Model):
_name = 'res.partner'
_inherit = 'res.partner'
partner_gid = fields.Integer('Company database ID')
additional_info = fields.Char('Additional info')
@api.model
def _replace_location_code_by_id(self, record):
record['country_id'], record['state_id'] = self._find_country_data(
state_code=record.pop('state_code', False),
state_name=record.pop('state_name', False),
country_code=record.pop('country_code', False),
country_name=record.pop('country_name', False)
)
return record
@api.model
def _format_data_company(self, company):
self._replace_location_code_by_id(company)
if company.get('child_ids'):
child_ids = []
for child in company.get('child_ids'):
child_ids.append(self._replace_location_code_by_id(child))
company['child_ids'] = child_ids
if company.get('additional_info'):
company['additional_info'] = json.dumps(company['additional_info'])
return company
@api.model
def _find_country_data(self, state_code, state_name, country_code, country_name):
country = self.env['res.country'].search([['code', '=ilike', country_code]])
if not country:
country = self.env['res.country'].search([['name', '=ilike', country_name]])
state_id = False
country_id = False
if country:
country_id = {
'id': country.id,
'display_name': country.display_name
}
if state_name or state_code:
state = self.env['res.country.state'].search([
('country_id', '=', country_id.get('id')),
'|',
('name', '=ilike', state_name),
('code', '=ilike', state_code)
], limit=1)
if state:
state_id = {
'id': state.id,
'display_name': state.display_name
}
else:
_logger.info('Country code not found: %s', country_code)
return country_id, state_id
@api.model
def get_endpoint(self):
url = self.env['ir.config_parameter'].sudo().get_param('iap.partner_autocomplete.endpoint', DEFAULT_ENDPOINT)
url += '/iap/partner_autocomplete'
return url
@api.model
def _rpc_remote_api(self, action, params, timeout=15):
if self.env.registry.in_test_mode() :
return False, 'Insufficient Credit'
url = '%s/%s' % (self.get_endpoint(), action)
account = self.env['iap.account'].get('partner_autocomplete')
if not account.account_token:
return False, 'No Account Token'
params.update({
'db_uuid': self.env['ir.config_parameter'].sudo().get_param('database.uuid'),
'account_token': account.account_token,
'country_code': self.env.company.country_id.code,
'zip': self.env.company.zip,
})
try:
return iap_tools.iap_jsonrpc(url=url, params=params, timeout=timeout), False
except (ConnectionError, HTTPError, exceptions.AccessError, exceptions.UserError) as exception:
_logger.error('Autocomplete API error: %s' % str(exception))
return False, str(exception)
except iap_tools.InsufficientCreditError as exception:
_logger.warning('Insufficient Credits for Autocomplete Service: %s' % str(exception))
return False, 'Insufficient Credit'
@api.model
def autocomplete(self, query):
suggestions, error = self._rpc_remote_api('search', {
'query': query,
})
if suggestions:
results = []
for suggestion in suggestions:
results.append(self._format_data_company(suggestion))
return results
else:
return []
@api.model
def enrich_company(self, company_domain, partner_gid, vat):
response, error = self._rpc_remote_api('enrich', {
'domain': company_domain,
'partner_gid': partner_gid,
'vat': vat,
})
if response and response.get('company_data'):
result = self._format_data_company(response.get('company_data'))
else:
result = {}
if response and response.get('credit_error'):
result.update({
'error': True,
'error_message': 'Insufficient Credit'
})
elif error:
result.update({
'error': True,
'error_message': error
})
return result
@api.model
def read_by_vat(self, vat):
vies_vat_data, error = self._rpc_remote_api('search_vat', {
'vat': vat,
})
if vies_vat_data:
return [self._format_data_company(vies_vat_data)]
else:
return []
@api.model
def _is_company_in_europe(self, country_code):
country = self.env['res.country'].search([('code', '=ilike', country_code)])
if country:
country_id = country.id
europe = self.env.ref('base.europe')
if not europe:
europe = self.env["res.country.group"].search([('name', '=', 'Europe')], limit=1)
if not europe or country_id not in europe.country_ids.ids:
return False
return True
def _is_vat_syncable(self, vat):
vat_country_code = vat[:2]
partner_country_code = self.country_id.code if self.country_id else ''
return self._is_company_in_europe(vat_country_code) and (partner_country_code == vat_country_code or not partner_country_code)
def _is_synchable(self):
already_synched = self.env['res.partner.autocomplete.sync'].search([('partner_id', '=', self.id), ('synched', '=', True)])
return self.is_company and self.partner_gid and not already_synched
def _update_autocomplete_data(self, vat):
self.ensure_one()
if vat and self._is_synchable() and self._is_vat_syncable(vat):
self.env['res.partner.autocomplete.sync'].sudo().add_to_queue(self.id)
@api.model_create_multi
def create(self, vals_list):
partners = super(ResPartner, self).create(vals_list)
if len(vals_list) == 1:
partners._update_autocomplete_data(vals_list[0].get('vat', False))
if partners.additional_info:
template_values = json.loads(partners.additional_info)
template_values['flavor_text'] = _("Partner created by Odoo Partner Autocomplete Service")
partners.message_post_with_view(
'iap_mail.enrich_company',
values=template_values,
subtype_id=self.env.ref('mail.mt_note').id,
)
partners.write({'additional_info': False})
return partners
def write(self, values):
res = super(ResPartner, self).write(values)
if len(self) == 1:
self._update_autocomplete_data(values.get('vat', False))
return res
|
