Skip to content

Commit

Permalink
[REF] l10n_br_nfe: DFe implementation
Browse files Browse the repository at this point in the history
  • Loading branch information
@CristianoMafraJunior
CristianoMafraJunior committed Jan 31, 2025
1 parent 376fced commit 202dc35
Show file tree
Hide file tree
Showing 20 changed files with 579 additions and 708 deletions.
5 changes: 3 additions & 2 deletions l10n_br_nfe/__manifest__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,15 +29,16 @@
"views/res_company_view.xml",
"views/nfe_document_view.xml",
"views/res_config_settings_view.xml",
"views/mde/mde_views.xml",
"views/dfe/dfe_views.xml",
"views/nfe_recipient_manifestation_event/nfe_recipient_manifestation_event_view.xml",
"views/dfe/dfe_access_key_views.xml",
"views/supplier_info_view.xml",
# Report
"report/reports.xml",
"report/danfe_nfce.xml",
"report/danfe_report.xml",
# Wizards
"wizards/import_document.xml",
"wizards/nfe_recipient_manifestation_event_wizard.xml",
# Actions,
"views/nfe_action.xml",
# Menus
Expand Down
20 changes: 0 additions & 20 deletions l10n_br_nfe/constants/mde.py
View file
Original file line number Diff line number Diff line change
@@ -1,34 +1,14 @@
# Copyright (C) 2023 KMEE Informática LTDA
# License AGPL-3 - See http://www.gnu.org/licenses/agpl-3.0.html

SIT_MANIF_PENDENTE = ("pendente", "Pendente")
SIT_MANIF_CIENTE = ("ciente", "Ciente da Operação")
SIT_MANIF_CONFIRMADO = ("confirmado", "Confirmada operação")
SIT_MANIF_DESCONHECIDO = ("desconhecido", "Desconhecimento")
SIT_MANIF_NAO_REALIZADO = ("nao_realizado", "Não realizado")

SCHEMA_RESNFE = ("resNFe", "NFe Resumida")
SCHEMA_RESEVENTO = ("resEvento", "Evento de NFe Resumido")
SCHEMA_PROCNFE = ("procNFe", "NFe Completa")
SCHEMA_PROCEVENTONFE = ("procEventoNFe", "Evento de NFe Completo")

SCHEMAS = [SCHEMA_RESNFE, SCHEMA_RESEVENTO, SCHEMA_PROCNFE, SCHEMA_PROCEVENTONFE]

SITUACAO_MANIFESTACAO = [
SIT_MANIF_PENDENTE,
SIT_MANIF_CIENTE,
SIT_MANIF_CONFIRMADO,
SIT_MANIF_DESCONHECIDO,
SIT_MANIF_NAO_REALIZADO,
]

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]

OP_TYPE_ENTRADA = ("0", "Entrada")
OP_TYPE_SAIDA = ("1", "Saída")

OPERATION_TYPE = [OP_TYPE_ENTRADA, OP_TYPE_SAIDA]
2 changes: 1 addition & 1 deletion l10n_br_nfe/models/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@
from . import res_config_settings
from . import cfop
from . import invalidate_number
from . import nfe_recipient_manifestation_event
from . import dfe
from . import mde

spec_schema = "nfe"
spec_version = "40"
139 changes: 16 additions & 123 deletions l10n_br_nfe/models/dfe.py
View file
Original file line number Diff line number Diff line change
@@ -1,131 +1,24 @@
# Copyright (C) 2023 KMEE Informatica LTDA
# License AGPL-3 or later (http://www.gnu.org/licenses/agpl)

from datetime import datetime

from lxml import objectify
from nfelib.nfe.bindings.v4_0.leiaute_nfe_v4_00 import TnfeProc

from odoo import api, fields, models

from odoo.addons.l10n_br_fiscal_dfe.tools import utils
# Copyright (C) 2025-Today - Engenere (<https://engenere.one>).
# License AGPL-3.0 or later (https://www.gnu.org/licenses/agpl).
from odoo import _, models


class DFe(models.Model):
_inherit = "l10n_br_fiscal.dfe"

mde_ids = fields.One2many(
comodel_name="l10n_br_nfe.mde",
inverse_name="dfe_id",
string="Manifestações do Destinatário Importadas",
)

def _process_distribution(self, result):
for doc in result.resposta.loteDistDFeInt.docZip:
xml = utils.parse_gzip_xml(doc.valueOf_).read()
root = objectify.fromstring(xml)

