from __future__ import annotations import hashlib import re from dataclasses import dataclass, field from datetime import date, datetime from decimal import Decimal from pathlib import Path from xml.etree import ElementTree as ET from src.fiscal.entities import Empresa NS_CTE = "http://www.portalfiscal.inf.br/cte" VERSAO_CTE = "4.00" 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 _fmt_dec(v: Decimal, casas: int = 2) -> str: return f"{v:.{casas}f}" def _limpar(s: str) -> str: return re.sub(r"[^\w\s\-.,/@]", "", str(s))[:60] # --------------------------------------------------------------------------- # Dataclasses auxiliares # --------------------------------------------------------------------------- @dataclass class CargaCTe: descricao: str produto: str # código produto predominante valor_carga: Decimal peso_kg: Decimal = field(default_factory=lambda: Decimal("0")) volume_m3: Decimal = field(default_factory=lambda: Decimal("0")) @dataclass class DocumentoReferenciado: chave_nfe: str # chave NF-e 44 dígitos @dataclass class DadosTransporte: rntrc: str # Registro Nacional de Transportadores Rodoviários de Cargas placa_veiculo: str uf_veiculo: str data_prevista_entrega: date @dataclass class ParteCTe: # remetente, destinatário, etc. cnpj: str razao_social: str endereco: str municipio: str uf: str cep: str cod_municipio: str # --------------------------------------------------------------------------- # Gerador principal # --------------------------------------------------------------------------- class GeradorCTe: """ Gera o XML de CT-e conforme o leiaute 4.00 da SEFAZ. O XML gerado deve ser assinado digitalmente com certificado ICP-Brasil antes da transmissão. """ VERSAO = "4.00" MOD = "57" _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 __init__( self, emitente: Empresa, remetente: ParteCTe, destinatario: ParteCTe, carga: CargaCTe, documentos: list[DocumentoReferenciado], transporte: DadosTransporte, numero: str = "1", serie: str = "001", cfop: str = "6352", natureza_operacao: str = "PRESTAÇÃO DE SERVIÇO DE TRANSPORTE", aliq_icms: Decimal = Decimal("12"), ambiente: str = "2", ): self.emitente = emitente self.remetente = remetente self.destinatario = destinatario self.carga = carga self.documentos = documentos self.transporte = transporte self.numero = numero self.serie = serie self.cfop = cfop self.natureza_operacao = natureza_operacao self.aliq_icms = aliq_icms self.ambiente = ambiente # 1=Produção, 2=Homologação self._data_emissao: date = date.today() # ------------------------------------------------------------------ # Utilitários internos # ------------------------------------------------------------------ def _cuf(self, uf: str) -> str: return self._CUF_MAP.get(uf.upper(), "35") def _gerar_chave(self) -> str: """ Gera a chave de acesso de 44 dígitos do CT-e. Estrutura: cUF(2) + AAMM(4) + CNPJ(14) + mod(2=57) + serie(3) + nCT(9) + tpEmis(1) + cCT(8) + cDV(1) """ uf = self._cuf(self.emitente.endereco.uf.value) aamm = self._data_emissao.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" c_ct = hashlib.md5(f"{cnpj}{numero}".encode()).hexdigest()[:8].upper() c_ct_digits = "".join(str(int(c, 16) % 10) for c in c_ct)[:8] chave_sem_dv = f"{uf}{aamm}{cnpj}{mod}{serie}{numero}{tp_emis}{c_ct_digits}" 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) def _fmt_dhemi(self) -> str: return ( datetime.combine(self._data_emissao, datetime.min.time()) .strftime("%Y-%m-%dT%H:%M:%S") + "-03:00" ) # ------------------------------------------------------------------ # Blocos XML # ------------------------------------------------------------------ def _ide(self, inf_cte: ET.Element, chave: str) -> None: emit_uf = self.emitente.endereco.uf.value cuf = self._cuf(emit_uf) ide = _sub(inf_cte, "ide") _sub(ide, "cUF", cuf) _sub(ide, "cCT", chave[35:43]) _sub(ide, "CFOP", self.cfop) _sub(ide, "natOp", _limpar(self.natureza_operacao)) _sub(ide, "mod", self.MOD) _sub(ide, "serie", self.serie.zfill(3)) _sub(ide, "nCT", self.numero.zfill(9)) _sub(ide, "dhEmi", self._fmt_dhemi()) _sub(ide, "tpImp", "1") # 1=DACTE normal retrato _sub(ide, "tpEmis", "1") # 1=Emissão normal _sub(ide, "cDV", chave[-1]) _sub(ide, "tpAmb", self.ambiente) _sub(ide, "tpCT", "0") # 0=CT-e normal _sub(ide, "tpServ", "0") # 0=Normal cod_mun_emit = self.emitente.endereco.cod_municipio or "3550308" _sub(ide, "cMunEnv", cod_mun_emit) _sub(ide, "xMunEnv", _limpar(self.emitente.endereco.municipio)) _sub(ide, "UFEnv", emit_uf) _sub(ide, "modal", "01") # 01=Rodoviário _sub(ide, "dhSaidaOrig", self._fmt_dhemi()) _sub(ide, "tpMultimodal", "0") _sub(ide, "UFIni", self.remetente.uf) _sub(ide, "UFFim", self.destinatario.uf) def _compl(self, inf_cte: ET.Element) -> None: compl = _sub(inf_cte, "compl") _sub(compl, "xObs", _limpar(self.carga.descricao)[:160] if self.carga.descricao else "") def _emit(self, inf_cte: ET.Element) -> None: emp = self.emitente end = emp.endereco emit = _sub(inf_cte, "emit") _sub(emit, "CNPJ", emp.cnpj) if emp.ie: _sub(emit, "IE", re.sub(r"\D", "", emp.ie)) _sub(emit, "xNome", _limpar(emp.razao_social)[:60]) if emp.nome_fantasia: _sub(emit, "xFant", _limpar(emp.nome_fantasia)[:60]) ender_emit = _sub(emit, "enderEmit") _sub(ender_emit, "xLgr", _limpar(end.logradouro)) _sub(ender_emit, "nro", end.numero) if end.complemento: _sub(ender_emit, "xCpl", _limpar(end.complemento)) _sub(ender_emit, "xBairro", _limpar(end.bairro)) _sub(ender_emit, "cMun", end.cod_municipio or "3550308") _sub(ender_emit, "xMun", _limpar(end.municipio)) _sub(ender_emit, "CEP", re.sub(r"\D", "", end.cep)) _sub(ender_emit, "UF", end.uf.value) _sub(ender_emit, "cPais", "1058") _sub(ender_emit, "xPais", "Brasil") if emp.contato.telefone: _sub(emit, "fone", re.sub(r"\D", "", emp.contato.telefone)) def _parte(self, inf_cte: ET.Element, tag: str, ender_tag: str, parte: ParteCTe) -> None: el = _sub(inf_cte, tag) _sub(el, "CNPJ", re.sub(r"\D", "", parte.cnpj).zfill(14)) _sub(el, "xNome", _limpar(parte.razao_social)[:60]) ender = _sub(el, ender_tag) _sub(ender, "xLgr", _limpar(parte.endereco)) _sub(ender, "nro", "S/N") _sub(ender, "xBairro", "CENTRO") _sub(ender, "cMun", parte.cod_municipio) _sub(ender, "xMun", _limpar(parte.municipio)) _sub(ender, "CEP", re.sub(r"\D", "", parte.cep).zfill(8)) _sub(ender, "UF", parte.uf) _sub(ender, "cPais", "1058") _sub(ender, "xPais", "Brasil") def _v_prest(self, inf_cte: ET.Element) -> None: v_prest = _sub(inf_cte, "vPrest") _sub(v_prest, "vTPrest", _fmt_dec(self.carga.valor_carga)) _sub(v_prest, "vRec", _fmt_dec(self.carga.valor_carga)) def _imp(self, inf_cte: ET.Element) -> None: imp = _sub(inf_cte, "imp") icms = _sub(imp, "ICMS") icms00 = _sub(icms, "ICMS00") _sub(icms00, "CST", "00") v_bc = self.carga.valor_carga v_icms = (v_bc * self.aliq_icms / Decimal("100")).quantize(Decimal("0.01")) _sub(icms00, "vBC", _fmt_dec(v_bc)) _sub(icms00, "pICMS", _fmt_dec(self.aliq_icms, 4)) _sub(icms00, "vICMS", _fmt_dec(v_icms)) def _inf_cte_norm(self, inf_cte: ET.Element) -> None: inf_cte_norm = _sub(inf_cte, "infCTeNorm") inf_carga = _sub(inf_cte_norm, "infCarga") _sub(inf_carga, "vCarga", _fmt_dec(self.carga.valor_carga)) _sub(inf_carga, "proPred", _limpar(self.carga.produto)[:60]) inf_q = _sub(inf_carga, "infQ") _sub(inf_q, "cUnid", "01") # 01=KG _sub(inf_q, "tpMed", "PESO BRUTO") _sub(inf_q, "qCarga", _fmt_dec(self.carga.peso_kg, 4)) if self.documentos: inf_doc = _sub(inf_cte_norm, "infDoc") for doc in self.documentos: inf_nfe = _sub(inf_doc, "infNFe") _sub(inf_nfe, "chave", doc.chave_nfe) _sub(inf_nfe, "pin", "") inf_modal = _sub(inf_cte_norm, "infModal") inf_modal.set("versaoModal", "3.00") rodo = _sub(inf_modal, "rodo") _sub(rodo, "RNTRC", self.transporte.rntrc) veic = _sub(rodo, "veicTracao") _sub(veic, "placa", self.transporte.placa_veiculo.upper()) _sub(veic, "UF", self.transporte.uf_veiculo.upper()) _sub(veic, "prop") # tag vazia occ = _sub(rodo, "occ") _sub(occ, "serie", "001") _sub(occ, "nOcc", "1") _sub(occ, "dEmi", self._data_emissao.strftime("%Y-%m-%d")) emi_occ = _sub(occ, "emiOcc") _sub(emi_occ, "CNPJ", self.emitente.cnpj) _sub(emi_occ, "xNome", _limpar(self.emitente.razao_social)[:60]) _sub(emi_occ, "cInt", "") _sub(emi_occ, "IE", "") # ------------------------------------------------------------------ # API pública # ------------------------------------------------------------------ def gerar_xml(self) -> str: """Retorna o XML do CT-e não assinado como string.""" chave = self._gerar_chave() cte_proc = ET.Element("cteProc") cte_proc.set("xmlns", NS_CTE) cte_proc.set("versao", self.VERSAO) cte = _sub(cte_proc, "CTe") cte.set("xmlns", NS_CTE) inf_cte = _sub(cte, "infCte") inf_cte.set("versao", self.VERSAO) inf_cte.set("Id", f"CTe{chave}") self._ide(inf_cte, chave) self._compl(inf_cte) self._emit(inf_cte) self._parte(inf_cte, "rem", "enderReme", self.remetente) self._parte(inf_cte, "dest", "enderDest", self.destinatario) self._v_prest(inf_cte) self._imp(inf_cte) self._inf_cte_norm(inf_cte) _sub(cte, "Signature") ET.indent(cte_proc, space=" ") return '\n' + ET.tostring(cte_proc, encoding="unicode") def salvar(self, diretorio: str | Path = ".") -> Path: """Salva o XML em diretório informado e retorna o caminho do arquivo.""" cnpj = self.emitente.cnpj caminho = Path(diretorio) / f"CTe_{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