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 auth, fichas, historico, unidades, vinculos from custos_precificacao.api.deps import get_session from custos_precificacao.api.schemas import ( DuplicarIn, FichaOut, LinhaFichaOut, LinhaRoteiroIn, LinhaRoteiroOut, LinhaRoteiroPatch, MarcarTabelaDePrecosIn, ProdutoIn, ProdutoOut, RoteiroOut, VersaoProdutoOut, VersaoRoteiroOut, ) from custos_precificacao.models import Atividade, Insumo, Produto, RoteiroInsumo router = APIRouter(tags=["produtos"]) TIPOS = ("acabado", "semi_acabado", "revenda") MSG_LINHA_TOTVS = "Esta linha veio do TOTVS: altere no TOTVS e reimporte a estrutura" def _mes(data) -> str: return f"{data.year:04d}-{data.month:02d}" def _erro(exc: Exception) -> HTTPException: return HTTPException(status_code=422, detail=str(exc)) def _numero(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: raise HTTPException(status_code=422, detail=f"{rotulo} não pode ser negativo") return valor def _obter(session: Session, codigo: str) -> Produto: produto = session.query(Produto).filter_by(codigo=codigo).one_or_none() if produto is None: raise HTTPException(status_code=404, detail=f"Produto {codigo!r} não encontrado") return produto def _rendimento_real(p: Produto) -> Decimal | None: if p.base_quantidade: return p.base_quantidade * (1 - (p.perda_pct or 0)) return p.rendimento_un if unidades.normalizar(p.unidade) == "un" else p.rendimento_kg def _fracao_opcional(texto: str | None, rotulo: str) -> Decimal | None: if texto is None or texto.strip() == "": return None return _numero(texto, rotulo) def _aplicar_formulacao(produto: Produto, corpo: ProdutoIn) -> None: """Unidade base e bacia/perda/rendimento real (dois dos três calculam o terceiro). Não versionados: gravam direto.""" enviados = corpo.model_fields_set if "unidade" in enviados: unidade = unidades.normalizar(corpo.unidade, "") if not unidade or len(unidade) > 10: raise HTTPException(status_code=422, detail="Informe a unidade (até 10 caracteres)") produto.unidade = unidade if not ({"base_quantidade", "perda_pct", "rendimento_real"} & enviados): return base = _fracao_opcional(corpo.base_quantidade, "base") if "base_quantidade" in enviados else None perda = _fracao_opcional(corpo.perda_pct, "perda") if "perda_pct" in enviados else None real = _fracao_opcional(corpo.rendimento_real, "rendimento real") if "rendimento_real" in enviados else None if "base_quantidade" in enviados and base is None: # base vazia remove a bacia (volta ao rendimento cadastrado) produto.base_quantidade, produto.perda_pct = None, Decimal(0) return # o que não veio, mas é necessário, sai do cadastro atual if base is None and produto.base_quantidade: base = produto.base_quantidade if perda is None and real is None: perda = produto.perda_pct or Decimal(0) try: base, perda, real = unidades.reconciliar_rendimento(base, perda, real) except unidades.ErroDeUnidade as exc: raise _erro(exc) from exc if base is None: raise HTTPException(status_code=422, detail="Informe a base (bacia) para calcular a perda ou o rendimento real") produto.base_quantidade, produto.perda_pct = base, perda def _travar_cadastro_totvs(produto: Produto, corpo: ProdutoIn) -> None: """Produto da ingestão: nome, unidade e bacia vêm do TOTVS (perda do processo, rendimento real e custo seguem editáveis).""" if produto.origem != "totvs": return enviados = corpo.model_fields_set travados = [] if "nome" in enviados and (corpo.nome or "").strip() != produto.nome: travados.append("nome") if "unidade" in enviados and unidades.normalizar(corpo.unidade, "") != unidades.normalizar(produto.unidade): travados.append("unidade") if "base_quantidade" in enviados: novo = _fracao_opcional(corpo.base_quantidade, "base") if novo != produto.base_quantidade: travados.append("bacia (base)") if travados: raise HTTPException( status_code=409, detail=f"Produto {produto.codigo!r} veio do TOTVS: {', '.join(travados)} só mudam no TOTVS (reimporte a estrutura)", ) def _saida(session: Session, p: Produto, na_tabela: bool | None = None) -> ProdutoOut: if na_tabela is None: na_tabela = p.tabela_de_precos linhas = sum( session.query(func.count(modelo.id)).filter_by(produto_id=p.id).scalar() for modelo in vinculos.ROTEIRO_MODELS.values() ) como_insumo = ( session.query(func.count(RoteiroInsumo.id)) .filter(RoteiroInsumo.insumo_produto_id == p.id, RoteiroInsumo.produto_id != p.id) .scalar() ) return ProdutoOut( codigo=p.codigo, nome=p.nome, tipo=p.tipo, rendimento_kg=p.rendimento_kg, rendimento_un=p.rendimento_un, qtd_estimada_producao_mes=p.qtd_estimada_producao_mes, atualizado_em=_mes(p.atualizado_em), unidade=unidades.normalizar(p.unidade), base_quantidade=p.base_quantidade, perda_pct=p.perda_pct or Decimal(0), rendimento_real=_rendimento_real(p), linhas_roteiro=linhas, usado_como_insumo=como_insumo, custo_compra=p.custo_compra, origem=p.origem or "manual", sem_formulacao=p.tipo != "revenda" and linhas == 0, na_tabela_de_precos=na_tabela, digitado_por=auth.NomesDeAutores(session)(p.digitado_por), digitado_em=p.digitado_em, ) @router.get("/produtos", response_model=list[ProdutoOut]) def listar_produtos(session: Session = Depends(get_session)) -> list[ProdutoOut]: return [_saida(session, p, p.tabela_de_precos) for p in session.query(Produto).order_by(Produto.codigo)] @router.post("/produtos/tabela-de-precos") def marcar_tabela_de_precos(corpo: MarcarTabelaDePrecosIn, session: Session = Depends(get_session)) -> dict[str, int]: """Marca ou desmarca vários produtos como cabeça da Tabela de Preços (tudo ou nada: código inexistente recusa o lote).""" produtos = {p.codigo: p for p in session.query(Produto).filter(Produto.codigo.in_(corpo.codigos))} faltando = [c for c in corpo.codigos if c not in produtos] if faltando: raise HTTPException(status_code=404, detail=f"Produto(s) não encontrado(s): {', '.join(faltando[:10])}") alterados = 0 for produto in produtos.values(): if produto.tabela_de_precos != corpo.marcado: produto.tabela_de_precos = corpo.marcado alterados += 1 session.commit() return {"alterados": alterados} @router.post("/produtos", response_model=ProdutoOut, status_code=201) def criar_produto(corpo: ProdutoIn, session: Session = Depends(get_session)) -> ProdutoOut: codigo = (corpo.codigo or "").strip() nome = (corpo.nome or "").strip() if not codigo or not nome: raise HTTPException(status_code=422, detail="codigo e nome são obrigatórios") if len(codigo) > 20 or len(nome) > 300: raise HTTPException(status_code=422, detail="codigo aceita até 20 e nome até 300 caracteres") if corpo.tipo not in TIPOS: raise HTTPException(status_code=422, detail="tipo deve ser 'acabado', 'semi_acabado' ou 'revenda'") valores = {} for campo in ("rendimento_kg", "rendimento_un", "qtd_estimada_producao_mes"): bruto = getattr(corpo, campo) valores[campo] = None if bruto is None or bruto.strip() == "" else _numero(bruto, campo) try: mes = historico.normalizar_mes(corpo.atualizado_em) except historico.ErroDeHistorico as exc: raise _erro(exc) from exc cabeca = corpo.na_tabela_de_precos if corpo.na_tabela_de_precos is not None else corpo.tipo in ("acabado", "revenda") produto = Produto(codigo=codigo, nome=nome, tipo=corpo.tipo, atualizado_em=mes, tabela_de_precos=cabeca, **valores) if corpo.tipo == "revenda": # comprado pronto: 1 un por "bacia", custo = custo de compra; sem formulação produto.unidade, produto.base_quantidade, produto.perda_pct = "un", Decimal(1), Decimal(0) produto.custo_compra = _fracao_opcional(corpo.custo_compra, "custo de compra") else: _aplicar_formulacao(produto, corpo) session.add(produto) try: session.commit() except IntegrityError as exc: session.rollback() raise HTTPException(status_code=409, detail=f"Já existe um Produto com código {codigo!r}") from exc return _saida(session, produto) @router.delete("/produtos/{codigo}", status_code=204) def excluir_produto(codigo: str, session: Session = Depends(get_session)) -> None: _obter(session, codigo) try: fichas.excluir_produto(session, codigo) except ValueError as exc: # ainda há dados de negócio presos ao Produto session.rollback() raise HTTPException(status_code=409, detail=str(exc)) from exc @router.post("/produtos/{codigo}/duplicar", response_model=ProdutoOut, status_code=201) def duplicar_produto(codigo: str, corpo: DuplicarIn, session: Session = Depends(get_session)) -> ProdutoOut: """Novo Produto com o mesmo tipo, rendimentos e as linhas atuais do roteiro (valendo de `atualizado_em`).""" _obter(session, codigo) novo_codigo, novo_nome = corpo.codigo.strip(), corpo.nome.strip() if not novo_codigo or not novo_nome: raise HTTPException(status_code=422, detail="codigo e nome são obrigatórios") if len(novo_codigo) > 20 or len(novo_nome) > 300: raise HTTPException(status_code=422, detail="codigo aceita até 20 e nome até 300 caracteres") try: novo = fichas.duplicar_produto(session, codigo, novo_codigo, novo_nome, corpo.atualizado_em) except historico.ErroDeHistorico as exc: session.rollback() raise _erro(exc) from exc except ValueError as exc: session.rollback() raise HTTPException(status_code=409, detail=str(exc)) from exc return _saida(session, novo) @router.get("/produtos/{codigo}/ficha", response_model=FichaOut) def obter_ficha(codigo: str, competencia: str | None = None, session: Session = Depends(get_session)) -> FichaOut: """Roteiro com o custo de cada linha e os totais, como seriam no mês `competencia` (padrão o atual).""" _obter(session, codigo) try: f = fichas.montar_ficha(session, codigo, competencia) except historico.ErroDeHistorico as exc: raise _erro(exc) from exc origens = {i: o for i, o in session.query(RoteiroInsumo.id, RoteiroInsumo.origem).filter_by(produto_id=f.produto.id)} return FichaOut( produto_codigo=f.produto.codigo, produto_nome=f.produto.nome, competencia=_mes(f.competencia), linhas=[ LinhaFichaOut( origem=(origens.get(l.linha_id) or "manual") if l.tipo == "insumo" else "manual", tipo=l.tipo, linha_id=l.linha_id, descricao=l.descricao, atividade_codigo=l.atividade_codigo, quantidade=l.quantidade, unidade=l.unidade, base=l.base, valor_producao=l.valor_producao, valor_variavel=l.valor_variavel, atualizado_em=_mes(l.atualizado_em), editavel=l.editavel, quantidade_informada=l.quantidade_informada, unidade_informada=l.unidade_informada, perda_pct=l.perda_pct, direcionador_codigo=l.direcionador_codigo, direcionador_nome=l.direcionador_nome, produto_origem_codigo=l.produto_origem_codigo, ) for l in f.linhas ], custo_producao_lote=f.custo_producao_lote, custo_variavel_lote=f.custo_variavel_lote, custo_producao_unitario=f.custo_producao_unitario, custo_variavel_unitario=f.custo_variavel_unitario, rendimento=f.rendimento, rendimento_unidade=f.rendimento_unidade, aviso=f.aviso, ) @router.patch("/produtos/{codigo}", response_model=ProdutoOut) def atualizar_produto(codigo: str, corpo: ProdutoIn, session: Session = Depends(get_session)) -> ProdutoOut: """Nome e produção estimada mudam direto; rendimento muda com histórico.""" produto = _obter(session, codigo) _travar_cadastro_totvs(produto, corpo) enviados = corpo.model_fields_set novos: dict = {} if "nome" in enviados: nome = (corpo.nome or "").strip() if not nome: raise HTTPException(status_code=422, detail="nome é obrigatório") novos["nome"] = nome for campo in ("rendimento_kg", "rendimento_un", "qtd_estimada_producao_mes"): if campo in enviados: bruto = getattr(corpo, campo) novos[campo] = None if bruto is None or bruto.strip() == "" else _numero(bruto, campo) if corpo.na_tabela_de_precos is not None: produto.tabela_de_precos = corpo.na_tabela_de_precos if "custo_compra" in enviados: if produto.tipo != "revenda": raise HTTPException(status_code=422, detail="Só produtos de revenda têm custo de compra") novos["custo_compra"] = _fracao_opcional(corpo.custo_compra, "custo de compra") if produto.tipo == "revenda": if {"unidade", "base_quantidade", "perda_pct", "rendimento_real"} & enviados: raise HTTPException(status_code=422, detail="Produto de revenda: 1 un por compra, sem bacia nem perda") else: _aplicar_formulacao(produto, corpo) try: historico.atualizar_com_historico(session, produto, novos, corpo.atualizado_em) except historico.ErroDeHistorico as exc: session.rollback() raise _erro(exc) from exc return _saida(session, produto) @router.get("/produtos/{codigo}/historico", response_model=list[VersaoProdutoOut]) def historico_do_produto(codigo: str, session: Session = Depends(get_session)) -> list[VersaoProdutoOut]: produto = _obter(session, codigo) nome = auth.NomesDeAutores(session) versoes = [ VersaoProdutoOut( atualizado_em=_mes(v.inicio), substituido_em=_mes(v.fim) if v.fim else None, rendimento_kg=v.valores["rendimento_kg"], rendimento_un=v.valores["rendimento_un"], autor=nome(v.autor), digitado_em=v.digitado_em, substituido_por=nome(v.substituido_por), ) for v in historico.versoes_com_autoria(session, produto) ] return list(reversed(versoes)) # --- Roteiro ----------------------------------------------------------------- def _tipo(tipo: str) -> type: if tipo not in vinculos.ROTEIRO_MODELS: raise HTTPException( status_code=404, detail=f"Tipo {tipo!r} inválido. Opções: {', '.join(vinculos.ROTEIRO_MODELS)}" ) return vinculos.ROTEIRO_MODELS[tipo] def _descricao(session: Session, tipo: str, dados) -> str: """Texto do recurso/insumo da linha. `dados` é a linha atual ou o dicionário do histórico.""" pega = (lambda c: dados.get(c)) if isinstance(dados, dict) else (lambda c: getattr(dados, c)) if tipo == "insumo": if pega("insumo_id") is not None: i = session.get(Insumo, pega("insumo_id")) return f"{i.codigo} - {i.nome}" p = session.get(Produto, pega("insumo_produto_id")) return f"{p.codigo} - {p.nome}" config = vinculos.ROTEIRO_RECURSOS[tipo] recurso = session.get(config.recurso_model, pega(config.campo_fk)) return f"{getattr(recurso, config.chave_recurso)} - {recurso.nome}" def _perda(texto: str | None) -> str: return str(_numero(texto, "perda")) if texto not in (None, "") else "0" def _quantidade(tipo: str, dados) -> Decimal: campo = "quantidade_kg" if tipo == "insumo" else "tempo_horas" return dados[campo] if isinstance(dados, dict) else getattr(dados, campo) @router.get("/produtos/{codigo}/roteiro", response_model=RoteiroOut) def obter_roteiro(codigo: str, session: Session = Depends(get_session)) -> RoteiroOut: produto = _obter(session, codigo) atividades = {a.id: a.codigo for a in session.query(Atividade)} nome = auth.NomesDeAutores(session) linhas, antigas = [], [] for tipo, modelo in vinculos.ROTEIRO_MODELS.items(): unidade = "kg" if tipo == "insumo" else "h" for l in session.query(modelo).filter_by(produto_id=produto.id).order_by(modelo.id): linhas.append(LinhaRoteiroOut( id=l.id, tipo=tipo, atividade_codigo=atividades.get(l.atividade_id, ""), descricao=_descricao(session, tipo, l), quantidade=_quantidade(tipo, l), unidade=(unidades.normalizar(l.unidade) if tipo == "insumo" else unidade), perda_pct=(l.perda_pct or Decimal(0)) if tipo == "insumo" else Decimal(0), atualizado_em=_mes(l.atualizado_em), origem=getattr(l, "origem", None) or "manual", digitado_por=nome(l.digitado_por), digitado_em=l.digitado_em, )) for r in historico.historico_de_linhas(session, modelo, produto.id): dados = {c: (Decimal(v) if c in ("tempo_horas", "quantidade_kg") and v is not None else v) for c, v in r.valores.items()} antigas.append(VersaoRoteiroOut( tipo=tipo, registro_id=r.registro_id, descricao=_descricao(session, tipo, dados), quantidade=_quantidade(tipo, dados), unidade=unidade, atualizado_em=_mes(r.atualizado_em), substituido_em=_mes(r.substituido_em), acao=r.acao, registrado_em=r.registrado_em, autor=nome(r.autor), digitado_em=r.digitado_em, substituido_por=nome(r.substituido_por), )) antigas.sort(key=lambda v: (v.substituido_em, v.registro_id), reverse=True) return RoteiroOut(produto_codigo=produto.codigo, linhas=linhas, historico=antigas) @router.post("/produtos/{codigo}/roteiro/{tipo}", response_model=RoteiroOut, status_code=201) def adicionar_linha( codigo: str, tipo: str, corpo: LinhaRoteiroIn, session: Session = Depends(get_session) ) -> RoteiroOut: """Nova linha de roteiro, valendo a partir do mês `atualizado_em` (padrão o atual).""" if _obter(session, codigo).tipo == "revenda": raise HTTPException(status_code=422, detail="Produto de revenda não tem formulação: o custo é o custo de compra") _tipo(tipo) quantidade = str(_numero(corpo.quantidade, "quantidade")) try: if tipo == "insumo": vinculos.adicionar_roteiro_insumo( session, codigo, corpo.atividade_codigo, quantidade, insumo_codigo=corpo.insumo_codigo or None, insumo_produto_codigo=corpo.insumo_produto_codigo or None, atualizado_em=corpo.atualizado_em, unidade=corpo.unidade, perda_pct=_perda(corpo.perda_pct), ) else: if not corpo.recurso_codigo: raise HTTPException(status_code=422, detail="recurso_codigo é obrigatório") vinculos.adicionar_roteiro_recurso( session, tipo, codigo, corpo.atividade_codigo, corpo.recurso_codigo, quantidade, atualizado_em=corpo.atualizado_em, ) except historico.ErroDeHistorico as exc: session.rollback() raise _erro(exc) from exc except ValueError as exc: # inclui ErroDeUnidade (conversão não cadastrada, perda inválida) session.rollback() raise HTTPException(status_code=404 if "não encontrado" in str(exc) else 422, detail=str(exc)) from exc return obter_roteiro(codigo, session) def _recusar_linha_totvs(linha) -> None: if getattr(linha, "origem", "manual") == "totvs": raise HTTPException(status_code=409, detail=MSG_LINHA_TOTVS) def _linha_do_produto(session: Session, codigo: str, tipo: str, linha_id: int): produto = _obter(session, codigo) linha = session.get(_tipo(tipo), linha_id) if linha is None or linha.produto_id != produto.id: raise HTTPException(status_code=404, detail=f"Linha {linha_id} não pertence ao roteiro de {codigo!r}") return linha @router.patch("/produtos/{codigo}/roteiro/{tipo}/{linha_id}", response_model=RoteiroOut) def alterar_linha( codigo: str, tipo: str, linha_id: int, corpo: LinhaRoteiroPatch, session: Session = Depends(get_session) ) -> RoteiroOut: _recusar_linha_totvs(_linha_do_produto(session, codigo, tipo, linha_id)) quantidade = str(_numero(corpo.quantidade, "quantidade")) try: vinculos.alterar_linha_roteiro( session, tipo, linha_id, quantidade, atualizado_em=corpo.atualizado_em, unidade=corpo.unidade, perda_pct=None if corpo.perda_pct is None else _perda(corpo.perda_pct), ) except (historico.ErroDeHistorico, unidades.ErroDeUnidade) as exc: session.rollback() raise _erro(exc) from exc return obter_roteiro(codigo, session) @router.delete("/produtos/{codigo}/roteiro/{tipo}/{linha_id}", response_model=RoteiroOut) def remover_linha( codigo: str, tipo: str, linha_id: int, em: str | None = None, session: Session = Depends(get_session) ) -> RoteiroOut: """Encerra a linha a partir do mês `em` (padrão o atual); ela segue valendo nos meses anteriores.""" _recusar_linha_totvs(_linha_do_produto(session, codigo, tipo, linha_id)) try: vinculos.remover_linha_roteiro(session, tipo, linha_id, em=em) except historico.ErroDeHistorico as exc: session.rollback() raise _erro(exc) from exc return obter_roteiro(codigo, session)