1from __future__ import annotations23import asyncio4from dataclasses import dataclass, field5from datetime import date6from decimal import Decimal78from intellevo.sap import ODataClient, RfcError9from intellevo.fiscal import Evento, Regra, apurar_lote1011LOTE = 25012TENTATIVAS = 4131415@dataclass(slots=True)16class Apuracao:17 """Acumula os eventos de uma competencia por filial."""1819 competencia: date20 filial: str21 eventos: list[Evento] = field(default_factory=list)22 total: Decimal = Decimal("0.00")2324 @property25 def vazia(self) -> bool:26 return not self.eventos2728 def acumular(self, evento: Evento, regra: Regra) -> None:29 if evento.cancelado or not regra.alcanca(evento):30 return31 self.eventos.append(evento)32 self.total += evento.base * regra.aliquota(evento.cfop)333435async def coletar(cliente: ODataClient, filial: str, quando: date) -> list[Evento]:36 pagina, coletados = 0, []37 while True:38 try:39 lote = await cliente.get(40 "/ZFI_EVENTOS",41 params={"filial": filial, "skip": pagina * LOTE},42 )43 except RfcError as erro:44 if pagina >= TENTATIVAS:45 raise46 await asyncio.sleep(2 ** pagina)47 continue48 coletados.extend(Evento.de_odata(linha) for linha in lote["value"])49 if len(lote["value"]) < LOTE:50 return coletados51 pagina += 1525354async def fechar_competencia(filiais: list[str], quando: date) -> dict[str, Decimal]:55 async with ODataClient.do_ambiente() as cliente:56 tarefas = [coletar(cliente, filial, quando) for filial in filiais]57 resultados = await asyncio.gather(*tarefas)5859 fechamento: dict[str, Decimal] = {}60 for filial, eventos in zip(filiais, resultados):61 apuracao = Apuracao(competencia=quando, filial=filial)62 for evento in eventos:63 apuracao.acumular(evento, Regra.vigente_em(quando))64 fechamento[filial] = apurar_lote(apuracao)65 return fechamento
1from __future__ import annotations23import asyncio4from dataclasses import dataclass, field5from datetime import date6from decimal import Decimal78from intellevo.sap import ODataClient, RfcError9from intellevo.fiscal import Evento, Regra, apurar_lote1011LOTE = 25012TENTATIVAS = 4131415@dataclass(slots=True)16class Apuracao:17 """Acumula os eventos de uma competencia por filial."""1819 competencia: date20 filial: str21 eventos: list[Evento] = field(default_factory=list)22 total: Decimal = Decimal("0.00")2324 @property25 def vazia(self) -> bool:26 return not self.eventos2728 def acumular(self, evento: Evento, regra: Regra) -> None:29 if evento.cancelado or not regra.alcanca(evento):30 return31 self.eventos.append(evento)32 self.total += evento.base * regra.aliquota(evento.cfop)333435async def coletar(cliente: ODataClient, filial: str, quando: date) -> list[Evento]:36 pagina, coletados = 0, []37 while True:38 try:39 lote = await cliente.get(40 "/ZFI_EVENTOS",41 params={"filial": filial, "skip": pagina * LOTE},42 )43 except RfcError as erro:44 if pagina >= TENTATIVAS:45 raise46 await asyncio.sleep(2 ** pagina)47 continue48 coletados.extend(Evento.de_odata(linha) for linha in lote["value"])49 if len(lote["value"]) < LOTE:50 return coletados51 pagina += 1525354async def fechar_competencia(filiais: list[str], quando: date) -> dict[str, Decimal]:55 async with ODataClient.do_ambiente() as cliente:56 tarefas = [coletar(cliente, filial, quando) for filial in filiais]57 resultados = await asyncio.gather(*tarefas)5859 fechamento: dict[str, Decimal] = {}60 for filial, eventos in zip(filiais, resultados):61 apuracao = Apuracao(competencia=quando, filial=filial)62 for evento in eventos:63 apuracao.acumular(evento, Regra.vigente_em(quando))64 fechamento[filial] = apurar_lote(apuracao)65 return fechamento
