import json
import uuid as uuid_generator
from collections import defaultdict
from copy import deepcopy
from datetime import datetime
from django.core.exceptions import ValidationError as coreValidation
from django.db import transaction
from django.db.models import (
Case,
CharField,
Count,
OuterRef,
Q,
Subquery,
Value,
When,
)
from django.forms import ValidationError
from django_filters import rest_framework as filters
from rest_framework import generics, mixins, serializers, status
from rest_framework.decorators import action
from rest_framework.permissions import IsAdminUser, IsAuthenticated
from rest_framework.response import Response
from rest_framework.status import HTTP_201_CREATED, HTTP_400_BAD_REQUEST
from rest_framework.viewsets import GenericViewSet
from xworkflows import InvalidTransitionError
from src.dados_comuns import constants
from src.dados_comuns.api.paginations import HistoricoDietasPagination
from src.dados_comuns.fluxo_status import DietaEspecialWorkflow
from src.dados_comuns.models import LogSolicitacoesUsuario
from src.dados_comuns.permissions import (
PermissaoHistoricoDietasEspeciais,
PermissaoParaRecuperarDietaEspecial,
PermissaoRelatorioDietasEspeciais,
PermissaoRelatorioRecreioNasFerias,
UsuarioCODAEDietaEspecial,
UsuarioEmpresaGenerico,
UsuarioEscolaDiretaParceira,
UsuarioEscolaTercTotal,
)
from src.dados_comuns.services import enviar_email_codae_atualiza_protocolo
from src.dados_comuns.utils import convert_dict_to_querydict
from src.dieta_especial.forms import RelatorioDietaForm
from src.dieta_especial.protocolo_padrao.api.serializers import (
AlimentoSerializer,
)
from src.dieta_especial.protocolo_padrao.models import (
Alimento,
ProtocoloPadraoDietaEspecial,
SubstituicaoAlimento,
)
from src.dieta_especial.solicitacao_dieta_especial.api.filters import (
AlimentoFilter,
DietaEspecialFilter,
MotivoNegacaoFilter,
)
from src.dieta_especial.solicitacao_dieta_especial.api.forms import (
NegaDietaEspecialForm,
RelatorioQuantitativoSolicDietaEspForm,
SolicitacoesAtivasInativasPorAlunoForm,
)
from src.dieta_especial.solicitacao_dieta_especial.api.serializers import (
AlergiaIntoleranciaSerializer,
ClassificacaoDietaSerializer,
MotivoAlteracaoUESerializer,
MotivoNegacaoSerializer,
RelatorioQuantitativoSolicDietaEspSerializer,
SolicitacaoDietaEspecialAutorizarSerializer,
SolicitacaoDietaEspecialRecreioNasFeriasSerializer,
SolicitacaoDietaEspecialRelatorioTercSerializer,
SolicitacaoDietaEspecialSerializer,
SolicitacaoDietaEspecialSimplesSerializer,
SolicitacaoDietaEspecialUpdateSerializer,
SolicitacoesAtivasInativasPorAlunoSerializer,
UnidadeEducacionalSerializer,
)
from src.dieta_especial.solicitacao_dieta_especial.api.serializers_create import (
AlteracaoUESerializer,
SolicitacaoDietaEspecialCreateSerializer,
)
from src.dieta_especial.solicitacao_dieta_especial.models import (
AlergiaIntolerancia,
Anexo,
ClassificacaoDieta,
MotivoAlteracaoUE,
MotivoNegacao,
SolicitacaoDietaEspecial,
)
from src.dieta_especial.solicitacao_dieta_especial.services import (
atualiza_historico_protocolo,
)
from src.dieta_especial.tasks import (
gera_pdf_relatorio_dieta_especial_async,
gera_pdf_relatorio_dietas_especiais_terceirizadas_async,
gera_pdf_relatorio_historico_dietas_especiais_async,
gera_pdf_relatorio_recreio_nas_ferias_async,
gera_xlsx_relatorio_dietas_especiais_terceirizadas_async,
gera_xlsx_relatorio_historico_dietas_especiais_async,
gera_xlsx_relatorio_recreio_nas_ferias_async,
)
from src.dieta_especial.utils.filtra_relatorio_recreio_nas_ferias import (
filtra_relatorio_recreio_nas_ferias,
)
from src.dieta_especial.utils.gera_dicionario_historico_dietas import (
gera_dicionario_historico_dietas,
)
from src.dieta_especial.utils.gerar_filtros_relatorio_historico import (
gerar_filtros_relatorio_historico,
)
from src.escola.models import (
Aluno,
DiretoriaRegional,
Escola,
Lote,
TipoGestao,
)
from src.paineis_consolidados.api.constants import FILTRO_CODIGO_EOL_ALUNO
from src.paineis_consolidados.models import SolicitacoesCODAE
from src.relatorios.relatorios import (
relatorio_dieta_especial,
relatorio_dieta_especial_historico,
relatorio_dieta_especial_protocolo,
relatorio_quantitativo_classificacao_dieta_especial,
relatorio_quantitativo_diag_dieta_especial,
relatorio_quantitativo_diag_dieta_especial_somente_dietas_ativas,
relatorio_quantitativo_solic_dieta_especial,
)
from .pagination import RelatorioPagination
from .utils import (
filtrar_alunos_com_dietas_nos_status_e_rastro_escola,
trata_lotes_dict_duplicados,
)
[docs]
class SolicitacaoDietaEspecialViewSet(
mixins.RetrieveModelMixin,
mixins.ListModelMixin,
mixins.CreateModelMixin,
mixins.UpdateModelMixin,
GenericViewSet,
):
lookup_field = "uuid"
permission_classes = (IsAuthenticated,)
queryset = SolicitacaoDietaEspecial.objects.all()
filter_backends = (filters.DjangoFilterBackend,)
filterset_class = DietaEspecialFilter
[docs]
def get_queryset(self):
if self.action in [
"relatorio_dieta_especial",
"imprime_relatorio_dieta_especial",
]:
return self.queryset.select_related(
"rastro_escola__diretoria_regional"
).order_by("criado_em")
if self.action in ["relatorio_dieta_especial_terceirizada"]:
return self.queryset.select_related("rastro_terceirizada").order_by(
"-criado_em"
)
return super().get_queryset()
[docs]
def get_permissions(self):
PERMISSAO_PADRAO = (
IsAuthenticated,
PermissaoParaRecuperarDietaEspecial,
)
permission_map = {
"list": PERMISSAO_PADRAO,
"retrieve": PERMISSAO_PADRAO,
"imprime_relatorio_dieta_especial": PERMISSAO_PADRAO,
"update": (
IsAdminUser,
UsuarioCODAEDietaEspecial,
),
"create": [UsuarioEscolaTercTotal | UsuarioEscolaDiretaParceira],
"relatorio_dieta_especial_terceirizada": (
IsAuthenticated,
PermissaoRelatorioDietasEspeciais,
),
}
self.permission_classes = permission_map.get(
self.action, self.permission_classes
)
return super().get_permissions()
[docs]
def get_serializer_class(self):
RELATORIOS_QUANTITATIVOS = {
"relatorio_quantitativo_solic_dieta_esp",
"relatorio_quantitativo_diag_dieta_esp",
"relatorio_quantitativo_classificacao_dieta_esp",
}
AUTORIZACAO_ACTIONS = {"autorizar", "atualiza_protocolo"}
UPDATE_ACTIONS = {"update", "partial_update"}
if self.action in RELATORIOS_QUANTITATIVOS:
return RelatorioQuantitativoSolicDietaEspSerializer
if self.action in AUTORIZACAO_ACTIONS:
return SolicitacaoDietaEspecialAutorizarSerializer
if self.action in UPDATE_ACTIONS:
return SolicitacaoDietaEspecialUpdateSerializer
serializer_map = {
"create": SolicitacaoDietaEspecialCreateSerializer,
"relatorio_dieta_especial": SolicitacaoDietaEspecialSimplesSerializer,
"relatorio_dieta_especial_terceirizada": SolicitacaoDietaEspecialRelatorioTercSerializer,
"alteracao_ue": AlteracaoUESerializer,
"relatorio-historico-dieta-especial": UnidadeEducacionalSerializer,
"relatorio_recreio_nas_ferias": SolicitacaoDietaEspecialRecreioNasFeriasSerializer,
}
return serializer_map.get(self.action, SolicitacaoDietaEspecialSerializer)
[docs]
def atualiza_solicitacao(self, solicitacao, request):
texto_html = atualiza_historico_protocolo(solicitacao, request.data)
if (
solicitacao.aluno.possui_dieta_especial_ativa
and not solicitacao.tipo_solicitacao == "ALTERACAO_UE"
):
solicitacao.aluno.inativar_dieta_especial()
if not solicitacao.tipo_solicitacao == "ALTERACAO_UE":
serializer = self.get_serializer()
data_copy = deepcopy(request.data)
serializer.update(solicitacao, request.data)
self.verifica_se_aluno_possui_solicitacao_alteracao_ue_autorizada(
request, solicitacao.aluno, serializer, data_copy
)
solicitacao.ativo = True
self.salva_log_transicao(solicitacao, request.user, texto_html)
if solicitacao.aluno.escola:
enviar_email_codae_atualiza_protocolo(solicitacao)
if not solicitacao.data_inicio:
solicitacao.data_inicio = datetime.now().strftime("%Y-%m-%d")
solicitacao.save()
[docs]
def salva_log_transicao(self, solicitacao, user, justificativa):
solicitacao.salvar_log_transicao(
status_evento=LogSolicitacoesUsuario.CODAE_ATUALIZOU_PROTOCOLO,
usuario=user,
justificativa=justificativa,
)
[docs]
@action(
detail=False,
methods=["get"],
url_path=f"solicitacoes-aluno/{FILTRO_CODIGO_EOL_ALUNO}",
)
def solicitacoes_vigentes(self, request, codigo_eol_aluno=None):
codigo_eol_escola = request.query_params.get("codigo_eol_escola")
ativas_view = (
(
SolicitacoesCODAE.get_autorizados_dieta_especial()
| SolicitacoesCODAE.get_autorizadas_temporariamente_dieta_especial()
)
.filter(codigo_eol_aluno=codigo_eol_aluno)
.distinct()
)
inativas_view = (
(
SolicitacoesCODAE.get_cancelados_dieta_especial()
| SolicitacoesCODAE.get_inativas_dieta_especial()
| SolicitacoesCODAE.get_inativas_temporariamente_dieta_especial()
)
.filter(codigo_eol_aluno=codigo_eol_aluno)
.distinct()
)
dietas_view = (ativas_view | inativas_view).distinct()
if codigo_eol_escola:
escola_uuid = (
Escola.objects.filter(codigo_eol=codigo_eol_escola)
.values_list("uuid", flat=True)
.first()
)
if escola_uuid:
escola_uuid_str = str(escola_uuid)
dietas_view = dietas_view.filter(Q(escola_destino_uuid=escola_uuid_str))
else:
dietas_view = dietas_view.none()
uuids_inativas = set(
str(uuid) for uuid in inativas_view.values_list("uuid", flat=True)
)
uuids_dietas = list(dietas_view.values_list("uuid", flat=True).distinct())
log_mais_recente_subquery = (
LogSolicitacoesUsuario.objects.filter(uuid_original=OuterRef("uuid"))
.order_by("-criado_em")
.values("criado_em")[:1]
)
solicitacoes = (
SolicitacaoDietaEspecial.objects.filter(uuid__in=uuids_dietas)
.annotate(data_log_mais_recente=Subquery(log_mais_recente_subquery))
.order_by("-data_log_mais_recente", "-criado_em")
)
page = self.paginate_queryset(solicitacoes)
for solicitacao in page:
if str(solicitacao.uuid) in uuids_inativas:
solicitacao.ativo = False
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
[docs]
@action(
detail=False,
methods=("GET",),
url_path="solicitacoes-aluno-nao-matriculado",
)
def solicitacoes_vigentes_aluno_nao_matriculado(self, request):
try:
codigo_eol_escola = request.query_params.get("codigo_eol_escola", None)
nome_aluno = request.query_params.get("nome_aluno", False)
if not codigo_eol_escola:
raise ValidationError(
"`codigo_eol_escola` como query_param é obrigatório"
)
if not nome_aluno:
raise ValidationError("`nome_aluno` como query_param é obrigatório")
solicitacoes = (
SolicitacaoDietaEspecial.objects.filter(
aluno__nao_matriculado=True,
escola_destino__codigo_eol=codigo_eol_escola,
aluno__nome=nome_aluno,
tipo_solicitacao="ALUNO_NAO_MATRICULADO",
)
.exclude(
status=SolicitacaoDietaEspecial.workflow_class.CODAE_A_AUTORIZAR
)
.order_by("-criado_em")
)
page = self.paginate_queryset(solicitacoes)
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
except ValidationError as e:
return Response(e, status=status.HTTP_400_BAD_REQUEST)
[docs]
def verifica_se_aluno_possui_solicitacao_alteracao_ue_autorizada(
self, request, aluno, serializer, data
):
dieta_alteracao_ue_autorizada = (
aluno.dietas_especiais.filter(
status=SolicitacaoDietaEspecial.workflow_class.CODAE_AUTORIZADO,
tipo_solicitacao=SolicitacaoDietaEspecial.ALTERACAO_UE,
)
.filter()
.first()
)
if dieta_alteracao_ue_autorizada:
data_sem_datas = {
k: v
for k, v in data.items()
if k not in ("data_inicio", "data_termino")
}
if dieta_alteracao_ue_autorizada.data_termino:
data_sem_datas["data_termino"] = (
dieta_alteracao_ue_autorizada.data_termino.strftime("%Y-%m-%d")
)
serializer.update(dieta_alteracao_ue_autorizada, data_sem_datas)
[docs]
@transaction.atomic
@action(
detail=True, methods=["patch"], permission_classes=(UsuarioCODAEDietaEspecial,)
)
def autorizar(self, request, uuid=None): # noqa C901
solicitacao = self.get_object()
dieta_ativa = solicitacao.aluno.obter_dieta_especial_ativa
if (
dieta_ativa
and solicitacao.tipo_solicitacao == SolicitacaoDietaEspecial.COMUM
):
dieta_inativada = solicitacao.aluno.inativar_dieta_especial()
dieta_inativada.salvar_log_transicao(
status_evento=LogSolicitacoesUsuario.CODAE_INATIVOU,
usuario=request.user,
)
serializer = self.get_serializer()
try:
if solicitacao.tipo_solicitacao != SolicitacaoDietaEspecial.ALTERACAO_UE:
data_copy = deepcopy(request.data)
serializer.update(solicitacao, request.data)
self.verifica_se_aluno_possui_solicitacao_alteracao_ue_autorizada(
request, solicitacao.aluno, serializer, data_copy
)
solicitacao.ativo = True
solicitacao.codae_autoriza(user=request.user)
if not solicitacao.data_inicio:
solicitacao.data_inicio = datetime.now().strftime("%Y-%m-%d")
solicitacao.save()
return Response(
{"detail": "Autorização de Dieta Especial realizada com sucesso!"}
)
except InvalidTransitionError as e:
return Response(
{"detail": f"Erro na transição de estado {e}"},
status=HTTP_400_BAD_REQUEST,
)
except serializers.ValidationError as e:
return Response(
{"detail": f"Dados inválidos {e}"}, status=HTTP_400_BAD_REQUEST
)
[docs]
@transaction.atomic
@action(
detail=True,
methods=["patch"],
url_path=constants.CODAE_ATUALIZA_PROTOCOLO,
permission_classes=(UsuarioCODAEDietaEspecial,),
)
def atualiza_protocolo(self, request, uuid=None):
solicitacao = self.get_object()
try:
self.atualiza_solicitacao(solicitacao, request)
return Response({"detail": "Edição realizada com sucesso!"})
except serializers.ValidationError as e:
return Response(
{"detail": f"Dados inválidos {e}"}, status=HTTP_400_BAD_REQUEST
)
except coreValidation as e:
return Response(
dict(detail=e.messages[0]), status=status.HTTP_400_BAD_REQUEST
)
[docs]
@action(
detail=True,
methods=["POST"],
url_path=constants.ESCOLA_SOLICITA_INATIVACAO,
permission_classes=[UsuarioEscolaTercTotal | UsuarioEscolaDiretaParceira],
)
def escola_solicita_inativacao(self, request, uuid=None):
# TODO: colocar essa lógica dentro de um serializer
dieta_cancelada = self.get_object()
solicitacoes_de_cancelamento = SolicitacaoDietaEspecial.objects.filter(
tipo_solicitacao="CANCELAMENTO_DIETA",
dieta_alterada=dieta_cancelada,
status=SolicitacaoDietaEspecial.workflow_class.ESCOLA_SOLICITOU_INATIVACAO,
)
if solicitacoes_de_cancelamento:
return Response(
dict(
detail="Já existe uma solicitação de cancelamento para essa dieta"
),
status=HTTP_400_BAD_REQUEST,
)
justificativa = request.data.get("justificativa", "")
substituicoes = SubstituicaoAlimento.objects.filter(
solicitacao_dieta_especial=dieta_cancelada
)
anexos = Anexo.objects.filter(solicitacao_dieta_especial=dieta_cancelada)
solicitacao_cancelamento = deepcopy(dieta_cancelada)
solicitacao_cancelamento.id = None
solicitacao_cancelamento.status = None
solicitacao_cancelamento.ativo = False
solicitacao_cancelamento.uuid = uuid_generator.uuid4()
solicitacao_cancelamento.tipo_solicitacao = "CANCELAMENTO_DIETA"
solicitacao_cancelamento.dieta_alterada = dieta_cancelada
solicitacao_cancelamento.save()
solicitacao_cancelamento.alergias_intolerancias.add(
*dieta_cancelada.alergias_intolerancias.all()
)
solicitacao_cancelamento.cria_anexos_inativacao(request.data.get("anexos"))
solicitacao_cancelamento.inicia_fluxo_inativacao(
user=request.user, justificativa=justificativa
)
for substituicao in substituicoes:
substituicao_cancelamento = deepcopy(substituicao)
substituicao_cancelamento.id = None
substituicao_cancelamento.solicitacao_dieta_especial = (
solicitacao_cancelamento
)
substituicao_cancelamento.save()
substituicao_cancelamento.substitutos.add(*substituicao.substitutos.all())
substituicao_cancelamento.alimentos_substitutos.add(
*substituicao.alimentos_substitutos.all()
)
for anexo in anexos:
anexo_cancelamento = deepcopy(anexo)
anexo_cancelamento.id = None
anexo_cancelamento.solicitacao_dieta_especial = solicitacao_cancelamento
anexo_cancelamento.save()
serializer = self.get_serializer(solicitacao_cancelamento)
return Response(serializer.data)
[docs]
@action(
detail=True,
methods=["patch"],
url_path=constants.CODAE_AUTORIZA_INATIVACAO,
permission_classes=(UsuarioCODAEDietaEspecial,),
)
def codae_autoriza_inativacao(self, request, uuid=None):
solicitacao_dieta_especial = self.get_object()
try:
solicitacao_dieta_especial.codae_autoriza_inativacao(user=request.user)
solicitacao_dieta_especial.ativo = False
solicitacao_dieta_especial.save()
if solicitacao_dieta_especial.tipo_solicitacao == "ALTERACAO_UE":
solicitacao_dieta_especial.dieta_alterada.ativo = True
solicitacao_dieta_especial.dieta_alterada.save()
serializer = self.get_serializer(solicitacao_dieta_especial)
return Response(serializer.data)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@action(
detail=True,
methods=["patch"],
url_path=constants.CODAE_NEGA_INATIVACAO,
permission_classes=(UsuarioCODAEDietaEspecial,),
)
def codae_nega_inativacao(self, request, uuid=None):
solicitacao_dieta_especial = self.get_object()
justificativa = request.data.get("justificativa_negacao", "")
try:
solicitacao_dieta_especial.codae_nega_inativacao(
user=request.user, justificativa=justificativa
)
serializer = self.get_serializer(solicitacao_dieta_especial)
return Response(serializer.data)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@action(
detail=True,
methods=["patch"],
url_path=constants.TERCEIRIZADA_TOMOU_CIENCIA_INATIVACAO,
permission_classes=(UsuarioEmpresaGenerico,),
)
def terceirizada_toma_ciencia_inativacao(self, request, uuid=None):
solicitacao_dieta_especial = self.get_object()
try:
solicitacao_dieta_especial.terceirizada_toma_ciencia_inativacao(
user=request.user
)
serializer = self.get_serializer(solicitacao_dieta_especial)
return Response(serializer.data)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@action(
detail=True, methods=["post"], permission_classes=(UsuarioCODAEDietaEspecial,)
)
def negar(self, request, uuid=None):
solicitacao = self.get_object()
form = NegaDietaEspecialForm(request.data, instance=solicitacao)
if not form.is_valid():
return Response(form.errors)
solicitacao.codae_nega(user=request.user)
return Response({"mensagem": "Solicitação de Dieta Especial Negada"})
[docs]
@action(detail=True, methods=["post"], permission_classes=(UsuarioEmpresaGenerico,))
def tomar_ciencia(self, request, uuid=None):
solicitacao = self.get_object()
try:
solicitacao.terceirizada_toma_ciencia(user=request.user)
return Response({"mensagem": "Ciente da solicitação de dieta especial"})
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
) # noqa
[docs]
@action(
detail=True,
url_path=constants.RELATORIO,
methods=["get"],
permission_classes=(IsAuthenticated,),
)
def relatorio(self, request, uuid=None):
return relatorio_dieta_especial(request, solicitacao=self.get_object())
[docs]
@action(
detail=True,
url_path=constants.RELATORIO_HISTORICO_DIETA,
methods=["get"],
permission_classes=(IsAuthenticated,),
)
def relatorio_historico(self, request, uuid=None):
return relatorio_dieta_especial_historico(
request, solicitacao=self.get_object()
)
[docs]
@action(
detail=True,
url_path=constants.PROTOCOLO,
methods=["get"],
permission_classes=(IsAuthenticated,),
)
def protocolo(self, request, uuid=None):
sem_foto = request.query_params.get("sem_foto", "false").lower() in [
"true",
"1",
"yes",
]
return relatorio_dieta_especial_protocolo(
request, solicitacao=self.get_object(), sem_foto=sem_foto
)
[docs]
@action(
detail=True, methods=["post"], url_path=constants.ESCOLA_CANCELA_DIETA_ESPECIAL
) # noqa C901
def escola_cancela_solicitacao(self, request, uuid=None):
justificativa = request.data.get("justificativa", "")
solicitacao = self.get_object()
alta_medica = (
solicitacao.status
== DietaEspecialWorkflow.states.ESCOLA_SOLICITOU_INATIVACAO
)
pendente_autorizacao = (
solicitacao.status == DietaEspecialWorkflow.states.CODAE_A_AUTORIZAR
)
try:
solicitacao.cancelar_pedido(
user=request.user,
justificativa=justificativa,
alta_medica=alta_medica,
pendente_autorizacao=pendente_autorizacao,
)
solicitacao.ativo = False
solicitacao.save()
if solicitacao.tipo_solicitacao == "CANCELAMENTO_DIETA":
if request.user.tipo_usuario == "escola":
solicitacao.dieta_alterada.ativo = True
else:
solicitacao.dieta_alterada.cancelar_pedido(
user=request.user,
justificativa=solicitacao.logs.first().justificativa,
alta_medica=alta_medica,
)
solicitacao.dieta_alterada.save()
if solicitacao.tipo_solicitacao == "ALTERACAO_UE":
solicitacao.dieta_alterada.ativo = True
solicitacao.dieta_alterada.save()
serializer = self.get_serializer(solicitacao)
return Response(serializer.data)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
) # noqa
[docs]
@action(
detail=True, methods=["post"], url_path=constants.CODAE_NEGA_CANCELAMENTO_DIETA
)
def codae_nega_cancelamento(self, request, uuid=None):
justificativa = request.data.get("justificativa", "")
motivo_negacao = request.data.get("motivo_negacao", "")
solicitacao = self.get_object()
try:
solicitacao.negar_cancelamento_pedido(
user=request.user, justificativa=justificativa
)
solicitacao.motivo_negacao = MotivoNegacao.objects.get(id=motivo_negacao)
solicitacao.save()
serializer = self.get_serializer(solicitacao)
return Response(serializer.data)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@action(
detail=True,
methods=["patch"],
url_path=constants.MARCAR_CONFERIDA,
permission_classes=(UsuarioEmpresaGenerico,),
)
def terceirizada_marca_solicitacao_como_conferida(self, request, uuid=None):
solicitacao_dieta_especial: SolicitacaoDietaEspecial = self.get_object()
try:
solicitacao_dieta_especial.conferido = True
solicitacao_dieta_especial.save()
serializer = self.get_serializer(solicitacao_dieta_especial)
return Response(serializer.data)
except Exception as e:
return Response(
dict(detail=f"Erro ao marcar solicitação como conferida: {e}"),
status=HTTP_400_BAD_REQUEST,
) # noqa
[docs]
def get_queryset_relatorio_quantitativo_solic_dieta_esp( # noqa: C901
self, form, campos
):
user = self.request.user
qs = self.get_queryset()
if user.tipo_usuario == "escola":
qs = qs.filter(aluno__escola=user.vinculo_atual.instituicao)
elif form.cleaned_data["escola"]:
qs = qs.filter(aluno__escola__in=form.cleaned_data["escola"])
elif user.tipo_usuario == "diretoriaregional":
qs = qs.filter(
aluno__escola__diretoria_regional=user.vinculo_atual.instituicao
) # noqa
elif form.cleaned_data["dre"]:
qs = qs.filter(
aluno__escola__diretoria_regional__in=form.cleaned_data["dre"]
) # noqa
if form.cleaned_data["data_inicial"]:
qs = qs.filter(
criado_em__date__gte=form.cleaned_data["data_inicial"]
) # noqa
if form.cleaned_data["data_final"]:
qs = qs.filter(criado_em__date__lte=form.cleaned_data["data_final"]) # noqa
if form.cleaned_data["diagnostico"]:
qs = qs.filter(
alergias_intolerancias__in=form.cleaned_data["diagnostico"]
) # noqa
if form.cleaned_data["classificacao"]:
qs = qs.filter(classificacao__in=form.cleaned_data["classificacao"]) # noqa
STATUS_PENDENTE = ["CODAE_A_AUTORIZAR"]
STATUS_ATIVA = [
"CODAE_AUTORIZADO",
"TERCEIRIZADA_TOMOU_CIENCIA",
"ESCOLA_SOLICITOU_INATIVACAO",
"CODAE_NEGOU_INATIVACAO",
]
STATUS_INATIVA = [
"CODAE_AUTORIZOU_INATIVACAO",
"TERCEIRIZADA_TOMOU_CIENCIA_INATIVACAO",
"TERMINADA_AUTOMATICAMENTE_SISTEMA",
"CANCELADO_ALUNO_NAO_PERTENCE_REDE",
]
when_data = {
"ativas": When(status__in=STATUS_ATIVA, then=Value("Ativa")),
"inativas": When(status__in=STATUS_INATIVA, then=Value("Inativa")),
"pendentes": When(status__in=STATUS_PENDENTE, then=Value("Pendente")),
}
if form.cleaned_data["status"] == "":
whens = when_data.values()
else:
whens = [when_data[form.cleaned_data["status"]]]
qs = (
qs.annotate(
status_simples=Case(
*whens, default=Value("Outros"), output_field=CharField()
)
)
.exclude(status_simples="Outros")
.values(*campos)
.annotate(
qtde_ativas=Count(
"status_simples", filter=Q(status_simples="Ativa")
), # noqa
qtde_inativas=Count(
"status_simples", filter=Q(status_simples="Inativa")
), # noqa
qtde_pendentes=Count(
"status_simples", filter=Q(status_simples="Pendente")
),
)
.order_by(*campos)
)
return qs
[docs]
def get_campos_relatorio_quantitativo_solic_dieta_esp(self, filtros):
campos = ["aluno__escola__diretoria_regional__nome"]
if len(filtros["escola"]) > 0:
campos.append("aluno__escola__nome")
return campos
[docs]
def get_campos_relatorio_quantitativo_diag_dieta_esp(self, filtros):
user = self.request.user
campos = []
if (
user.tipo_usuario != "diretoriaregional" and len(filtros["escola"]) == 0
): # noqa
campos.append("aluno__escola__diretoria_regional__nome")
else:
if user.tipo_usuario != "diretoriaregional":
campos.append("aluno__escola__diretoria_regional__nome")
if len(filtros["escola"]) > 0:
campos.append("aluno__escola__nome")
campos.append("alergias_intolerancias__descricao")
campos.append("aluno__data_nascimento__year")
return campos
[docs]
def get_campos_relatorio_quantitativo_classificacao_dieta_esp(self, filtros):
user = self.request.user
campos = []
if (
user.tipo_usuario != "diretoriaregional" and len(filtros["escola"]) == 0
): # noqa
campos.append("aluno__escola__diretoria_regional__nome")
else:
if user.tipo_usuario != "diretoriaregional":
campos.append("aluno__escola__diretoria_regional__nome")
if len(filtros["escola"]) > 0:
campos.append("aluno__escola__nome")
campos.append("classificacao__nome")
return campos
[docs]
@action(
detail=False,
methods=["POST"],
url_path="relatorio-quantitativo-solic-dieta-esp",
)
def relatorio_quantitativo_solic_dieta_esp(self, request):
form = RelatorioQuantitativoSolicDietaEspForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
campos = self.get_campos_relatorio_quantitativo_solic_dieta_esp(
form.cleaned_data
)
qs = self.get_queryset_relatorio_quantitativo_solic_dieta_esp(form, campos)
self.pagination_class = RelatorioPagination
page = self.paginate_queryset(qs)
serializer = self.get_serializer(page if page is not None else qs, many=True)
return self.get_paginated_response(serializer.data)
[docs]
@action(
detail=False, methods=["POST"], url_path="relatorio-quantitativo-diag-dieta-esp"
)
def relatorio_quantitativo_diag_dieta_esp(self, request):
form = RelatorioQuantitativoSolicDietaEspForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
if self.request.data.get("somente_dietas_ativas"):
campos = ["alergias_intolerancias__descricao"]
else:
campos = self.get_campos_relatorio_quantitativo_diag_dieta_esp(
form.cleaned_data
) # noqa
qs = self.get_queryset_relatorio_quantitativo_solic_dieta_esp(
form, campos
) # noqa
self.pagination_class = RelatorioPagination
page = self.paginate_queryset(qs)
serializer = self.get_serializer(page if page is not None else qs, many=True)
return self.get_paginated_response(serializer.data)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="relatorio-quantitativo-classificacao-dieta-esp",
)
def relatorio_quantitativo_classificacao_dieta_esp(self, request):
form = RelatorioQuantitativoSolicDietaEspForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
campos = self.get_campos_relatorio_quantitativo_classificacao_dieta_esp(
form.cleaned_data
) # noqa
qs = self.get_queryset_relatorio_quantitativo_solic_dieta_esp(
form, campos
) # noqa
self.pagination_class = RelatorioPagination
page = self.paginate_queryset(qs)
serializer = self.get_serializer(page if page is not None else qs, many=True)
return self.get_paginated_response(serializer.data)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="imprime-relatorio-quantitativo-solic-dieta-esp",
)
def imprime_relatorio_quantitativo_solic_dieta_esp(self, request):
form = RelatorioQuantitativoSolicDietaEspForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
campos = self.get_campos_relatorio_quantitativo_solic_dieta_esp(
form.cleaned_data
)
qs = self.get_queryset_relatorio_quantitativo_solic_dieta_esp(form, campos)
user = self.request.user
return relatorio_quantitativo_solic_dieta_especial(campos, form, qs, user)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="imprime-relatorio-quantitativo-classificacao-dieta-esp",
)
def imprime_relatorio_quantitativo_classificacao_dieta_esp(self, request):
form = RelatorioQuantitativoSolicDietaEspForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
campos = self.get_campos_relatorio_quantitativo_classificacao_dieta_esp(
form.cleaned_data
)
qs = self.get_queryset_relatorio_quantitativo_solic_dieta_esp(form, campos)
user = self.request.user
return relatorio_quantitativo_classificacao_dieta_especial(
campos, form, qs, user
)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="imprime-relatorio-quantitativo-diag-dieta-esp",
)
def imprime_relatorio_quantitativo_diag_dieta_esp(self, request):
form = RelatorioQuantitativoSolicDietaEspForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
campos = self.get_campos_relatorio_quantitativo_diag_dieta_esp(
form.cleaned_data
) # noqa
qs = self.get_queryset_relatorio_quantitativo_solic_dieta_esp(
form, campos
) # noqa
user = self.request.user
return relatorio_quantitativo_diag_dieta_especial(campos, form, qs, user)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="imprime-relatorio-quantitativo-diag-dieta-esp/somente-dietas-ativas",
)
def imprime_relatorio_quantitativo_diag_dieta_esp_somente_dietas_ativas(
self, request
): # noqa
form = RelatorioQuantitativoSolicDietaEspForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
if self.request.data.get("somente_dietas_ativas"):
campos = ["alergias_intolerancias__descricao"]
else:
campos = self.get_campos_relatorio_quantitativo_diag_dieta_esp(
form.cleaned_data
) # noqa
qs = self.get_queryset_relatorio_quantitativo_solic_dieta_esp(
form, campos
) # noqa
user = self.request.user
return relatorio_quantitativo_diag_dieta_especial_somente_dietas_ativas(
campos, form, qs, user
) # noqa
[docs]
@action(
detail=False, methods=["POST"], url_path="relatorio-dieta-especial"
) # noqa C901
def relatorio_dieta_especial(self, request):
self.pagination_class = RelatorioPagination
form = RelatorioDietaForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
queryset = self.filter_queryset(self.get_queryset())
data = form.cleaned_data
filtros = {}
if data["escola"]:
filtros["rastro_escola__uuid__in"] = [
escola.uuid for escola in data["escola"]
]
if data["diagnostico"]:
filtros["alergias_intolerancias__id__in"] = [
disgnostico.id for disgnostico in data["diagnostico"]
]
page = self.paginate_queryset(queryset.filter(**filtros))
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
[docs]
@action(
detail=False, methods=["POST"], url_path="imprime-relatorio-dieta-especial"
) # noqa C901
def imprime_relatorio_dieta_especial(self, request):
try:
form = RelatorioDietaForm(self.request.data)
if not form.is_valid():
raise ValidationError(form.errors)
queryset = self.filter_queryset(self.get_queryset())
data = form.cleaned_data
filtros = {}
if data["escola"]:
filtros["rastro_escola__uuid__in"] = [
escola.uuid for escola in data["escola"]
]
if data["diagnostico"]:
filtros["alergias_intolerancias__id__in"] = [
disgnostico.id for disgnostico in data["diagnostico"]
]
user = request.user.get_username()
ids_dietas = list(queryset.filter(**filtros).values_list("id", flat=True))
gera_pdf_relatorio_dieta_especial_async.delay(
user=user,
nome_arquivo="relatorio_dieta_especial.pdf",
data=self.request.data,
ids_dietas=ids_dietas,
)
return Response(
dict(detail="Solicitação de geração de arquivo recebida com sucesso."),
status=status.HTTP_200_OK,
)
except ValidationError as error:
return Response(error, status=status.HTTP_400_BAD_REQUEST)
[docs]
def filtrar_queryset_polo_recreio_ferias(self, query_set, query_params):
motivos = []
if query_params.get("outro") == "true":
motivos.append("outro")
if not motivos:
return query_set.order_by("motivo_alteracao_ue__nome")
filtro = Q()
for motivo in motivos:
filtro |= Q(motivo_alteracao_ue__nome__icontains=motivo)
return query_set.filter(filtro).order_by("motivo_alteracao_ue__nome")
[docs]
def filtra_por_serie(self, request, query_set):
series = request.query_params.getlist("serie")
if series:
q_obj = Q()
for s in series:
q_obj |= Q(aluno__serie__icontains=s)
query_set = query_set.filter(q_obj)
return query_set
[docs]
def filtrar_queryset_relatorio_dieta_especial(self, request, eh_relatorio=False):
query_set = self.filter_queryset(self.get_queryset())
lotes_filtro = request.query_params.getlist("lotes_selecionados[]", None)
instituicao = request.user.vinculo_atual.instituicao
if not lotes_filtro and isinstance(instituicao, DiretoriaRegional):
lotes_list = list(instituicao.lotes.all().values_list("uuid"))
lotes_filtro = [str(u[0]) for u in lotes_list]
campo_escola_destino = (
"escola_destino__uuid__in"
if eh_relatorio
else "escola_destino__codigo_eol__in"
)
map_filtros = {
"protocolo_padrao__uuid__in": request.query_params.getlist(
"protocolos_padrao_selecionados[]", None
),
"alergias_intolerancias__id__in": request.query_params.getlist(
"alergias_intolerancias_selecionadas[]", None
),
"classificacao__id__in": request.query_params.getlist(
"classificacoes_selecionadas[]", None
),
"escola_destino__lote__uuid__in": lotes_filtro,
campo_escola_destino: request.query_params.getlist(
"unidades_educacionais_selecionadas[]", None
),
"escola_destino__codigo_eol": request.query_params.get("codigo_eol"),
"escola_destino__tipo_unidade__uuid__in": request.query_params.getlist(
"tipos_unidades_selecionadas[]", None
),
}
if isinstance(instituicao, DiretoriaRegional):
map_filtros["escola_destino__diretoria_regional__uuid"] = instituicao.uuid
filtros = {
key: value for key, value in map_filtros.items() if value not in [None, []]
}
query_set = query_set.filter(**filtros)
query_set = self.filtra_por_serie(request, query_set)
data = request.query_params
query_set = self.filtrar_queryset_polo_recreio_ferias(query_set, data)
if data.get("status_selecionado") == "AUTORIZADAS":
query_set = query_set.filter(
status=SolicitacaoDietaEspecial.workflow_class.states.CODAE_AUTORIZADO,
ativo=True,
)
elif data.get("status_selecionado") == "CANCELADAS":
data_inicial_str = data.get("data_cancelamento_inicial")
data_final_str = data.get("data_cancelamento_final")
formato = "%d/%m/%Y"
data_inicial = (
datetime.strptime(data_inicial_str, formato)
if data_inicial_str
else None
)
data_final = (
datetime.strptime(data_final_str, formato) if data_final_str else None
)
params = {
"criado_em__gte": data_inicial,
"criado_em__lte": data_final,
"status_evento__in": [
LogSolicitacoesUsuario.ESCOLA_CANCELOU,
LogSolicitacoesUsuario.CODAE_AUTORIZOU_CANCELAMENTO_DIETA_ESPECIAL,
LogSolicitacoesUsuario.CODAE_AUTORIZOU_INATIVACAO,
LogSolicitacoesUsuario.CANCELADO_ALUNO_MUDOU_ESCOLA,
LogSolicitacoesUsuario.CANCELADO_ALUNO_NAO_PERTENCE_REDE,
LogSolicitacoesUsuario.TERMINADA_AUTOMATICAMENTE_SISTEMA,
],
}
log_filtros = {
key: value for key, value in params.items() if value is not None
}
logs = (
LogSolicitacoesUsuario.objects.filter(**log_filtros)
.values("uuid_original")
.distinct()
)
query_set = query_set.filter(uuid__in=Subquery(logs))
return query_set
[docs]
@action(detail=False, methods=("get",), url_path="filtros-relatorio-dieta-especial")
def filtros_relatorio_dieta_especial(self, request):
query_set = self.filtrar_queryset_relatorio_dieta_especial(request)
instituicao = request.user.vinculo_atual.instituicao
lotes_dres = []
if isinstance(instituicao, DiretoriaRegional):
for lote in instituicao.lotes.all().values_list("nome", "uuid"):
lotes_dres.append(tuple([f"{lote[0]} - {instituicao.nome}", lote[1]]))
lotes_dict = dict(lotes_dres)
else:
for s in query_set.values_list(
"escola_destino__lote__nome",
"escola_destino__lote__uuid",
"escola_destino__diretoria_regional__nome",
):
lotes_dres.append(tuple([f"{s[0]} - {s[2]}", s[1]]))
lotes_dict = dict(lotes_dres)
lotes_dict = trata_lotes_dict_duplicados(lotes_dict)
lotes = sorted(
[
{"nome": key, "uuid": lotes_dict[key]}
for key in lotes_dict.keys()
if key
],
key=lambda lote: lote["nome"],
)
classificacoes_dict = dict(
set(query_set.values_list("classificacao__nome", "classificacao__id"))
)
classificacoes = sorted(
[
{"nome": key, "id": classificacoes_dict[key]}
for key in classificacoes_dict.keys()
if key
],
key=lambda classificacao: classificacao["nome"],
)
protocolos_padrao_dict = dict(
set(
query_set.values_list(
"protocolo_padrao__nome_protocolo", "protocolo_padrao__uuid"
)
)
)
protocolos_padrao = sorted(
[
{"nome": key, "uuid": protocolos_padrao_dict[key]}
for key in protocolos_padrao_dict.keys()
if key
],
key=lambda protocolo_padrao: protocolo_padrao["nome"],
)
alergias_intolerancias_dict = dict(
set(
query_set.values_list(
"alergias_intolerancias__descricao", "alergias_intolerancias__id"
)
)
)
alergias_intolerancias = sorted(
[
{"nome": key, "id": alergias_intolerancias_dict[key]}
for key in alergias_intolerancias_dict.keys()
if key
],
key=lambda alergia_intolerancia: alergia_intolerancia["nome"],
)
tipos_gestao_dict = dict(
set(
TipoGestao.objects.exclude(nome__in=["DIRETA", "MISTA"]).values_list(
"nome", "uuid"
)
)
)
tipos_gestao = [
{"nome": key, "uuid": tipos_gestao_dict[key]}
for key in tipos_gestao_dict.keys()
if key
]
tipos_unidades_dict = dict(
set(
query_set.values_list(
"escola_destino__tipo_unidade__iniciais",
"escola_destino__tipo_unidade__uuid",
)
)
)
tipos_unidades = [
{"nome": key, "uuid": tipos_unidades_dict[key]}
for key in tipos_unidades_dict.keys()
if key
]
return Response(
{
"alergias_intolerancias": alergias_intolerancias,
"classificacoes": classificacoes,
"lotes": lotes,
"protocolos_padrao": protocolos_padrao,
"tipos_gestao": tipos_gestao,
"tipos_unidades": tipos_unidades,
},
status=status.HTTP_200_OK,
)
[docs]
@action(
detail=False, methods=("get",), url_path="relatorio-dieta-especial-terceirizada"
)
def relatorio_dieta_especial_terceirizada(self, request):
query_set = self.filtrar_queryset_relatorio_dieta_especial(
request, True
).distinct()
page = self.paginate_queryset(query_set)
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
[docs]
@action(detail=False, methods=["POST"], url_path="alteracao-ue")
def alteracao_ue(self, request):
serializer = self.get_serializer(data=request.data)
if serializer.is_valid():
serializer.save()
return Response(status=HTTP_201_CREATED)
else:
return Response(serializer.errors, status=HTTP_400_BAD_REQUEST)
[docs]
@action(detail=False, methods=["GET"], url_path="exportar-xlsx")
def exportar_xlsx(self, request):
"""TODO: ver porque nao pode importar o pandas via pytest."""
query_set = self.filtrar_queryset_relatorio_dieta_especial(request, True)
data = request.query_params
user = request.user.get_username()
ids_dietas = list(query_set.values_list("id", flat=True))
lotes = data.getlist("lotes_selecionados[]", None)
classificacoes = data.getlist("classificacoes_selecionadas[]", None)
protocolos_padrao = data.getlist("protocolos_padrao_selecionados[]", None)
gera_xlsx_relatorio_dietas_especiais_terceirizadas_async.delay(
user=user,
nome_arquivo="relatorio_dietas_especiais.xlsx",
ids_dietas=ids_dietas,
data=request.query_params,
lotes=lotes,
classificacoes=classificacoes,
protocolos_padrao=protocolos_padrao,
)
return Response(
dict(detail="Solicitação de geração de arquivo recebida com sucesso."),
status=status.HTTP_200_OK,
)
[docs]
def build_texto(
self,
lotes,
classificacoes,
protocolos,
alergias_intolerancias,
data_inicial,
data_final,
):
partes = []
configs = [
(lotes, Lote, "uuid__in", "nome", None),
(
classificacoes,
ClassificacaoDieta,
"id__in",
"nome",
"Classificação(ões) da dieta",
),
(
protocolos,
ProtocoloPadraoDietaEspecial,
"uuid__in",
"nome_protocolo",
"Protocolo(s) padrão(ões)",
),
(
alergias_intolerancias,
AlergiaIntolerancia,
"id__in",
"descricao",
"Diagnóstico(s) da dieta",
),
]
for valores, model, filtro, field, label in configs:
if not valores:
continue
nomes = model.objects.filter(**{filtro: valores}).values_list(
field, flat=True
)
texto = ", ".join(nomes)
if label:
texto = f"{label}: {texto}"
partes.append(texto)
if data_inicial:
partes.append(f"Data inicial: {data_inicial}")
if data_final:
partes.append(f"Data final: {data_final}")
return " | ".join(partes)
[docs]
@action(detail=False, methods=["GET"], url_path="exportar-pdf")
def exportar_pdf(self, request):
user = request.user.get_username()
query_set = self.filtrar_queryset_relatorio_dieta_especial(request, True)
ids_dietas = list(query_set.values_list("id", flat=True))
filtros = self.build_texto(
request.query_params.getlist("lotes_selecionados[]", None),
request.query_params.getlist("classificacoes_selecionadas[]", None),
request.query_params.getlist("protocolos_padrao_selecionados[]", None),
request.query_params.getlist("alergias_intolerancias_selecionadas[]", None),
request.query_params.get("data_cancelamento_inicial", None),
request.query_params.get("data_cancelamento_final", None),
)
gera_pdf_relatorio_dietas_especiais_terceirizadas_async.delay(
user=user,
data=request.query_params,
nome_arquivo="relatorio_dietas_especiais.pdf",
ids_dietas=ids_dietas,
filtros=filtros,
)
return Response(
dict(detail="Solicitação de geração de arquivo recebida com sucesso."),
status=status.HTTP_200_OK,
)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="relatorio-historico-dieta-especial",
permission_classes=(PermissaoHistoricoDietasEspeciais,),
)
def relatorio_historico_dieta_especial(self, request):
try:
filtros, data_dieta = gerar_filtros_relatorio_historico(
convert_dict_to_querydict(request.data)
)
dietas = gera_dicionario_historico_dietas(filtros)
paginator = HistoricoDietasPagination()
page = paginator.paginate_queryset(dietas["resultados"], request)
serializer = UnidadeEducacionalSerializer(page, many=True)
return paginator.get_paginated_response(
serializer.data, dietas["total_dietas"], data_dieta
)
except ValidationError as e:
return Response(dict(detail=e.messages[0]), status=HTTP_400_BAD_REQUEST)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="relatorio-historico-dieta-especial/exportar-pdf",
permission_classes=(PermissaoHistoricoDietasEspeciais,),
)
def relatorio_historico_dieta_especial_exportar_pdf(self, request):
try:
query_dict = request.data
data = {
key: query_dict.getlist(key) if "[]" in key else query_dict.get(key)
for key in query_dict.keys()
}
data_str = data.get("data").replace("/", "_")
user = request.user.get_username()
gera_pdf_relatorio_historico_dietas_especiais_async.delay(
user=user,
nome_arquivo=f"relatorio_historico_dietas_especiais_{data_str}.pdf",
data=json.dumps(data),
)
return Response(
dict(detail="Solicitação de geração de arquivo recebida com sucesso."),
status=status.HTTP_200_OK,
)
except ValidationError as e:
return Response(dict(detail=e.messages[0]), status=HTTP_400_BAD_REQUEST)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="relatorio-historico-dieta-especial/exportar-excel",
permission_classes=(PermissaoHistoricoDietasEspeciais,),
)
def relatorio_historico_dieta_especial_exportar_excel(self, request):
try:
query_dict = request.data
data = {
key: query_dict.getlist(key) if "[]" in key else query_dict.get(key)
for key in query_dict.keys()
}
data_str = data.get("data").replace("/", "_")
user = request.user.get_username()
gera_xlsx_relatorio_historico_dietas_especiais_async.delay(
user=user,
nome_arquivo=f"relatorio_historico_dietas_especiais_{data_str}.xlsx",
data=json.dumps(data),
)
return Response(
dict(detail="Solicitação de geração de arquivo recebida com sucesso."),
status=status.HTTP_200_OK,
)
except ValidationError as e:
return Response(dict(detail=e.messages[0]), status=HTTP_400_BAD_REQUEST)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="relatorio-recreio-nas-ferias",
permission_classes=(PermissaoRelatorioRecreioNasFerias,),
)
def relatorio_recreio_nas_ferias(self, request):
self.pagination_class = RelatorioPagination
try:
query_params = convert_dict_to_querydict(request.data)
for pag_param in ["page", "page_size"]:
if pag_param in request.data and pag_param not in request.query_params:
request._request.GET = request._request.GET.copy()
request._request.GET[pag_param] = request.data[pag_param]
queryset = filtra_relatorio_recreio_nas_ferias(query_params)
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
return self.get_paginated_response(serializer.data)
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
except ValidationError as e:
return Response(dict(detail=e.messages[0]), status=HTTP_400_BAD_REQUEST)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="relatorio-recreio-nas-ferias/exportar-pdf",
permission_classes=(PermissaoRelatorioRecreioNasFerias,),
)
def relatorio_recreio_nas_ferias_exportar_pdf(self, request):
try:
user = request.user.get_username()
gera_pdf_relatorio_recreio_nas_ferias_async.delay(
user=user,
nome_arquivo="relatorio_recreio_nas_ferias.pdf",
params=request.data,
)
return Response(
dict(detail="Solicitação de geração de arquivo recebida com sucesso."),
status=status.HTTP_200_OK,
)
except ValidationError as e:
return Response(dict(detail=e.messages[0]), status=HTTP_400_BAD_REQUEST)
[docs]
@action(
detail=False,
methods=["POST"],
url_path="relatorio-recreio-nas-ferias/exportar-excel",
permission_classes=(PermissaoRelatorioRecreioNasFerias,),
)
def relatorio_recreio_nas_ferias_exportar_xlsx(self, request):
try:
user = request.user.get_username()
gera_xlsx_relatorio_recreio_nas_ferias_async.delay(
user=user,
nome_arquivo="relatorio_recreio_nas_ferias.xlsx",
params=request.data,
)
return Response(
dict(detail="Solicitação de geração de arquivo recebida com sucesso."),
status=status.HTTP_200_OK,
)
except ValidationError as e:
return Response(dict(detail=e.messages[0]), status=HTTP_400_BAD_REQUEST)
MSG_DRE_NAO_INFORMADA = "(DRE não informada)"
[docs]
class SolicitacoesAtivasInativasPorAlunoView(generics.ListAPIView):
serializer_class = SolicitacoesAtivasInativasPorAlunoSerializer
def _qs_ativas_view(self):
return (
SolicitacoesCODAE.get_autorizados_dieta_especial()
| SolicitacoesCODAE.get_autorizadas_temporariamente_dieta_especial()
).distinct()
def _qs_inativas_view(self):
return (
SolicitacoesCODAE.get_cancelados_dieta_especial()
| SolicitacoesCODAE.get_inativas_dieta_especial()
| SolicitacoesCODAE.get_inativas_temporariamente_dieta_especial()
).distinct()
[docs]
def aplicar_filtros_escola_view(
self,
qs,
escola_filtrada=None,
codigo_eol_escola=None,
dre_filtrada=None,
user=None,
):
if escola_filtrada:
return self._filtrar_por_escola(qs, escola_filtrada)
if codigo_eol_escola:
escola_uuid = (
Escola.objects.filter(codigo_eol=codigo_eol_escola)
.values_list("uuid", flat=True)
.first()
)
if escola_uuid:
escola_uuid_str = str(escola_uuid)
return qs.filter(
Q(escola_uuid=escola_uuid_str)
| Q(escola_destino_uuid=escola_uuid_str)
)
return qs.none()
if user:
return self._filtrar_por_usuario(qs, user)
if dre_filtrada:
return self._filtrar_por_dre(qs, dre_filtrada)
return qs
def _filtrar_por_escola(self, qs, escola):
escola_uuid = getattr(escola, "uuid", escola)
return qs.filter(escola_uuid=escola_uuid)
def _filtrar_por_dre(self, qs, dre):
dre_uuid = getattr(dre, "uuid", dre)
return qs.filter(dre_uuid=dre_uuid)
def _filtrar_por_usuario(self, qs, user):
tipo_usuario = getattr(user, "tipo_usuario", None)
if tipo_usuario == "escola":
return self._filtrar_por_escola_usuario(qs, user)
if tipo_usuario == "diretoriaregional":
return self._filtrar_por_dre_usuario(qs, user)
return qs
def _filtrar_por_escola_usuario(self, qs, user):
instituicao = getattr(user.vinculo_atual, "instituicao", None)
escola_uuid = getattr(instituicao, "uuid", None)
if escola_uuid:
return qs.filter(escola_uuid=escola_uuid)
return qs
def _filtrar_por_dre_usuario(self, qs, user):
instituicao = getattr(user.vinculo_atual, "instituicao", None)
dre_uuid = getattr(instituicao, "uuid", None)
if dre_uuid:
return qs.filter(dre_uuid=dre_uuid)
return qs
[docs]
def calcular_totais(self, alunos_qs, **filtros):
codigos_eol = alunos_qs.values_list("codigo_eol", flat=True).distinct()
nomes_alunos = alunos_qs.values_list("nome", flat=True).distinct()
ativas_qs = self._qs_ativas_view().filter(
Q(codigo_eol_aluno__in=codigos_eol)
| Q(nome_aluno__in=nomes_alunos, codigo_eol_aluno__isnull=True)
)
ativas_qs = self.aplicar_filtros_escola_view(ativas_qs, **filtros)
total_ativas = ativas_qs.values("uuid").distinct().count()
inativas_qs = self._qs_inativas_view().filter(
Q(codigo_eol_aluno__in=codigos_eol)
| Q(nome_aluno__in=nomes_alunos, codigo_eol_aluno__isnull=True)
)
inativas_qs = self.aplicar_filtros_escola_view(inativas_qs, **filtros)
total_inativas = inativas_qs.values("uuid").distinct().count()
return total_ativas, total_inativas
[docs]
def list(self, request, *args, **kwargs):
queryset = self.filter_queryset(self.get_queryset())
alunos_ids = queryset.values_list("id", flat=True).distinct()
alunos_qs = Aluno.objects.filter(id__in=alunos_ids)
filtros = self._obter_filtros(request)
total_ativas, total_inativas = self.calcular_totais(alunos_qs, **filtros)
escolas_por_aluno = self._obter_escolas_por_aluno(alunos_qs, filtros)
alunos_por_escola = self._criar_alunos_por_escola(alunos_qs, escolas_por_aluno)
return self._criar_resposta(
request, alunos_por_escola, total_ativas, total_inativas
)
def _obter_filtros(self, request):
form = SolicitacoesAtivasInativasPorAlunoForm(request.GET)
form.is_valid()
return {
"escola_filtrada": form.cleaned_data.get("escola"),
"dre_filtrada": form.cleaned_data.get("dre"),
"codigo_eol_escola": request.query_params.get("codigo_eol_escola"),
"user": request.user,
}
def _obter_escolas_por_aluno(self, alunos_qs, filtros):
codigos_eol = alunos_qs.values_list("codigo_eol", flat=True).distinct()
nomes_alunos = alunos_qs.values_list("nome", flat=True).distinct()
dietas_relevantes = (self._qs_ativas_view() | self._qs_inativas_view()).filter(
Q(codigo_eol_aluno__in=codigos_eol)
| Q(nome_aluno__in=nomes_alunos, codigo_eol_aluno__isnull=True)
)
dietas_relevantes = self.aplicar_filtros_escola_view(
dietas_relevantes, **filtros
)
dietas_data = list(
dietas_relevantes.values(
"codigo_eol_aluno",
"nome_aluno",
"escola_uuid",
"escola_nome",
"dre_nome",
"tipo_solicitacao_dieta",
"escola_destino_uuid",
).distinct()
)
escolas_map = self._criar_mapa_escolas(dietas_data)
return self._agrupar_escolas_por_aluno(dietas_data, escolas_map)
def _criar_mapa_escolas(self, dietas_data):
todas_escolas_uuids = set()
for row in dietas_data:
escola_uuid = row.get("escola_uuid")
if escola_uuid:
todas_escolas_uuids.add(str(escola_uuid))
destino_uuid = row.get("escola_destino_uuid")
if destino_uuid:
todas_escolas_uuids.add(str(destino_uuid))
escolas_map = {}
if todas_escolas_uuids:
escolas = Escola.objects.filter(
uuid__in=todas_escolas_uuids
).select_related("diretoria_regional")
for escola in escolas:
escolas_map[str(escola.uuid)] = {
"nome": escola.nome,
"dre": (
escola.diretoria_regional.nome
if escola.diretoria_regional
else MSG_DRE_NAO_INFORMADA
),
"codigo_eol": escola.codigo_eol,
}
return escolas_map
def _agrupar_escolas_por_aluno(self, dietas_data, escolas_map):
escolas_por_aluno = defaultdict(set)
for row in dietas_data:
eol = row.get("codigo_eol_aluno")
nome_aluno = row.get("nome_aluno")
chave_aluno = eol if eol else f"nome_{nome_aluno}"
if not chave_aluno:
continue
self._adicionar_escola_origem(
escolas_por_aluno, row, escolas_map, chave_aluno
)
self._adicionar_escola_destino(
escolas_por_aluno, row, escolas_map, chave_aluno
)
return escolas_por_aluno
def _adicionar_escola_origem(self, escolas_por_aluno, row, escolas_map, eol):
escola_uuid = row.get("escola_uuid")
if not escola_uuid:
return
escola_uuid_str = str(escola_uuid)
escola_info = escolas_map.get(
escola_uuid_str,
{
"nome": row.get("escola_nome") or "(Escola não encontrada)",
"dre": row.get("dre_nome") or MSG_DRE_NAO_INFORMADA,
"codigo_eol": None,
},
)
escolas_por_aluno[eol].add(
(
escola_uuid_str,
escola_info["nome"],
escola_info["dre"],
escola_info["codigo_eol"],
)
)
def _adicionar_escola_destino(self, escolas_por_aluno, row, escolas_map, eol):
if row.get("tipo_solicitacao_dieta") != "ALTERACAO_UE":
return
escola_destino_uuid = row.get("escola_destino_uuid")
if not escola_destino_uuid:
return
escola_destino_uuid_str = str(escola_destino_uuid)
escola_info = escolas_map.get(
escola_destino_uuid_str,
{
"nome": "(Escola não encontrada)",
"dre": MSG_DRE_NAO_INFORMADA,
"codigo_eol": None,
},
)
escolas_por_aluno[eol].add(
(
escola_destino_uuid_str,
escola_info["nome"],
escola_info["dre"],
escola_info["codigo_eol"],
)
)
def _criar_alunos_por_escola(self, alunos_qs, escolas_por_aluno):
alunos_por_escola = []
for aluno in alunos_qs:
chave_aluno = aluno.codigo_eol if aluno.codigo_eol else f"nome_{aluno.nome}"
escolas_aluno = escolas_por_aluno.get(chave_aluno, set())
if not escolas_aluno:
alunos_por_escola.append(self._criar_copia_aluno_sem_escola(aluno))
else:
for escola_uuid, escola_nome, dre_nome, codigo_eol in escolas_aluno:
alunos_por_escola.append(
self._criar_copia_aluno_com_escola(
aluno, escola_uuid, escola_nome, dre_nome, codigo_eol
)
)
alunos_por_escola.sort(
key=lambda x: ((x._escola_nome or "").upper(), x.nome.upper())
)
return alunos_por_escola
def _criar_copia_aluno_sem_escola(self, aluno):
from copy import copy
copia = copy(aluno)
copia._escola_contexto_id = None
copia._escola_nome = "(Sem escola vinculada)"
copia._escola_dre = MSG_DRE_NAO_INFORMADA
copia._escola_codigo_eol = None
return copia
def _criar_copia_aluno_com_escola(
self, aluno, escola_uuid, escola_nome, dre_nome, codigo_eol
):
from copy import copy
copia = copy(aluno)
copia._escola_contexto_id = escola_uuid
copia._escola_nome = escola_nome
copia._escola_dre = dre_nome
copia._escola_codigo_eol = codigo_eol
return copia
def _criar_resposta(self, request, alunos_por_escola, total_ativas, total_inativas):
paginar = request.GET.get("page", False)
if paginar:
return self._criar_resposta_paginada(
request, alunos_por_escola, total_ativas, total_inativas
)
serializer = self.get_serializer(alunos_por_escola, many=True)
return Response(
{
"total_ativas": total_ativas,
"total_inativas": total_inativas,
"solicitacoes": serializer.data,
}
)
def _criar_resposta_paginada(
self, request, alunos_por_escola, total_ativas, total_inativas
):
self.pagination_class = RelatorioPagination
page = self.paginate_queryset(alunos_por_escola)
serializer = self.get_serializer(page, many=True, context={"request": request})
return self.get_paginated_response(
{
"total_ativas": total_ativas,
"total_inativas": total_inativas,
"solicitacoes": serializer.data,
}
)
[docs]
def get_queryset(self): # noqa C901
form = SolicitacoesAtivasInativasPorAlunoForm(self.request.GET)
if not form.is_valid():
raise ValidationError(form.errors)
user = self.request.user
STATUS_DIETA_ESPECIAL = [
"CODAE_AUTORIZADO",
"CODAE_AUTORIZOU_INATIVACAO",
"TERMINADA_AUTOMATICAMENTE_SISTEMA",
"ESCOLA_CANCELOU",
"CANCELADO_ALUNO_NAO_PERTENCE_REDE",
]
qs = Aluno.objects.filter(dietas_especiais__status__in=STATUS_DIETA_ESPECIAL)
incluir_alteracao_ue = (
self.request.query_params.get("incluir_alteracao_ue", None) == "true"
)
if user.tipo_usuario == "escola":
if incluir_alteracao_ue:
qs = qs.filter(
Q(
dietas_especiais__rastro_escola=user.vinculo_atual.instituicao,
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="COMUM",
)
| Q(
dietas_especiais__escola_destino=user.vinculo_atual.instituicao,
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="ALTERACAO_UE",
)
)
else:
qs = qs.filter(
dietas_especiais__rastro_escola=user.vinculo_atual.instituicao
)
else:
if form.cleaned_data["escola"]:
if incluir_alteracao_ue:
qs = qs.filter(
Q(
dietas_especiais__rastro_escola=form.cleaned_data["escola"],
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="COMUM",
)
| Q(
dietas_especiais__escola_destino=form.cleaned_data[
"escola"
],
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="ALTERACAO_UE",
)
)
else:
qs = qs.filter(
dietas_especiais__rastro_escola=form.cleaned_data["escola"]
)
qs = filtrar_alunos_com_dietas_nos_status_e_rastro_escola(
qs, STATUS_DIETA_ESPECIAL, form.cleaned_data["escola"]
)
elif user.tipo_usuario == "diretoriaregional":
if incluir_alteracao_ue:
qs = qs.filter(
Q(
dietas_especiais__rastro_escola__diretoria_regional=user.vinculo_atual.instituicao,
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="COMUM",
)
| Q(
dietas_especiais__escola_destino__diretoria_regional=user.vinculo_atual.instituicao,
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="ALTERACAO_UE",
)
)
else:
qs = qs.filter(
dietas_especiais__rastro_escola__diretoria_regional=user.vinculo_atual.instituicao
)
elif form.cleaned_data["dre"]:
if incluir_alteracao_ue:
qs = qs.filter(
Q(
dietas_especiais__rastro_escola__diretoria_regional=form.cleaned_data[
"dre"
],
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="COMUM",
)
| Q(
dietas_especiais__escola_destino__diretoria_regional=form.cleaned_data[
"dre"
],
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="ALTERACAO_UE",
)
)
else:
qs = qs.filter(
dietas_especiais__rastro_escola__diretoria_regional=form.cleaned_data[
"dre"
]
)
else:
codigo_eol_escola = self.request.query_params.get("codigo_eol_escola")
if codigo_eol_escola:
if incluir_alteracao_ue:
qs = qs.filter(
Q(
dietas_especiais__rastro_escola__codigo_eol=codigo_eol_escola,
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="COMUM",
)
| Q(
dietas_especiais__escola_destino__codigo_eol=codigo_eol_escola,
dietas_especiais__status__in=["CODAE_AUTORIZADO"],
dietas_especiais__ativo=True,
dietas_especiais__tipo_solicitacao="ALTERACAO_UE",
)
)
else:
qs = qs.filter(
dietas_especiais__rastro_escola__codigo_eol=codigo_eol_escola
)
if form.cleaned_data["codigo_eol"]:
codigo_eol = f"{int(form.cleaned_data['codigo_eol']):06d}"
qs = qs.filter(codigo_eol=codigo_eol)
elif form.cleaned_data["nome_aluno"]:
qs = qs.filter(nome__icontains=form.cleaned_data["nome_aluno"])
series = self.request.query_params.getlist("serie")
if series:
qs = qs.filter(Q(*[Q(serie__icontains=s) for s in series], _connector=Q.OR))
if self.request.user.tipo_usuario == "dieta_especial":
return qs.distinct().order_by(
"escola__diretoria_regional__nome", "escola__nome", "nome"
)
elif self.request.user.tipo_usuario == "diretoriaregional":
return qs.distinct().order_by("escola__nome", "nome")
return qs.distinct().order_by("nome")
[docs]
class AlergiaIntoleranciaViewSet(
mixins.ListModelMixin, mixins.RetrieveModelMixin, GenericViewSet
):
queryset = AlergiaIntolerancia.objects.all()
serializer_class = AlergiaIntoleranciaSerializer
pagination_class = None
[docs]
class ClassificacaoDietaViewSet(
mixins.ListModelMixin, mixins.RetrieveModelMixin, GenericViewSet
):
queryset = ClassificacaoDieta.objects.order_by("nome")
serializer_class = ClassificacaoDietaSerializer
pagination_class = None
[docs]
class MotivoNegacaoViewSet(
mixins.ListModelMixin, mixins.RetrieveModelMixin, GenericViewSet
):
queryset = MotivoNegacao.objects.all()
serializer_class = MotivoNegacaoSerializer
pagination_class = None
filter_backends = (filters.DjangoFilterBackend,)
filterset_class = MotivoNegacaoFilter
[docs]
class AlimentoViewSet(mixins.ListModelMixin, mixins.RetrieveModelMixin, GenericViewSet):
queryset = Alimento.objects.all().order_by("nome")
serializer_class = AlimentoSerializer
pagination_class = None
filter_backends = (filters.DjangoFilterBackend,)
filterset_class = AlimentoFilter
[docs]
class MotivoAlteracaoUEViewSet(mixins.ListModelMixin, GenericViewSet):
queryset = MotivoAlteracaoUE.objects.filter(ativo=True).order_by("nome")
serializer_class = MotivoAlteracaoUESerializer