src/generators/efd_reinf.py

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 typing import Optional
from xml.etree import ElementTree as ET

NS_REINF = "http://www.reinf.esocial.gov.br/schema/evt"


VERSAO = "2.01.02"


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}"


@dataclass
class PrestadorServico:
    cnpj_cpf: str
    nome: str
    tipo: str = "J"  # J=PJ, F=PF
    valor_bruto: Decimal = Decimal("0")
    valor_bc_csll: Decimal = Decimal("0")
    valor_bc_irrf: Decimal = Decimal("0")
    valor_bc_pis: Decimal = Decimal("0")
    valor_bc_cofins: Decimal = Decimal("0")
    aliq_csll: Decimal = Decimal("1.0")
    aliq_irrf: Decimal = Decimal("1.5")
    aliq_pis: Decimal = Decimal("0.65")
    aliq_cofins: Decimal = Decimal("3.0")
    cod_servico: str = "0001"
    nota_fiscal: str = ""
    data_pagamento: Optional[date] = None

    @property
    def valor_csll(self) -> Decimal:
        return (self.valor_bc_csll * self.aliq_csll / 100).quantize(Decimal("0.01"))

    @property
    def valor_irrf(self) -> Decimal:
        return (self.valor_bc_irrf * self.aliq_irrf / 100).quantize(Decimal("0.01"))

    @property
    def valor_pis(self) -> Decimal:
        return (self.valor_bc_pis * self.aliq_pis / 100).quantize(Decimal("0.01"))

    @property
    def valor_cofins(self) -> Decimal:
        return (self.valor_bc_cofins * self.aliq_cofins / 100).quantize(Decimal("0.01"))

    @property
    def total_retencoes(self) -> Decimal:
        return self.valor_csll + self.valor_irrf + self.valor_pis + self.valor_cofins


@dataclass
class EventoR2010:
    
    cnpj_tomador: str
    periodo: str  # YYYY-MM
    prestadores: list[PrestadorServico] = field(default_factory=list)
    numero_evento: int = 1


@dataclass
class EventoR2020:
    """R-2020: Retenções na fonte — Serviços prestados para PJ."""
    cnpj_prestador: str
    periodo: str  # YYYY-MM
    tomadores: list[PrestadorServico] = field(default_factory=list)
    numero_evento: int = 1


@dataclass
class EventoR2060:
    """R-2060: Contribuição Previdenciária sobre a Receita Bruta (CPRB)."""
    cnpj: str
    periodo: str
    receita_bruta_total: Decimal = Decimal("0")
    receita_bruta_excluida: Decimal = Decimal("0")
    aliq_apur: Decimal = Decimal("0")
    numero_evento: int = 1

    @property
    def base_calculo(self) -> Decimal:
        return self.receita_bruta_total - self.receita_bruta_excluida

    @property
    def valor_contribuicao(self) -> Decimal:
        return (self.base_calculo * self.aliq_apur / 100).quantize(Decimal("0.01"))