mde_id = self.env["l10n_br_nfe.mde"].search(
[
("nsu", "=", utils.format_nsu(doc.NSU)),
("company_id", "=", self.company_id.id),
],
limit=1,
def import_document(self):
self.ensure_one()
try:
document = self.dfe_monitor_id._download_document(self.key)
document_id = self.dfe_monitor_id._parse_xml_document(document)
except Exception as e:
self.message_post(
body=_("Error importing document: \n\n %(error)s", error=e)
)
if not mde_id:
mde_id = self._create_mde_from_schema(doc.schema, root)
if mde_id:
mde_id.nsu = doc.NSU
mde_id.create_xml_attachment(xml)

@api.model
def _create_mde_from_schema(self, schema, root):
schema_type = schema.split("_")[0]
method = "_create_mde_from_%s" % schema_type
if not hasattr(self, method):
return
if document_id:
self.document_id = document_id

return getattr(self, method)(root)

@api.model
def _create_mde_from_procNFe(self, root):
nfe_key = root.protNFe.infProt.chNFe
mde_id = self.find_mde_by_key(nfe_key)
if mde_id:
return mde_id

supplier_cnpj = utils.mask_cnpj("%014d" % root.NFe.infNFe.emit.CNPJ)
partner = self.env["res.partner"].search([("cnpj_cpf", "=", supplier_cnpj)])

return self.env["l10n_br_nfe.mde"].create(
{
"number": root.NFe.infNFe.ide.nNF,
"key": nfe_key,
"operation_type": str(root.NFe.infNFe.ide.tpNF),
"document_value": root.NFe.infNFe.total.ICMSTot.vNF,
"state": "pendente",
"inclusion_datetime": datetime.now(),
"cnpj_cpf": supplier_cnpj,
"ie": root.NFe.infNFe.emit.IE,
"partner_id": partner.id,
"emission_datetime": datetime.strptime(
str(root.NFe.infNFe.ide.dhEmi)[:19],
"%Y-%m-%dT%H:%M:%S",
),
"company_id": self.company_id.id,
"dfe_id": self.id,
"inclusion_mode": "Verificação agendada",
"schema": "procNFe",
}
)

@api.model
def _create_mde_from_resNFe(self, root):
nfe_key = root.chNFe
mde_id = self.find_mde_by_key(nfe_key)
if mde_id:
return mde_id

supplier_cnpj = utils.mask_cnpj("%014d" % root.CNPJ)
partner_id = self.env["res.partner"].search([("cnpj_cpf", "=", supplier_cnpj)])

return self.env["l10n_br_nfe.mde"].create(
{
"key": nfe_key,
"emitter": root.xNome,
"operation_type": str(root.tpNF),
"document_value": root.vNF,
"document_state": str(root.cSitNFe),
"state": "pendente",
"inclusion_datetime": datetime.now(),
"cnpj_cpf": supplier_cnpj,
"ie": root.IE,
"partner_id": partner_id.id,
"emission_datetime": datetime.strptime(
str(root.dhEmi)[:19], "%Y-%m-%dT%H:%M:%S"
),
"company_id": self.company_id.id,
"dfe_id": self.id,
"inclusion_mode": "Verificação agendada - manifestada por outro app",
"schema": "resNFe",
}
)

@api.model
def find_mde_by_key(self, key):
mde_id = self.env["l10n_br_nfe.mde"].search([("key", "=", key)])
if not mde_id:
return False

if mde_id not in self.mde_ids:
mde_id.dfe_id = self.id
return mde_id

def import_documents(self):
for record in self:
record.mde_ids.import_document_multi()

@api.model
def parse_procNFe(self, xml):
binding = TnfeProc.from_xml(xml.read().decode())
return self.env["l10n_br_fiscal.document"].import_binding_nfe(binding)
def import_document_multi(self):
for rec in self:
rec.import_document()
13 changes: 13 additions & 0 deletions l10n_br_nfe/models/dfe_access_key.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
# 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 DFeAccessKey(models.Model):
_inherit = "l10n_br_fiscal.dfe_access_key"

nfe_recipient_manifestation_event_ids = fields.One2many(
comodel_name="l10n_br_nfe.recipient_manifestation_event",
inverse_name="dfe_access_key_id",
string="Manifestações do Destinatário Importadas",
)
Loading

0 comments on commit 202dc35

Please sign in to comment.