trabalho em desenvolvimento

Author: CristianoMafraJunior

l10n_br_fiscal/views/cfop_view.xml

@@ -87,6 +87,7 @@
                             name="tax_definition_ids"
                             nolabel="1"
                             context="{'tree_view_ref': 'l10n_br_fiscal.tax_definition_tree','form_view_ref': 'l10n_br_fiscal.tax_definition_form', 'default_cfop_id': id}"
+                            colspan="2"
                         />
                     </group>
                 </sheet>

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

@@ -2,3 +2,5 @@
 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,100 @@
-# 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).
+from odoo import fields, models
 
-import logging
-import re
-
-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"
-    _inherit = ["mail.thread", "mail.activity.mixin"]
-    _description = "Consult DF-e"
-    _order = "id desc"
-    _rec_name = "display_name"
+    _description = "DF-e"
 
-    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"
+    )
 
-    company_id = fields.Many2one(comodel_name="res.company", string="Company")
+    key = fields.Char(string="Access Key", size=44, related="dfe_access_key_id.key")
 
-    version = fields.Selection(related="company_id.dfe_version")
+    serie = fields.Char(size=3, index=True)
 
-    environment = fields.Selection(related="company_id.dfe_environment")
+    number = fields.Float(string="Document Number", index=True, digits=(18, 0))
 
-    last_nsu = fields.Char(string="Last NSU", size=25, default="0")
+    emitter = fields.Char(size=60)
 
-    last_query = fields.Datetime(string="Last query")
+    cnpj_cpf = fields.Char(string="CNPJ/CPF", size=18)
 
-    imported_document_ids = fields.One2many(
-        comodel_name="l10n_br_fiscal.document",
-        inverse_name="dfe_id",
-        string="Imported Documents",
-    )
+    nsu = fields.Char(string="NSU", size=25, index=True)
 
-    use_cron = fields.Boolean(
-        default=False,
-        string="Download new documents automatically",
-        help="If activated, allows new manifestations to be automatically "
-        "searched with a Cron",
+    operation_type = fields.Selection(
+        selection=OPERATION_TYPE,
     )
 
-    @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,
-        )
-
-    @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,
-                    )
-                )
-            )
-
-        return valid
-
-    @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
-
-            self.write(
-                {
-                    "last_nsu": result.resposta.ultNSU,
-                    "last_query": fields.Datetime.now(),
-                }
-            )
+    document_value = fields.Float(
+        string="Document Total Value",
+        readonly=True,
+        digits=(18, 2),
+    )
 
-            if not self.validate_distribution_response(result):
-                break
+    ie = fields.Char(string="Inscrição estadual", size=18)
 
-            self._process_distribution(result)
+    partner_id = fields.Many2one(
+        comodel_name="res.partner",
+        string="Supplier (partner)",
+    )
 
-            maxNSU = result.resposta.maxNSU
+    company_id = fields.Many2one(
+        comodel_name="res.company",
+        string="Company",
+        default=lambda self: self.env.company,
+        readonly=True,
+    )
 
-    @api.model
-    def _process_distribution(self, result):
-        """Method to process the distribution data."""
+    emission_datetime = fields.Datetime(
+        string="Emission Date",
+        index=True,
+        default=fields.Datetime.now,
+    )
 
-    @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
+    inclusion_datetime = fields.Datetime(
+        string="Inclusion Date",
+        index=True,
+        default=fields.Datetime.now,
+    )
 
-        xml = utils.parse_gzip_xml(document.valueOf_)
-        return getattr(self, method)(xml)
+    inclusion_mode = fields.Char(size=255)
 
-    @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)
-            )
-        except Exception as e:
-            self.message_post(
-                body=_("Error on searching documents.\n%(error)s", error=e)
-            )
-            return
+    document_state = fields.Selection(
+        selection=SITUACAO_NFE,
+        index=True,
+    )
 
-        if not self.validate_distribution_response(result):
-            return
+    cfop_ids = fields.Many2many(
+        comodel_name="l10n_br_fiscal.cfop",
+        string="CFOPs",
+    )
 
-        return result.resposta.loteDistDFeInt.docZip[0]
+    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 _cron_search_documents(self):
-        self.search([("use_cron", "=", True)]).search_documents()
+    dfe_monitor_id = fields.Many2one(
+        comodel_name="l10n_br_fiscal.dfe_monitor",
+        string="DFe Monitor",
+    )
 
-    def search_documents(self):
-        for record in self:
-            record._document_distribution()
+    def name_get(self):
+        return [
+            (
+                rec.id,
+                f"{rec.emitter} - {rec.key}",
+            )
+            for rec in self
+        ]

l10n_br_fiscal_dfe/models/dfe_access_key.py

@@ -0,0 +1,37 @@
+# Copyright (C) 2025-Today - Engenere (<https://engenere.one>).
+# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
+from odoo import fields, models
+
+
+class AccessKey(models.Model):
+    _name = "l10n_br_fiscal.dfe_access_key"
+    _description = ""
+
+    key = fields.Char(size=44, required=True)
+
+    dfe_ids = fields.One2many(
+        comodel_name="l10n_br_fiscal.dfe",
+        inverse_name="dfe_access_key_id",
+        string="DFe",
+    )
+
+    emitter = fields.Char(related="dfe_ids.emitter")
+
+    cnpj_cpf = fields.Char(related="dfe_ids.cnpj_cpf")
+
+    document_value = fields.Float(
+        string="Document Total Value", digits=(18, 2), related="dfe_ids.document_value"
+    )
+
+    dfe_monitor_id = fields.Many2one(
+        comodel_name="l10n_br_fiscal.dfe_monitor",
+        string="Monitor de DFe",
+    )
+
+    _sql_constraints = [
+        (
+            "unique_key",
+            "UNIQUE(key)",
+            "The access key already exists",
+        ),
+    ]