80 lines
3.6 KiB
Python
80 lines
3.6 KiB
Python
"""Lógica de reabertura automática: quando um item (Ocorrência ou Ação) já
|
|
marcado como "tratado" pela Qualidade recebe, numa importação nova, algo
|
|
diferente do que existia no momento do tratamento, ele volta sozinho para
|
|
"pendente" — porque a tratativa real (prorrogar/finalizar uma ação, decidir
|
|
sobre uma análise) acontece no Sigsistem, sistema externo sem integração; o
|
|
Portal só monitora e precisa saber quando "o que a Qualidade já viu" ficou
|
|
desatualizado.
|
|
|
|
`snapshot_*()` é o que fica congelado em `NCOcorrencia.snapshot_tratativa`/
|
|
`NCAcao.snapshot_tratativa` no momento em que o item é marcado como tratado.
|
|
`decide_reabertura_*()` compara esse snapshot contra o estado atual (recém
|
|
extraído de uma importação nova) e devolve o motivo textual da reabertura, ou
|
|
`None` se nada relevante mudou."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import re
|
|
from datetime import date
|
|
|
|
from .modelos import AcaoExtraida, OcorrenciaExtraida
|
|
|
|
|
|
def _normaliza_texto(texto: str) -> str:
|
|
return re.sub(r"\s+", " ", (texto or "").strip().lower())
|
|
|
|
|
|
def hash_texto(texto: str) -> str:
|
|
return hashlib.sha256(_normaliza_texto(texto).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def snapshot_ocorrencia(ocorrencia: OcorrenciaExtraida) -> dict:
|
|
return {
|
|
"tem_analise": bool(ocorrencia.descricao_analise.strip()),
|
|
"hash_analise": hash_texto(ocorrencia.descricao_analise),
|
|
"codigos_acao": sorted(a.codigo for a in ocorrencia.acoes),
|
|
}
|
|
|
|
|
|
def decide_reabertura_ocorrencia(snapshot_congelado: dict, snapshot_atual: dict) -> str | None:
|
|
if snapshot_atual["tem_analise"] and not snapshot_congelado.get("tem_analise"):
|
|
return "Análise foi preenchida após o tratamento."
|
|
if snapshot_congelado.get("tem_analise") and snapshot_atual["hash_analise"] != snapshot_congelado.get(
|
|
"hash_analise"
|
|
):
|
|
return "O texto da análise mudou após o tratamento."
|
|
novos = sorted(set(snapshot_atual["codigos_acao"]) - set(snapshot_congelado.get("codigos_acao", [])))
|
|
if novos:
|
|
return "Nova(s) ação(ões) aberta(s): " + ", ".join(str(c) for c in novos) + "."
|
|
return None
|
|
|
|
|
|
def snapshot_acao(acao: AcaoExtraida) -> dict:
|
|
ultimo = acao.ultimo_acompanhamento
|
|
return {
|
|
"ultimo_acompanhamento_em": ultimo.data.isoformat() if ultimo else None,
|
|
"situacao": acao.situacao,
|
|
"fase": acao.fase,
|
|
"prazo_prorrogado": acao.prazo_prorrogado.isoformat() if acao.prazo_prorrogado else None,
|
|
"data_conclusao": acao.data_conclusao.isoformat() if acao.data_conclusao else None,
|
|
}
|
|
|
|
|
|
def decide_reabertura_acao(snapshot_congelado: dict, snapshot_atual: dict) -> str | None:
|
|
congelado_str = snapshot_congelado.get("ultimo_acompanhamento_em")
|
|
atual_str = snapshot_atual.get("ultimo_acompanhamento_em")
|
|
congelado = date.fromisoformat(congelado_str) if congelado_str else None
|
|
atual = date.fromisoformat(atual_str) if atual_str else None
|
|
# Comparação por DATA (não texto) do último acompanhamento — robusto
|
|
# contra reformatação cosmética do export, mesmo raciocínio que a skill
|
|
# original já usa pra identificar "o último acompanhamento".
|
|
if atual and (not congelado or atual > congelado):
|
|
return f"Novo acompanhamento registrado em {atual:%d/%m/%Y}."
|
|
rotulos = {"situacao": "Situação", "fase": "Fase", "prazo_prorrogado": "Prazo Prorrogado"}
|
|
for campo, rotulo in rotulos.items():
|
|
valor_congelado = snapshot_congelado.get(campo)
|
|
valor_atual = snapshot_atual.get(campo)
|
|
if valor_atual != valor_congelado:
|
|
return f'{rotulo} mudou no Sigsistem (de "{valor_congelado or "—"}" para "{valor_atual or "—"}").'
|
|
return None
|