from __future__ import annotations import hashlib import random import re from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from xml.etree import ElementTree as ET from src.fiscal.entities import Empresa NS_MDFE = "http://www.portalfiscal.inf.br/mdfe" _CUF_MAP = { "AC": "12", "AL": "27", "AP": "16", "AM": "13", "BA": "29", "CE": "23", "DF": "53", "ES": "32", "GO": "52", "MA": "21", "MT": "51", "MS": "50", "MG": "31", "PA": "15", "PB": "25", "PR": "41", "PE": "26", "PI": "22", "RJ": "33", "RN": "24", "RS": "43", "RO": "11", "RR": "14", "SC": "42", "SP": "35", "SE": "28", "TO": "17", } def _sub(pai: ET.Element, tag: str, texto: str = "") -> ET.Element: el = ET.SubElement(pai, tag) if texto: el.text = str(texto) return el def _limpar(s: str) -> str: return re.sub(r"[^\w\s\-.,/@]", "", str(s))[:60] def _cuf(uf: str) -> str: return _CUF_MAP.get(uf.upper(), "35") @dataclass class DocumentoMDFe: """Referência a CT-e ou NF-e incluído no MDF-e.""" chave: str # 44 dígitos tipo: str # "CTe" ou "NFe" @dataclass class MunicipioDescarga: cod_municipio: str nome_municipio: str uf: str documentos: list[DocumentoMDFe] = field(default_factory=list) @dataclass class ConductorMDFe: nome: str cpf: str @dataclass class SeguroMDFe: resp_seg: str # "1"=emitente, "2"=contratante cnpj_seg: str # CNPJ seguradora nome_seg: str nApol: str # número apólice nCT: list[str] = field(default_factory=list) # números averbação class GeradorMDFe: VERSAO = "3.00" MOD = "58" def __init__( self, emitente: Empresa, municipios_descarrega: list[MunicipioDescarga], condutores: list[ConductorMDFe], seguro: SeguroMDFe, uf_ini: str, uf_fim: str, placa_veiculo: str, uf_veiculo: str, rntrc: str, numero: str = "1", serie: str = "1", ambiente: str = "2", ): self.emitente = emitente self.municipios_descarrega = municipios_descarrega self.condutores = condutores self.seguro = seguro self.uf_ini = uf_ini.upper() self.uf_fim = uf_fim.upper() self.placa_veiculo = placa_veiculo.upper() self.uf_veiculo = uf_veiculo.upper() self.rntrc = rntrc self.numero = numero self.serie = serie self.ambiente = ambiente # 1=Produção, 2=Homologação self._dhEmi = datetime.now(tz=timezone.utc).astimezone().strftime("%Y-%m-%dT%H:%M:%S") + "-03:00" # ------------------------------------------------------------------ # Chave de acesso # ------------------------------------------------------------------ def _gerar_chave(self) -> str: """Gera a chave de acesso de 44 dígitos do MDF-e (módulo 11).""" uf = _cuf(self.emitente.endereco.uf.value) aamm = datetime.now().strftime("%y%m") cnpj = self.emitente.cnpj.zfill(14) mod = self.MOD.zfill(2) serie = self.serie.zfill(3) numero = self.numero.zfill(9) tp_emis = "1" # Emissão normal # cMDF — 8 dígitos aleatórios c_mdf = str(random.randint(10000000, 99999999)) chave_sem_dv = f"{uf}{aamm}{cnpj}{mod}{serie}{numero}{tp_emis}{c_mdf}" # Módulo 11 para dígito verificador (mesmo algoritmo da NF-e) pesos = [2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 4] soma = sum(int(chave_sem_dv[i]) * pesos[i] for i in range(43)) resto = soma % 11 dv = 0 if resto in (0, 1) else 11 - resto return chave_sem_dv + str(dv) # ------------------------------------------------------------------ # Blocos XML # ------------------------------------------------------------------ def _ide(self, inf_mdfe: ET.Element, chave: str) -> None: ide = _sub(inf_mdfe, "ide") _sub(ide, "cUF", _cuf(self.emitente.endereco.uf.value)) _sub(ide, "tpAmb", self.ambiente) _sub(ide, "tpEmit", "1") # 1=Transportador _sub(ide, "tpTransp", "2") # 2=Carga própria _sub(ide, "mod", self.MOD) _sub(ide, "serie", self.serie.zfill(3)) _sub(ide, "nMDF", self.numero.zfill(9)) _sub(ide, "cMDF", chave[35:43]) _sub(ide, "cDV", chave[-1]) _sub(ide, "modal", "01") # 01=Rodoviário _sub(ide, "dhEmi", self._dhEmi) _sub(ide, "tpEmis", "1") # 1=Emissão normal _sub(ide, "procEmi", "0") # 0=Aplicativo do contribuinte _sub(ide, "verProc", "1.0.0") _sub(ide, "UFIni", self.uf_ini) _sub(ide, "UFFim", self.uf_fim) _sub(ide, "dhIniViagem", self._dhEmi) def _emit(self, inf_mdfe: ET.Element) -> None: emp = self.emitente end = emp.endereco emit = _sub(inf_mdfe, "emit") _sub(emit, "CNPJ", emp.cnpj.zfill(14)) if emp.ie: _sub(emit, "IE", re.sub(r"\D", "", emp.ie)) _sub(emit, "xNome", _limpar(emp.razao_social)[:60]) ender = _sub(emit, "enderEmit") _sub(ender, "xLgr", _limpar(end.logradouro)) _sub(ender, "nro", end.numero) _sub(ender, "xBairro", _limpar(end.bairro)) _sub(ender, "cMun", end.cod_municipio or "3550308") _sub(ender, "xMun", _limpar(end.municipio)) _sub(ender, "CEP", re.sub(r"\D", "", end.cep)) _sub(ender, "UF", end.uf.value) _sub(ender, "cPais", "1058") _sub(ender, "xPais", "Brasil") if emp.contato.telefone: _sub(ender, "fone", re.sub(r"\D", "", emp.contato.telefone)) def _inf_modal(self, inf_mdfe: ET.Element) -> None: inf_modal = _sub(inf_mdfe, "infModal") inf_modal.set("versaoModal", self.VERSAO) rodo = _sub(inf_modal, "rodo") _sub(rodo, "RNTRC", self.rntrc) veic = _sub(rodo, "veicTracao") _sub(veic, "cInt", "") _sub(veic, "placa", self.placa_veiculo) _sub(veic, "tara", "0") _sub(veic, "tpRod", "06") # 06=Caminhão _sub(veic, "tpCar", "00") # 00=Não aplicável _sub(veic, "UF", self.uf_veiculo) for condutor in self.condutores: cond_el = _sub(rodo, "condutor") _sub(cond_el, "xNome", _limpar(condutor.nome)[:60]) _sub(cond_el, "CPF", re.sub(r"\D", "", condutor.cpf).zfill(11)) def _inf_doc(self, inf_mdfe: ET.Element) -> None: inf_doc = _sub(inf_mdfe, "infDoc") for mun in self.municipios_descarrega: mun_el = _sub(inf_doc, "infMunDescarga") _sub(mun_el, "cMunDescarga", mun.cod_municipio) _sub(mun_el, "xMunDescarga", _limpar(mun.nome_municipio)[:60]) for doc in mun.documentos: if doc.tipo == "CTe": doc_el = _sub(mun_el, "infCTe") _sub(doc_el, "chCTe", doc.chave) else: doc_el = _sub(mun_el, "infNFe") _sub(doc_el, "chNFe", doc.chave) def _seg(self, inf_mdfe: ET.Element) -> None: seg = self.seguro seg_el = _sub(inf_mdfe, "seg") _sub(seg_el, "respSeg", seg.resp_seg) _sub(seg_el, "xSeg", _limpar(seg.nome_seg)[:60]) _sub(seg_el, "CNPJ", re.sub(r"\D", "", seg.cnpj_seg).zfill(14)) _sub(seg_el, "nApol", seg.nApol[:20]) for numero_ct in seg.nCT: _sub(seg_el, "nCT", numero_ct[:20]) inf_carga = _sub(seg_el, "infCarga") _sub(inf_carga, "xProd", "CARGA GERAL") def _tot(self, inf_mdfe: ET.Element) -> None: todos_docs = [d for m in self.municipios_descarrega for d in m.documentos] q_cte = sum(1 for d in todos_docs if d.tipo == "CTe") q_nfe = sum(1 for d in todos_docs if d.tipo == "NFe") tot = _sub(inf_mdfe, "tot") _sub(tot, "qCTe", str(q_cte)) _sub(tot, "qNFe", str(q_nfe)) _sub(tot, "qMDFe", "0") _sub(tot, "vCarga", "0.00") _sub(tot, "cUnid", "01") # 01=KG _sub(tot, "qCarga", "0.0000") # ------------------------------------------------------------------ # Geração e persistência # ------------------------------------------------------------------ def gerar_xml(self) -> str: chave = self._gerar_chave() raiz = ET.Element("mdfeProc") raiz.set("xmlns", NS_MDFE) raiz.set("versao", self.VERSAO) mdfe = _sub(raiz, "MDFe") mdfe.set("xmlns", NS_MDFE) inf_mdfe = _sub(mdfe, "infMDFe") inf_mdfe.set("versao", self.VERSAO) inf_mdfe.set("Id", f"MDFe{chave}") self._ide(inf_mdfe, chave) self._emit(inf_mdfe) self._inf_modal(inf_mdfe) self._inf_doc(inf_mdfe) self._seg(inf_mdfe) self._tot(inf_mdfe) _sub(inf_mdfe, "Signature") ET.indent(raiz, space=" ") return '\n' + ET.tostring(raiz, encoding="unicode") def salvar(self, diretorio: str | Path = ".") -> Path: caminho = Path(diretorio) / f"MDFe_{self.emitente.cnpj}_{self.numero.zfill(9)}.xml" caminho.parent.mkdir(parents=True, exist_ok=True) caminho.write_text(self.gerar_xml(), encoding="utf-8") return caminho