[IMP] l10n_br_fiscal_dfe: DFe implementation

Author: CristianoMafraJunior

l10n_br_fiscal_dfe/README.rst

@@ -56,11 +56,15 @@ Authors
 Contributors
 ------------
 
--  `KMEE <https://www.kmee.com.br>`__:
+- `KMEE <https://www.kmee.com.br>`__:
 
-   -  Luis Felipe Miléo <mileo@kmee.com.br>
-   -  Gabriel Cardoso <gabriel.cardoso@kmee.com.br>
-   -  Felipe Zago <felipe.zago@kmee.com.br>
+  - Luis Felipe Miléo <mileo@kmee.com.br>
+  - Gabriel Cardoso <gabriel.cardoso@kmee.com.br>
+  - Felipe Zago <felipe.zago@kmee.com.br>
+
+- `Engenere <https://engenere.one>`__:
+
+  - Cristiano Mafra Junior
 
 Maintainers
 -----------

l10n_br_fiscal_dfe/__manifest__.py

@@ -10,9 +10,15 @@
     "website": "https://github.com/OCA/l10n-brazil",
     "depends": ["l10n_br_fiscal", "l10n_br_fiscal_certificate"],
     "data": [
+        # Data
         "data/ir_cron.xml",
+        # Security
+        "security/dfe_security.xml",
         "security/ir.model.access.csv",
+        # Views
+        "views/dfe/dfe_monitor_views.xml",
         "views/dfe/dfe_views.xml",
+        "views/dfe/dfe_access_key.xml",
         "views/l10n_br_fiscal_menu.xml",
         "views/res_company_view.xml",
     ],

l10n_br_fiscal_dfe/constants/dfe.py

@@ -8,3 +8,14 @@
 DFE_ENVIRONMENTS = [("1", "Produção"), ("2", "Homologação")]
 
 DFE_ENVIRONMENT_DEFAULT = "2"
+
+OP_TYPE_ENTRADA = ("0", "Entrada")
+OP_TYPE_SAIDA = ("1", "Saída")
+
+OPERATION_TYPE = [OP_TYPE_ENTRADA, OP_TYPE_SAIDA]
+
+
+SIT_NFE_AUTORIZADA = ("1", "Autorizada")
+SIT_NFE_CANCELADA = ("2", "Cancelada")
+SIT_NFE_DENEGADA = ("3", "Denegada")
+SITUACAO_NFE = [SIT_NFE_AUTORIZADA, SIT_NFE_CANCELADA, SIT_NFE_DENEGADA]

l10n_br_fiscal_dfe/models/__init__.py

@@ -1,4 +1,5 @@
 from . import dfe
-from . import document
 from . import attachment
 from . import res_company
+from . import dfe_monitor
+from . import dfe_access_key

l10n_br_fiscal_dfe/models/dfe.py

@@ -1,157 +1,151 @@
-# Copyright (C) 2023 KMEE Informatica LTDA
-# License AGPL-3 or later (http://www.gnu.org/licenses/agpl)
+# Copyright (C) 2025-Today - Engenere (<https://engenere.one>).
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+import base64
 
-import logging
-import re
+from odoo import fields, models
 
-from erpbrasil.transmissao import TransmissaoSOAP
-from nfelib.nfe.ws.edoc_legacy import NFeAdapter as edoc_nfe
-from requests import Session
-
-from odoo import _, api, fields, models
-
-from ..tools import utils
-
-_logger = logging.getLogger(__name__)
+from ..constants.dfe import (
+    OPERATION_TYPE,
+    SITUACAO_NFE,
+)
 
 
 class DFe(models.Model):
     _name = "l10n_br_fiscal.dfe"
+    _description = "DF-e"
     _inherit = ["mail.thread", "mail.activity.mixin"]
-    _description = "Consult DF-e"
     _order = "id desc"
     _rec_name = "display_name"
 
-    display_name = fields.Char(compute="_compute_display_name")
+    dfe_access_key_id = fields.Many2one(
+        comodel_name="l10n_br_fiscal.dfe_access_key", string="Chave de Acesso"
+    )
+
+    key = fields.Char(string="Access Key", size=44, related="dfe_access_key_id.key")
 
-    company_id = fields.Many2one(comodel_name="res.company", string="Company")
+    serie = fields.Char(size=3, index=True)
 
-    version = fields.Selection(related="company_id.dfe_version")
+    document_number = fields.Float(index=True, digits=(18, 0))
 
-    environment = fields.Selection(related="company_id.dfe_environment")
+    emitter = fields.Char(size=60)
 
-    last_nsu = fields.Char(string="Last NSU", size=25, default="0")
+    cnpj_cpf = fields.Char(string="CNPJ/CPF", size=18)
 
-    last_query = fields.Datetime(string="Last query")
+    nsu = fields.Char(string="NSU", size=25, index=True)
 
