from contextlib import contextmanager from decimal import Decimal, InvalidOperation from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import func from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from custos_precificacao import arvore, despesas, impostos from custos_precificacao.api.deps import get_session from custos_precificacao.api.schemas import ( DespesaIn, DespesaOut, DespesasOut, ResumoDespesasOut, TotalPlanoOut, ) from custos_precificacao.models import ( CenarioPrecificacao, CentroDeCusto, Despesa, Direcionador, PlanoDeContas, ) router = APIRouter(tags=["despesas"]) TAM_CODIGO = 20 # --- helpers ----------------------------------------------------------------- def _valor(texto: str | None) -> Decimal: try: valor = Decimal((texto or "").strip().replace(",", ".")) except InvalidOperation as exc: raise HTTPException(status_code=422, detail="valor precisa ser um número") from exc if not valor.is_finite() or valor < 0: raise HTTPException(status_code=422, detail="valor não pode ser negativo") return valor def _codigo_novo(texto: str | None) -> str: codigo = (texto or "").strip() if not codigo: raise HTTPException(status_code=422, detail="codigo é obrigatório") if len(codigo) > TAM_CODIGO: raise HTTPException(status_code=422, detail=f"codigo pode ter no máximo {TAM_CODIGO} caracteres") return codigo def _fk(session: Session, modelo, codigo: str | None, rotulo: str, *, obrigatorio: bool = True): codigo = (codigo or "").strip() if not codigo: if obrigatorio: raise HTTPException(status_code=422, detail=f"{rotulo} é obrigatório") return None instancia = session.query(modelo).filter_by(codigo=codigo).one_or_none() if instancia is None and hasattr(modelo, "chave"): instancia = session.query(modelo).filter_by(chave=codigo).one_or_none() # código antigo if instancia is None: raise HTTPException(status_code=404, detail=f"{rotulo} {codigo!r} não encontrado") try: arvore.exigir_lancavel(instancia, rotulo) # só nó analítico e ativo recebe lançamento except arvore.ErroDeArvore as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc return instancia 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 encontrada") return instancia def _commit_unico(session: Session, codigo: str, rotulo: str) -> None: try: session.commit() except IntegrityError as exc: session.rollback() raise HTTPException(status_code=409, detail=f"Já existe {rotulo} com código {codigo!r}") from exc @contextmanager def _rollback_em_erro(session: Session): """Descarta alterações pendentes se uma validação falhar no meio de uma edição.""" try: yield except HTTPException: session.rollback() raise def _fracao(texto: str | None, rotulo: str) -> Decimal: try: valor = Decimal((texto or "").strip().replace(",", ".")) except InvalidOperation as exc: raise HTTPException(status_code=422, detail=f"{rotulo} precisa ser um número") from exc if not valor.is_finite() or valor <= 0 or valor > 1: raise HTTPException(status_code=422, detail=f"{rotulo} deve ficar entre 0 e 100%") return valor def _classes_validas(creditos: list[str]) -> list[str]: invalidas = [c for c in creditos if c not in impostos.CLASSES] if invalidas: raise HTTPException(status_code=422, detail=f"Classe de crédito inválida: {', '.join(invalidas)}. Opções: {', '.join(impostos.CLASSES)}") return list(dict.fromkeys(creditos)) def _gravar_creditos(session: Session, despesa: Despesa, creditos: list[str]) -> None: """Marca exatamente as classes pedidas, preservando a alíquota/origem própria das que já estavam marcadas.""" atuais = impostos.classes_do_item(session, "despesa", despesa.id) impostos.definir_classes_do_item(session, "despesa", despesa.id, {c: atuais.get(c) for c in _classes_validas(creditos)}) def _validar_despesa(d: Despesa) -> None: try: despesas.validar_despesa(d) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc # --- saída ------------------------------------------------------------------- def _despesa_out(session: Session, d: Despesa) -> DespesaOut: plano = session.get(PlanoDeContas, d.plano_de_contas_id) centro = session.get(CentroDeCusto, d.centro_de_custo_id) dr = session.get(Direcionador, d.direcionador_id) if d.direcionador_id else None return DespesaOut( codigo=d.codigo, titulo=d.titulo, tipo=d.tipo, plano_de_contas_codigo=plano.codigo, plano_de_contas_nome=plano.nome, centro_de_custo_codigo=centro.codigo, centro_de_custo_nome=centro.nome, direcionador_codigo=dr.codigo if dr else None, direcionador_nome=dr.nome if dr else None, direcionador_base=dr.base if dr else None, valor=d.valor, percentual_receita=d.percentual_receita, descricao=d.descricao, creditos=sorted(impostos.classes_do_item(session, "despesa", d.id)), cenario_codigo=session.get(CenarioPrecificacao, d.cenario_id).codigo if d.cenario_id else None, ) @router.get("/despesas", response_model=DespesasOut) def listar_despesas(cenario: str | None = None, session: Session = Depends(get_session)) -> DespesasOut: """As despesas do Planejamento (sem `cenario`) ou as adicionais de um Centro de Resultados (`cenario` = código).""" consulta = session.query(Despesa) if cenario: alvo = _obter(session, CenarioPrecificacao, cenario, "Centro de Resultados") consulta = consulta.filter(Despesa.cenario_id == alvo.id) else: consulta = consulta.filter(Despesa.cenario_id.is_(None)) despesas = consulta.order_by(Despesa.codigo).all() # fixas e variáveis, sem competência por_plano: dict[int, Decimal] = {} for d in despesas: if d.tipo == "fixa": por_plano[d.plano_de_contas_id] = por_plano.get(d.plano_de_contas_id, Decimal(0)) + d.valor planos = {p.id: p for p in session.query(PlanoDeContas).filter(PlanoDeContas.id.in_(por_plano)).all()} if por_plano else {} return DespesasOut( despesas=[_despesa_out(session, d) for d in despesas], resumo=ResumoDespesasOut( total_despesas=sum(por_plano.values(), Decimal(0)), percentual_variaveis=sum((d.percentual_receita or Decimal(0) for d in despesas if d.tipo == "variavel"), Decimal(0)), por_plano=[ TotalPlanoOut(plano_de_contas_codigo=planos[pid].codigo, plano_de_contas_nome=planos[pid].nome, total=total) for pid, total in sorted(por_plano.items(), key=lambda kv: kv[1], reverse=True) ], ), ) # --- Despesa ----------------------------------------------------------------- @router.post("/despesas", response_model=DespesaOut, status_code=201) def criar_despesa(corpo: DespesaIn, session: Session = Depends(get_session)) -> DespesaOut: codigo = _codigo_novo(corpo.codigo) tipo = (corpo.tipo or "").strip() direcionador = _fk(session, Direcionador, corpo.direcionador_codigo, "Direcionador", obrigatorio=False) centro = _obter(session, CenarioPrecificacao, corpo.cenario_codigo.strip(), "Centro de Resultados") if (corpo.cenario_codigo or "").strip() else None if centro is not None and centro.a_alocar: raise HTTPException(status_code=422, detail="O Centro 'Produção a alocar' não recebe despesas adicionais: ele segue a precificação primária") despesa = Despesa( cenario_id=centro.id if centro else None, codigo=codigo, titulo=(corpo.titulo or "").strip(), tipo=tipo, plano_de_contas_id=_fk(session, PlanoDeContas, corpo.plano_de_contas_codigo, "Plano de Contas").id, centro_de_custo_id=_fk(session, CentroDeCusto, corpo.centro_de_custo_codigo, "Centro de Custo").id, direcionador_id=direcionador.id if direcionador else None, valor=_valor(corpo.valor) if tipo == "fixa" else Decimal(0), percentual_receita=_fracao(corpo.percentual_receita, "O % sobre a receita") if tipo == "variavel" else None, descricao=(corpo.descricao or "").strip() or None, ) _validar_despesa(despesa) if corpo.creditos: _classes_validas(corpo.creditos) session.add(despesa) try: session.flush() except IntegrityError as exc: session.rollback() raise HTTPException(status_code=409, detail=f"Já existe uma Despesa com código {codigo!r}") from exc if corpo.creditos: _gravar_creditos(session, despesa, corpo.creditos) _commit_unico(session, codigo, "uma Despesa") return _despesa_out(session, despesa) @router.patch("/despesas/{codigo}", response_model=DespesaOut) def atualizar_despesa(codigo: str, corpo: DespesaIn, session: Session = Depends(get_session)) -> DespesaOut: despesa = _obter(session, Despesa, codigo, "Despesa") enviados = corpo.model_fields_set with _rollback_em_erro(session): if "codigo" in enviados and (corpo.codigo or "").strip() not in ("", codigo): raise HTTPException(status_code=422, detail="O código não pode ser alterado") if "plano_de_contas_codigo" in enviados: despesa.plano_de_contas_id = _fk(session, PlanoDeContas, corpo.plano_de_contas_codigo, "Plano de Contas").id if "centro_de_custo_codigo" in enviados: despesa.centro_de_custo_id = _fk(session, CentroDeCusto, corpo.centro_de_custo_codigo, "Centro de Custo").id if "titulo" in enviados: despesa.titulo = (corpo.titulo or "").strip() if "tipo" in enviados: despesa.tipo = (corpo.tipo or "").strip() if "direcionador_codigo" in enviados: direcionador = _fk(session, Direcionador, corpo.direcionador_codigo, "Direcionador", obrigatorio=False) despesa.direcionador_id = direcionador.id if direcionador else None if "valor" in enviados and despesa.tipo == "fixa": despesa.valor = _valor(corpo.valor) if "percentual_receita" in enviados and despesa.tipo == "variavel": despesa.percentual_receita = _fracao(corpo.percentual_receita, "O % sobre a receita") if "descricao" in enviados: despesa.descricao = (corpo.descricao or "").strip() or None _validar_despesa(despesa) if corpo.creditos is not None: _gravar_creditos(session, despesa, corpo.creditos) session.commit() return _despesa_out(session, despesa) @router.delete("/despesas/{codigo}", status_code=204) def excluir_despesa(codigo: str, session: Session = Depends(get_session)) -> None: session.delete(_obter(session, Despesa, codigo, "Despesa")) session.commit()