import datetime
import uuid as uuid_lib
from math import ceil
from django.core.exceptions import ObjectDoesNotExist
from django.db import transaction
from django.db.models import QuerySet
from django_filters import rest_framework as filters
from rest_framework import mixins, viewsets
from rest_framework.decorators import action
from rest_framework.exceptions import ValidationError
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.status import (
HTTP_200_OK,
HTTP_400_BAD_REQUEST,
HTTP_401_UNAUTHORIZED,
HTTP_406_NOT_ACCEPTABLE,
)
from xworkflows.base import InvalidTransitionError
from src.dados_comuns.constants import ADMINISTRADOR_EMPRESA
from src.dados_comuns.fluxo_status import (
CronogramaWorkflow,
)
from src.dados_comuns.permissions import (
PermissaoParaAnalisarDilogAbastecimentoSolicitacaoAlteracaoCronograma,
PermissaoParaAnalisarDilogSolicitacaoAlteracaoCronograma,
PermissaoParaAssinarCronogramaUsuarioDilog,
PermissaoParaAssinarCronogramaUsuarioFornecedor,
PermissaoParaCriarCronograma,
PermissaoParaCriarSolicitacoesAlteracaoCronograma,
PermissaoParaDarCienciaAlteracaoCronograma,
PermissaoParaDashboardCronograma,
PermissaoParaListarDashboardSolicitacaoAlteracaoCronograma,
PermissaoParaVisualizarCalendarioCronograma,
PermissaoParaVisualizarCronograma,
PermissaoParaVisualizarRelatorioCronograma,
PermissaoParaVisualizarSolicitacoesAlteracaoCronograma,
UsuarioDilogAbastecimento,
ViewSetActionPermissionMixin,
)
from src.pos_recebimento.api.permissions import (
PermissaoParaCadastrarTermoRecebimentoDefinitivo,
)
from src.pre_recebimento.base.api.paginations import (
PreRecebimentoPagination,
)
from src.pre_recebimento.cronograma_entrega.api.filters import (
CronogramaFilter,
SolicitacaoAlteracaoCronogramaFilter,
)
from src.pre_recebimento.cronograma_entrega.api.helpers import (
filtrar_etapas,
totalizador_relatorio_cronograma,
)
from src.pre_recebimento.cronograma_entrega.api.serializers.serializer_create import (
CronogramaCreateSerializer,
CronogramaPontoAPontoCreateSerializer,
SolicitacaoDeAlteracaoCronogramaCreateSerializer,
)
from src.pre_recebimento.cronograma_entrega.api.serializers.serializers import (
CronogramaComLogSerializer,
CronogramaDetalhePosRecebimentoSerializer,
CronogramaFichaDeRecebimentoSerializer,
CronogramaRascunhosSerializer,
CronogramaRelatorioSerializer,
CronogramaSerializer,
CronogramaSimplesSerializer,
EtapasDoCronogramaCalendarioSerializer,
InterrupcaoProgramadaEntregaCreateSerializer,
InterrupcaoProgramadaEntregaSerializer,
PainelCronogramaSerializer,
PainelSolicitacaoAlteracaoCronogramaSerializer,
SolicitacaoAlteracaoCronogramaCompletoSerializer,
SolicitacaoAlteracaoCronogramaSerializer,
)
from src.pre_recebimento.cronograma_entrega.api.services import (
ServiceDashboardSolicitacaoAlteracaoCronogramaProfiles,
ServiceQuerysetAlteracaoCronograma,
)
from src.pre_recebimento.cronograma_entrega.models import (
Cronograma,
EtapasDoCronograma,
InterrupcaoProgramadaEntrega,
SolicitacaoAlteracaoCronograma,
)
from src.pre_recebimento.tasks import (
gerar_relatorio_cronogramas_pdf_async,
gerar_relatorio_cronogramas_xlsx_async,
)
from src.relatorios.relatorios import (
get_pdf_cronograma,
get_pdf_cronograma_ponto_a_ponto_flv,
)
from ....dados_comuns.models import LogSolicitacoesUsuario
from .validators import valida_parametros_calendario
def _uuid_valido(valor):
"""Verifica se o parâmetro recebido é um UUID válido."""
if not valor:
return False
try:
uuid_lib.UUID(str(valor))
return True
except (ValueError, AttributeError, TypeError):
return False
[docs]
class CronogramaModelViewSet(ViewSetActionPermissionMixin, viewsets.ModelViewSet):
"""ViewSet para CRUD e gerenciamento de Cronogramas.
Endpoints principais:
- ``GET /`` — Lista cronogramas (com filtros por número, empresa, produto, status).
- ``POST /`` — Cria um novo cronograma (rascunho).
- ``GET /dashboard/`` — Dashboard agrupado por status.
- ``GET /dashboard-com-filtro/`` — Dashboard com filtros.
- ``PATCH /<uuid>/fornecedor-assina-cronograma/`` — Assinatura do fornecedor.
- ``PATCH /<uuid>/abastecimento-assina/`` — Assinatura DILOG Abastecimento.
- ``PATCH /<uuid>/codae-assina/`` — Assinatura CODAE.
- ``GET /<uuid>/gerar-pdf-cronograma/`` — Geração de PDF.
- ``GET /rascunhos/`` — Lista cronogramas em rascunho.
- ``GET /listagem-relatorio/`` — Relatório de cronogramas com filtros.
Permissões são definidas dinamicamente por ``permission_action_classes``
e ``ViewSetActionPermissionMixin``.
"""
lookup_field = "uuid"
queryset = Cronograma.objects.all()
serializer_class = CronogramaSerializer
filter_backends = (filters.DjangoFilterBackend,)
filterset_class = CronogramaFilter
pagination_class = PreRecebimentoPagination
permission_classes = (PermissaoParaVisualizarCronograma,)
permission_action_classes = {
"create": [PermissaoParaCriarCronograma],
"delete": [PermissaoParaCriarCronograma],
}
[docs]
def get_serializer_class(self):
"""Retorna o serializer adequado conforme a ação.
- ``retrieve`` / ``list``: CronogramaSerializer (leitura).
- ``create`` com ``ponto_a_ponto=True``: CronogramaPontoAPontoCreateSerializer.
- demais ações: CronogramaCreateSerializer.
"""
if self.action in ["retrieve", "list"]:
return CronogramaSerializer
if self.request and self.request.data.get("ponto_a_ponto"):
return CronogramaPontoAPontoCreateSerializer
return CronogramaCreateSerializer
[docs]
def get_queryset(self):
"""Retorna todos os cronogramas ordenados por data de criação (decrescente)."""
return Cronograma.objects.all().order_by("-criado_em")
[docs]
def get_lista_status(self):
"""Retorna os status que aparecem no dashboard do cronograma."""
lista_status = [
Cronograma.workflow_class.ASSINADO_E_ENVIADO_AO_FORNECEDOR,
Cronograma.workflow_class.ASSINADO_FORNECEDOR,
Cronograma.workflow_class.ASSINADO_DILOG_ABASTECIMENTO,
Cronograma.workflow_class.ASSINADO_CODAE,
]
return lista_status
[docs]
def get_default_sql(self, workflow, query_set, use_raw):
"""Busca cronogramas por status ordenados pelo log mais recente.
Se ``use_raw=True`` utiliza SQL raw; caso contrário ordena em
memória via Python.
"""
workflow = workflow if isinstance(workflow, list) else [workflow]
if use_raw:
data = {
"logs": LogSolicitacoesUsuario._meta.db_table,
"cronograma": Cronograma._meta.db_table,
"status": workflow,
}
raw_sql = (
"SELECT %(cronograma)s.* FROM %(cronograma)s "
"JOIN (SELECT uuid_original, MAX(criado_em) AS log_criado_em FROM %(logs)s "
"GROUP BY uuid_original) "
"AS most_recent_log "
"ON %(cronograma)s.uuid = most_recent_log.uuid_original "
"WHERE %(cronograma)s.status = '%(status)s' "
)
raw_sql += "ORDER BY log_criado_em DESC"
return query_set.raw(raw_sql % data)
else:
qs = sorted(
query_set.filter(status__in=workflow).distinct().all(),
key=lambda x: (
x.log_mais_recente.criado_em if x.log_mais_recente else "-criado_em"
),
reverse=True,
)
return qs
[docs]
def dados_dashboard(self, request, query_set: QuerySet, use_raw) -> list:
"""Monta os dados do dashboard agrupados por status.
Aceita parâmetros ``limit``, ``offset`` e ``status`` via query params.
"""
limit = int(request.query_params.get("limit", 10))
offset = int(request.query_params.get("offset", 0))
status = request.query_params.getlist("status", None)
sumario = []
if status:
qs = self.get_default_sql(
workflow=status, query_set=query_set, use_raw=use_raw
)
sumario.append(
{
"status": status,
"total": len(qs),
"dados": PainelCronogramaSerializer(
qs[offset : limit + offset],
context={"request": self.request, "workflow": status},
many=True,
).data,
}
)
else:
for workflow in self.get_lista_status():
qs = self.get_default_sql(
workflow=workflow, query_set=query_set, use_raw=use_raw
)
sumario.append(
{
"status": workflow,
"dados": PainelCronogramaSerializer(
qs[:6],
context={"request": self.request, "workflow": workflow},
many=True,
).data,
}
)
return sumario
[docs]
@action(
detail=False,
methods=["GET"],
url_path="dashboard",
permission_classes=(PermissaoParaDashboardCronograma,),
)
def dashboard(self, request):
"""Dashboard de cronogramas agrupados por status.
Retorna os cronogramas mais recentes de cada status
(ASSINADO_E_ENVIADO_AO_FORNECEDOR, ASSINADO_FORNECEDOR,
ASSINADO_DILOG_ABASTECIMENTO, ASSINADO_CODAE) com até 6
registros por grupo.
"""
query_set = self.get_queryset()
response = {
"results": self.dados_dashboard(
query_set=query_set, request=request, use_raw=False
)
}
return Response(response)
[docs]
@action(
detail=False,
methods=["GET"],
url_path="dashboard-com-filtro",
permission_classes=(PermissaoParaDashboardCronograma,),
)
def dashboard_com_filtro(self, request):
"""Dashboard de cronogramas com filtros.
Similar ao ``dashboard``, mas permite filtrar por número do
cronograma, nome do produto e nome do fornecedor via query params.
"""
query_set = self.get_queryset()
numero_cronograma = request.query_params.get("numero_cronograma", None)
produto = request.query_params.get("nome_produto", None)
fornecedor = request.query_params.get("nome_fornecedor", None)
if numero_cronograma:
query_set = query_set.filter(numero__icontains=numero_cronograma)
if produto:
query_set = query_set.filter(
ficha_tecnica__produto__nome__icontains=produto
)
if fornecedor:
query_set = query_set.filter(empresa__razao_social__icontains=fornecedor)
response = {
"results": self.dados_dashboard(
query_set=query_set, request=request, use_raw=False
)
}
return Response(response)
[docs]
def list(self, request, *args, **kwargs):
"""Lista cronogramas com paginação.
Para usuários do perfil ADMINISTRADOR_EMPRESA que são
fornecedores, filtra apenas os cronogramas da sua empresa.
"""
vinculo = self.request.user.vinculo_atual
queryset = self.filter_queryset(self.get_queryset())
queryset = queryset.order_by("-alterado_em").distinct()
if (
vinculo.perfil.nome == ADMINISTRADOR_EMPRESA
and vinculo.instituicao.eh_fornecedor
):
queryset = queryset.filter(empresa=vinculo.instituicao)
page = self.paginate_queryset(queryset)
if page is not None:
serializer = self.get_serializer(page, many=True)
response = self.get_paginated_response(serializer.data)
return response
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
[docs]
@action(
detail=False,
permission_classes=(PermissaoParaVisualizarRelatorioCronograma,),
methods=["GET"],
url_path="listagem-relatorio",
)
def lista_relatorio(self, request, *args, **kwargs):
"""Retorna listagem de cronogramas para relatório com filtros.
Aceita filtros por ``situacao`` (lista), ``data_inicial`` e
``data_final`` via query params. Aplica paginação e retorna
totalizadores por status.
"""
queryset = (
self.filter_queryset(self.get_queryset())
.order_by("-alterado_em")
.distinct()
)
situacoes = request.query_params.getlist("situacao", [])
data_inicial = request.query_params.get("data_inicial")
data_final = request.query_params.get("data_final")
tem_filtro_pos_serializacao = (
(bool(situacoes) and len(situacoes) < 3)
or bool(data_inicial)
or bool(data_final)
)
if tem_filtro_pos_serializacao:
serializer = CronogramaRelatorioSerializer(queryset, many=True)
filtros = {
"situacao": situacoes,
"data_inicial": data_inicial,
"data_final": data_final,
}
dados_filtrados = filtrar_etapas(serializer.data, filtros)
page_size = self.pagination_class().page_size
page_number = int(request.query_params.get("page", 1))
start_index = (page_number - 1) * page_size
end_index = start_index + page_size
dados_paginados = dados_filtrados[start_index:end_index]
total_items = len(dados_filtrados)
total_pages = ceil(total_items / page_size) if page_size > 0 else 1
has_next = page_number < total_pages
has_previous = page_number > 1
response_data = {
"count": total_items,
"next": f"?page={page_number + 1}" if has_next else None,
"previous": f"?page={page_number - 1}" if has_previous else None,
"results": dados_paginados,
"totalizadores": totalizador_relatorio_cronograma(dados_filtrados),
}
return Response(response_data)
else:
page = self.paginate_queryset(queryset)
if page is not None:
serializer = CronogramaRelatorioSerializer(page, many=True)
response = self.get_paginated_response(serializer.data)
response.data["totalizadores"] = totalizador_relatorio_cronograma(
queryset
)
return response
else:
serializer = CronogramaRelatorioSerializer(queryset, many=True)
return Response(serializer.data)
[docs]
@action(detail=False, url_path="opcoes-etapas")
def etapas(self, _):
"""Retorna lista de opções de etapas (1 a 100) para formulários."""
return Response(EtapasDoCronograma.etapas_to_json())
[docs]
@action(detail=False, methods=["GET"], url_path="rascunhos")
def rascunhos(self, _):
"""Retorna lista de cronogramas em status RASCUNHO."""
queryset = self.get_queryset().filter(status__in=[CronogramaWorkflow.RASCUNHO])
response = {"results": CronogramaRascunhosSerializer(queryset, many=True).data}
return Response(response)
[docs]
@transaction.atomic
@action(
detail=True,
permission_classes=(PermissaoParaAssinarCronogramaUsuarioFornecedor,),
methods=["patch"],
url_path="fornecedor-assina-cronograma",
)
def fornecedor_assina(self, request, uuid=None):
"""Registra a assinatura do fornecedor no cronograma.
Valida a senha do usuário antes de prosseguir. Avança o fluxo
do cronograma para ASSINADO_FORNECEDOR.
"""
usuario = request.user
if not usuario.verificar_autenticidade(request.data.get("password")):
return Response(
dict(
detail="Assinatura do cronograma não foi validada. Verifique sua senha."
),
status=HTTP_401_UNAUTHORIZED,
)
try:
cronograma = Cronograma.objects.get(uuid=uuid)
cronograma.fornecedor_assina(
user=usuario,
)
serializer = CronogramaSerializer(cronograma)
return Response(serializer.data)
except ObjectDoesNotExist as e:
return Response(
dict(detail=f"Cronograma informado não é valido: {e}"),
status=HTTP_406_NOT_ACCEPTABLE,
)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@transaction.atomic
@action(
detail=True,
permission_classes=(UsuarioDilogAbastecimento,),
methods=["patch"],
url_path="abastecimento-assina",
)
def abastecimento_assina(self, request, uuid):
"""Registra a assinatura da DILOG Abastecimento no cronograma.
Valida a senha do usuário antes de prosseguir. Avança o fluxo
do cronograma para ASSINADO_DILOG_ABASTECIMENTO.
"""
usuario = request.user
if not usuario.verificar_autenticidade(request.data.get("password")):
return Response(
dict(
detail="Assinatura do cronograma não foi validada. Verifique sua senha."
),
status=HTTP_401_UNAUTHORIZED,
)
try:
cronograma = Cronograma.objects.get(uuid=uuid)
cronograma.dilog_abastecimento_assina(user=usuario)
serializer = CronogramaSerializer(cronograma)
return Response(serializer.data)
except ObjectDoesNotExist as e:
return Response(
dict(detail=f"Cronograma informado não é valido: {e}"),
status=HTTP_406_NOT_ACCEPTABLE,
)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@transaction.atomic
@action(
detail=True,
permission_classes=(PermissaoParaAssinarCronogramaUsuarioDilog,),
methods=["patch"],
url_path="codae-assina",
)
def codae_assina(self, request, uuid):
"""Registra a assinatura da CODAE no cronograma.
Valida a senha do usuário antes de prosseguir. Avança o fluxo
do cronograma para ASSINADO_CODAE (última etapa do fluxo).
"""
usuario = request.user
if not usuario.verificar_autenticidade(request.data.get("password")):
return Response(
dict(
detail="Assinatura do cronograma não foi validada. Verifique sua senha."
),
status=HTTP_401_UNAUTHORIZED,
)
try:
cronograma = Cronograma.objects.get(uuid=uuid)
cronograma.codae_assina(user=usuario)
serializer = CronogramaSerializer(cronograma)
return Response(serializer.data)
except ObjectDoesNotExist as e:
return Response(
dict(detail=f"Cronograma informado não é valido: {e}"),
status=HTTP_406_NOT_ACCEPTABLE,
)
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=["GET"], url_path="gerar-pdf-cronograma")
def gerar_pdf_cronograma(self, request, uuid=None):
"""Gera o PDF do cronograma.
Para cronogramas Ponto a Ponto (FLV) utiliza o template específico
``get_pdf_cronograma_ponto_a_ponto_flv``. Para os demais, utiliza
o template ``get_pdf_cronograma`` padrão.
"""
cronograma = self.get_object()
if cronograma.ponto_a_ponto:
return get_pdf_cronograma_ponto_a_ponto_flv(request, cronograma)
return get_pdf_cronograma(request, cronograma)
[docs]
@action(detail=True, methods=["GET"], url_path="detalhar-com-log")
def detalhar_com_log(self, request, uuid=None):
"""Retorna os detalhes do cronograma com o histórico de logs."""
cronograma = self.get_object()
response = CronogramaComLogSerializer(cronograma, many=False).data
return Response(response)
[docs]
@action(detail=False, methods=["GET"], url_path="lista-cronogramas-cadastro")
def lista_cronogramas_para_cadastro(self, request):
"""Lista cronogramas para usar em campos de seleção em cadastros.
Para fornecedores, retorna apenas os cronogramas da sua empresa.
"""
user = self.request.user
if user.eh_fornecedor:
cronogramas = Cronograma.objects.filter(
empresa=user.vinculo_atual.instituicao
).order_by("-criado_em")
else:
cronogramas = self.get_queryset()
serializer = CronogramaSimplesSerializer(cronogramas, many=True).data
response = {"results": serializer}
return Response(response)
[docs]
@action(
detail=False,
methods=["GET"],
url_path="lista-cronogramas-ficha-recebimento",
)
def lista_cronogramas_ficha_recebimento(self, request):
"""Lista cronogramas aptos para gerar ficha de recebimento.
Retorna apenas cronogramas com status ASSINADO_CODAE.
"""
qs = self.get_queryset().filter(status=CronogramaWorkflow.ASSINADO_CODAE)
return Response({"results": CronogramaSimplesSerializer(qs, many=True).data})
[docs]
@action(
detail=True,
methods=["GET"],
url_path="dados-cronograma-ficha-recebimento",
)
def dados_cronograma_ficha_recebimento(self, request, uuid):
"""Retorna os dados completos do cronograma para a ficha de recebimento."""
return Response(
{"results": CronogramaFichaDeRecebimentoSerializer(self.get_object()).data}
)
[docs]
@action(
detail=False,
methods=["GET"],
url_path="lista-cronogramas-pos-recebimento",
permission_classes=(PermissaoParaCadastrarTermoRecebimentoDefinitivo,),
)
def lista_cronogramas_pos_recebimento(self, request):
"""Cronogramas vinculados ao contrato selecionado (query param
``contrato_id``) para o cadastro do Termo de Recebimento Definitivo
(Pós-Recebimento)."""
contrato_uuid = request.query_params.get("contrato_id")
if not _uuid_valido(contrato_uuid):
queryset = Cronograma.objects.none()
else:
queryset = Cronograma.objects.filter(contrato__uuid=contrato_uuid).order_by(
"numero"
)
return Response(
{"results": CronogramaSimplesSerializer(queryset, many=True).data}
)
[docs]
@action(
detail=True,
methods=["GET"],
url_path="dados-cronograma-pos-recebimento",
permission_classes=(PermissaoParaCadastrarTermoRecebimentoDefinitivo,),
)
def dados_cronograma_pos_recebimento(self, request, uuid):
"""Dados do cronograma (produto, processo SEI, unidade de medida)
para preenchimento automático no cadastro do Termo de Recebimento
Definitivo (Pós-Recebimento)."""
return Response(
CronogramaDetalhePosRecebimentoSerializer(self.get_object()).data
)
[docs]
@action(
detail=False,
permission_classes=(PermissaoParaVisualizarRelatorioCronograma,),
methods=["GET"],
url_path="gerar-relatorio-xlsx-async",
)
def gerar_relatorio_xlsx_async(self, request):
"""Solicita a geração assíncrona de relatório de cronogramas em XLSX.
Os filtros (situação, data_inicial, data_final) são passados
via query params.
"""
ids_cronogramas = list(
(
self.filter_queryset(self.get_queryset())
.order_by("-alterado_em")
.distinct()
).values_list("id", flat=True)
)
filtros = {
"situacao": request.query_params.getlist("situacao", []),
"data_inicial": request.query_params.get("data_inicial"),
"data_final": request.query_params.get("data_final"),
}
gerar_relatorio_cronogramas_xlsx_async.delay(
request.user.username,
ids_cronogramas,
filtros,
)
return Response(
{"detail": "Solicitação de geração de arquivo recebida com sucesso."},
status=HTTP_200_OK,
)
[docs]
@action(
detail=False,
permission_classes=(PermissaoParaVisualizarRelatorioCronograma,),
methods=["GET"],
url_path="gerar-relatorio-pdf-async",
)
def gerar_relatorio_pdf_async(self, request):
"""Solicita a geração assíncrona de relatório de cronogramas em PDF.
Os filtros (situação, data_inicial, data_final) são passados
via query params.
"""
ids_cronogramas = list(
(
self.filter_queryset(self.get_queryset())
.order_by("-alterado_em")
.distinct()
).values_list("id", flat=True)
)
filtros = {
"situacao": request.query_params.getlist("situacao", []),
"data_inicial": request.query_params.get("data_inicial"),
"data_final": request.query_params.get("data_final"),
}
gerar_relatorio_cronogramas_pdf_async.delay(
request.user.username,
ids_cronogramas,
filtros,
)
return Response(
{"detail": "Solicitação de geração de arquivo recebida com sucesso."},
status=HTTP_200_OK,
)
[docs]
class SolicitacaoDeAlteracaoCronogramaViewSet(viewsets.ModelViewSet):
"""ViewSet para CRUD e gerenciamento de solicitações de alteração de cronograma.
Permite que fornecedores e CODAE criem solicitações de alteração de
cronogramas já assinados. Inclui fluxo de aprovação com análise da
DILOG Abastecimento e DILOG, ciência do cronograma e dashboard.
Endpoints principais:
- ``GET /`` — Lista solicitações (ordenadas por prioridade do perfil).
- ``POST /`` — Cria nova solicitação de alteração.
- ``GET /dashboard/`` — Dashboard por status.
- ``PATCH /<uuid>/cronograma-ciente/`` — Confirma ciência do cronograma.
- ``PATCH /<uuid>/analise-abastecimento/`` — Análise DILOG Abastecimento.
- ``PATCH /<uuid>/analise-dilog/`` — Análise DILOG.
- ``PATCH /<uuid>/fornecedor-ciente/`` — Ciência do fornecedor.
Permissões são ajustadas por ação via ``get_permissions()``.
"""
lookup_field = "uuid"
filter_backends = (filters.DjangoFilterBackend,)
pagination_class = PreRecebimentoPagination
permission_classes = (IsAuthenticated,)
filterset_class = SolicitacaoAlteracaoCronogramaFilter
[docs]
def get_queryset(self):
"""Retorna as solicitações de alteração.
Para fornecedores, filtra apenas as solicitações dos cronogramas
da sua empresa.
"""
user = self.request.user
if user.eh_fornecedor:
return SolicitacaoAlteracaoCronograma.objects.filter(
cronograma__empresa=user.vinculo_atual.instituicao
).order_by("-criado_em")
return SolicitacaoAlteracaoCronograma.objects.all().order_by("-criado_em")
[docs]
def get_serializer_class(self):
"""Retorna o serializer adequado conforme a ação.
- ``list``: SolicitacaoAlteracaoCronogramaSerializer (resumo).
- ``retrieve``: SolicitacaoAlteracaoCronogramaCompletoSerializer (detalhes + logs).
- demais: SolicitacaoDeAlteracaoCronogramaCreateSerializer.
"""
serializer_classes_map = {
"list": SolicitacaoAlteracaoCronogramaSerializer,
"retrieve": SolicitacaoAlteracaoCronogramaCompletoSerializer,
}
return serializer_classes_map.get(
self.action, SolicitacaoDeAlteracaoCronogramaCreateSerializer
)
[docs]
def get_permissions(self):
"""Retorna as permissões específicas para cada ação.
- ``list`` / ``retrieve``: PermissaoParaVisualizarSolicitacoesAlteracaoCronograma.
- ``create``: PermissaoParaCriarSolicitacoesAlteracaoCronograma.
"""
permission_classes_map = {
"list": (PermissaoParaVisualizarSolicitacoesAlteracaoCronograma,),
"retrieve": (PermissaoParaVisualizarSolicitacoesAlteracaoCronograma,),
"create": (PermissaoParaCriarSolicitacoesAlteracaoCronograma,),
}
action_permissions = permission_classes_map.get(self.action, [])
self.permission_classes = (*self.permission_classes, *action_permissions)
return super(SolicitacaoDeAlteracaoCronogramaViewSet, self).get_permissions()
[docs]
def list(self, request, *args, **kwargs):
"""Lista solicitações de alteração ordenadas por prioridade do perfil.
Utiliza o ServiceQuerysetAlteracaoCronograma para ordenar:
primeiro os status prioritários do perfil, depois os demais.
"""
queryset = ServiceQuerysetAlteracaoCronograma(
request=self.request
).get_queryset(filter=self.filter_queryset)
page = self.paginate_queryset(queryset)
if page is not None:
serializer = SolicitacaoAlteracaoCronogramaSerializer(page, many=True)
response = self.get_paginated_response(serializer.data)
return response
serializer = SolicitacaoAlteracaoCronogramaSerializer(queryset, many=True)
return Response(serializer.data)
def _dados_dashboard(self, request, filtros=None):
"""Monta os dados do dashboard de solicitações de alteração.
Aceita parâmetros ``limit``, ``offset`` e ``status``.
Os status exibidos variam conforme o perfil do usuário.
"""
limit = (
int(request.query_params.get("limit", 10))
if "limit" in request.query_params
else 6
)
offset = (
int(request.query_params.get("offset", 0))
if "offset" in request.query_params
else 0
)
status = request.query_params.getlist("status", None)
dados_dashboard = []
lista_status = (
[status]
if status
else ServiceDashboardSolicitacaoAlteracaoCronogramaProfiles.get_dashboard_status(
self.request.user
)
)
dados_dashboard = [
{
"status": status,
"dados": SolicitacaoAlteracaoCronograma.objects.filtrar_por_status(
status, filtros, offset, limit + offset
),
}
for status in lista_status
]
if status:
dados_dashboard[0]["total"] = (
SolicitacaoAlteracaoCronograma.objects.filtrar_por_status(
status, filtros
).count()
)
return dados_dashboard
[docs]
@action(
detail=False,
methods=["GET"],
url_path="dashboard",
permission_classes=(
PermissaoParaListarDashboardSolicitacaoAlteracaoCronograma,
),
)
def dashboard(self, request):
"""Dashboard de solicitações de alteração agrupadas por status.
Os status exibidos variam conforme o perfil do usuário logado
(fornecedor, DILOG Abastecimento, DILOG, CODAE, etc.).
"""
serialized_data = PainelSolicitacaoAlteracaoCronogramaSerializer(
self._dados_dashboard(request), many=True
).data
return Response({"results": serialized_data})
[docs]
@action(
detail=False,
methods=["GET"],
url_path="dashboard-com-filtro",
permission_classes=(
PermissaoParaListarDashboardSolicitacaoAlteracaoCronograma,
),
)
def dashboard_com_filtro(self, request):
"""Dashboard com filtros por parâmetros de consulta.
Aceita os mesmos filtros do ``SolicitacaoAlteracaoCronogramaFilter``
via query params (número do cronograma, fornecedor, data, status).
"""
filtros = request.query_params
serialized_data = PainelSolicitacaoAlteracaoCronogramaSerializer(
self._dados_dashboard(request, filtros), many=True
).data
return Response({"results": serialized_data})
[docs]
@transaction.atomic
@action(
detail=True,
permission_classes=(PermissaoParaDarCienciaAlteracaoCronograma,),
methods=["patch"],
url_path="cronograma-ciente",
)
def cronograma_ciente(self, request, uuid):
"""Registra a ciência do cronograma sobre a alteração solicitada.
Recebe as novas etapas e programações de recebimento, substitui
as anteriores e avança o fluxo da solicitação.
"""
usuario = request.user
justificativa = request.data.get("justificativa_cronograma")
etapas = request.data.get("etapas", [])
programacoes = request.data.get("programacoes_de_recebimento", [])
try:
solicitacao_alteracao = SolicitacaoAlteracaoCronograma.objects.get(
uuid=uuid
)
solicitacao_alteracao.cronograma_confirma_ciencia(
justificativa, usuario, etapas, programacoes
)
serializer = SolicitacaoAlteracaoCronogramaSerializer(solicitacao_alteracao)
return Response(serializer.data)
except ObjectDoesNotExist as e:
return Response(
dict(detail=f"Solicitação Cronograma informado não é valido: {e}"),
status=HTTP_406_NOT_ACCEPTABLE,
)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@transaction.atomic
@action(
detail=True,
permission_classes=(
PermissaoParaAnalisarDilogAbastecimentoSolicitacaoAlteracaoCronograma,
),
methods=["patch"],
url_path="analise-abastecimento",
)
def analise_abastecimento(self, request, uuid):
"""Análise da solicitação pela DILOG Abastecimento.
Recebe o parâmetro ``aprovado`` (bool) no body. Se ``True``,
aprova a solicitação; se ``False``, reprova com justificativa
(campo ``justificativa_abastecimento``).
"""
usuario = request.user
aprovado = request.data.get(("aprovado"), "aprovado")
justificativa = request.data.get("justificativa_abastecimento", "")
try:
solicitacao_cronograma = SolicitacaoAlteracaoCronograma.objects.get(
uuid=uuid
)
if aprovado is True:
solicitacao_cronograma.dilog_abastecimento_aprova(
user=usuario, justificativa=justificativa
)
elif aprovado is False:
justificativa = request.data.get("justificativa_abastecimento")
solicitacao_cronograma.dilog_abastecimento_reprova(
user=usuario, justificativa=justificativa
)
else:
raise ValidationError("Parametro aprovado deve ser true ou false.")
solicitacao_cronograma.save()
serializer = SolicitacaoAlteracaoCronogramaSerializer(
solicitacao_cronograma
)
return Response(serializer.data)
except ObjectDoesNotExist as e:
return Response(
dict(detail=f"Solicitação Cronograma informado não é valido: {e}"),
status=HTTP_406_NOT_ACCEPTABLE,
)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@transaction.atomic
@action(
detail=True,
permission_classes=(PermissaoParaAnalisarDilogSolicitacaoAlteracaoCronograma,),
methods=["patch"],
url_path="analise-dilog",
)
def analise_dilog(self, request, uuid):
"""Análise final da solicitação pela DILOG.
Recebe o parâmetro ``aprovado`` (bool) no body. Se aprovada,
substitui as etapas e programações do cronograma original pelas
novas. Se reprovada, registra a justificativa (campo
``justificativa_dilog``).
"""
usuario = request.user
aprovado = request.data.get(("aprovado"), "aprovado")
try:
solicitacao_cronograma = SolicitacaoAlteracaoCronograma.objects.get(
uuid=uuid
)
if aprovado is True:
solicitacao_cronograma.dilog_aprova(user=usuario)
elif aprovado is False:
justificativa = request.data.get("justificativa_dilog")
solicitacao_cronograma.dilog_reprova(
user=usuario, justificativa=justificativa
)
else:
raise ValidationError("Parametro aprovado deve ser true ou false.")
solicitacao_cronograma.save()
solicitacao_cronograma.cronograma.finaliza_solicitacao_alteracao(
user=usuario,
justificativa=str(solicitacao_cronograma.uuid),
)
serializer = SolicitacaoAlteracaoCronogramaSerializer(
solicitacao_cronograma
)
return Response(serializer.data)
except ObjectDoesNotExist as e:
return Response(
dict(detail=f"Solicitação Cronograma informado não é valido: {e}"),
status=HTTP_406_NOT_ACCEPTABLE,
)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
@transaction.atomic
@action(
detail=True,
permission_classes=(PermissaoParaAssinarCronogramaUsuarioFornecedor,),
methods=["patch"],
url_path="fornecedor-ciente",
)
def fornecedor_ciente(self, request, uuid):
"""Registra a ciência do fornecedor sobre a alteração.
Avança o fluxo da solicitação e, se o cronograma não estiver
no status ``ASSINADO_CODAE``, finaliza a solicitação de alteração.
"""
usuario = request.user
try:
solicitacao_cronograma = SolicitacaoAlteracaoCronograma.objects.get(
uuid=uuid
)
solicitacao_cronograma.fornecedor_ciente(user=usuario)
cronograma = solicitacao_cronograma.cronograma
solicitacao_cronograma.save()
if cronograma.status != Cronograma.workflow_class.ASSINADO_CODAE:
usuario_codae = cronograma.logs[len(cronograma.logs) - 2].usuario
cronograma.finaliza_solicitacao_alteracao(
user=usuario_codae,
justificativa=str(solicitacao_cronograma.uuid),
)
serializer = SolicitacaoAlteracaoCronogramaSerializer(
solicitacao_cronograma
)
return Response(serializer.data)
except ObjectDoesNotExist as e:
return Response(
dict(detail=f"Solicitação Cronograma informado não é valido: {e}"),
status=HTTP_406_NOT_ACCEPTABLE,
)
except InvalidTransitionError as e:
return Response(
dict(detail=f"Erro de transição de estado: {e}"),
status=HTTP_400_BAD_REQUEST,
)
[docs]
class CalendarioCronogramaViewset(viewsets.ReadOnlyModelViewSet):
"""ViewSet para consulta de etapas de cronograma em formato de calendário.
Endpoints:
- ``GET /etapas-por-mes/`` — Etapas agrupadas por mês/ano.
- ``GET /datas-bloqueadas-armazenavel/`` — Datas bloqueadas (tipo armazenável).
Utilizado pelo frontend para exibir cronogramas em visualização de calendário.
"""
queryset = EtapasDoCronograma.objects.filter(cronograma__isnull=False).order_by(
"-criado_em"
)
serializer_class = EtapasDoCronogramaCalendarioSerializer
permission_classes = (PermissaoParaVisualizarCalendarioCronograma,)
[docs]
def get_queryset(self):
"""Retorna etapas de cronograma filtradas por mês/ano.
Valida os parâmetros ``mes`` e ``ano`` (obrigatórios) e filtra
apenas cronogramas com status ASSINADO_CODAE, ALTERACAO_CODAE
ou SOLICITADO_ALTERACAO.
"""
mes = self.request.query_params.get("mes", None)
ano = self.request.query_params.get("ano", None)
valida_parametros_calendario(mes, ano)
status_necessarios_cronogramas = [
CronogramaWorkflow.ASSINADO_CODAE,
CronogramaWorkflow.ALTERACAO_CODAE,
CronogramaWorkflow.SOLICITADO_ALTERACAO,
]
queryset = EtapasDoCronograma.objects.filter(
cronograma__isnull=False,
cronograma__status__in=status_necessarios_cronogramas,
data_programada__month=mes,
data_programada__year=ano,
).order_by("-criado_em")
return queryset
[docs]
class InterrupcaoProgramadaEntregaViewSet(
mixins.CreateModelMixin,
mixins.ListModelMixin,
mixins.DestroyModelMixin,
viewsets.GenericViewSet,
):
"""ViewSet para cadastro e listagem de Interrupções Programadas de Entregas."""
queryset = InterrupcaoProgramadaEntrega.objects.all()
permission_classes = (PermissaoParaVisualizarCalendarioCronograma,)
lookup_field = "uuid"
[docs]
def get_serializer_class(self):
"""Retorna o serializer conforme a ação.
- ``create``: InterrupcaoProgramadaEntregaCreateSerializer.
- demais: InterrupcaoProgramadaEntregaSerializer (leitura).
"""
if self.action == "create":
return InterrupcaoProgramadaEntregaCreateSerializer
return InterrupcaoProgramadaEntregaSerializer
[docs]
def get_queryset(self):
"""Retorna interrupções filtradas por mês, ano e tipo de calendário."""
qs = super().get_queryset()
mes = self.request.query_params.get("mes")
ano = self.request.query_params.get("ano")
motivo = self.request.query_params.getlist("motivo[]")
tipo_calendario = self.request.query_params.getlist("tipo_calendario[]")
if mes and ano:
qs = qs.filter(data__month=mes, data__year=ano)
if motivo:
qs = qs.filter(motivo__in=motivo)
if tipo_calendario:
qs = qs.filter(tipo_calendario__in=tipo_calendario)
return qs
[docs]
@action(detail=False, methods=["GET"], url_path="motivos")
def motivos(self, request):
"""Retorna lista de motivos disponíveis para interrupção."""
return Response(
[
{"value": c[0], "label": c[1]}
for c in InterrupcaoProgramadaEntrega.MOTIVO_CHOICES
if c[0] != InterrupcaoProgramadaEntrega.MOTIVO_FERIADO
]
)
[docs]
@action(detail=False, methods=["GET"], url_path="datas-bloqueadas-armazenavel")
def datas_bloqueadas_armazenavel(self, request):
"""Retorna lista de datas bloqueadas para cadastro de etapas (tipo ARMAZENAVEL)."""
ano_atual = datetime.datetime.now().year
ano_proximo = ano_atual + 1
datas = InterrupcaoProgramadaEntrega.objects.filter(
tipo_calendario=InterrupcaoProgramadaEntrega.TIPO_CALENDARIO_ARMAZENAVEL,
data__year__in=[ano_atual, ano_proximo],
).values_list("data", flat=True)
return Response({"results": list(datas)}, status=HTTP_200_OK)