from datetime import date
from rest_framework import serializers
from rest_framework.exceptions import NotAuthenticated
from xworkflows.base import InvalidTransitionError
from src.dados_comuns.models import LogSolicitacoesUsuario
from src.dados_comuns.utils import (
update_instance_from_dict,
)
from src.pre_recebimento.base.models import (
UnidadeMedida,
)
from src.pre_recebimento.cronograma_entrega.models import (
Cronograma,
EtapasDoCronograma,
ProgramacaoDoRecebimentoDoCronograma,
SolicitacaoAlteracaoCronograma,
)
from src.pre_recebimento.ficha_tecnica.models import FichaTecnicaDoProduto
from src.pre_recebimento.qualidade.models import TipoEmbalagemQld
from src.terceirizada.models import Contrato, Terceirizada
from ..helpers import (
cria_etapas_de_cronograma,
cria_programacao_de_cronograma,
)
from ..validators import (
contrato_pertence_a_empresa,
)
[docs]
class ProgramacaoDoRecebimentoDoCronogramaCreateSerializer(serializers.ModelSerializer):
data_programada = serializers.CharField(required=False)
tipo_carga = serializers.ChoiceField(
choices=ProgramacaoDoRecebimentoDoCronograma.TIPO_CARGA_CHOICES,
required=False,
allow_blank=True,
)
[docs]
class EtapasDoCronogramaCreateSerializer(serializers.ModelSerializer):
numero_empenho = serializers.CharField(required=False)
etapa = serializers.IntegerField(required=False, allow_null=True)
parte = serializers.IntegerField(required=False, allow_null=True)
data_programada = serializers.CharField(required=False)
quantidade = serializers.FloatField(required=False)
total_embalagens = serializers.FloatField(required=False)
qtd_total_empenho = serializers.FloatField(required=False)
[docs]
class CronogramaCreateSerializer(serializers.ModelSerializer):
armazem = serializers.SlugRelatedField(
slug_field="uuid",
required=False,
queryset=Terceirizada.objects.filter(
tipo_servico=Terceirizada.DISTRIBUIDOR_ARMAZEM
),
allow_null=True,
)
empresa = serializers.SlugRelatedField(
slug_field="uuid",
required=False,
queryset=Terceirizada.objects.filter(
tipo_servico__in=[
Terceirizada.FORNECEDOR,
Terceirizada.FORNECEDOR_E_DISTRIBUIDOR,
]
),
)
contrato = serializers.SlugRelatedField(
slug_field="uuid",
required=False,
queryset=Contrato.objects.all(),
allow_null=True,
)
unidade_medida = serializers.SlugRelatedField(
slug_field="uuid",
required=False,
queryset=UnidadeMedida.objects.all(),
allow_null=True,
)
password = serializers.CharField(required=False)
qtd_total_programada = serializers.FloatField(required=False)
etapas = EtapasDoCronogramaCreateSerializer(many=True, required=False)
programacoes_de_recebimento = ProgramacaoDoRecebimentoDoCronogramaCreateSerializer(
many=True, required=False
)
cadastro_finalizado = serializers.BooleanField(required=False)
ficha_tecnica = serializers.SlugRelatedField(
slug_field="uuid",
required=False,
queryset=FichaTecnicaDoProduto.objects.all(),
allow_null=True,
)
tipo_embalagem_secundaria = serializers.SlugRelatedField(
slug_field="uuid",
queryset=TipoEmbalagemQld.objects.all(),
required=False,
allow_null=True,
)
custo_unitario_produto = serializers.FloatField(required=False)
[docs]
def gera_proximo_numero_cronograma(self):
ano = date.today().year
# Filtra apenas cronogramas com sufixo 'A' do ano atual e ordena pelo número
# para encontrar o último sequencial corretamente, independente da ordem de inserção.
ultimo_cronograma = (
Cronograma.objects.filter(numero__endswith=f"{ano}A")
.order_by("numero")
.last()
)
if ultimo_cronograma:
try:
sequencial = int(ultimo_cronograma.numero.split("/")[0])
return f"{str(sequencial + 1).zfill(3)}/{ano}A"
except (ValueError, IndexError):
pass
return f"001/{ano}A"
[docs]
def validate(self, attrs):
user = self.context["request"].user
cadastro_finalizado = attrs.get("cadastro_finalizado", None)
if cadastro_finalizado and not user.verificar_autenticidade(
attrs.pop("password", None)
):
raise NotAuthenticated(
"Assinatura do cronograma não foi validada. Verifique sua senha."
)
contrato = attrs.get("contrato", None)
empresa = attrs.get("empresa", None)
contrato_pertence_a_empresa(contrato, empresa)
return super().validate(attrs)
[docs]
def create(self, validated_data):
user = self.context["request"].user
cadastro_finalizado = validated_data.pop("cadastro_finalizado", None)
etapas = validated_data.pop("etapas", [])
programacoes_de_recebimento = validated_data.pop(
"programacoes_de_recebimento", []
)
numero_cronograma = self.gera_proximo_numero_cronograma()
cronograma = Cronograma.objects.create(
numero=numero_cronograma, **validated_data
)
cronograma.salvar_log_transicao(
status_evento=LogSolicitacoesUsuario.CRONOGRAMA_CRIADO, usuario=user
)
cria_etapas_de_cronograma(etapas, cronograma)
cria_programacao_de_cronograma(programacoes_de_recebimento, cronograma)
if cadastro_finalizado:
cronograma.inicia_fluxo(user=user)
return cronograma
[docs]
def update(self, instance, validated_data):
user = self.context["request"].user
cadastro_finalizado = validated_data.pop("cadastro_finalizado", None)
etapas = validated_data.pop("etapas", [])
programacoes_de_recebimento = validated_data.pop(
"programacoes_de_recebimento", []
)
instance.etapas.all().delete()
instance.programacoes_de_recebimento.all().delete()
update_instance_from_dict(instance, validated_data, save=True)
cria_etapas_de_cronograma(etapas, instance)
cria_programacao_de_cronograma(programacoes_de_recebimento, instance)
if cadastro_finalizado:
instance.inicia_fluxo(user=user)
return instance
[docs]
def novo_numero_solicitacao(objeto):
# Nova regra para sequência de numeração.
objeto.numero_solicitacao = f"{str(objeto.pk).zfill(8)}-ALT"
objeto.save()
[docs]
class SolicitacaoDeAlteracaoCronogramaCreateSerializer(serializers.ModelSerializer):
cronograma = serializers.UUIDField()
etapas = serializers.JSONField(write_only=True)
programacoes_de_recebimento = serializers.JSONField(write_only=True, required=False)
[docs]
def validate_cronograma(self, value):
cronograma = Cronograma.objects.filter(uuid=value)
if not cronograma:
raise serializers.ValidationError("Cronograma não existe")
if not cronograma.first().status == Cronograma.workflow_class.ASSINADO_CODAE:
raise serializers.ValidationError(
"Não é possivel criar Solicitação de alteração neste momento!"
)
return value
[docs]
def valida_campo_etapa(self, etapa, campo):
if not etapa[campo]:
raise serializers.ValidationError({campo: ["Este campo é obrigatório."]})
[docs]
def validate(self, attrs):
for etapa in attrs["etapas"]:
self.valida_campo_etapa(etapa, "etapa")
self.valida_campo_etapa(etapa, "data_programada")
self.valida_campo_etapa(etapa, "quantidade")
self.valida_campo_etapa(etapa, "total_embalagens")
return super().validate(attrs)
def _criar_etapas(self, etapas):
etapas_created = []
for etapa in etapas:
etapas_created.append(EtapasDoCronograma.objects.create(**etapa))
return etapas_created
[docs]
def create(self, validated_data):
user = self.context["request"].user
uuid_cronograma = validated_data.pop("cronograma", None)
etapas = validated_data.pop("etapas", [])
programacoes = validated_data.pop("programacoes_de_recebimento", [])
cronograma = Cronograma.objects.get(uuid=uuid_cronograma)
alteracao_cronograma = SolicitacaoAlteracaoCronograma.objects.create(
usuario_solicitante=user,
cronograma=cronograma,
**validated_data,
)
alteracao_cronograma.etapas_antigas.set(cronograma.etapas.all())
etapas_created = cria_etapas_de_cronograma(etapas)
alteracao_cronograma.etapas_novas.set(etapas_created)
programacoes_criadas = cria_programacao_de_cronograma(programacoes)
alteracao_cronograma.programacoes_novas.set(programacoes_criadas)
self._alterna_estado_cronograma(cronograma, user, alteracao_cronograma)
self._alterna_estado_solicitacao_alteracao_cronograma(
alteracao_cronograma, user, validated_data
)
return alteracao_cronograma
def _alterna_estado_cronograma(self, cronograma, user, alteracao_cronograma):
try:
if user.eh_fornecedor:
cronograma.fornecedor_solicita_alteracao(
user=user, justificativa=alteracao_cronograma.uuid
)
else:
cronograma.codae_realiza_alteracao(
user=user, justificativa=alteracao_cronograma.uuid
)
cronograma.finaliza_solicitacao_alteracao(
user=user,
justificativa=str(alteracao_cronograma.uuid),
)
except InvalidTransitionError as e:
raise serializers.ValidationError(
f"Erro de transição de estado do cronograma: {e}"
)
def _alterna_estado_solicitacao_alteracao_cronograma(
self, alteracao_cronograma, user, validated_data
):
try:
if user.eh_fornecedor:
alteracao_cronograma.inicia_fluxo(
user=user, justificativa=validated_data.get("justificativa", "")
)
else:
alteracao_cronograma.inicia_fluxo_codae(
user=user, justificativa=validated_data.get("justificativa", "")
)
except InvalidTransitionError as e:
raise serializers.ValidationError(
f"Erro de transição de estado da alteração: {e}"
)
[docs]
class EtapasDoCronogramaPontoAPontoCreateSerializer(serializers.ModelSerializer):
"""
Serializer simplificado para etapas do cronograma Ponto a Ponto (FLV).
Omitidos campos de empenho e total de embalagens.
"""
data_programada = serializers.CharField(required=False)
quantidade = serializers.FloatField(required=False)
[docs]
def validate_data_programada(self, value):
"""Converte MM/YYYY para o dia 01 do mês para salvamento."""
try:
parts = value.split("/")
if len(parts) != 2:
raise ValueError
mes, ano = map(int, parts)
return date(ano, mes, 1)
except (ValueError, TypeError):
raise serializers.ValidationError("Formato de data inválido. Use MM/YYYY.")
[docs]
class CronogramaPontoAPontoCreateSerializer(serializers.ModelSerializer):
"""
Serializer para criação de cronogramas FLV Ponto a Ponto.
- Omitidos: Armazém, Embalagem Secundária e Programações de Recebimento.
- Numeração com sufixo "P".
"""
empresa = serializers.SlugRelatedField(
slug_field="uuid",
queryset=Terceirizada.objects.filter(
tipo_servico__in=[
Terceirizada.FORNECEDOR,
Terceirizada.FORNECEDOR_E_DISTRIBUIDOR,
]
),
)
contrato = serializers.SlugRelatedField(
slug_field="uuid",
queryset=Contrato.objects.all(),
allow_null=True,
)
unidade_medida = serializers.SlugRelatedField(
slug_field="uuid",
queryset=UnidadeMedida.objects.all(),
allow_null=True,
required=False,
)
ficha_tecnica = serializers.SlugRelatedField(
slug_field="uuid",
queryset=FichaTecnicaDoProduto.objects.all(),
allow_null=True,
)
etapas = EtapasDoCronogramaPontoAPontoCreateSerializer(many=True, required=False)
cadastro_finalizado = serializers.BooleanField(required=False)
password = serializers.CharField(required=False)
[docs]
def gera_proximo_numero_cronograma(self):
ano = date.today().year
# Filtra apenas cronogramas com sufixo 'P' do ano atual e ordena pelo número
# para encontrar o último sequencial corretamente, independente da ordem de inserção.
ultimo_cronograma = (
Cronograma.objects.filter(numero__endswith=f"{ano}P")
.order_by("numero")
.last()
)
if ultimo_cronograma:
try:
sequencial = int(ultimo_cronograma.numero.split("/")[0])
return f"{str(sequencial + 1).zfill(3)}/{ano}P"
except (ValueError, IndexError):
pass
return f"001/{ano}P"
def _validate_etapas_finalizacao(self, etapas):
etapas_errors = []
has_errors = False
for etapa in etapas:
etapa_error = {}
if not etapa.get("data_programada"):
etapa_error["data_programada"] = ["Este campo é obrigatório."]
if not etapa.get("quantidade"):
etapa_error["quantidade"] = ["Este campo é obrigatório."]
if etapa_error:
has_errors = True
etapas_errors.append(etapa_error)
return etapas_errors if has_errors else None
def _validate_finalizacao(self, attrs):
user = self.context["request"].user
if not user.verificar_autenticidade(attrs.pop("password", None)):
raise NotAuthenticated(
"Assinatura do cronograma não foi validada. Verifique sua senha."
)
errors = {}
if not attrs.get("unidade_medida"):
errors["unidade_medida"] = ["Este campo é obrigatório."]
etapas = attrs.get("etapas", [])
if not etapas:
errors["etapas"] = ["Pelo menos uma etapa deve ser informada."]
else:
etapas_errors = self._validate_etapas_finalizacao(etapas)
if etapas_errors:
errors["etapas"] = etapas_errors
if errors:
raise serializers.ValidationError(errors)
[docs]
def validate(self, attrs):
cadastro_finalizado = attrs.get("cadastro_finalizado", False)
if cadastro_finalizado:
self._validate_finalizacao(attrs)
contrato = attrs.get("contrato", None)
empresa = attrs.get("empresa", None)
contrato_pertence_a_empresa(contrato, empresa)
return super().validate(attrs)
[docs]
def create(self, validated_data):
user = self.context["request"].user
cadastro_finalizado = validated_data.pop("cadastro_finalizado", None)
etapas = validated_data.pop("etapas", [])
numero_cronograma = self.gera_proximo_numero_cronograma()
cronograma = Cronograma.objects.create(
numero=numero_cronograma, **validated_data
)
cronograma.salvar_log_transicao(
status_evento=LogSolicitacoesUsuario.CRONOGRAMA_CRIADO, usuario=user
)
cria_etapas_de_cronograma(etapas, cronograma)
if cadastro_finalizado:
cronograma.inicia_fluxo(user=user)
return cronograma
[docs]
def update(self, instance, validated_data):
user = self.context["request"].user
cadastro_finalizado = validated_data.pop("cadastro_finalizado", None)
etapas = validated_data.pop("etapas", [])
# Para Ponto a Ponto, não temos programacoes_de_recebimento
instance.etapas.all().delete()
# Remove campos que não devem ser atualizados via dict se necessário
# mas aqui o validate já filtrou
for attr, value in validated_data.items():
setattr(instance, attr, value)
instance.save()
cria_etapas_de_cronograma(etapas, instance)
if cadastro_finalizado:
instance.inicia_fluxo(user=user)
return instance