class GeradorEFDReinf:
    """
    Gera os eventos da EFD-Reinf em XML conforme os schemas da Receita Federal.

    Eventos implementados:
    - R-1000: Informações do contribuinte
    - R-2010: Retenções — serviços tomados de PJ
    - R-2020: Retenções — serviços prestados para PJ
    - R-2060: CPRB (Contribuição Previdenciária sobre Receita Bruta)
    - R-2099: Fechamento dos eventos periódicos
    """

    def __init__(
        self,
        cnpj: str,
        ambiente: str = "2",  # 1=Produção, 2=Testes
        transmissor_cnpj: Optional[str] = None,
    ):
        self.cnpj = re.sub(r"\D", "", cnpj)
        self.ambiente = ambiente
        self.transmissor_cnpj = transmissor_cnpj or self.cnpj
        self._numero_lote = 1

    def _cabecalho(self, evt: ET.Element, tp_inscricao: str = "1") -> None:
        """Gera o elemento ideContri (identificação do contribuinte)."""
        ide = _sub(evt, "ideContri")
        _sub(ide, "tpInsc", tp_inscricao)  # 1=CNPJ
        _sub(ide, "nrInsc", self.cnpj[:8])  # Raiz do CNPJ

    def _ide_evento(self, evt: ET.Element, nr_evento: int, periodo: str) -> None:
        ide = _sub(evt, "ideEvento")
        _sub(ide, "nrRec", "")  # Número do recibo (preenchido após envio)
        _sub(ide, "perApur", periodo)
        _sub(ide, "tpAmb", self.ambiente)
        _sub(ide, "procEmi", "1")  # 1=Aplicativo do contribuinte
        _sub(ide, "verProc", "1.0")

    def gerar_r1000(self, periodo_inicio: date, periodo_fim: date) -> str:
        """R-1000: Informações do Contribuinte (envio único ou atualização)."""
        root = ET.Element("Reinf")
        root.set("xmlns", NS_REINF)
        root.set("xmlns:xsi", "http://www.w3.org/2001/XMLSchema-instance")

        evt = _sub(root, "evtInfoContri")
        evt.set("id", f"ID1{self.cnpj}{datetime.now().strftime('%Y%m%d%H%M%S')}00001")

        ide = _sub(evt, "ideEvento")
        _sub(ide, "tpAmb", self.ambiente)
        _sub(ide, "procEmi", "1")
        _sub(ide, "verProc", "1.0")

        self._cabecalho(evt)

        info_contrib = _sub(evt, "infoContri")
        inclusao = _sub(info_contrib, "inclusao")
        ide_contrib = _sub(inclusao, "ideContri")
        _sub(ide_contrib, "tpInsc", "1")
        _sub(ide_contrib, "nrInsc", self.cnpj)

        dados = _sub(inclusao, "dadosContri")
        _sub(dados, "classTrib", "99")   # 99=Outros
        _sub(dados, "indEscrituracao", "1")  # 1=Lucro Real
        _sub(dados, "indDesoneracao", "0")   # 0=Não desonerado
        _sub(dados, "indAcordoIsenMulta", "N")
        _sub(dados, "indSitPJ", "0")  # 0=Normal
        _sub(dados, "dtTrans11096", periodo_inicio.strftime("%Y-%m-%d"))

        ET.indent(root, space="  ")
        return ET.tostring(root, encoding="unicode", xml_declaration=True)

    def gerar_r2010(self, evento: EventoR2010) -> str:
        """R-2010: Retenções na fonte — serviços tomados de PJ."""
        root = ET.Element("Reinf")
        root.set("xmlns", NS_REINF)

        evt = _sub(root, "evtServTom")
        evt.set("id", f"ID1{self.cnpj}{datetime.now().strftime('%Y%m%d%H%M%S')}{str(evento.numero_evento).zfill(5)}")

        self._ide_evento(evt, evento.numero_evento, evento.periodo)
        self._cabecalho(evt)

        ide_tomador = _sub(evt, "ideTomador")
        _sub(ide_tomador, "tpInsc", "1")
        _sub(ide_tomador, "nrInsc", re.sub(r"\D", "", evento.cnpj_tomador)[:14])

        for prest in evento.prestadores:
            ide_estab = _sub(evt, "ideEstab")
            _sub(ide_estab, "tpInsc", "1")
            _sub(ide_estab, "nrInsc", re.sub(r"\D", "", evento.cnpj_tomador)[:14])

            ide_prest = _sub(ide_estab, "idePrestServ")
            _sub(ide_prest, "tpInsc", "1" if prest.tipo == "J" else "2")
            _sub(ide_prest, "nrInsc", re.sub(r"\D", "", prest.cnpj_cpf)[:14])

            if prest.nota_fiscal:
                docs = _sub(ide_prest, "ideDocs")
                doc = _sub(docs, "ideDoc")
                _sub(doc, "nrDoc", prest.nota_fiscal)
                _sub(doc, "vlrBruto", _fmt_dec(prest.valor_bruto))

            if prest.total_retencoes > 0:
                info_ret = _sub(ide_prest, "infoRet")
                if prest.valor_csll > 0:
                    _sub(info_ret, "vlrBaseCSLL", _fmt_dec(prest.valor_bc_csll))
                    _sub(info_ret, "vlrCSLL", _fmt_dec(prest.valor_csll))
                if prest.valor_irrf > 0:
                    _sub(info_ret, "vlrBaseIRRF", _fmt_dec(prest.valor_bc_irrf))
                    _sub(info_ret, "vlrIRRF", _fmt_dec(prest.valor_irrf))
                if prest.valor_pis > 0:
                    _sub(info_ret, "vlrBasePIS", _fmt_dec(prest.valor_bc_pis))
                    _sub(info_ret, "vlrPIS", _fmt_dec(prest.valor_pis))
                if prest.valor_cofins > 0:
                    _sub(info_ret, "vlrBaseCOFINS", _fmt_dec(prest.valor_bc_cofins))
                    _sub(info_ret, "vlrCOFINS", _fmt_dec(prest.valor_cofins))

        ET.indent(root, space="  ")
        return ET.tostring(root, encoding="unicode", xml_declaration=True)

    def gerar_r2020(self, evento: EventoR2020) -> str:
        """R-2020: Retenções na fonte — serviços prestados para PJ."""
        root = ET.Element("Reinf")
        root.set("xmlns", NS_REINF)

        evt = _sub(root, "evtServPrest")
        evt.set("id", f"ID1{self.cnpj}{datetime.now().strftime('%Y%m%d%H%M%S')}{str(evento.numero_evento).zfill(5)}")

        self._ide_evento(evt, evento.numero_evento, evento.periodo)
        self._cabecalho(evt)

        ide_prest = _sub(evt, "idePrestServ")
        _sub(ide_prest, "tpInsc", "1")
        _sub(ide_prest, "nrInsc", re.sub(r"\D", "", evento.cnpj_prestador)[:14])

        for tom in evento.tomadores:
            ide_tom = _sub(evt, "ideTomador")
            _sub(ide_tom, "tpInsc", "1" if tom.tipo == "J" else "2")
            _sub(ide_tom, "nrInsc", re.sub(r"\D", "", tom.cnpj_cpf)[:14])

            if tom.total_retencoes > 0:
                info_ret = _sub(ide_tom, "infoRet")
                _sub(info_ret, "vlrBruto", _fmt_dec(tom.valor_bruto))
                if tom.valor_csll > 0:
                    _sub(info_ret, "vlrBaseCSLL", _fmt_dec(tom.valor_bc_csll))
                    _sub(info_ret, "vlrCSLL", _fmt_dec(tom.valor_csll))
                if tom.valor_irrf > 0:
                    _sub(info_ret, "vlrBaseIRRF", _fmt_dec(tom.valor_bc_irrf))
                    _sub(info_ret, "vlrIRRF", _fmt_dec(tom.valor_irrf))
                if tom.valor_pis > 0:
                    _sub(info_ret, "vlrBasePIS", _fmt_dec(tom.valor_bc_pis))
                    _sub(info_ret, "vlrPIS", _fmt_dec(tom.valor_pis))
                if tom.valor_cofins > 0:
                    _sub(info_ret, "vlrBaseCOFINS", _fmt_dec(tom.valor_bc_cofins))
                    _sub(info_ret, "vlrCOFINS", _fmt_dec(tom.valor_cofins))

        ET.indent(root, space="  ")
        return ET.tostring(root, encoding="unicode", xml_declaration=True)

    def gerar_r2060(self, evento: EventoR2060) -> str:
        """R-2060: Contribuição Previdenciária sobre a Receita Bruta (CPRB)."""
        root = ET.Element("Reinf")
        root.set("xmlns", NS_REINF)

        evt = _sub(root, "evtCPRB")
        evt.set("id", f"ID1{self.cnpj}{datetime.now().strftime('%Y%m%d%H%M%S')}{str(evento.numero_evento).zfill(5)}")

        self._ide_evento(evt, evento.numero_evento, evento.periodo)
        self._cabecalho(evt)

        ide_estab = _sub(evt, "ideEstab")
        _sub(ide_estab, "tpInsc", "1")
        _sub(ide_estab, "nrInsc", re.sub(r"\D", "", evento.cnpj)[:14])

        info_cprb = _sub(ide_estab, "infoCPRB")
        _sub(info_cprb, "indAcConstitucionalidade", "N")
        _sub(info_cprb, "vlrRecBrtTotal", _fmt_dec(evento.receita_bruta_total))
        _sub(info_cprb, "vlrRecBrtExclTotal", _fmt_dec(evento.receita_bruta_excluida))
        _sub(info_cprb, "vlrRecBrtBase", _fmt_dec(evento.base_calculo))
        _sub(info_cprb, "aliqApur", _fmt_dec(evento.aliq_apur, 4))
        _sub(info_cprb, "vlrCPRBApur", _fmt_dec(evento.valor_contribuicao))
        _sub(info_cprb, "vlrCPRBSusp", "0.00")
        _sub(info_cprb, "vlrCPRBRec", _fmt_dec(evento.valor_contribuicao))

        ET.indent(root, space="  ")
        return ET.tostring(root, encoding="unicode", xml_declaration=True)

    def gerar_r2099(self, periodo: str, hash_total: str = "") -> str:
        """R-2099: Fechamento dos Eventos Periódicos."""
        root = ET.Element("Reinf")
        root.set("xmlns", NS_REINF)

        evt = _sub(root, "evtFechamento")
        evt.set("id", f"ID1{self.cnpj}{datetime.now().strftime('%Y%m%d%H%M%S')}99999")

        self._ide_evento(evt, 99999, periodo)
        self._cabecalho(evt)

        ideRespInf = _sub(evt, "ideRespInf")
        _sub(ideRespInf, "nrInsc", self.transmissor_cnpj[:14])
        _sub(ideRespInf, "nome", "RESPONSAVEL INFORMACAO")

        fechamento = _sub(evt, "fechamento")
        _sub(fechamento, "evtServTom", "S")     # Tem R-2010?
        _sub(fechamento, "evtServPrest", "N")   # Tem R-2020?
        _sub(fechamento, "evtAssocDespRec", "N")
        _sub(fechamento, "evtCPRB", "N")
        _sub(fechamento, "evtRRA", "N")
        _sub(fechamento, "evtRecurso", "N")
        _sub(fechamento, "evtComProd", "N")
        _sub(fechamento, "evtInfoComplPer", "N")

        ET.indent(root, space="  ")
        return ET.tostring(root, encoding="unicode", xml_declaration=True)

    def salvar_eventos(
        self,
        diretorio: str | Path,
        eventos_r2010: Optional[list[EventoR2010]] = None,
        eventos_r2020: Optional[list[EventoR2020]] = None,
        eventos_r2060: Optional[list[EventoR2060]] = None,
        periodo: str = "",
    ) -> list[Path]:
        diretorio = Path(diretorio)
        diretorio.mkdir(parents=True, exist_ok=True)
        arquivos = []

        
        r1000 = diretorio / f"R1000_{self.cnpj}_{periodo}.xml"
        r1000.write_text(self.gerar_r1000(date.today(), date.today()), encoding="utf-8")
        arquivos.append(r1000)

        
        for i, evt in enumerate(eventos_r2010 or [], 1):
            evt.numero_evento = i
            p = diretorio / f"R2010_{self.cnpj}_{periodo}_{i:04d}.xml"
            p.write_text(self.gerar_r2010(evt), encoding="utf-8")
            arquivos.append(p)

        
        for i, evt in enumerate(eventos_r2020 or [], 1):
            evt.numero_evento = i
            p = diretorio / f"R2020_{self.cnpj}_{periodo}_{i:04d}.xml"
            p.write_text(self.gerar_r2020(evt), encoding="utf-8")
            arquivos.append(p)

        
        for i, evt in enumerate(eventos_r2060 or [], 1):
            evt.numero_evento = i
            p = diretorio / f"R2060_{self.cnpj}_{periodo}_{i:04d}.xml"
            p.write_text(self.gerar_r2060(evt), encoding="utf-8")
            arquivos.append(p)

        
        r2099 = diretorio / f"R2099_{self.cnpj}_{periodo}.xml"
        r2099.write_text(self.gerar_r2099(periodo), encoding="utf-8")
        arquivos.append(r2099)

        return arquivos