"""Resolve as entradas de custo para o custeio (§9): valores atuais ou "como eram em tal mês". `Entradas(session)` (competência None) lê os valores **atuais** direto dos modelos — o comportamento de sempre. `Entradas(session, competencia)` resolve, pelo histórico (`historico.py`), as versões vigentes naquele mês: Gastos de Funcionário/Máquina/Ambiente, horas, preço do Insumo, rendimento do Produto e as linhas de roteiro. Com `registrar=True` anota tudo o que serviu ao cálculo, e `snapshot()` devolve isso como um dicionário JSON — é o que `CusteioExecucao.entradas` guarda, pra cada execução ser auditável e reproduzível. """ from datetime import date from decimal import Decimal from typing import Any from sqlalchemy.orm import Session from custos_precificacao import historico from custos_precificacao.models import ( Ambiente, AmbienteGasto, Funcionario, FuncionarioGasto, GrupoMaoDeObra, Insumo, Maquina, MaquinaGasto, PlanoDeContas, Produto, RoteiroAmbiente, RoteiroInsumo, RoteiroMaoDeObra, RoteiroMaquina, ) # tipo de recurso -> (modelo do recurso, modelo do gasto, chave do recurso) RECURSOS: dict[str, tuple[type, type, str]] = { "maquina": (Maquina, MaquinaGasto, "codigo"), "ambiente": (Ambiente, AmbienteGasto, "codigo"), "funcionario": (Funcionario, FuncionarioGasto, "codigo_fopag"), # o grupo Depto-Cargo soma os colaboradores (os gastos ficam em cada um): ver `_resolver_recurso` "grupo": (GrupoMaoDeObra, FuncionarioGasto, "codigo"), } ROTEIROS: dict[str, type] = { "maquina": RoteiroMaquina, "ambiente": RoteiroAmbiente, "mao-de-obra": RoteiroMaoDeObra, "insumo": RoteiroInsumo, } VERSAO_SNAPSHOT = 1 def _num(valor: Decimal | None) -> str | None: return None if valor is None else str(valor) class Entradas: def __init__(self, session: Session, competencia: date | str | None = None, *, registrar: bool = False): self.session = session self.competencia = None if competencia is None else historico.normalizar_mes(competencia) self.registrar = registrar self._indice = historico.IndiceHistorico(session) self._cache_custo: dict[tuple[str, int], tuple[Decimal, Decimal, Decimal]] = {} self._planos: dict[int, PlanoDeContas] = {} self._recursos_usados: dict[str, dict[str, Any]] = {} self._insumos_usados: dict[str, dict[str, Any]] = {} self._produtos_usados: dict[str, dict[str, Any]] = {} # --- acesso a registros ------------------------------------------------------------------ def recurso(self, tipo: str, recurso_id: int): return self.session.get(RECURSOS[tipo][0], recurso_id) def _plano(self, plano_id: int) -> PlanoDeContas | None: if plano_id not in self._planos: self._planos[plano_id] = self.session.get(PlanoDeContas, plano_id) return self._planos[plano_id] # --- recursos: gastos, horas, custo/hora ------------------------------------------------- def _gastos(self, tipo: str, recurso) -> list[Any]: if self.competencia is None: return list(recurso.gastos) return historico.linhas_em(self.session, RECURSOS[tipo][1], recurso.id, self.competencia, self._indice) def _horas_uteis(self, tipo: str, recurso) -> Decimal: if self.competencia is None: return recurso.horas_uteis_mensais valores = historico.valores_em(self.session, recurso, self.competencia, self._indice) if tipo == "funcionario": return valores["horas_contratadas_mensais"] * (1 - valores["absenteismo_pct"]) return valores["horas_uteis_mensais"] def _resolver_recurso(self, tipo: str, recurso) -> tuple[Decimal, Decimal, Decimal]: """(custo mensal, horas úteis, custo/hora) — o cálculo é o mesmo das propriedades dos modelos.""" chave = (tipo, recurso.id) if tipo == "grupo" and chave not in self._cache_custo: membros = [self._resolver_recurso("funcionario", f) for f in recurso.funcionarios] custo_mensal = sum((m[0] for m in membros), Decimal(0)) horas = sum((m[1] for m in membros), Decimal(0)) custo_hora = custo_mensal / horas if horas else Decimal(0) self._cache_custo[chave] = (custo_mensal, horas, custo_hora) if self.registrar: self._recursos_usados[f"grupo:{recurso.codigo}"] = { "tipo": "grupo", "nome": recurso.nome, "classificacao": recurso.classificacao, "horas_uteis": _num(horas), "custo_mensal": _num(custo_mensal), "custo_hora": _num(custo_hora), "gastos": [], "membros": sorted(f.codigo_fopag for f in recurso.funcionarios), } if chave not in self._cache_custo: gastos = self._gastos(tipo, recurso) custo_mensal = sum((g.desembolso + g.retencao for g in gastos), Decimal(0)) horas = self._horas_uteis(tipo, recurso) custo_hora = custo_mensal / horas if horas else Decimal(0) self._cache_custo[chave] = (custo_mensal, horas, custo_hora) if self.registrar: self._recursos_usados[f"{tipo}:{getattr(recurso, RECURSOS[tipo][2])}"] = { "tipo": tipo, "nome": recurso.nome, "classificacao": recurso.classificacao, "horas_uteis": _num(horas), "custo_mensal": _num(custo_mensal), "custo_hora": _num(custo_hora), "gastos": [ { "plano_de_contas": getattr(self._plano(g.plano_de_contas_id), "codigo", None), "plano_de_contas_nome": getattr(self._plano(g.plano_de_contas_id), "nome", None), "desembolso": _num(g.desembolso), "retencao": _num(g.retencao), "atualizado_em": g.atualizado_em.isoformat(), } for g in gastos ], } return self._cache_custo[chave] def horas_uteis(self, tipo: str, recurso) -> Decimal: return self._resolver_recurso(tipo, recurso)[1] def custo_mensal(self, tipo: str, recurso) -> Decimal: return self._resolver_recurso(tipo, recurso)[0] def custo_hora(self, tipo: str, recurso) -> Decimal: return self._resolver_recurso(tipo, recurso)[2] # --- insumos e produtos ------------------------------------------------------------------ def custo_final(self, insumo: Insumo) -> Decimal: if self.competencia is None: custo = insumo.custo_final valores = None else: valores = historico.valores_em(self.session, insumo, self.competencia, self._indice) preco = valores["valor_compra_unitario"] custo = preco + valores["frete"] + preco * valores["ipi_pct"] - preco * valores["icms_pct"] if self.registrar: v = valores or { "valor_compra_unitario": insumo.valor_compra_unitario, "frete": insumo.frete, "ipi_pct": insumo.ipi_pct, "icms_pct": insumo.icms_pct, } self._insumos_usados[insumo.codigo] = { "nome": insumo.nome, "valor_compra_unitario": _num(v["valor_compra_unitario"]), "frete": _num(v["frete"]), "ipi_pct": _num(v["ipi_pct"]), "icms_pct": _num(v["icms_pct"]), "custo_final": _num(custo), } return custo def custo_compra(self, produto: Produto) -> Decimal: """Custo de compra de 1 un de um produto de revenda, como era na competência (versões antigas sem o campo valem 0).""" if self.competencia is None: valor = produto.custo_compra else: valor = historico.valores_em(self.session, produto, self.competencia, self._indice).get("custo_compra") valor = valor or Decimal(0) if self.registrar: self._produto_registro(produto)["custo_compra"] = _num(valor) return valor def rendimentos(self, produto: Produto) -> tuple[Decimal | None, Decimal | None]: """(rendimento_un, rendimento_kg) do Produto.""" if self.competencia is None: un, kg = produto.rendimento_un, produto.rendimento_kg else: v = historico.valores_em(self.session, produto, self.competencia, self._indice) un, kg = v["rendimento_un"], v["rendimento_kg"] if produto.base_quantidade: # formulação com bacia: o rendimento real é base × (1 − perda), na unidade do produto real = produto.base_quantidade * (1 - (produto.perda_pct or 0)) if (produto.unidade or "kg").strip().lower() == "un": un = real else: kg = real if self.registrar: self._produto_registro(produto)["rendimento_un"] = _num(un) self._produto_registro(produto)["rendimento_kg"] = _num(kg) return un, kg def roteiro(self, produto: Produto, tipo: str) -> list[Any]: """Linhas de roteiro do Produto (`tipo`: maquina, ambiente, mao-de-obra, insumo).""" if self.competencia is None: linhas = list(getattr(produto, { "maquina": "roteiro_maquina", "ambiente": "roteiro_ambiente", "mao-de-obra": "roteiro_mao_de_obra", "insumo": "roteiro_insumo", }[tipo])) else: linhas = historico.linhas_em(self.session, ROTEIROS[tipo], produto.id, self.competencia, self._indice) if self.registrar: self._produto_registro(produto).setdefault("roteiro", {})[tipo] = [self._linha_roteiro(tipo, l) for l in linhas] return linhas def _produto_registro(self, produto: Produto) -> dict[str, Any]: return self._produtos_usados.setdefault(produto.codigo, {"nome": produto.nome}) def _linha_roteiro(self, tipo: str, linha: Any) -> dict[str, Any]: if tipo == "insumo": if linha.insumo_id is not None: fonte = {"insumo": self.session.get(Insumo, linha.insumo_id).codigo} else: fonte = {"produto": self.session.get(Produto, linha.insumo_produto_id).codigo} return { **fonte, "quantidade_kg": _num(linha.quantidade_kg), "unidade": getattr(linha, "unidade", None) or "kg", "perda_pct": _num(Decimal(getattr(linha, "perda_pct", None) or 0)), } tipo_recurso = "grupo" if tipo == "mao-de-obra" else tipo recurso = self.recurso(tipo_recurso, getattr(linha, f"{tipo_recurso}_id")) return {tipo_recurso: getattr(recurso, RECURSOS[tipo_recurso][2]), "tempo_horas": _num(linha.tempo_horas)} # --- snapshot ---------------------------------------------------------------------------- def snapshot(self, rateio) -> dict[str, Any]: """O que foi usado no cálculo, pronto pra guardar em JSON (`CusteioExecucao.entradas`).""" def codigos(modelo: type, ids: set[int], chave: str) -> list[str]: if not ids: return [] return sorted(getattr(r, chave) for r in self.session.query(modelo).filter(modelo.id.in_(ids))) return { "versao": VERSAO_SNAPSHOT, "competencia": f"{self.competencia.year:04d}-{self.competencia.month:02d}" if self.competencia else None, "rateio": { "ambientes_desligados": codigos(Ambiente, rateio.ambientes_desligados, "codigo"), "maquinas_desligadas": codigos(Maquina, rateio.maquinas_desligadas, "codigo"), "grupos_desligados": codigos(GrupoMaoDeObra, rateio.grupos_desligados, "codigo"), }, "recursos": dict(sorted(self._recursos_usados.items())), "insumos": dict(sorted(self._insumos_usados.items())), "produtos": dict(sorted(self._produtos_usados.items())), }