-    imported_document_ids = fields.One2many(
-        comodel_name="l10n_br_fiscal.document",
-        inverse_name="dfe_id",
-        string="Imported Documents",
+    operation_type = fields.Selection(
+        selection=OPERATION_TYPE,
     )
 
-    use_cron = fields.Boolean(
-        default=False,
-        string="Download new documents automatically",
-        help="If activated, allows new manifestations to be automatically "
-        "searched with a Cron",
+    document_amount = fields.Float(
+        string="Document Total Value",
+        readonly=True,
+        digits=(18, 2),
     )
 
-    @api.depends("company_id.name", "last_nsu")
-    def name_get(self):
-        return self.mapped(lambda d: (d.id, f"{d.company_id.name} - NSU: {d.last_nsu}"))
-
-    @api.model
-    def _get_processor(self):
-        certificado = self.env.company._get_br_ecertificate()
-        session = Session()
-        session.verify = False
-        return edoc_nfe(
-            TransmissaoSOAP(certificado, session),
-            self.company_id.state_id.ibge_code,
-            versao=self.version,
-            ambiente=self.environment,
-        )
+    ie = fields.Char(string="Inscrição estadual", size=18)
 
-    @api.model
-    def validate_distribution_response(self, result):
-        valid = False
-        message = result.resposta.xMotivo
-        if result.retorno.status_code != 200:
-            code = result.retorno.status_code
-        elif result.resposta.cStat != "138":
-            code = result.resposta.cStat
-        else:
-            valid = True
-
-        if not valid:
-            self.message_post(
-                body=_(
-                    _(
-                        "Error validating document distribution:"
-                        "\n\n%(code)s - %(message)s",
-                        code=code,
-                        message=message,
-                    )
-                )
-            )
+    partner_id = fields.Many2one(
+        comodel_name="res.partner",
+        string="Supplier (partner)",
+    )
 
-        return valid
+    company_id = fields.Many2one(
+        comodel_name="res.company",
+        string="Company",
+        default=lambda self: self.env.company,
+        readonly=True,
+    )
 
-    @api.model
-    def _document_distribution(self):
-        maxNSU = ""
-        while maxNSU != self.last_nsu:
-            try:
-                result = self._get_processor().consultar_distribuicao(
-                    cnpj_cpf=re.sub("[^0-9]", "", self.company_id.cnpj_cpf),
-                    ultimo_nsu=utils.format_nsu(self.last_nsu),
-                )
-            except Exception as e:
-                self.message_post(
-                    body=_("Error on searching documents.\n%(error)s", error=e)
-                )
-                break
+    emission_datetime = fields.Datetime(
+        string="Emission Date",
+        index=True,
+        default=fields.Datetime.now,
+    )
 
-            self.write(
-                {
-                    "last_nsu": result.resposta.ultNSU,
-                    "last_query": fields.Datetime.now(),
-                }
-            )
+    inclusion_datetime = fields.Datetime(
+        string="Inclusion Date",
+        index=True,
+        default=fields.Datetime.now,
+    )
 
-            if not self.validate_distribution_response(result):
-                break
+    inclusion_mode = fields.Char(size=255)
 
-            self._process_distribution(result)
+    document_state = fields.Selection(
+        selection=SITUACAO_NFE,
+        index=True,
+    )
 
-            maxNSU = result.resposta.maxNSU
+    cfop_ids = fields.Many2many(
+        comodel_name="l10n_br_fiscal.cfop",
+        string="CFOPs",
+    )
 
-    @api.model
-    def _process_distribution(self, result):
-        """Method to process the distribution data."""
+    dfe_nfe_document_type = fields.Selection(
+        selection=[
+            ("dfe_nfe_complete", "NF-e Completa"),
+            ("dfe_nfe_summary", "Resumo da NF-e"),
+            ("dfe_nfe_event", "Evento da NF-e"),
+        ],
+        string="DFe Document Type",
+    )
 
-    @api.model
-    def _parse_xml_document(self, document):
-        schema_type = document.schema.split("_")[0]
-        method = "parse_%s" % schema_type
-        if not hasattr(self, method):
-            return
+    dfe_monitor_id = fields.Many2one(
+        comodel_name="l10n_br_fiscal.dfe_monitor",
+        string="DFe Monitor",
+    )
 
-        xml = utils.parse_gzip_xml(document.valueOf_)
-        return getattr(self, method)(xml)
+    attachment_id = fields.Many2one(comodel_name="ir.attachment")
 
-    @api.model
-    def _download_document(self, nfe_key):
-        try:
-            result = self._get_processor().consultar_distribuicao(
-                chave=nfe_key, cnpj_cpf=re.sub("[^0-9]", "", self.company_id.cnpj_cpf)
+    document_id = fields.Many2one(
+        comodel_name="l10n_br_fiscal.document",
+        string="Fiscal Document",
+    )
+
+    def name_get(self):
+        result = []
+        for rec in self:
+            document_type = dict(rec._fields["dfe_nfe_document_type"].selection).get(
+                rec.dfe_nfe_document_type
             )
-        except Exception as e:
-            self.message_post(
-                body=_("Error on searching documents.\n%(error)s", error=e)
+            result.append(
+                (
+                    rec.id,
+                    f"{rec.key} - {document_type}",
+                )
             )
-            return
-
-        if not self.validate_distribution_response(result):
-            return
-
-        return result.resposta.loteDistDFeInt.docZip[0]
+        return result
+
+    def create_xml_attachment(self, xml):
+        file_name = "NFe%s.xml" % self.key
+        self.attachment_id = self.env["ir.attachment"].create(
+            {
+                "name": file_name,
+                "datas": base64.b64encode(xml),
+                "store_fname": file_name,
+                "description": "NFe via Manifesto",
+                "res_model": self._name,
+                "res_id": self.id,
+            }
+        )
 
-    @api.model
-    def _cron_search_documents(self):
-        self.search([("use_cron", "=", True)]).search_documents()
+    def action_download_xml(self):
+        if len(self) == 1:
+            return self.download_attachment(self.attachment_id)
 
-    def search_documents(self):
-        for record in self:
-            record._document_distribution()
+        compressed_attachment_id = (
+            self.env["l10n_br_fiscal.attachment"]
+            .create([])
+            .build_compressed_attachment(self.mapped("attachment_id"))
+        )
+        return self.download_attachment(compressed_attachment_id)
+
+    def download_attachment(self, attachment_id):
+        return {
+            "type": "ir.actions.act_url",
+            "url": (
+                f"/web/content/{attachment_id.id}"
+                f"/{attachment_id.name}?download=true"
+            ),
+            "target": "self",
+        }