"""Ficha técnica de um Produto (§9): o roteiro com o custo de cada linha, num mês escolhido; e as operações de duplicar e excluir um Produto com seu roteiro. A ficha usa as mesmas contas do custeio (`calcular_custo_produto`) com as versões de custo vigentes no mês (`Entradas`, §4.3), mas é independente de Cenário: considera **todos os recursos ligados** (o rateio por Cenário só existe no custeio de um Cenário). """ from dataclasses import dataclass, field from datetime import date from decimal import Decimal from sqlalchemy import func, inspect from sqlalchemy.orm import Session from custos_precificacao import historico from custos_precificacao.custeio import LinhaCusto, RateioCenario, calcular_custo_produto from custos_precificacao.entradas import ROTEIROS, Entradas from custos_precificacao.models import ( Atividade, CusteioProduto, Direcionador, Produto, RoteiroInsumo, VolumeProjetado, ) ORDEM_TIPOS = {tipo: i for i, tipo in enumerate(ROTEIROS)} @dataclass class LinhaFicha: tipo: str linha_id: int descricao: str atividade_codigo: str quantidade: Decimal unidade: str base: Decimal # custo/hora ou custo/kg valor_producao: Decimal valor_variavel: Decimal atualizado_em: date editavel: bool # a versão mostrada é a atual (só essas se alteram/encerram pela tela) quantidade_informada: Decimal | None = None # insumo: quantidade e unidade digitadas na formulação unidade_informada: str | None = None perda_pct: Decimal = Decimal(0) direcionador_codigo: str | None = None # elementos: o direcionador da alocação direcionador_nome: str | None = None produto_origem_codigo: str | None = None # insumo que é intermediário: o produto de origem (atalho para a formulação dele) def produtos_na_tabela_de_precos(session: Session) -> set[int]: """Produtos "cabeça" da Tabela de Preços (marcados em `Produto.tabela_de_precos`): o custeio parte deles.""" return {produto_id for (produto_id,) in session.query(Produto.id).filter(Produto.tabela_de_precos.is_(True))} @dataclass class Ficha: produto: Produto competencia: date linhas: list[LinhaFicha] = field(default_factory=list) custo_producao_lote: Decimal | None = None custo_variavel_lote: Decimal | None = None custo_producao_unitario: Decimal | None = None custo_variavel_unitario: Decimal | None = None rendimento: Decimal | None = None rendimento_unidade: str | None = None # 'un' | 'kg' aviso: str | None = None def _obter(session: Session, codigo: str) -> Produto: produto = session.query(Produto).filter_by(codigo=codigo).one_or_none() if produto is None: raise ValueError(f"Produto com codigo={codigo!r} não encontrado") return produto def montar_ficha(session: Session, produto_codigo: str, competencia: date | str | None = None) -> Ficha: """Roteiro do Produto com o custo de cada linha e os totais do lote e unitários, no mês `competencia` (padrão o atual). Se algo não calcula (ex.: semiacabado sem rendimento em kg) os totais ficam None e `aviso` explica — as linhas já calculadas continuam aparecendo.""" produto = _obter(session, produto_codigo) mes = historico.normalizar_mes(competencia) entradas = Entradas(session, mes) coletadas: list[LinhaCusto] = [] ficha = Ficha(produto=produto, competencia=mes) try: custo = calcular_custo_produto( session, produto, RateioCenario(set(), set(), set()), entradas=entradas, _linhas=coletadas ) except (ValueError, RuntimeError) as exc: custo = None ficha.aviso = str(exc) atividades = {a.id: a.codigo for a in session.query(Atividade)} direcionadores = {d.id: d for d in session.query(Direcionador)} produtos = {p.id: p.codigo for p in session.query(Produto.id, Produto.codigo)} if any(l.produto_origem_id for l in coletadas) else {} for l in sorted(coletadas, key=lambda x: (ORDEM_TIPOS[x.tipo], x.linha_id)): atual = session.get(ROTEIROS[l.tipo], l.linha_id) ficha.linhas.append(LinhaFicha( tipo=l.tipo, linha_id=l.linha_id, descricao=l.descricao, atividade_codigo=atividades.get(l.atividade_id, ""), quantidade=l.quantidade, unidade=l.unidade, base=l.base, valor_producao=l.valor_producao, valor_variavel=l.valor_variavel, atualizado_em=l.atualizado_em, editavel=atual is not None and atual.atualizado_em == l.atualizado_em, quantidade_informada=l.quantidade_informada, unidade_informada=l.unidade_informada, perda_pct=l.perda_pct, direcionador_codigo=direcionadores[l.direcionador_id].codigo if l.direcionador_id in direcionadores else None, direcionador_nome=direcionadores[l.direcionador_id].nome if l.direcionador_id in direcionadores else None, produto_origem_codigo=produtos.get(l.produto_origem_id), )) if custo is not None: ficha.custo_producao_lote = custo.custo_producao_lote ficha.custo_variavel_lote = custo.custo_variavel_lote ficha.custo_producao_unitario = custo.custo_producao_unitario(produto) ficha.custo_variavel_unitario = custo.custo_variavel_unitario(produto) ficha.rendimento = custo.rendimento un, kg = entradas.rendimentos(produto) ficha.rendimento_unidade = "un" if un else ("kg" if kg else None) if not custo.rendimento: ficha.aviso = "Produto sem rendimento definido: o custo unitário é igual ao custo do lote." return ficha def duplicar_produto( session: Session, origem_codigo: str, novo_codigo: str, novo_nome: str, atualizado_em: date | str | None = None, ) -> Produto: """Cria um Produto novo copiando tipo, rendimentos, produção estimada e as **linhas atuais** do roteiro (os 4 tipos). Tudo vale a partir de `atualizado_em` (mês, padrão o atual); o histórico antigo não é copiado.""" origem = _obter(session, origem_codigo) if session.query(Produto).filter_by(codigo=novo_codigo).first() is not None: raise ValueError(f"Já existe um Produto com código {novo_codigo!r}") mes = historico.normalizar_mes(atualizado_em) novo = Produto( codigo=novo_codigo, nome=novo_nome, tipo=origem.tipo, rendimento_kg=origem.rendimento_kg, rendimento_un=origem.rendimento_un, qtd_estimada_producao_mes=origem.qtd_estimada_producao_mes, tabela_de_precos=origem.tabela_de_precos, atualizado_em=mes, ) session.add(novo) session.flush() for modelo in ROTEIROS.values(): colunas = [c.key for c in inspect(modelo).columns if c.key not in ("id", "produto_id", "atualizado_em")] for linha in session.query(modelo).filter_by(produto_id=origem.id): copiadas = {c: getattr(linha, c) for c in colunas} if "origem" in copiadas: copiadas["origem"] = "manual" # a cópia é um produto novo: não herda a trava das linhas do TOTVS session.add(modelo(produto_id=novo.id, atualizado_em=mes, **copiadas)) session.commit() return novo @dataclass class Removido: linhas_roteiro: int execucoes: int # linhas de custeio (`CusteioProduto`) apagadas def excluir_produto(session: Session, produto_codigo: str) -> Removido: """Exclui o Produto e o que é derivado dele (linhas de roteiro e o histórico delas, resultados de custeio e o histórico do próprio Produto). ValueError se ele ainda tem *dados de negócio* — Volume Projetado, preços ou uso como semiacabado no roteiro de outro Produto — que não somem em silêncio.""" produto = _obter(session, produto_codigo) def contar(consulta) -> int: return consulta.count() bloqueios = { "Volume(s) Projetado(s)": contar(session.query(VolumeProjetado).filter_by(produto_id=produto.id)), "uso como semiacabado em outro roteiro": contar( session.query(RoteiroInsumo).filter( RoteiroInsumo.insumo_produto_id == produto.id, RoteiroInsumo.produto_id != produto.id ) ), } pendentes = {nome: qtd for nome, qtd in bloqueios.items() if qtd} if pendentes: detalhe = ", ".join(f"{qtd} {nome}" for nome, qtd in pendentes.items()) raise ValueError(f"O Produto {produto_codigo!r} ainda tem {detalhe} — remova ou mova antes de excluir") historico.apagar_historico(session, produto) # versões do Produto e das linhas dele linhas = 0 for modelo in ROTEIROS.values(): linhas += session.query(modelo).filter_by(produto_id=produto.id).delete(synchronize_session=False) execucoes = session.query(CusteioProduto).filter_by(produto_id=produto.id).delete(synchronize_session=False) session.delete(produto) session.commit() return Removido(linhas_roteiro=linhas, execucoes=execucoes)