import re
from collections import Counter
from datetime import datetime
from typing import Union
from django.db.models.query import QuerySet
from src.dados_comuns.fluxo_status import (
CronogramaWorkflow,
)
from src.pre_recebimento.cronograma_entrega.models import (
EtapasDoCronograma,
ProgramacaoDoRecebimentoDoCronograma,
)
from rest_framework.exceptions import ValidationError
from src.recebimento.models import FichaDeRecebimento
[docs]
def cria_etapas_de_cronograma(etapas, cronograma=None):
"""Cria registros de EtapasDoCronograma a partir de uma lista de dicionários.
Retorna a lista das etapas criadas, já persistidas no banco.
"""
etapas_criadas = []
for etapa in etapas:
etapas_criadas.append(
EtapasDoCronograma.objects.create(cronograma=cronograma, **etapa)
)
return etapas_criadas
[docs]
def cria_programacao_de_cronograma(programacoes, cronograma=None):
"""Cria registros de ProgramacaoDoRecebimentoDoCronograma.
Retorna a lista das programações criadas, já persistidas no banco.
"""
programacoes_criadas = []
for programacao in programacoes:
programacoes_criadas.append(
ProgramacaoDoRecebimentoDoCronograma.objects.create(
cronograma=cronograma, **programacao
)
)
return programacoes_criadas
[docs]
def totalizador_relatorio_cronograma(
data: Union[QuerySet, list[dict]],
) -> dict[str, int]:
"""Calcula a quantidade de itens por status do cronograma.
Aceita tanto um QuerySet (usando ORM) quanto uma lista de dicionários
já serializados, mantendo o mesmo formato de saída.
Retorna dicionário ordenado de forma decrescente pela quantidade,
onde a chave é o título do status e o valor é a contagem.
"""
is_queryset = isinstance(data, QuerySet)
if is_queryset:
status_count = {
CronogramaWorkflow.states[s].title: data.filter(status=s).count()
for s in CronogramaWorkflow.states
}
else:
counter = Counter(item.get("status") for item in data)
status_count = {
CronogramaWorkflow.states[s].title: counter.get(
CronogramaWorkflow.states[s].title, 0
)
for s in CronogramaWorkflow.states
}
return dict(sorted(status_count.items(), key=lambda e: e[1], reverse=True))
[docs]
def parse_date(date_str):
"""Converte string de data DD/MM/YYYY para datetime.date.
Retorna None se a string for inválida.
"""
try:
return datetime.strptime(date_str, "%d/%m/%Y").date()
except (ValueError, AttributeError):
return None
[docs]
def passa_filtro_data_etapa(etapa_data, data_inicio_obj, data_fim_obj):
"""Verifica se a data da etapa está dentro do intervalo informado."""
data_programada = etapa_data.get("data_programada")
if not data_programada:
return False
try:
data_etapa = datetime.strptime(data_programada, "%d/%m/%Y").date()
if data_inicio_obj and data_etapa < data_inicio_obj:
return False
if data_fim_obj and data_etapa > data_fim_obj:
return False
return True
except (ValueError, AttributeError):
return False
[docs]
def processar_situacao_recebido(
situacoes,
tem_fichas_sem_ocorrencia,
fichas_recebimento,
incluir_etapa,
fichas_finais,
):
"""Processa o filtro de situação \"Recebido\" nas fichas de recebimento."""
if "Recebido" in situacoes and tem_fichas_sem_ocorrencia:
incluir_etapa = True
fichas_finais = [
f for f in fichas_recebimento if f.get("houve_ocorrencia") in (None, False)
]
return incluir_etapa, fichas_finais
[docs]
def processar_situacao_ocorrencia(
situacoes,
tem_fichas_com_ocorrencia,
tem_fichas_sem_ocorrencia,
fichas_recebimento,
incluir_etapa,
fichas_finais,
):
"""Processa o filtro de situação \"Ocorrência\" nas fichas de recebimento."""
if "Ocorrência" in situacoes and tem_fichas_com_ocorrencia:
incluir_etapa = True
fichas_com_ocorrencia = [
f for f in fichas_recebimento if f.get("houve_ocorrencia") is True
]
if "Recebido" in situacoes and tem_fichas_sem_ocorrencia:
fichas_finais = fichas_finais + fichas_com_ocorrencia
else:
fichas_finais = fichas_com_ocorrencia
return incluir_etapa, fichas_finais
[docs]
def processar_situacao_a_receber(
situacoes,
fichas_recebimento,
tem_fichas_com_ocorrencia,
tem_fichas_sem_ocorrencia,
incluir_etapa,
fichas_finais,
):
"""Processa o filtro de situação \"A Receber\" nas fichas de recebimento."""
if "A Receber" in situacoes:
if not fichas_recebimento or (
tem_fichas_com_ocorrencia and not tem_fichas_sem_ocorrencia
):
incluir_etapa = True
if "Ocorrência" not in situacoes:
fichas_finais = []
return incluir_etapa, fichas_finais
[docs]
def aplicar_filtros_etapa(
etapa_data, data_inicio_obj, data_fim_obj, situacoes, tem_filtro_situacao
):
"""Aplica os filtros de data e situação a uma etapa do relatório."""
if data_inicio_obj or data_fim_obj:
if not passa_filtro_data_etapa(etapa_data, data_inicio_obj, data_fim_obj):
return False
if tem_filtro_situacao:
return passa_filtro_situacao(etapa_data, situacoes)
return True
[docs]
def passa_filtro_situacao(etapa_data, situacoes):
"""Verifica se a etapa atende ao filtro de situação selecionado."""
if not situacoes or len(situacoes) == 3:
return True
fichas_recebimento = etapa_data.get("fichas_recebimento", [])
tem_fichas_com_ocorrencia = any(
f.get("houve_ocorrencia") is True for f in fichas_recebimento
)
tem_fichas_sem_ocorrencia = any(
f.get("houve_ocorrencia") in (None, False) for f in fichas_recebimento
)
incluir_etapa = False
fichas_finais = []
incluir_etapa, fichas_finais = processar_situacao_recebido(
situacoes,
tem_fichas_sem_ocorrencia,
fichas_recebimento,
incluir_etapa,
fichas_finais,
)
incluir_etapa, fichas_finais = processar_situacao_ocorrencia(
situacoes,
tem_fichas_com_ocorrencia,
tem_fichas_sem_ocorrencia,
fichas_recebimento,
incluir_etapa,
fichas_finais,
)
incluir_etapa, fichas_finais = processar_situacao_a_receber(
situacoes,
fichas_recebimento,
tem_fichas_com_ocorrencia,
tem_fichas_sem_ocorrencia,
incluir_etapa,
fichas_finais,
)
if incluir_etapa:
etapa_data["fichas_recebimento"] = fichas_finais
return incluir_etapa
[docs]
def filtrar_etapas(serialized_data, filtros):
"""Filtra uma lista de cronogramas por data e situação das etapas."""
data_inicial = filtros.get("data_inicial")
data_final = filtros.get("data_final")
situacoes = filtros.get("situacao", [])
sem_datas = not data_inicial and not data_final
sem_filtros = (sem_datas and not situacoes) or (sem_datas and len(situacoes) == 3)
if sem_filtros:
return serialized_data
data_inicio_obj = parse_date(data_inicial) if data_inicial else None
data_fim_obj = parse_date(data_final) if data_final else None
tem_filtro_situacao = situacoes and len(situacoes) < 3
for i in range(len(serialized_data) - 1, -1, -1):
cronograma_data = serialized_data[i]
etapas = cronograma_data.get("etapas", [])
etapas[:] = [
etapa
for etapa in etapas
if aplicar_filtros_etapa(
etapa,
data_inicio_obj,
data_fim_obj,
situacoes,
tem_filtro_situacao,
)
]
if not etapas:
serialized_data.pop(i)
return serialized_data
[docs]
def converter_para_numero(quantidade_str):
"""Converte string formatada para número float.
Exemplo: ``"10.000,00 kg"`` → ``10000.00``.
Retorna None se a conversão falhar.
"""
numero_str = extrair_numero_quantidade(quantidade_str)
if not numero_str:
return None
try:
numero_limpo = numero_str.replace(".", "").replace(",", ".")
return float(numero_limpo)
except (ValueError, AttributeError):
return None
[docs]
def migrar_fichas_para_etapas_novas(etapas_antigas, etapas_novas):
for indice, etapa_antiga in enumerate(etapas_antigas):
try:
etapa_nova = etapas_novas[indice]
except IndexError:
raise ValidationError(
f"Não foi possível migrar todas as fichas de recebimento. "
f"A etapa de índice {indice} não possui correspondente nas novas etapas."
)
FichaDeRecebimento.objects.filter(etapa=etapa_antiga).update(
etapa=etapa_nova
)
etapa_antiga.cronograma = None
etapa_antiga.save(update_fields=["cronograma"])