|
|
# -*- coding: utf-8 -*-# Python source code encoding : https://www.python.org/dev/peps/pep-0263/################################################################################ OpenERP, Open Source Management Solution# This module copyright :# (c) 2015 Antiun Ingenieria, SL (Madrid, Spain, http://www.antiun.com)# Antonio Espinosa <antonioea@antiun.com>## This program is free software: you can redistribute it and/or modify# it under the terms of the GNU Affero General Public License as# published by the Free Software Foundation, either version 3 of the# License, or (at your option) any later version.## This program is distributed in the hope that it will be useful,# but WITHOUT ANY WARRANTY; without even the implied warranty of# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the# GNU Affero General Public License for more details.## You should have received a copy of the GNU Affero General Public License# along with this program. If not, see <http://www.gnu.org/licenses/>.###############################################################################
from openerp import models, api, _from openerp.exceptions import Warningimport requestsimport reimport loggingfrom lxml import etreefrom collections import OrderedDict
from pprint import pformat
logger = logging.getLogger(__name__)
class NaceImport(models.TransientModel): _name = 'nace.import' _description = 'Import NACE activities from European RAMON service' _parents = [False, False, False, False] _available_langs = { 'bg_BG': 'BG', # Bulgarian 'cs_CZ': 'CZ', # Czech 'da_DK': 'DA', # Danish 'de_DE': 'DE', # German 'et_EE': 'EE', # Estonian 'el_GR': 'EL', # Greek 'es_AR': 'ES', # Spanish (AR) 'es_BO': 'ES', # Spanish (BO) 'es_CL': 'ES', # Spanish (CL) 'es_CO': 'ES', # Spanish (CO) 'es_CR': 'ES', # Spanish (CR) 'es_DO': 'ES', # Spanish (DO) 'es_EC': 'ES', # Spanish (EC) 'es_GT': 'ES', # Spanish (GT) 'es_HN': 'ES', # Spanish (HN) 'es_MX': 'ES', # Spanish (MX) 'es_NI': 'ES', # Spanish (NI) 'es_PA': 'ES', # Spanish (PA) 'es_PE': 'ES', # Spanish (PE) 'es_PR': 'ES', # Spanish (PR) 'es_PY': 'ES', # Spanish (PY) 'es_SV': 'ES', # Spanish (SV) 'es_UY': 'ES', # Spanish (UY) 'es_VE': 'ES', # Spanish (VE) 'es_PY': 'ES', # Spanish (PY) 'es_ES': 'ES', # Spanish 'fi_FI': 'FI', # Finnish 'fr_BE': 'FR', # French (FR) 'fr_CA': 'FR', # French (CA) 'fr_CH': 'FR', # French (CH) 'fr_FR': 'FR', # French 'hr_HR': 'HR', # Croatian 'hu_HU': 'HU', # Hungarian 'it_IT': 'IT', # Italian 'lt_LT': 'LT', # Lithuanian 'lv_LV': 'LV', # Latvian # '': 'MT', # Il-Malti, has no language in Odoo 'nl_BE': 'NL', # Dutch (BE) 'nl_NL': 'NL', # Dutch 'nb_NO': 'NO', # Norwegian Bokmål 'pl_PL': 'PL', # Polish 'pt_BR': 'PT', # Portuguese (BR) 'pt_PT': 'PT', # Portuguese 'ro_RO': 'RO', # Romanian 'ru_RU': 'RU', # Russian 'sl_SI': 'SI', # Slovenian 'sk_SK': 'SK', # Slovak 'sv_SE': 'SV', # Swedish 'tr_TR': 'TR', # Turkish } _map = OrderedDict([ ('level', {'xpath': '', 'attrib': 'idLevel', 'type': 'integer', 'translate': False, 'required': True}), ('code', {'xpath': '', 'attrib': 'id', 'type': 'string', 'translate': False, 'required': True}), ('name', {'xpath': './Label/LabelText', 'type': 'string', 'translate': True, 'required': True}), ('generic', { 'xpath': './Property[@name="Generic"]' '/PropertyQualifier[@name="Value"]/PropertyText', 'type': 'string', 'translate': False, 'required': False}), ('rules', { 'xpath': './Property[@name="ExplanatoryNote"]' '/PropertyQualifier[@name="Rules"]/PropertyText', 'type': 'string', 'translate': False, 'required': False}), ('central_content', { 'xpath': './Property[@name="ExplanatoryNote"]' '/PropertyQualifier[@name="CentralContent"]/PropertyText', 'type': 'string', 'translate': True, 'required': False}), ('limit_content', { 'xpath': './Property[@name="ExplanatoryNote"]' '/PropertyQualifier[@name="LimitContent"]/PropertyText', 'type': 'string', 'translate': True, 'required': False}), ('exclusions', { 'xpath': './Property[@name="ExplanatoryNote"]' '/PropertyQualifier[@name="Exclusions"]/PropertyText', 'type': 'string', 'translate': True, 'required': False}), ])
def _check_node(self, node): if node.get('id') and node.get('idLevel'): return True return False
def _mapping(self, node, translate): item = {} for k, v in self._map.iteritems(): field_translate = v.get('translate', False) field_xpath = v.get('xpath', '') field_attrib = v.get('attrib', False) field_type = v.get('type', 'string') field_required = v.get('required', False) if field_translate == translate: # logger.info("_mapping = '%s', config = '%s', node = '%s'" % # (k, pformat(v), pformat(node))) # logger.info("_mapping:children = '%s'" % # (pformat(node.getchildren()))) # for x in node.getchildren(): # logger.info("_mapping:grandchildren = '%s'" % # (pformat(x.getchildren()))) # logger.info("_mapping:LabelText = '%s'" % # (pformat(node.find('./Label/LabelText')))) value = '' if field_xpath: n = node.find(field_xpath) # logger.info("_mapping:Finding = '%s', result='%s'" % # (field_xpath, pformat(n))) else: n = node # logger.info("_mapping:Using same node='%s'" % # (pformat(n))) if n is not None: if field_attrib: value = n.get(field_attrib, '') else: value = n.text if field_type == 'integer': try: value = int(value) except: value = 0 else: logger.debug("xpath = '%s', not found" % field_xpath) if field_required and not value: raise Warning( _('Value not found for mandatory field %s' % k)) item[k] = value return item
def _download_nace(self, lang_code): url_base = 'http://ec.europa.eu' url_path = '/eurostat/ramon/nomenclatures/index.cfm' url_params = { 'TargetUrl': 'ACT_OTH_CLS_DLD', 'StrNom': 'NACE_REV2', 'StrFormat': 'XML', 'StrLanguageCode': lang_code, # 'IntKey': '', # 'IntLevel': '', # 'TxtDelimiter': ';', # 'bExport': '', } url = url_base + url_path + '?' url += '&'.join([k + '=' + v for k, v in url_params.iteritems()]) logger.info('Starting to download %s' % url) try: res_request = requests.get(url) except Exception, e: raise Warning( _('Got an error when trying to download the file: %s.') % str(e)) if res_request.status_code != requests.codes.ok: raise Warning( _('Got an error %d when trying to download the file %s.') % (res_request.status_code, url)) logger.info('Download successfully %d bytes' % len(res_request.content)) # Workaround XML: Remove all characters before <?xml pattern = re.compile(r'^.*<\?xml', re.DOTALL) content_fixed = re.sub(pattern, '<?xml', res_request.content) if not re.match(r'<\?xml', content_fixed): raise Warning(_('Downloaded file is not a valid XML file')) return content_fixed
@api.model def create_or_update_nace(self, node): if not self._check_node(node): return False
nace_model = self.env['res.partner.nace'] data = self._mapping(node, False) data.update(self._mapping(node, True)) level = data.get('level', 0) if level >= 2 and level <= 5: data['parent_id'] = self._parents[level - 2] nace = nace_model.search([('level', '=', data['level']), ('code', '=', data['code'])]) if nace: nace.write(data) else: nace = nace_model.create(data) if level >= 1 and level <= 4: self._parents[level - 1] = nace.id return nace
@api.model def translate_nace(self, node, lang): translation_model = self.env['ir.translation'] nace_model = self.env['res.partner.nace'] index = self._mapping(node, False) trans = self._mapping(node, True) nace = nace_model.search([('level', '=', index['level']), ('code', '=', index['code'])]) if nace: for field, value in trans.iteritems(): name = 'res.partner.nace,' + field query = [('res_id', '=', nace.id), ('type', '=', 'model'), ('name', '=', name), ('lang', '=', lang.code)] translation = translation_model.search(query) data = { 'value': value, 'state': 'translated' } if translation: translation_model.write(data) else: data['res_id'] = nace.id data['type'] = 'model' data['name'] = name data['lang'] = lang.code translation_model.create(data)
@api.one def run_import(self): nace_model = self.env['res.partner.nace'] lang_model = self.env['res.lang'] # Available lang list langs = lang_model.search( [('code', 'in', self._available_langs.keys()), ('active', '=', True)]) # All current NACEs, delete if not found above naces_to_delete = nace_model.search([]) # Download NACEs in english, create or update logger.info('Import NACE Rev.2 English') xmlcontent = self._download_nace('EN') dom = etree.fromstring(xmlcontent) for node in dom.iter('Item'): logger.info('Reading level=%s, code=%s' % (node.get('idLevel', 'N/A'), node.get('id', 'N/A'))) nace = self.create_or_update_nace(node) if nace and nace in naces_to_delete: naces_to_delete -= nace # Download NACEs in other languages, translate them for lang in langs: logger.info('Import NACE Rev.2 %s' % lang.code) nace_lang = self._available_langs[lang.code] xmlcontent = self._download_nace(nace_lang) dom = etree.fromstring(xmlcontent) for node in dom.iter('Item'): logger.info('Reading lang=%s, level=%s, code=%s' % (nace_lang, node.get('idLevel', 'N/A'), node.get('id', 'N/A'))) self.translate_nace(node, lang) # Delete obsolete NACEs if naces_to_delete: naces_to_delete.unlink() logger.info('%d NACEs entries deleted' % len(naces_to_delete)) logger.info( 'The wizard to create NACEs entries from RAMON ' 'has been successfully completed.')
return True
|
