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 xml.etree import ElementTree as ET from src.fiscal.constants import COD_RECEITA_DARF from src.fiscal.entities import Empresa 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) -> str: return f"{v:.2f}" @dataclass class DebitoDCTF: """Representa um débito tributário na DCTF.""" codigo_receita: str periodo_apuracao: str # YYYY-MM ou YYYY-TT (trimestre) valor_debito: Decimal valor_suspenso: Decimal = Decimal("0") numero_processo: str = "" tipo_suspensao: str = "" # "D"=Decisão Judicial, "A"=Administrativo @dataclass class PagamentoDCTF: """Pagamento (DARF) vinculado a um débito.""" numero_darf: str data_pagamento: date valor_principal: Decimal valor_multa: Decimal = Decimal("0") valor_juros: Decimal = Decimal("0") @property def valor_total(self) -> Decimal: return self.valor_principal + self.valor_multa + self.valor_juros @dataclass class CompensacaoDCTF: """Compensação de crédito vinculada a um débito.""" numero_per_comp: str # Número do PER/COMP data_transmissao: date valor_compensado: Decimal codigo_receita_credito: str @dataclass class ItemDCTF: """Item de débito com seus pagamentos e compensações.""" debito: DebitoDCTF pagamentos: list[PagamentoDCTF] = field(default_factory=list) compensacoes: list[CompensacaoDCTF] = field(default_factory=list) @property def total_pago(self) -> Decimal: return sum(p.valor_total for p in self.pagamentos) @property def total_compensado(self) -> Decimal: return sum(c.valor_compensado for c in self.compensacoes) @property def saldo_a_pagar(self) -> Decimal: return max( Decimal("0"), self.debito.valor_debito - self.debito.valor_suspenso - self.total_pago - self.total_compensado, ) @dataclass class ConfigDCTF: finalidade: str = "0" # 0=Original, 1=Retificadora situacao_especial: str = "" # Em branco=sem situação especial declarante_pj_inativa: bool = False class GeradorDCTF: """ Gera o arquivo XML da DCTF para importação no PGD DCTF Web. Periodicidade: mensal. Prazo: 15º dia útil do 2º mês subsequente ao período de apuração. Obrigação: pessoas jurídicas com débitos de tributos federais. """ VERSAO = "1.0" def __init__( self, empresa: Empresa, periodo: str, # "YYYY-MM" itens: Optional[list[ItemDCTF]] = None, cfg: Optional[ConfigDCTF] = None, ): self.empresa = empresa self.periodo = periodo self.itens = itens or [] self.cfg = cfg or ConfigDCTF() def _resumo_por_codigo(self) -> dict[str, dict]: resumo: dict[str, dict] = {} for item in self.itens: cod = item.debito.codigo_receita if cod not in resumo: resumo[cod] = { "debito": Decimal("0"), "suspenso": Decimal("0"), "pago": Decimal("0"), "compensado": Decimal("0"), "saldo": Decimal("0"), "nome": COD_RECEITA_DARF.get(cod, cod), } resumo[cod]["debito"] += item.debito.valor_debito resumo[cod]["suspenso"] += item.debito.valor_suspenso resumo[cod]["pago"] += item.total_pago resumo[cod]["compensado"] += item.total_compensado resumo[cod]["saldo"] += item.saldo_a_pagar return resumo def gerar_xml(self) -> str: """Gera o XML da DCTF para importação no PGD DCTF Web.""" root = ET.Element("DCTF") root.set("xmlns:xsi", "http://www.w3.org/2001/XMLSchema-instance") root.set("versao", self.VERSAO) ide = _sub(root, "ideDeclarante") _sub(ide, "CNPJ", self.empresa.cnpj) _sub(ide, "nome", self.empresa.razao_social[:100]) _sub(ide, "perApuracao", self.periodo) _sub(ide, "dtInicio", f"{self.periodo}-01") _sub(ide, "dtFim", f"{self.periodo}-{self._ultimo_dia()}") _sub(ide, "situacaoEspecial", self.cfg.situacao_especial) _sub(ide, "indDeclaranteInativo", "S" if self.cfg.declarante_pj_inativa else "N") _sub(ide, "finalidade", self.cfg.finalidade) for item in self.itens: deb_el = _sub(root, "debito") _sub(deb_el, "codReceita", item.debito.codigo_receita) _sub(deb_el, "perApuracao", item.debito.periodo_apuracao) _sub(deb_el, "vlrDebito", _fmt_dec(item.debito.valor_debito)) if item.debito.valor_suspenso > 0: susp = _sub(deb_el, "suspensao") _sub(susp, "tipoSuspensao", item.debito.tipo_suspensao or "D") _sub(susp, "nrProcesso", item.debito.numero_processo) _sub(susp, "vlrSuspenso", _fmt_dec(item.debito.valor_suspenso)) for pag in item.pagamentos: pag_el = _sub(deb_el, "pagamento") _sub(pag_el, "nrDarf", pag.numero_darf) _sub(pag_el, "dtPagamento", pag.data_pagamento.strftime("%Y-%m-%d")) _sub(pag_el, "vlrPrincipal", _fmt_dec(pag.valor_principal)) _sub(pag_el, "vlrMulta", _fmt_dec(pag.valor_multa)) _sub(pag_el, "vlrJuros", _fmt_dec(pag.valor_juros)) _sub(pag_el, "vlrTotal", _fmt_dec(pag.valor_total)) for comp in item.compensacoes: comp_el = _sub(deb_el, "compensacao") _sub(comp_el, "nrPERCOMP", comp.numero_per_comp) _sub(comp_el, "dtTransmissao", comp.data_transmissao.strftime("%Y-%m-%d")) _sub(comp_el, "vlrCompensado", _fmt_dec(comp.valor_compensado)) _sub(comp_el, "codReceitaCredito", comp.codigo_receita_credito) _sub(deb_el, "vlrSaldoPagar", _fmt_dec(item.saldo_a_pagar)) resumo = self._resumo_por_codigo() total_el = _sub(root, "totalDebitos") _sub(total_el, "vlrTotalDebitos", _fmt_dec(sum(r["debito"] for r in resumo.values()))) _sub(total_el, "vlrTotalSuspenso", _fmt_dec(sum(r["suspenso"] for r in resumo.values()))) _sub(total_el, "vlrTotalPago", _fmt_dec(sum(r["pago"] for r in resumo.values()))) _sub(total_el, "vlrTotalCompensado", _fmt_dec(sum(r["compensado"] for r in resumo.values()))) _sub(total_el, "vlrTotalSaldo", _fmt_dec(sum(r["saldo"] for r in resumo.values()))) ET.indent(root, space=" ") return '\n' + ET.tostring(root, encoding="unicode") def _ultimo_dia(self) -> str: ano, mes = map(int, self.periodo.split("-")) if mes == 12: return "31" ultimo = (date(ano, mes + 1, 1) - date(ano, mes, 1)).days return str(ultimo).zfill(2) def salvar(self, diretorio: str | Path = ".") -> Path: caminho = Path(diretorio) / f"DCTF_{self.empresa.cnpj}_{self.periodo}.xml" caminho.parent.mkdir(parents=True, exist_ok=True) caminho.write_text(self.gerar_xml(), encoding="utf-8") return caminho def relatorio_resumo(self) -> str: resumo = self._resumo_por_codigo() linhas = [ f"DCTF — {self.empresa.razao_social} — Período: {self.periodo}", "=" * 70, ] total_geral = Decimal("0") for cod, dados in resumo.items(): linhas.append( f"{cod} — {dados['nome']}" ) linhas.append(f" Débito: R$ {dados['debito']:>12,.2f}") if dados["suspenso"] > 0: linhas.append(f" Suspenso: R$ {dados['suspenso']:>12,.2f}") if dados["pago"] > 0: linhas.append(f" Pago (DARF): R$ {dados['pago']:>12,.2f}") if dados["compensado"] > 0: linhas.append(f" Compensado: R$ {dados['compensado']:>12,.2f}") linhas.append(f" Saldo: R$ {dados['saldo']:>12,.2f}") total_geral += dados["saldo"] linhas += [ "=" * 70, f"TOTAL A PAGAR: R$ {total_geral:>12,.2f}", ] return "\n".join(linhas) def montar_dctf_do_periodo( empresa: Empresa, periodo: str, valor_irpj: Decimal = Decimal("0"), valor_csll: Decimal = Decimal("0"), valor_pis: Decimal = Decimal("0"), valor_cofins: Decimal = Decimal("0"), valor_cpp: Decimal = Decimal("0"), valor_irrf: Decimal = Decimal("0"), darfs_pagos: Optional[list[PagamentoDCTF]] = None, ) -> GeradorDCTF: """ Função auxiliar: monta a DCTF automaticamente a partir dos valores apurados. """ darfs_pagos = darfs_pagos or [] itens: list[ItemDCTF] = [] pares = [ ("6912", valor_irpj), ("2484", valor_csll), ("8109", valor_pis), ("2172", valor_cofins), ("2100", valor_cpp), ("1038", valor_irrf), ] for cod, valor in pares: if valor > 0: itens.append(ItemDCTF( debito=DebitoDCTF( codigo_receita=cod, periodo_apuracao=periodo, valor_debito=valor, ), pagamentos=[p for p in darfs_pagos], )) return GeradorDCTF(empresa=empresa, periodo=periodo, itens=itens)