from __future__ import annotations from dataclasses import dataclass, field from datetime import date from decimal import Decimal from pathlib import Path from typing import Optional from src.fiscal.entities import Empresa, NotaFiscal, PeriodoApuracao, Produto from src.generators.sped_writer import ArquivoSPED, RegistroSPED, criar_registro VERSAO_LEIAUTE = "017" CODIGO_FINALIDADE = "0" # 0=Original @dataclass class ConfigEFD: ind_perfil: str = "A" ind_atividade: str = "0" versao: str = VERSAO_LEIAUTE finalidade: str = CODIGO_FINALIDADE class GeradorEFDICMSIPI: def __init__(self, periodo: PeriodoApuracao, contador: Optional[object] = None, cfg: Optional[ConfigEFD] = None): self.periodo = periodo self.contador = contador self.cfg = cfg or ConfigEFD() self._arquivo = ArquivoSPED( f"EFD_ICMS_IPI_{periodo.empresa.cnpj}_{periodo.data_inicio.strftime('%Y%m')}.txt" ) self._qt_linhas_bloco: dict[str, int] = {} self._participantes: dict[str, Empresa] = {} self._produtos: dict[str, Produto] = {} def _coletar_participantes_produtos(self) -> None: for nf in self.periodo.notas_saida + self.periodo.notas_entrada: cnpj_dest = nf.destinatario.cnpj if cnpj_dest not in self._participantes: self._participantes[cnpj_dest] = nf.destinatario for item in nf.itens: cod = item.produto.codigo if cod not in self._produtos: self._produtos[cod] = item.produto def _bloco_0(self) -> None: emp = self.periodo.empresa end = emp.endereco self._arquivo.adicionar_linha_raw(criar_registro( "0000", self.cfg.versao, self.cfg.finalidade, self.periodo.data_inicio, self.periodo.data_fim, emp.razao_social, emp.cnpj, end.uf.value, emp.ie, end.cod_municipio, emp.im, emp.suframa, self.cfg.ind_perfil, self.cfg.ind_atividade, )) self._arquivo.adicionar_linha_raw(criar_registro("0001", "0")) self._arquivo.adicionar_linha_raw(criar_registro( "0005", emp.nome_fantasia or emp.razao_social, end.cep, end.logradouro, end.numero, end.complemento, end.bairro, end.municipio, end.uf.value, emp.contato.telefone, emp.contato.fax, emp.contato.email, )) if self.contador: c = self.contador self._arquivo.adicionar_linha_raw(criar_registro( "0100", c.nome, c.cpf, c.crc, c.cnpj_escritorio, getattr(c.endereco, "cep", "") if c.endereco else "", getattr(c.endereco, "logradouro", "") if c.endereco else "", getattr(c.endereco, "numero", "") if c.endereco else "", getattr(c.endereco, "complemento", "") if c.endereco else "", getattr(c.endereco, "bairro", "") if c.endereco else "", c.contato.telefone, c.contato.fax, c.contato.email, getattr(c.endereco, "cod_municipio", "") if c.endereco else "", )) for cnpj, part in self._participantes.items(): self._arquivo.adicionar_linha_raw(criar_registro( "0150", cnpj, part.razao_social, "1058", # Brasil cnpj, "", # CPF part.ie, part.endereco.cod_municipio, part.suframa, part.endereco.logradouro, part.endereco.numero, part.endereco.complemento, part.endereco.bairro, )) for cod, prod in self._produtos.items(): self._arquivo.adicionar_linha_raw(criar_registro( "0200", cod, prod.descricao, prod.cod_barras, "", # cod anterior prod.unidade, prod.tipo_item, prod.ncm, prod.ex_ipi, prod.cod_gen, "", # cod_lst prod.aliq_icms, prod.cest, )) qt = self._arquivo.total_linhas() + 1 self._arquivo.adicionar_linha_raw(criar_registro("0990", qt)) def _bloco_c(self) -> None: self._arquivo.adicionar_linha_raw(criar_registro("C001", "0")) qt_inicio = self._arquivo.total_linhas() for nf in self.periodo.notas_saida + self.periodo.notas_entrada: ind_oper = "1" if nf in self.periodo.notas_saida else "0" ind_emit = "0" if nf in self.periodo.notas_saida else "1" self._arquivo.adicionar_linha_raw(criar_registro( "C100", ind_oper, ind_emit, nf.destinatario.cnpj if ind_oper == "1" else nf.emitente.cnpj, nf.modelo, nf.serie, nf.numero, nf.chave_acesso, nf.data_emissao, nf.data_saida_entrada, nf.valor_produtos, Decimal("0"), # bc_icms nf.valor_icms, Decimal("0"), # bc_icms_st Decimal("0"), # vl_icms_st nf.valor_ipi, Decimal("0"), # vl_pis Decimal("0"), # vl_cofins Decimal("0"), # vl_pis_st Decimal("0"), # vl_cofins_st )) if nf.info_complementar: self._arquivo.adicionar_linha_raw(criar_registro( "C110", "001", nf.info_complementar )) for item in nf.itens: self._arquivo.adicionar_linha_raw(criar_registro( "C170", item.numero_item, item.produto.codigo, item.produto.descricao, item.quantidade, item.produto.unidade, item.valor_unitario, item.desconto, item.cfop, item.produto.ncm, Decimal("0"), # ex_ipi item.produto.cst_icms, item.base_icms, item.produto.aliq_icms, item.valor_icms, Decimal("0"), # bc_icms_st Decimal("0"), # aliq_icms_st Decimal("0"), # vl_icms_st Decimal("0"), # ind_apur item.produto.cst_ipi, item.produto.ex_ipi, item.base_ipi, item.produto.aliq_ipi, item.valor_ipi, item.produto.cst_pis, Decimal("0"), # bc_pis item.produto.aliq_icms, # aliq_pis (simplificado) Decimal("0"), # vl_pis item.produto.cst_cofins, Decimal("0"), # bc_cofins Decimal("0"), # aliq_cofins Decimal("0"), # vl_cofins item.outras_despesas, )) self._arquivo.adicionar_linha_raw(criar_registro( "C190", item.produto.cst_icms if nf.itens else "000", item.cfop if nf.itens else "", item.produto.aliq_icms if nf.itens else Decimal("0"), nf.valor_produtos, nf.valor_icms, Decimal("0"), # vl_bc_icms_st Decimal("0"), # vl_icms_st Decimal("0"), # vl_red_bc nf.valor_ipi, "0", # cod_obs )) qt_fim = self._arquivo.total_linhas() + 1 self._arquivo.adicionar_linha_raw(criar_registro("C990", qt_fim - qt_inicio + 1)) def _bloco_e(self) -> None: """Bloco E - Apuração do ICMS e IPI.""" self._arquivo.adicionar_linha_raw(criar_registro("E001", "0")) self._arquivo.adicionar_linha_raw(criar_registro( "E100", self.periodo.data_inicio, self.periodo.data_fim, )) total_debitos = sum(nf.valor_icms for nf in self.periodo.notas_saida) total_creditos = sum(nf.valor_icms for nf in self.periodo.notas_entrada) saldo_devedor = max(Decimal("0"), total_debitos - total_creditos) saldo_credor = max(Decimal("0"), total_creditos - total_debitos) self._arquivo.adicionar_linha_raw(criar_registro( "E110", total_debitos, # VL_TOT_DEBITOS Decimal("0"), # VL_AJ_DEBITOS Decimal("0"), # VL_TOT_AJ_DEBITOS Decimal("0"), # VL_ESTORNOS_CRED total_creditos, # VL_TOT_CREDITOS Decimal("0"), # VL_AJ_CREDITOS Decimal("0"), # VL_TOT_AJ_CREDITOS Decimal("0"), # VL_ESTORNOS_DEB Decimal("0"), # VL_SLD_CREDOR_ANT saldo_devedor, # VL_SLD_APURADO Decimal("0"), # VL_TOT_DED saldo_devedor, # VL_ICMS_RECOLHER saldo_credor, # VL_SLD_CREDOR_TRANSPORTAR Decimal("0"), # DEB_ESP )) self._arquivo.adicionar_linha_raw(criar_registro( "E500", self.periodo.data_inicio, self.periodo.data_fim, "999", # cod_ind_apur - outros )) total_ipi_debitos = sum(nf.valor_ipi for nf in self.periodo.notas_saida) total_ipi_creditos = sum(nf.valor_ipi for nf in self.periodo.notas_entrada) saldo_ipi = max(Decimal("0"), total_ipi_debitos - total_ipi_creditos) self._arquivo.adicionar_linha_raw(criar_registro( "E510", "50", # CST_IPI total_ipi_debitos, # VL_BC_CONS Decimal("0"), # VL_BC_PROD )) self._arquivo.adicionar_linha_raw(criar_registro( "E520", Decimal("0"), # VL_SD_ANT_IPI total_ipi_debitos, # VL_DEBITOS_IPI total_ipi_creditos, # VL_CREDITOS_IPI Decimal("0"), # VL_OD_IPI saldo_ipi, # VL_OC_IPI saldo_ipi, # VL_IPI_RECOLHER Decimal("0"), # VL_SD_CRED_TRANSPORTAR Decimal("0"), # VL_DEB_ESP )) self._arquivo.adicionar_linha_raw(criar_registro("E990", "0")) def _bloco_h(self) -> None: """Bloco H - Inventário Físico.""" self._arquivo.adicionar_linha_raw(criar_registro("H001", "1")) data_inv = self.periodo.data_fim total_inv = sum(m.valor_total for m in self.periodo.movimentacoes) self._arquivo.adicionar_linha_raw(criar_registro( "H005", data_inv, total_inv, "01" )) for mov in self.periodo.movimentacoes: self._arquivo.adicionar_linha_raw(criar_registro( "H010", mov.produto.codigo, mov.produto.unidade, mov.quantidade, mov.valor_unitario, mov.valor_total, "00", # ind_prop "", # cod_part "", # txt_compl "", # cod_cta "", # vl_item_ir )) self._arquivo.adicionar_linha_raw(criar_registro("H990", "0")) def _bloco_g(self) -> None: """ Bloco G — CIAP: Controle de Crédito do ICMS do Ativo Permanente. Escritura os créditos de ICMS sobre bens do ativo imobilizado (1/48 avos). IND_MOV=1 (sem movimento) quando não há ativos com ICMS a creditar. """ ind_mov = "1" self._arquivo.adicionar_linha_raw(criar_registro("G001", ind_mov)) if ind_mov == "0": # G110 — período de apuração do CIAP self._arquivo.adicionar_linha_raw(criar_registro( "G110", self.periodo.data_inicio, self.periodo.data_fim, "0,00", # VL_ICMS_ANT_P (saldo inicial) "48", # NUM_PARC (número de parcelas — bens do ativo: 48 meses) )) self._arquivo.adicionar_linha_raw(criar_registro( "G125", "", # COD_IND_BEM self.periodo.data_inicio, "AT", # COD_CCUS (centro de custo) "0,00", # VL_ICMS_OP "0,00", # VL_ICMS_ST "0,00", # VL_ICMS_FRT "0,00", # VL_ICMS_DIF "48", # NUM_PARC "0,00", # VL_PARC_PASS )) qt = self._arquivo.total_linhas() + 1 self._arquivo.adicionar_linha_raw(criar_registro("G990", qt)) def _bloco_k(self) -> None: """ Bloco K — Controle da Produção e do Estoque. Obrigatório para industriais/atacadistas acima do limite (IN RFB 1.600/2015). IND_MOV=1 quando não há movimentação de produção no período. """ # IND_MOV: 0=com dados, 1=sem movimento de produção ind_mov = "0" if self.periodo.movimentacoes else "1" self._arquivo.adicionar_linha_raw(criar_registro("K001", ind_mov)) if ind_mov == "0": # K010 — estabelecimento industrial # IND_TIPO_CONTRIBUINTE: 0=industrial, 1=industrial misto, 2=atacadista self._arquivo.adicionar_linha_raw(criar_registro("K010", "0")) # K100 — período de apuração self._arquivo.adicionar_linha_raw(criar_registro( "K100", self.periodo.data_inicio, self.periodo.data_fim, )) # K200 — estoque escriturado por produto for mov in self.periodo.movimentacoes: if hasattr(mov, "produto") and hasattr(mov, "quantidade"): self._arquivo.adicionar_linha_raw(criar_registro( "K200", self.periodo.data_fim, # DT_EST mov.produto.codigo, # COD_ITEM mov.quantidade, # QTD mov.produto.unidade, # UNID "0", # IND_EST: 0=estoque real )) qt = self._arquivo.total_linhas() + 1 self._arquivo.adicionar_linha_raw(criar_registro("K990", qt)) def _bloco_1(self) -> None: """Bloco 1 - Outras Informações.""" self._arquivo.adicionar_linha_raw(criar_registro("1001", "1")) self._arquivo.adicionar_linha_raw(criar_registro("1990", "2")) def _bloco_9(self) -> None: """Bloco 9 - Controle e Encerramento do Arquivo.""" self._arquivo.adicionar_linha_raw(criar_registro("9001", "0")) qt_total = self._arquivo.total_linhas() + 4 # registros de controle por bloco for reg, qt in [("0000", 1), ("C001", 1), ("E001", 1), ("G001", 1), ("H001", 1), ("K001", 1), ("1001", 1), ("9001", 1)]: self._arquivo.adicionar_linha_raw(criar_registro("9900", reg, qt)) self._arquivo.adicionar_linha_raw(criar_registro("9990", qt_total)) self._arquivo.adicionar_linha_raw(criar_registro("9999", qt_total + 1)) def gerar(self, diretorio: str | Path = ".") -> Path: self._coletar_participantes_produtos() self._bloco_0() self._bloco_c() self._bloco_e() self._bloco_g() self._bloco_h() self._bloco_k() self._bloco_1() self._bloco_9() return self._arquivo.salvar(diretorio)