[15.0] Fusão de classes herdadas internamente

l10n_br_cnpj_search/models/__init__.py

@@ -3,7 +3,5 @@
 from . import l10n_br_base_party_mixin
 from . import res_config_settings
 from . import cnpj_webservice
-from . import receitaws_webservice
-from . import serpro_webservice
 from . import res_partner
 from . import res_company

l10n_br_cnpj_search/models/cnpj_webservice.py

@@ -1,10 +1,21 @@
 # Copyright 2022 KMEE - Luis Felipe Mileo
 # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
 
+import csv
+import logging
+from os.path import dirname
+
 from erpbrasil.base.misc import punctuation_rm
 
 from odoo import _, api, models
-from odoo.exceptions import ValidationError
+from odoo.exceptions import UserError, ValidationError
+
+_logger = logging.getLogger(__name__)
+
+RECEITAWS_URL = "https://www.receitaws.com.br/v1/cnpj/"
+
+SERPRO_URL = "https://gateway.apiserpro.serpro.gov.br"
+QUALIFICACAO_CSV = dirname(__file__) + "/../data/serpro_qualificacao.csv"
 
 
 class CNPJWebservice(models.AbstractModel):
@@ -119,3 +130,267 @@
             )
 
         return cnae_id
+
+    #
+    # RECEITA WS
+    #
+
+    @api.model
+    def receitaws_get_api_url(self, cnpj):
+        return RECEITAWS_URL + cnpj
+
+    @api.model
+    def receitaws_get_headers(self):
+        return {"Accept": "application/json"}
+
+    @api.model
+    def receitaws_validate(self, response):
+        self._validate(response)
+        data = response.json()
+        if data.get("status") == "ERROR":
+            raise ValidationError(_(data.get("message")))
+
+        return data
+
+    @api.model
+    def _receitaws_import_data(self, data):
+        legal_name = self.get_data(data, "nome", title=True)
+        fantasy_name = self.get_data(data, "fantasia", title=True)
+        phone, mobile = self._receitaws_get_phones(data)
+        state_id, city_id = self._get_state_city(data)
+
+        res = {
+            "legal_name": legal_name,
+            "name": fantasy_name if fantasy_name else legal_name,
+            "email": self.get_data(data, "email", lower=True),
+            "street_name": self.get_data(data, "logradouro", title=True),
+            "street2": self.get_data(data, "complemento", title=True),
+            "district": self.get_data(data, "bairro", title=True),
+            "street_number": self.get_data(data, "numero"),
+            "zip": self.get_data(data, "cep"),
+            "legal_nature": self.get_data(data, "natureza_juridica"),
+            "phone": phone,
+            "mobile": mobile,
+            "state_id": state_id,
+            "city_id": city_id,
+            "equity_capital": self.get_data(data, "capital_social"),
+            "cnae_main_id": self._receitaws_get_cnae(data),
+            "cnae_secondary_ids": self._receitaws_get_secondary_cnae(data),
+        }
+
+        return res
+
+    @api.model
+    def _receitaws_get_phones(self, data):
+        """Get phones from data.
+        If there is more than one phone, the second is assigned to mobile."""
+        phone = False
+        mobile = False
+        if data.get("telefone"):
+            phones = data["telefone"].split("/")
+            phone = phones[0]
+            if len(phones) > 1:
+                mobile = phones[1][1:]  # Remove Empty space separation
+
+        return [phone, mobile]
+
+    @api.model
+    def _get_state_city(self, data):
+        state_id = False
+        city_id = False
+        if data.get("uf"):
+            state = self.env["res.country.state"].search(
+                [
+                    ("code", "=", data["uf"]),
+                    ("country_id.code", "=", "BR"),
+                ],
+                limit=1,
+            )
+            if state.id:
+                state_id = state.id
+
+            if data.get("municipio"):
+                city = self.env["res.city"].search(
+                    [
+                        ("name", "=ilike", data["municipio"].title()),
+                        ("state_id.id", "=", state_id),
+                    ]
+                )
+                if len(city) == 1:
+                    city_id = city.id
+
+        return [state_id, city_id]
+
+    @api.model
+    def _receitaws_get_cnae(self, data):
+        if data.get("atividade_principal"):
+            cnae_main = data.get("atividade_principal")[0]
+            cnae_code = self.get_data(cnae_main, "code")
+            return self._get_cnae(cnae_code)
+        return False
+
+    @api.model
+    def _receitaws_get_secondary_cnae(self, data):
+        cnae_secondary = []
+        for atividade in data.get("atividades_secundarias", []):
+            unformated = self.get_data(atividade, "code").split(".")
+            formatted = ""
+            for nums in unformated:
+                for num in nums.split("-"):
+                    formatted += num
+
+            if self._get_cnae(formatted) is not False:
+                cnae_secondary.append(self._get_cnae(formatted))
+
+        return cnae_secondary
+
+    #
+    # SERPRO
+    #
+
+    @api.model
+    def serpro_get_api_url(self, cnpj):
+        trial = self._get_cnpj_param("serpro_trial")
+        schema = self._get_cnpj_param("serpro_schema")
+
+        if trial:
+            url = SERPRO_URL + f"/consulta-cnpj-df-trial/v2/{schema}/{cnpj}"
+        else:
+            url = SERPRO_URL + f"/v2/{schema}/{cnpj}"
+
+        return url
+
+    @api.model
+    def serpro_get_headers(self):
+        token = self._get_cnpj_param("serpro_token")
+        return {"Authorization": "Bearer " + token}
+
+    @api.model
+    def serpro_validate(self, response):
+        self._validate(response)
+        data = response.json()
+        return data
+
+    @api.model
+    def _serpro_import_data(self, data):
+        schema = self._get_cnpj_param("serpro_schema")
+
+        legal_name = self.get_data(data, "nomeEmpresarial", title=True)
+        fantasy_name = self.get_data(data, "nomeFantasia", title=True)
+        name = fantasy_name if fantasy_name else legal_name
+        phone, mobile = self._serpro_get_phones(data)
+        address = data.get("endereco")
+        nature = data.get("naturezaJuridica")
+        cep = self.get_data(address, "cep")
+
+        res = {
+            "legal_name": legal_name,
+            "name": name,
+            "email": self.get_data(data, "correioEletronico"),
+            "street_name": self.get_data(address, "logradouro", title=True),
+            "street2": self.get_data(address, "complemento", title=True),
+            "district": self.get_data(address, "bairro", title=True),
+            "street_number": self.get_data(address, "numero"),
+            "legal_nature": self.get_data(nature, "codigo", title=True)
+            + self.get_data(nature, "descricao", title=True),
+            "zip": cep,
+            "phone": phone,
+            "mobile": mobile,
+            "state_id": self._get_state_id(address),
+            "city_id": self._get_city_id(cep),
+            "equity_capital": self.get_data(data, "capitalSocial"),
+            "cnae_main_id": self._serpro_get_cnae(data),
+        }
+
+        res.update(self._import_additional_info(data, schema))
+
+        return res
+
+    @api.model
+    def _import_additional_info(self, data, schema):
+        if schema not in ["empresa", "qsa"]:
+            return {}
+
+        partners = data.get("socios")
+        child_ids = []
+        for partner in partners:
+            partner_name = self.get_data(partner, "nome", title=True)
+            partner_qualification = self._get_qualification(partner)
+
+            values = {
+                "name": partner_name,
+                "function": partner_qualification,
+                "company_type": "person",
+            }
+
+            if schema == "empresa":
+                partner_cpf = self.get_data(partner, "cpf")
+                values.update({"cnpj_cpf": partner_cpf})
+
+            partner_id = self.env["res.partner"].create(values).id
+            child_ids.append(partner_id)
+
+        return {
+            "child_ids": [(6, 0, child_ids)],
+        }
+
+    @api.model
+    def _get_qualification(self, partner):
+        qualification = self.get_data(partner, "qualificacao")
+
+        with open(QUALIFICACAO_CSV) as csvfile:
+            reader = csv.reader(csvfile, delimiter=",")
+            next(reader)  # Remove header
+            for row in reader:
+                if row[0] == qualification:
+                    return row[1]
+        return ""
+
+    @api.model
+    def _serpro_get_phones(self, data):
+        """Get phones from data.
+        If there is more than one phone, the second is assigned to mobile and the rest
+        is ignored."""
+        phone = False
+        mobile = False
+        phones_data = data.get("telefones")
+        ddd = phones_data[0].get("ddd")
+        num = phones_data[0].get("numero")
+        phone = f"({ddd}) {num}"
+        if len(phones_data) == 2:
+            ddd = phones_data[1].get("ddd")
+            num = phones_data[1].get("numero")
+            mobile = f"({ddd}) {num}"
+
+        return phone, mobile
+
+    @api.model
+    def _get_state_id(self, address):
+        state_code = self.get_data(address, "uf")
+
+        return (
+            self.env["res.country.state"]
+            .search(
+                [("country_id.code", "=", "BR"), ("code", "=", state_code)], limit=1
+            )
+            .id
+        )
+
+    @api.model
+    def _get_city_id(self, cep):
+        # Get city from cep
+        # TODO Send message if address doesn't match CEP
+        try:
+            cep_values = self.env["l10n_br.zip"]._consultar_cep(cep)
+        except UserError as error:
+            _logger.warning(error.name)
+            return False
+
+        return cep_values.get("city_id")
+
+    @api.model
+    def _serpro_get_cnae(self, data):
+        cnae_main = data.get("cnaePrincipal")
+        cnae_code = self.get_data(cnae_main, "codigo")
+
+        return self._get_cnae(cnae_code)