from decimal import Decimal, InvalidOperation from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from custos_precificacao import impostos from custos_precificacao.api.deps import get_session from custos_precificacao.api.schemas import ( AliquotaVigenciaIn, AliquotaVigenciaOut, AliquotasOut, ClasseCreditoOut, ClassificacaoIn, CompraIcmsOut, ClassificacaoLoteIn, ClassificacaoOut, DesenvolveIn, DesenvolveOut, DestinoIcmsIn, DestinoIcmsOut, FaixaIcmsIn, FaixaIcmsOut, IcmsOut, ItemClassificavelOut, MixCenarioOut, MixIn, MixLinhaOut, OpcaoOut, ResultadoDesenvolveOut, SecaoClassificacaoOut, SimulacaoDesenvolveIn, ) from custos_precificacao.models import CenarioPrecificacao, Planejamento router = APIRouter(tags=["impostos"]) def _obter(session: Session, modelo, codigo: str, rotulo: str): instancia = session.query(modelo).filter_by(codigo=codigo).one_or_none() if instancia is None: raise HTTPException(status_code=404, detail=f"{rotulo} {codigo!r} não encontrado") return instancia def _cenarios(session: Session) -> dict[int, CenarioPrecificacao]: return {c.id: c for c in session.query(CenarioPrecificacao).order_by(CenarioPrecificacao.codigo)} # --- Classificação dos créditos --------------------------------------------------------------------------------------------------------- def _item_out(i: impostos.ItemClassificavel) -> ItemClassificavelOut: return ItemClassificavelOut( origem=i.origem, origem_id=i.origem_id, elemento_codigo=i.elemento_codigo, elemento_nome=i.elemento_nome, descricao=i.descricao, base=i.base, classes=[ClasseCreditoOut(classe=c, aliquota_pct=m.aliquota, faixa=m.faixa) for c, m in sorted(i.classes.items())], ) @router.get("/impostos/classificacao", response_model=ClassificacaoOut) def obter_classificacao(session: Session = Depends(get_session)) -> ClassificacaoOut: """Todos os itens que podem gerar crédito (insumos, gastos de mão de obra/instalações/máquinas, despesas, revenda) por sessão, com as classes marcadas.""" itens = impostos.listar_itens_classificaveis(session) return ClassificacaoOut(secoes=[ SecaoClassificacaoOut(chave=chave, titulo=titulo, itens=[_item_out(i) for i in itens if i.secao == chave]) for chave, titulo in impostos.SECOES ]) @router.get("/impostos/classificacao/itens", response_model=list[ItemClassificavelOut]) def itens_do_cadastro(secao: str, codigo: str, session: Session = Depends(get_session)) -> list[ItemClassificavelOut]: """Os itens classificáveis de um cadastro (o insumo, os gastos de uma máquina/instalação/colaborador, a despesa…) para marcar as classes no próprio cadastro.""" if secao not in dict(impostos.SECOES): raise HTTPException(status_code=422, detail=f"Sessão {secao!r} inválida. Opções: {', '.join(dict(impostos.SECOES))}") return [_item_out(i) for i in impostos.listar_itens_classificaveis(session) if i.secao == secao and i.elemento_codigo == codigo] @router.put("/impostos/classificacao/{origem}/{origem_id}", response_model=list[ClasseCreditoOut]) def definir_classificacao(origem: str, origem_id: int, corpo: ClassificacaoIn, session: Session = Depends(get_session)) -> list[ClasseCreditoOut]: """Define (substitui) as classes de crédito de um item; lista vazia = o item não gera crédito.""" try: impostos.definir_classes_do_item(session, origem, origem_id, {c.classe: {"aliquota_pct": c.aliquota_pct, "faixa": c.faixa} for c in corpo.classes}) session.commit() except ValueError as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc return [ClasseCreditoOut(classe=c, aliquota_pct=m.aliquota, faixa=m.faixa) for c, m in sorted(impostos.classes_do_item(session, origem, origem_id).items())] @router.post("/impostos/classificacao/lote") def marcar_classificacao_em_lote(corpo: ClassificacaoLoteIn, session: Session = Depends(get_session)) -> dict[str, int]: """Liga ou desliga uma classe de crédito em vários itens de uma vez (os "visíveis" da tela).""" try: mudou = impostos.marcar_lote(session, [(i.origem, i.id) for i in corpo.itens], corpo.classe, corpo.marcar) session.commit() except ValueError as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc return {"alterados": mudou} # --- Alíquotas por vigência (CBS, IR/CSLL, crédito padrão do ICMS) ------------------------------------------------------------------------ def _mes(data) -> str: return f"{data.year:04d}-{data.month:02d}" if data else None def _vigencias(linhas) -> list[AliquotaVigenciaOut]: return [AliquotaVigenciaOut(vigencia=_mes(a.vigencia), aliquota_pct=a.aliquota_pct, nota=a.nota) for a in linhas] @router.get("/impostos/aliquotas", response_model=AliquotasOut) def listar_aliquotas(session: Session = Depends(get_session)) -> AliquotasOut: por_classe = impostos.listar_aliquotas(session) return AliquotasOut( cbs=_vigencias(por_classe["cbs"]), ir_csll=_vigencias(por_classe["ir_csll"]), icms_credito=_vigencias(por_classe["icms_credito"]), competencia=_mes(impostos.competencia_de_referencia(session)), ) @router.put("/impostos/aliquotas/{classe}/{vigencia}", response_model=AliquotaVigenciaOut) def definir_aliquota_vigencia(classe: str, vigencia: str, corpo: AliquotaVigenciaIn, session: Session = Depends(get_session)) -> AliquotaVigenciaOut: try: linha = impostos.definir_aliquota(session, classe, vigencia, corpo.aliquota_pct, corpo.nota) session.commit() except ValueError as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc return _vigencias([linha])[0] @router.delete("/impostos/aliquotas/{classe}/{vigencia}", status_code=204) def remover_aliquota_vigencia(classe: str, vigencia: str, session: Session = Depends(get_session)) -> None: try: impostos.remover_aliquota(session, classe, vigencia) session.commit() except ValueError as exc: session.rollback() raise HTTPException(status_code=404, detail=str(exc)) from exc # --- ICMS: destinos, mix por Centro de Resultado e Desenvolve-BA ------------------------------------------------------------------------ def _destino_out(d) -> DestinoIcmsOut: return DestinoIcmsOut( uf=d.uf, nome=d.nome, faixa=d.faixa, aliquota_pct=d.aliquota_pct, reducao_base_pct=d.reducao_base_pct, frete_adicional_pct=d.frete_adicional_pct, ) def _desenvolve_registro_out(d) -> DesenvolveOut: return DesenvolveOut( ativo=d.ativo, pct_dilatado=d.pct_dilatado, pct_pago_a_vista=d.pct_pago_a_vista, pct_fundo_pobreza=d.pct_fundo_pobreza, plano_de_contas_codigo=d.plano_de_contas.codigo if d.plano_de_contas else None, plano_de_contas_nome=d.plano_de_contas.nome if d.plano_de_contas else None, observacao=d.observacao, ) def _desenvolve_out(r) -> ResultadoDesenvolveOut | None: if r is None: return None return ResultadoDesenvolveOut( saldo_devedor=r.saldo_devedor, fundo_pobreza=r.fundo_pobreza, parcela_dilatada=r.parcela_dilatada, parcela_normal=r.parcela_normal, pago_a_vista=r.pago_a_vista, beneficio=r.beneficio, icms_a_pagar=r.icms_a_pagar, debito_icms=r.debito_icms, credito_icms=r.credito_icms, ) def _mix_do_centro(session: Session, c: CenarioPrecificacao) -> MixCenarioOut: mix = impostos.obter_mix(session, c.id) return MixCenarioOut( cenario_codigo=c.codigo, cenario_nome=c.nome, linhas=[MixLinhaOut(uf=uf, percentual=p) for uf, p in mix.items()], total=sum(mix.values(), Decimal(0)), aliquota_efetiva=impostos.aliquota_icms_do_centro(session, c.id), ) def _icms_out(session: Session) -> IcmsOut: destinos = impostos.listar_destinos(session) compras = impostos.listar_compras(session) desenvolve = impostos.obter_desenvolve(session) session.commit() # grava o que nasceu agora (UFs e Desenvolve de partida) return IcmsOut( faixas=[FaixaIcmsOut(faixa=f, titulo=t) for f, t in impostos.FAIXAS.items()], destinos=[_destino_out(d) for d in destinos], compras=[CompraIcmsOut(faixa=c.faixa, titulo=impostos.FAIXAS[c.faixa], aliquota_pct=c.aliquota_pct) for c in compras], mix=[_mix_do_centro(session, c) for c in session.query(CenarioPrecificacao).order_by(CenarioPrecificacao.codigo)], desenvolve=_desenvolve_registro_out(desenvolve), ) @router.get("/impostos/icms", response_model=IcmsOut) def obter_icms(session: Session = Depends(get_session)) -> IcmsOut: """Tudo do ICMS da venda: faixas, UFs com alíquota, mix de destinos de cada Centro de Resultado e o Desenvolve-BA.""" return _icms_out(session) @router.put("/impostos/icms/destinos/{uf}", response_model=DestinoIcmsOut) def atualizar_destino_icms(uf: str, corpo: DestinoIcmsIn, session: Session = Depends(get_session)) -> DestinoIcmsOut: try: destino = impostos.atualizar_destino(session, uf, corpo.aliquota_pct, corpo.reducao_base_pct, corpo.frete_adicional_pct) session.commit() except ValueError as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc return _destino_out(destino) @router.put("/impostos/icms/faixas/{faixa}", response_model=list[DestinoIcmsOut]) def atualizar_faixa_icms(faixa: str, corpo: FaixaIcmsIn, session: Session = Depends(get_session)) -> list[DestinoIcmsOut]: """Muda a alíquota de todas as UFs de uma faixa de uma vez.""" try: impostos.atualizar_faixa(session, faixa, corpo.aliquota_pct) session.commit() except ValueError as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc return [_destino_out(d) for d in impostos.listar_destinos(session)] @router.put("/impostos/icms/compras/{faixa}", response_model=CompraIcmsOut) def atualizar_compra_icms(faixa: str, corpo: FaixaIcmsIn, session: Session = Depends(get_session)) -> CompraIcmsOut: """Alíquota do crédito de ICMS das compras de uma origem (dentro da Bahia, Sul/Sudeste exceto ES, demais).""" try: linha = impostos.atualizar_compra(session, faixa, corpo.aliquota_pct) session.commit() except ValueError as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc return CompraIcmsOut(faixa=linha.faixa, titulo=impostos.FAIXAS[linha.faixa], aliquota_pct=linha.aliquota_pct) @router.put("/impostos/icms/mix/{cenario}", response_model=MixCenarioOut) def definir_mix_icms(cenario: str, corpo: MixIn, session: Session = Depends(get_session)) -> MixCenarioOut: """Define o mix de destinos da venda do Centro de Resultado (fração da receita por UF; a soma fecha 100%; vazio limpa).""" centro = _obter(session, CenarioPrecificacao, cenario, "Centro de Resultado") try: impostos.definir_mix(session, centro.id, {l.uf: l.percentual for l in corpo.linhas}) session.commit() except ValueError as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc return _mix_do_centro(session, centro) @router.put("/impostos/icms/desenvolve", response_model=DesenvolveOut) def definir_desenvolve(corpo: DesenvolveIn, session: Session = Depends(get_session)) -> DesenvolveOut: try: linha = impostos.definir_desenvolve(session, **corpo.model_dump(exclude_unset=True)) session.commit() except (ValueError, TypeError) as exc: session.rollback() raise HTTPException(status_code=422, detail=str(exc)) from exc return _desenvolve_registro_out(linha) @router.post("/impostos/icms/desenvolve/simular", response_model=ResultadoDesenvolveOut) def simular_desenvolve(corpo: SimulacaoDesenvolveIn, session: Session = Depends(get_session)) -> ResultadoDesenvolveOut: """Aplica os parâmetros do Desenvolve a um saldo devedor mensal informado (a conta do benefício, sem depender do Planejamento).""" try: saldo = Decimal(corpo.saldo_devedor.strip().replace(",", ".")) except InvalidOperation as exc: raise HTTPException(status_code=422, detail="O saldo precisa ser um número") from exc parametros = impostos.obter_desenvolve(session) session.commit() return _desenvolve_out(impostos.calcular_desenvolve(saldo, parametros))