"""Ingestão do relatório de estruturas do TOTVS ("Estrutura simples") e conferência com o sistema. O relatório lista, para cada item pai (PA/PI), os componentes de um nível (`Codigo:` abre um bloco; as linhas abaixo trazem `Tp`, `Um`, `Perda %`, `Quantidade`…). Aqui: - `ler_relatorio` lê o .xlsx pelo **texto** dos cabeçalhos de coluna (e não pela posição); - `planejar` compara o relatório com o sistema **sem gravar** e devolve o que seria criado/alterado/removido, avisos e erros; - `aplicar` grava tudo-ou-nada, atualiza o espelho do último relatório (`TotvsEstruturaCab/Linha`) e registra a carga; - `conferir`/`conferir_pai` comparam o espelho do relatório com o que está no sistema (as linhas `origem='totvs'`). Mapeamento (decisões do usuário, 2026-10-29): MP/EM → Insumo; PA/PI → Produto (acabado/semiacabado, `Base Estrut.` = bacia, a perda do processo é nossa e nunca é sobrescrita); RV → Produto de revenda; item sem cabeçalho → produto sem formulação; MP/EM com estrutura → insumo sem estrutura. `Qtd. base` da linha é a base da estrutura do componente (informativo). Linhas `manuais` e elementos de custo (máquina, ambiente, mão de obra) nunca são tocados. """ import hashlib import re import unicodedata from dataclasses import dataclass, field from datetime import date, datetime from decimal import Decimal, InvalidOperation from io import BytesIO from typing import Any from sqlalchemy.orm import Session from custos_precificacao import historico, unidades from custos_precificacao.models import ( Atividade, Insumo, Processo, Produto, RoteiroInsumo, TotvsCarga, TotvsEstruturaCab, TotvsEstruturaLinha, ) LIMITE_BYTES = 25 * 1024 * 1024 ATIVIDADE_CODIGO = "TOTVS" ATIVIDADE_NOME = "Estrutura TOTVS" SEIS = Decimal("0.000001") QUATRO = Decimal("0.0001") TIPOS_INSUMO = {"MP": "ingrediente", "EM": "embalagem"} TIPOS_PRODUTO = {"PA": "acabado", "PI": "semi_acabado"} ROTULO_CAMPO = {"quantidade_kg": "quantidade", "perda_pct": "perda (fração)", "unidade": "unidade"} LIMITE_LISTA = 300 # avisos/erros/mudanças devolvidos por categoria (os totais vêm nos contadores) class ErroDeRelatorio(ValueError): """Arquivo que não é o relatório (ou mudou de layout): a mensagem diz o que faltou.""" # --------------------------------------------------------------------------- leitura @dataclass class CabTotvs: codigo: str descricao: str tipo: str grupo: str | None unidade: str base: Decimal @dataclass class LinhaTotvs: pai: str ordem: int componente: str tp: str descricao: str um: str quantidade: Decimal perda_percentual: Decimal qtde_necessaria: Decimal qtd_base: Decimal | None tipo_quantidade: str fim_validade: date | None linha_no_arquivo: int @dataclass class Relatorio: emitido_em: date | None cabs: list[CabTotvs] linhas: list[LinhaTotvs] def _norm(texto: Any) -> str: texto = unicodedata.normalize("NFKD", str(texto if texto is not None else "")).encode("ascii", "ignore").decode() return " ".join(texto.casefold().split()) def _texto(valor: Any) -> str: return "" if valor is None else str(valor).strip() def _dec(valor: Any, rotulo: str, linha: int, *, obrigatorio: bool = True) -> Decimal | None: if valor is None or (isinstance(valor, str) and not valor.strip()): if obrigatorio: raise ErroDeRelatorio(f"Linha {linha}: {rotulo} vazio") return None if isinstance(valor, (int, float, Decimal)): return Decimal(str(valor)) bruto = str(valor).strip().replace(" ", "") if "," in bruto: bruto = bruto.replace(".", "").replace(",", ".") try: return Decimal(bruto) except InvalidOperation as exc: raise ErroDeRelatorio(f"Linha {linha}: {rotulo} inválido ({valor!r})") from exc def _data(valor: Any) -> date | None: if isinstance(valor, datetime): return valor.date() if isinstance(valor, date): return valor texto = _texto(valor) for formato in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%d/%m/%Y"): try: return datetime.strptime(texto, formato).date() except ValueError: continue return None COLUNAS_OBRIGATORIAS = { "nivel": ("nivel",), "codigo": ("codigo",), "tp": ("tp",), "descricao": ("descricao",), "um": ("um",), "perda": ("perda %", "perda"), "quantidade": ("quantidade",), "qtde_necessaria": ("qtde.necessaria", "qtde. necessaria", "qtde necessaria"), } COLUNAS_OPCIONAIS = { "qtd_base": ("qtd. base", "qtd base"), "tipo_quantidade": ("tipo quantidade",), "fim": ("fim valid", "fim validade"), } ROTULOS_CABECALHO = {"codigo:": "codigo", "descricao:": "descricao", "tipo:": "tipo", "grupo:": "grupo", "unidade:": "unidade", "base estrut.:": "base"} def _mapear_colunas(linha: tuple) -> dict[str, int] | None: nomes = {_norm(c): i for i, c in enumerate(linha) if c is not None and _norm(c)} mapa: dict[str, int] = {} for chave, opcoes in {**COLUNAS_OBRIGATORIAS, **COLUNAS_OPCIONAIS}.items(): achou = next((nomes[o] for o in opcoes if o in nomes), None) if achou is not None: mapa[chave] = achou return mapa if all(k in mapa for k in COLUNAS_OBRIGATORIAS) else None def ler_relatorio(dados: bytes) -> Relatorio: from openpyxl import load_workbook if len(dados) > LIMITE_BYTES: raise ErroDeRelatorio(f"Arquivo grande demais; o limite é {LIMITE_BYTES // 1024 // 1024} MB") try: livro = load_workbook(BytesIO(dados), read_only=True, data_only=True) except Exception as exc: # openpyxl levanta vários tipos para arquivo corrompido raise ErroDeRelatorio("Não foi possível abrir a planilha. Use o .xlsx do relatório de estruturas do TOTVS.") from exc folha = livro.worksheets[0] linhas = list(folha.iter_rows(values_only=True)) mapa: dict[str, int] | None = None inicio = 0 emitido: date | None = None for i, linha in enumerate(linhas[:30]): for celula in linha: texto = _texto(celula) achou = re.search(r"(\d{2}/\d{2}/\d{4})", texto) if achou and ("emiss" in _norm(texto) or "dt.ref" in _norm(texto)) and emitido is None: emitido = _data(achou.group(1)) mapa = _mapear_colunas(linha) if mapa: inicio = i + 1 break if mapa is None: raise ErroDeRelatorio( "Cabeçalho não encontrado: o relatório precisa ter as colunas Nivel, Codigo, Tp, Descricao, Um, Perda %, Quantidade e Qtde.necessaria" ) cabs: list[CabTotvs] = [] itens: list[LinhaTotvs] = [] pai: str | None = None ordem = 0 for numero, linha in enumerate(linhas[inicio:], start=inicio + 1): if all(c is None or _texto(c) == "" for c in linha): continue if _norm(linha[0] if linha else "") == "codigo:": campos: dict[str, Any] = {} for j in range(len(linha) - 1): rotulo = ROTULOS_CABECALHO.get(_norm(linha[j])) if rotulo: campos[rotulo] = linha[j + 1] codigo = _texto(campos.get("codigo")) if not codigo: raise ErroDeRelatorio(f"Linha {numero}: bloco 'Codigo:' sem o código do item") base = _dec(campos.get("base"), "Base Estrut.", numero, obrigatorio=False) cabs.append(CabTotvs( codigo=codigo, descricao=_texto(campos.get("descricao")), tipo=_texto(campos.get("tipo")).upper(), grupo=_texto(campos.get("grupo")) or None, unidade=_texto(campos.get("unidade")).upper(), base=base if base is not None else Decimal(1), )) pai, ordem = codigo, 0 continue nivel = linha[mapa["nivel"]] if mapa["nivel"] < len(linha) else None if not _texto(nivel).isdigit(): continue # rodapé, totais, quebra de página… if pai is None: raise ErroDeRelatorio(f"Linha {numero}: componente antes de qualquer bloco 'Codigo:'") def cel(chave: str) -> Any: indice = mapa.get(chave) return linha[indice] if indice is not None and indice < len(linha) else None ordem += 1 itens.append(LinhaTotvs( pai=pai, ordem=ordem, componente=_texto(cel("codigo")), tp=_texto(cel("tp")).upper(), descricao=_texto(cel("descricao")), um=_texto(cel("um")).upper(), quantidade=_dec(cel("quantidade"), "Quantidade", numero), perda_percentual=_dec(cel("perda"), "Perda %", numero, obrigatorio=False) or Decimal(0), qtde_necessaria=_dec(cel("qtde_necessaria"), "Qtde.necessaria", numero), qtd_base=_dec(cel("qtd_base"), "Qtd. base", numero, obrigatorio=False), tipo_quantidade=_texto(cel("tipo_quantidade")), fim_validade=_data(cel("fim")), linha_no_arquivo=numero, )) if not cabs: raise ErroDeRelatorio("Nenhuma estrutura encontrada (nenhum bloco 'Codigo:').") return Relatorio(emitido_em=emitido, cabs=cabs, linhas=itens) # --------------------------------------------------------------------------- plano (sem gravar) @dataclass class Mudanca: objeto: str # insumo | produto | linha acao: str # criar | alterar | remover codigo: str detalhe: str = "" @dataclass class Plano: vigencia: date emitido_em: date | None mudancas: list[Mudanca] = field(default_factory=list) avisos: list[tuple[str | None, str]] = field(default_factory=list) erros: list[tuple[str | None, str]] = field(default_factory=list) contagem: dict[str, dict[str, int]] = field(default_factory=dict) nao_vieram: list[str] = field(default_factory=list) pais: int = 0 linhas_no_relatorio: int = 0 # operações para `aplicar` ops_insumos: list[tuple[str, int | None, dict]] = field(default_factory=list) ops_produtos: list[tuple[str, int | None, dict]] = field(default_factory=list) ops_linhas: list[tuple] = field(default_factory=list) @property def ok(self) -> bool: return not self.erros def contar(self, objeto: str, acao: str) -> None: self.contagem.setdefault(objeto, {}).setdefault(acao, 0) self.contagem[objeto][acao] += 1 def aviso(self, codigo: str | None, mensagem: str) -> None: self.avisos.append((codigo, mensagem)) def erro(self, codigo: str | None, mensagem: str) -> None: self.erros.append((codigo, mensagem)) def _q6(valor: Decimal) -> Decimal: return Decimal(valor).quantize(SEIS) def _fracao_perda(percentual: Decimal) -> Decimal: return (Decimal(percentual) / 100).quantize(QUATRO) def _unidade_normal(texto: str) -> str: return unidades.normalizar(texto, "un") def _atividade_padrao(session: Session) -> tuple[Atividade | None, Processo | None]: atividade = session.query(Atividade).filter_by(codigo=ATIVIDADE_CODIGO).one_or_none() if atividade is not None: return atividade, None return None, session.query(Processo).order_by(Processo.id).first() def _checar_unidades_da_linha( session: Session, plano: Plano, linha: LinhaTotvs, item_existente: Insumo | Produto | None, unidade_item: str ) -> None: """A unidade da linha precisa ser a do item, uma conversão universal (g↔kg, ml↔l) ou uma já cadastrada.""" unidade_linha = _unidade_normal(linha.um) if unidade_linha == unidade_item or (unidade_linha, unidade_item) in unidades._UNIVERSAIS: return if item_existente is not None: try: unidades.converter(session, item_existente, Decimal(1), unidade_linha, unidade_item) return except unidades.ErroDeUnidade: pass plano.erro( linha.componente, f"{linha.pai}: a linha usa {linha.um} mas o cadastro de {linha.componente} é {unidade_item.upper()}; cadastre a conversão " f"em Insumos e Embalagens → Unidades e conversões", ) def planejar(session: Session, relatorio: Relatorio, vigencia: date | str | None = None) -> Plano: """Compara o relatório com o sistema e monta o que seria feito, sem gravar nada.""" mes = historico.normalizar_mes(vigencia) plano = Plano(vigencia=mes, emitido_em=relatorio.emitido_em) atividade, processo = _atividade_padrao(session) if atividade is None and processo is None: plano.erro(None, "Cadastre ao menos um Processo (dicionário de processos): as linhas do TOTVS entram numa atividade 'Estrutura TOTVS'.") cabs: dict[str, CabTotvs] = {} for c in relatorio.cabs: if c.codigo in cabs: plano.erro(c.codigo, f"{c.codigo}: aparece com mais de um bloco 'Codigo:' no relatório") cabs[c.codigo] = c # linhas válidas (a validade expirada antes da emissão é ignorada) e checagens de dados linhas: list[LinhaTotvs] = [] expiradas = 0 for l in relatorio.linhas: if relatorio.emitido_em and l.fim_validade and l.fim_validade < relatorio.emitido_em: expiradas += 1 continue if not l.componente: plano.erro(l.pai, f"{l.pai}: linha {l.linha_no_arquivo} sem código de componente") continue if l.quantidade < 0: plano.erro(l.componente, f"{l.pai}: quantidade negativa em {l.componente}") continue if not (0 <= l.perda_percentual < 100): plano.erro(l.componente, f"{l.pai}: perda de {l.perda_percentual}% em {l.componente} fora do intervalo 0–100") continue linhas.append(l) # só as estruturas de PA/PI valem; a de MP/EM (ninguém a consome) é ignorada, e seus componentes não viram itens linhas = [l for l in linhas if cabs.get(l.pai) is not None and cabs[l.pai].tipo in TIPOS_PRODUTO] if expiradas: plano.aviso(None, f"{expiradas} linha(s) com validade (Fim valid) anterior à emissão do relatório foram ignoradas") fora_variavel = sum(1 for l in linhas if l.tipo_quantidade and _norm(l.tipo_quantidade) != "variavel") if fora_variavel: plano.aviso(None, f"{fora_variavel} linha(s) com 'Tipo quantidade' diferente de Variavel foram importadas como quantidade da bacia") # tipo e unidade de cada item tipo_do_item: dict[str, str] = {} descricao_do_item: dict[str, str] = {} unidades_das_linhas: dict[str, set[str]] = {} for c in relatorio.cabs: tipo_do_item[c.codigo] = c.tipo descricao_do_item[c.codigo] = c.descricao for l in linhas: unidades_das_linhas.setdefault(l.componente, set()).add(l.um) descricao_do_item.setdefault(l.componente, l.descricao) anterior = tipo_do_item.setdefault(l.componente, l.tp) if anterior != l.tp: plano.erro(l.componente, f"{l.componente}: tipo {anterior} no cabeçalho e {l.tp} em {l.pai}") for codigo, tipo in tipo_do_item.items(): if tipo not in TIPOS_INSUMO and tipo not in TIPOS_PRODUTO and tipo != "RV": plano.erro(codigo, f"{codigo}: tipo {tipo or '(vazio)'} não suportado (aceitos: MP, EM, PA, PI, RV)") insumos_atuais = {i.codigo: i for i in session.query(Insumo)} produtos_atuais = {p.codigo: p for p in session.query(Produto)} def unidade_do_item(codigo: str) -> str: if codigo in cabs and cabs[codigo].unidade: return _unidade_normal(cabs[codigo].unidade) usadas = unidades_das_linhas.get(codigo, set()) return _unidade_normal(sorted(usadas)[0]) if usadas else "un" # itens: insumos e produtos for codigo in sorted(tipo_do_item): tipo = tipo_do_item[codigo] nome = descricao_do_item.get(codigo, "") or codigo if tipo not in TIPOS_INSUMO and tipo not in TIPOS_PRODUTO and tipo != "RV": continue # tipo não suportado: já registrado como erro if tipo in TIPOS_INSUMO: if codigo in produtos_atuais: plano.erro(codigo, f"{codigo}: já existe como Produto no sistema, mas o TOTVS o trata como {tipo}") continue unidade = unidade_do_item(codigo) if len(nome) > 200: plano.aviso(codigo, f"{codigo}: nome com mais de 200 caracteres foi cortado") if len(codigo) > 20 or len(unidade) > 10: plano.erro(codigo, f"{codigo}: código (máx. 20) ou unidade (máx. 10) longos demais") continue if codigo in cabs: plano.aviso(codigo, f"{codigo}: {tipo} com estrutura no TOTVS; entra como insumo e a estrutura é ignorada") atributos = {"nome": nome[:200], "tipo": TIPOS_INSUMO[tipo], "unidade": unidade, "origem": "totvs"} existente = insumos_atuais.get(codigo) if existente is None: plano.ops_insumos.append((codigo, None, atributos)) plano.mudancas.append(Mudanca("insumo", "criar", codigo, f"{nome[:60]} ({unidade})")) plano.contar("insumos", "criar") else: diferencas = [c for c in ("nome", "tipo", "unidade", "origem") if getattr(existente, c) != atributos[c]] if diferencas: plano.ops_insumos.append((codigo, existente.id, atributos)) adotado = " (cadastro manual adotado pelo TOTVS)" if existente.origem != "totvs" else "" plano.mudancas.append(Mudanca("insumo", "alterar", codigo, ", ".join(c for c in diferencas if c != "origem") + adotado)) plano.contar("insumos", "alterar") else: plano.contar("insumos", "igual") else: if codigo in insumos_atuais: plano.erro(codigo, f"{codigo}: já existe como Insumo no sistema, mas o TOTVS o trata como {tipo}") continue if tipo == "RV": tipo_produto, unidade, base = "revenda", "un", Decimal(1) else: tipo_produto = TIPOS_PRODUTO[tipo] unidade = unidade_do_item(codigo) if codigo in cabs: base = cabs[codigo].base else: base = Decimal(1) plano.aviso(codigo, f"{codigo}: sem cabeçalho no relatório (usado como componente); entra como produto sem formulação, bacia 1 {unidade.upper()}") if len(codigo) > 20 or len(unidade) > 10: plano.erro(codigo, f"{codigo}: código (máx. 20) ou unidade (máx. 10) longos demais") continue if base <= 0: plano.erro(codigo, f"{codigo}: Base Estrut. {base} inválida") continue if len(nome) > 300: plano.aviso(codigo, f"{codigo}: nome com mais de 300 caracteres foi cortado") base = _q6(base) if tipo != "RV" and codigo in cabs and not any(l.pai == codigo for l in linhas): plano.aviso(codigo, f"{codigo}: estrutura vazia no TOTVS; entra como produto sem formulação") atributos = {"nome": nome[:300], "tipo": tipo_produto, "unidade": unidade, "base_quantidade": base, "origem": "totvs"} existente = produtos_atuais.get(codigo) if existente is None: plano.ops_produtos.append((codigo, None, atributos)) plano.mudancas.append(Mudanca("produto", "criar", codigo, f"{nome[:60]} ({tipo_produto}, bacia {base:f} {unidade})")) plano.contar("produtos", "criar") else: if (existente.tipo == "revenda") != (tipo_produto == "revenda"): plano.erro(codigo, f"{codigo}: o tipo mudaria de {existente.tipo} para {tipo_produto}; ajuste o cadastro antes") continue atual_base = _q6(existente.base_quantidade) if existente.base_quantidade is not None else None comparaveis = {"nome": existente.nome, "tipo": existente.tipo, "unidade": existente.unidade, "origem": existente.origem, "base_quantidade": atual_base} diferencas = [c for c in comparaveis if comparaveis[c] != atributos[c]] if diferencas: plano.ops_produtos.append((codigo, existente.id, atributos)) adotado = " (cadastro manual adotado pelo TOTVS)" if existente.origem != "totvs" else "" plano.mudancas.append(Mudanca("produto", "alterar", codigo, ", ".join(c for c in diferencas if c != "origem") + adotado)) plano.contar("produtos", "alterar") else: plano.contar("produtos", "igual") # unidades das linhas contra o cadastro do componente for l in linhas: if l.componente not in tipo_do_item: continue existente = insumos_atuais.get(l.componente) or produtos_atuais.get(l.componente) _checar_unidades_da_linha(session, plano, l, existente, unidade_do_item(l.componente)) # Qtd. base: base da estrutura do próprio componente (informativo); avisa onde foge da regra fora_da_regra = [] for l in linhas: if l.qtd_base is None: continue esperado = cabs[l.componente].base if l.componente in cabs and cabs[l.componente].tipo in TIPOS_PRODUTO else Decimal(1) if l.qtd_base != esperado: fora_da_regra.append(l) for l in fora_da_regra[:LIMITE_LISTA]: plano.aviso(l.componente, f"{l.pai}: 'Qtd. base' {l.qtd_base:f} de {l.componente} difere da base da estrutura do componente (informativo; a quantidade da linha vale como está)") # pais e linhas pais_com_estrutura = {c.codigo for c in relatorio.cabs if c.tipo in TIPOS_PRODUTO} plano.pais = len(pais_com_estrutura) plano.linhas_no_relatorio = len(linhas) por_pai: dict[str, list[LinhaTotvs]] = {} for l in linhas: if l.pai in pais_com_estrutura: por_pai.setdefault(l.pai, []).append(l) existentes_por_pai: dict[int, dict[tuple[str, str], RoteiroInsumo]] = {} ids_produto = {p.id: p.codigo for p in produtos_atuais.values()} ids_insumo = {i.id: i.codigo for i in insumos_atuais.values()} for linha in session.query(RoteiroInsumo).filter(RoteiroInsumo.origem == "totvs"): componente = ("insumo", ids_insumo.get(linha.insumo_id)) if linha.insumo_id else ("produto", ids_produto.get(linha.insumo_produto_id)) existentes_por_pai.setdefault(linha.produto_id, {})[componente] = linha for pai in sorted(pais_com_estrutura): produto = produtos_atuais.get(pai) existentes = existentes_por_pai.get(produto.id, {}) if produto else {} vistos: set[tuple[str, str]] = set() for l in por_pai.get(pai, []): tipo_comp = tipo_do_item.get(l.componente) if tipo_comp is None: continue chave = ("insumo", l.componente) if tipo_comp in TIPOS_INSUMO else ("produto", l.componente) if chave in vistos: plano.erro(l.componente, f"{pai}: {l.componente} aparece mais de uma vez na mesma estrutura") continue vistos.add(chave) quantidade = _q6(l.quantidade) if quantidade != l.quantidade: plano.aviso(l.componente, f"{pai}: quantidade {l.quantidade:f} de {l.componente} arredondada para 6 casas ({quantidade:f})") unidade = _unidade_normal(l.um) perda = _fracao_perda(l.perda_percentual) atual = existentes.get(chave) if atual is None: plano.ops_linhas.append(("criar", pai, chave, {"quantidade_kg": quantidade, "unidade": unidade, "perda_pct": perda, "ordem": l.ordem})) plano.mudancas.append(Mudanca("linha", "criar", f"{pai} ← {l.componente}", f"{quantidade:f} {unidade}" + (f", perda {l.perda_percentual:f}%" if perda else ""))) plano.contar("linhas", "criar") continue novos: dict[str, Any] = {} if _q6(atual.quantidade_kg) != quantidade: novos["quantidade_kg"] = quantidade if unidades.normalizar(atual.unidade) != unidade: novos["unidade"] = unidade if Decimal(atual.perda_pct or 0).quantize(QUATRO) != perda: novos["perda_pct"] = perda if novos: if mes < atual.atualizado_em: plano.erro(l.componente, f"{pai} ← {l.componente}: a vigência {mes:%m/%Y} é anterior à versão atual da linha ({atual.atualizado_em:%m/%Y})") continue plano.ops_linhas.append(("alterar", pai, chave, {"id": atual.id, **novos})) plano.mudancas.append(Mudanca("linha", "alterar", f"{pai} ← {l.componente}", ", ".join(f"{ROTULO_CAMPO.get(k, k)} → {v:f}" if isinstance(v, Decimal) else f"{ROTULO_CAMPO.get(k, k)} → {v}" for k, v in novos.items()))) plano.contar("linhas", "alterar") else: plano.contar("linhas", "igual") for chave, atual in existentes.items(): if chave not in vistos: if mes < atual.atualizado_em: plano.erro(chave[1], f"{pai} ← {chave[1]}: a vigência {mes:%m/%Y} é anterior à versão atual da linha ({atual.atualizado_em:%m/%Y})") continue plano.ops_linhas.append(("remover", pai, chave, {"id": atual.id})) plano.mudancas.append(Mudanca("linha", "remover", f"{pai} ← {chave[1]}", "saiu da estrutura no TOTVS")) plano.contar("linhas", "remover") # produtos da última carga que não vieram nesta (nada é apagado) anteriores = {c.codigo for c in session.query(TotvsEstruturaCab)} plano.nao_vieram = sorted(anteriores - set(cabs)) if plano.nao_vieram: plano.aviso(None, f"{len(plano.nao_vieram)} estrutura(s) da carga anterior não vieram neste relatório (nada foi apagado)") return plano # --------------------------------------------------------------------------- aplicação def hash_do_arquivo(dados: bytes) -> str: return hashlib.sha256(dados).hexdigest() def resumo_do_plano(plano: Plano) -> dict[str, Any]: return { "contagem": plano.contagem, "pais": plano.pais, "linhas_no_relatorio": plano.linhas_no_relatorio, "avisos": len(plano.avisos), "nao_vieram": len(plano.nao_vieram), } def aplicar( session: Session, relatorio: Relatorio, plano: Plano, *, arquivo: str | None = None, sha256: str = "", usuario: str | None = None ) -> TotvsCarga: """Grava o plano (tudo-ou-nada), substitui o espelho do relatório e registra a carga. `plano` vem de `planejar`.""" if plano.erros: raise ValueError("O plano tem erros; corrija o relatório antes de aplicar") try: atividade, processo = _atividade_padrao(session) if atividade is None: atividade = Atividade(codigo=ATIVIDADE_CODIGO, nome=ATIVIDADE_NOME, processo_id=processo.id) session.add(atividade) session.flush() insumos: dict[str, Insumo] = {} for codigo, existente_id, atributos in plano.ops_insumos: if existente_id is None: objeto = Insumo(codigo=codigo, tipo=atributos["tipo"], nome=atributos["nome"], unidade=atributos["unidade"], origem="totvs", atualizado_em=plano.vigencia) session.add(objeto) else: objeto = session.get(Insumo, existente_id) for campo, valor in atributos.items(): setattr(objeto, campo, valor) insumos[codigo] = objeto produtos: dict[str, Produto] = {} for codigo, existente_id, atributos in plano.ops_produtos: if existente_id is None: objeto = Produto(codigo=codigo, atualizado_em=plano.vigencia, tabela_de_precos=atributos.get("tipo") in ("acabado", "revenda"), **atributos) session.add(objeto) else: objeto = session.get(Produto, existente_id) for campo, valor in atributos.items(): setattr(objeto, campo, valor) produtos[codigo] = objeto session.flush() def resolver(codigo: str, tipo: str): if tipo == "insumo": return insumos.get(codigo) or session.query(Insumo).filter_by(codigo=codigo).one() return produtos.get(codigo) or session.query(Produto).filter_by(codigo=codigo).one() for op, pai, (tipo, componente), dados in plano.ops_linhas: if op == "remover": historico.remover_com_historico(session, session.get(RoteiroInsumo, dados["id"]), plano.vigencia, commit=False) elif op == "alterar": novos = {k: v for k, v in dados.items() if k != "id"} historico.atualizar_com_historico(session, session.get(RoteiroInsumo, dados["id"]), novos, plano.vigencia, commit=False) else: dono = resolver(pai, "produto") item = resolver(componente, tipo) session.add(RoteiroInsumo( produto_id=dono.id, atividade_id=atividade.id, ordem=dados["ordem"], insumo_id=item.id if tipo == "insumo" else None, insumo_produto_id=item.id if tipo == "produto" else None, quantidade_kg=dados["quantidade_kg"], unidade=dados["unidade"], perda_pct=dados["perda_pct"], atualizado_em=plano.vigencia, origem="totvs", )) session.flush() carga = TotvsCarga( fonte="estrutura", arquivo=arquivo, sha256=sha256, emitido_em=plano.emitido_em, vigencia=plano.vigencia, carregado_por=usuario, resumo=resumo_do_plano(plano), ) session.add(carga) session.flush() # espelho do relatório (só o último): cabeçalhos e linhas válidas session.query(TotvsEstruturaLinha).delete() session.query(TotvsEstruturaCab).delete() for c in relatorio.cabs: session.add(TotvsEstruturaCab( carga_id=carga.id, codigo=c.codigo, descricao=c.descricao[:300], tipo=c.tipo, grupo=c.grupo, unidade=c.unidade, base=c.base, )) pais_validos = {c.codigo for c in relatorio.cabs if c.tipo in TIPOS_PRODUTO} for l in relatorio.linhas: if l.pai not in pais_validos or (relatorio.emitido_em and l.fim_validade and l.fim_validade < relatorio.emitido_em): continue session.add(TotvsEstruturaLinha( carga_id=carga.id, pai_codigo=l.pai, ordem=l.ordem, componente_codigo=l.componente, tp=l.tp, descricao=l.descricao[:300], um=l.um, quantidade=l.quantidade, perda_percentual=l.perda_percentual, qtde_necessaria=l.qtde_necessaria, qtd_base=l.qtd_base, )) session.commit() return carga except Exception: session.rollback() raise # --------------------------------------------------------------------------- conferência @dataclass class LinhaConferencia: componente: str descricao: str situacao: str # igual | diferente | so_relatorio | so_sistema relatorio: dict[str, Any] | None sistema: dict[str, Any] | None diferencas: list[str] @dataclass class PaiConferencia: codigo: str descricao: str tipo: str situacao: str # ok | divergente | sem_produto linhas_relatorio: int linhas_sistema: int diferencas: int linhas_manuais: int def _linha_do_relatorio(l: TotvsEstruturaLinha) -> dict[str, Any]: return { "quantidade": l.quantidade, "um": l.um.lower(), "perda_percentual": l.perda_percentual, "qtde_necessaria": l.qtde_necessaria, "qtd_base": l.qtd_base, "tp": l.tp, } def _linha_do_sistema(linha: RoteiroInsumo) -> dict[str, Any]: perda = Decimal(linha.perda_pct or 0) fator = Decimal(1) / (1 - perda) if perda < 1 else Decimal(0) return { "quantidade": Decimal(linha.quantidade_kg), "um": unidades.normalizar(linha.unidade), "perda_percentual": perda * 100, "qtde_necessaria": Decimal(linha.quantidade_kg) * fator, "qtd_base": None, "tp": None, } def _diferencas(r: dict[str, Any], s: dict[str, Any]) -> list[str]: dif = [] if _q6(r["quantidade"]) != _q6(s["quantidade"]): dif.append("quantidade") if _unidade_normal(r["um"]) != _unidade_normal(s["um"]): dif.append("unidade") if Decimal(r["perda_percentual"]).quantize(QUATRO) != Decimal(s["perda_percentual"]).quantize(QUATRO): dif.append("perda") return dif def _sistema_por_pai(session: Session) -> tuple[dict[str, dict[str, RoteiroInsumo]], dict[str, int], dict[str, Produto]]: produtos = {p.id: p for p in session.query(Produto)} insumos = {i.id: i.codigo for i in session.query(Insumo)} por_pai: dict[str, dict[str, RoteiroInsumo]] = {} manuais: dict[str, int] = {} for linha in session.query(RoteiroInsumo): pai = produtos[linha.produto_id].codigo if linha.origem != "totvs": manuais[pai] = manuais.get(pai, 0) + 1 continue componente = insumos.get(linha.insumo_id) if linha.insumo_id else produtos[linha.insumo_produto_id].codigo por_pai.setdefault(pai, {})[componente] = linha return por_pai, manuais, {p.codigo: p for p in produtos.values()} def conferir(session: Session, *, busca: str = "", situacao: str = "") -> dict[str, Any]: """Compara, pai a pai, o último relatório aplicado com a formulação do sistema (linhas de origem TOTVS).""" cabs = [c for c in session.query(TotvsEstruturaCab).order_by(TotvsEstruturaCab.codigo) if c.tipo in TIPOS_PRODUTO] do_relatorio: dict[str, dict[str, TotvsEstruturaLinha]] = {} for l in session.query(TotvsEstruturaLinha): do_relatorio.setdefault(l.pai_codigo, {})[l.componente_codigo] = l no_sistema, manuais, produtos = _sistema_por_pai(session) pais: list[PaiConferencia] = [] for c in cabs: produto = produtos.get(c.codigo) r = do_relatorio.get(c.codigo, {}) s = no_sistema.get(c.codigo, {}) if produto is None: estado, n_dif = "sem_produto", len(r) else: n_dif = 0 if (produto.base_quantidade is None or _q6(produto.base_quantidade) != _q6(c.base)) or _unidade_normal(produto.unidade) != _unidade_normal(c.unidade): n_dif += 1 for comp in set(r) | set(s): if comp not in r or comp not in s: n_dif += 1 else: n_dif += 1 if _diferencas(_linha_do_relatorio(r[comp]), _linha_do_sistema(s[comp])) else 0 estado = "ok" if n_dif == 0 else "divergente" pais.append(PaiConferencia( codigo=c.codigo, descricao=c.descricao, tipo=c.tipo, situacao=estado, linhas_relatorio=len(r), linhas_sistema=len(s), diferencas=n_dif, linhas_manuais=manuais.get(c.codigo, 0), )) resumo = { "total": len(pais), "ok": sum(1 for p in pais if p.situacao == "ok"), "divergentes": sum(1 for p in pais if p.situacao == "divergente"), "sem_produto": sum(1 for p in pais if p.situacao == "sem_produto"), } termo = _norm(busca) visiveis = [ p for p in pais if (not termo or termo in _norm(p.codigo) or termo in _norm(p.descricao)) and (not situacao or p.situacao == situacao) ] return {"resumo": resumo, "pais": visiveis} def conferir_pai(session: Session, codigo: str) -> dict[str, Any] | None: cab = session.query(TotvsEstruturaCab).filter_by(codigo=codigo).one_or_none() if cab is None: return None relatorio = {l.componente_codigo: l for l in session.query(TotvsEstruturaLinha).filter_by(pai_codigo=codigo).order_by(TotvsEstruturaLinha.ordem)} no_sistema, manuais, produtos = _sistema_por_pai(session) sistema = no_sistema.get(codigo, {}) produto = produtos.get(codigo) linhas: list[LinhaConferencia] = [] for comp in [*relatorio, *[c for c in sistema if c not in relatorio]]: r = relatorio.get(comp) s = sistema.get(comp) rr = _linha_do_relatorio(r) if r else None ss = _linha_do_sistema(s) if s else None if rr and ss: dif = _diferencas(rr, ss) estado = "diferente" if dif else "igual" else: dif, estado = [], ("so_relatorio" if rr else "so_sistema") linhas.append(LinhaConferencia(comp, r.descricao if r else comp, estado, rr, ss, dif)) return { "codigo": cab.codigo, "descricao": cab.descricao, "tipo": cab.tipo, "unidade_relatorio": cab.unidade.lower(), "base_relatorio": cab.base, "produto_existe": produto is not None, "unidade_sistema": unidades.normalizar(produto.unidade) if produto else None, "base_sistema": produto.base_quantidade if produto else None, "linhas": linhas, "linhas_manuais": manuais.get(codigo, 0), }