from dataclasses import dataclass import re from typing import Any from celery.result import AsyncResult from django import forms from django.conf import settings from django.contrib import messages from django.contrib.auth.decorators import login_required from django.db.models import Model, Q from django.forms import modelform_factory from django.http import ( Http404, HttpRequest, HttpResponse, HttpResponseNotAllowed, JsonResponse, ) from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from django.utils.http import url_has_allowed_host_and_scheme from django.core.paginator import Paginator from .importers import build_aluno_export_file, import_alunos from .models import ( AnoCenso, Curso, FallbackCurso, Instituicao, Municipio, Pais, Pessoa, Polo, UF, Vinculo, ) from .tasks import ( import_carga_horaria_censup_task, import_pessoas_ingressantes_task, import_vinculos_censup_task, ) from .validation import validate_vinculo from .year import get_selected_ano_censo, set_selected_ano_censo TASK_OPERATIONS = { "pessoas-ingressantes": { "label": "Pessoas Ingressantes", "task": import_pessoas_ingressantes_task, }, "vinculos-censup": { "label": "Vinculos (censup_vinculos)", "task": import_vinculos_censup_task, }, "carga-horaria-censup": { "label": "Carga Horaria (censup_carga_horaria)", "task": import_carga_horaria_censup_task, }, } @dataclass(frozen=True) class CrudConfig: slug: str model: type[Model] title: str list_fields: tuple[str, ...] search_fields: tuple[tuple[str, str], ...] CRUD_CONFIGS = { "anos-censo": CrudConfig( slug="anos-censo", model=AnoCenso, title="Anos do Censo", list_fields=("ano", "ano_execucao"), search_fields=(("ano", "int"),), ), "ufs": CrudConfig( slug="ufs", model=UF, title="UFs", list_fields=("codigo", "nome"), search_fields=(("codigo", "int"), ("nome", "icontains")), ), "paises": CrudConfig( slug="paises", model=Pais, title="Paises", list_fields=("codigo", "nome"), search_fields=(("codigo", "icontains"), ("nome", "icontains")), ), "municipios": CrudConfig( slug="municipios", model=Municipio, title="Municipios", list_fields=("codigo", "nome", "uf"), search_fields=( ("codigo", "int"), ("nome", "icontains"), ("uf__nome", "icontains"), ), ), "cursos": CrudConfig( slug="cursos", model=Curso, title="Cursos", list_fields=( "codigo_mec", "descricao", "situacao_funcionamento_curso", "licenciatura", "ead", "polo", "carga_horaria_total", ), search_fields=( ("codigo_mec", "int"), ("descricao", "icontains"), ("situacao_funcionamento_curso", "icontains"), ("polo__descricao", "icontains"), ), ), "fallbacks-cursos": CrudConfig( slug="fallbacks-cursos", model=FallbackCurso, title="Fallbacks de Curso", list_fields=("id", "texto_api", "texto_local"), search_fields=( ("id", "int"), ("texto_api", "icontains"), ("texto_local", "icontains"), ), ), "instituicoes": CrudConfig( slug="instituicoes", model=Instituicao, title="Instituicoes", list_fields=("codigo_inep", "nome"), search_fields=(("codigo_inep", "int"), ("nome", "icontains")), ), "polos": CrudConfig( slug="polos", model=Polo, title="Polos", list_fields=("id", "codigo", "descricao"), search_fields=(("id", "int"), ("codigo", "int"), ("descricao", "icontains")), ), "pessoas": CrudConfig( slug="pessoas", model=Pessoa, title="Pessoas", list_fields=("id", "nome", "cpf", "data_nascimento", "nacionalidade"), search_fields=( ("id", "int"), ("nome", "icontains"), ("cpf", "icontains"), ), ), "vinculos": CrudConfig( slug="vinculos", model=Vinculo, title="Vinculos", list_fields=("id", "pessoa", "curso", "ano_censo", "situacao"), search_fields=( ("id", "int"), ("matricula", "icontains"), ("pessoa__nome", "icontains"), ("pessoa__cpf", "icontains"), ("curso__descricao", "icontains"), ("curso__codigo_mec", "int"), ("ano_censo", "int"), ), ), } CRUD_LINKS = tuple(CRUD_CONFIGS.values()) class ImportAlunosForm(forms.Form): ano_censo = forms.TypedChoiceField(coerce=int) arquivo = forms.FileField() def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self.fields["ano_censo"].choices = [ (ano.ano, ano.ano) for ano in AnoCenso.objects.order_by("-ano") ] self.fields["ano_censo"].widget.attrs["class"] = "form-select" self.fields["arquivo"].widget.attrs["class"] = "form-control" self.fields["arquivo"].widget.attrs["accept"] = ".txt,text/plain" class ProcessarSituacaoForm(forms.Form): matriculas = forms.CharField( label="Matricula", widget=forms.Textarea, ) situacao = forms.ChoiceField( label="Situacao", choices=Vinculo.Situacao.choices, ) def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self.fields["matriculas"].widget.attrs.update( { "class": "form-control", "rows": 10, "placeholder": "Informe uma ou mais matriculas", } ) self.fields["situacao"].widget.attrs["class"] = "form-select" def clean_matriculas(self) -> list[str]: raw_value = self.cleaned_data["matriculas"] matriculas = [ item.strip() for item in re.split(r"[\s,;]+", raw_value) if item.strip() ] deduplicated = list(dict.fromkeys(matriculas)) if not deduplicated: raise forms.ValidationError("Informe ao menos uma matricula.") return deduplicated def ano_ingresso(semestre_ingresso: str) -> int | None: year = (semestre_ingresso or "")[-4:] return int(year) if year.isdigit() else None EXPORT_FILTERS: dict[str, dict[str, Any]] = { "todos": {"label": "Todos", "situacao": None}, "ingressantes": {"label": "Apenas Ingressantes", "situacao": None}, "desvinculado": {"label": "Apenas Desvinculado", "situacao": Vinculo.Situacao.DESVINCULADO}, "cursando": {"label": "Apenas Cursando", "situacao": Vinculo.Situacao.CURSANDO}, "transferido": { "label": "Apenas Transferido", "situacao": Vinculo.Situacao.TRANSFERENCIA_INTERNA, }, "falecido": {"label": "Apenas Falecido", "situacao": Vinculo.Situacao.FALECIDO}, "formado": {"label": "Apenas Formado", "situacao": Vinculo.Situacao.FORMADO}, } @login_required def export_alunos_view(request: HttpRequest, filtro: str) -> HttpResponse: config = EXPORT_FILTERS.get(filtro) if config is None: raise Http404("Filtro de exportacao desconhecido.") try: ies_codigo = int(settings.CENSUP_IES_CODIGO) except (TypeError, ValueError): messages.error(request, "CENSUP_IES_CODIGO nao configurado.") return redirect("home") selected_year = get_selected_ano_censo(request) vinculos = Vinculo.objects.filter(ano_censo=selected_year) if filtro == "ingressantes": vinculos = vinculos.filter(semestre_ingresso__endswith=str(selected_year)) elif config["situacao"] is not None: vinculos = vinculos.filter(situacao=config["situacao"]) vinculos = vinculos.select_related( "pessoa", "pessoa__uf_nascimento", "pessoa__municipio_nascimento", "pessoa__pais_origem", "curso", "curso_origem", "polo", "mobilidade_ies_destino", "mobilidade_pais_destino", ).order_by("pessoa__nome", "pessoa_id") content = build_aluno_export_file(vinculos, ies_codigo) response = HttpResponse(content, content_type="text/plain; charset=utf-8") filename = f"aluno_exportacao_{filtro}_{selected_year}.txt" response["Content-Disposition"] = f'attachment; filename="{filename}"' return response @login_required def home(request: HttpRequest) -> HttpResponse: selected_year = get_selected_ano_censo(request) rows_by_course: dict[int, dict[str, Any]] = {} status_fields = { Vinculo.Situacao.DESVINCULADO: "desvinculado", Vinculo.Situacao.CURSANDO: "cursando", Vinculo.Situacao.TRANSFERENCIA_INTERNA: "transferido", Vinculo.Situacao.FALECIDO: "falecido", Vinculo.Situacao.FORMADO: "formado", } vinculos = ( Vinculo.objects.filter(ano_censo=selected_year) .select_related( "curso", "pessoa", "pessoa__pais_origem", "pessoa__uf_nascimento", "pessoa__municipio_nascimento", "polo", ) .order_by("curso__descricao", "pessoa__nome") ) for vinculo in vinculos: row = rows_by_course.setdefault( vinculo.curso_id, { "curso": vinculo.curso, "desvinculado": 0, "cursando": 0, "transferido": 0, "falecido": 0, "formado": 0, "pendente": 0, "ingressantes": 0, "total": 0, }, ) status_field = status_fields.get(vinculo.situacao) if status_field: row[status_field] += 1 if validate_vinculo(vinculo): row["pendente"] += 1 if ano_ingresso(vinculo.semestre_ingresso) == selected_year: row["ingressantes"] += 1 row["total"] += 1 return render( request, "core/home.html", { "selected_year": selected_year, "course_rows": rows_by_course.values(), }, ) @login_required def set_ano_censo(request: HttpRequest) -> HttpResponse: if request.method == "POST": try: ano_censo = int(request.POST.get("ano_censo", "")) except ValueError: messages.error(request, "Ano do censo invalido.") else: try: set_selected_ano_censo(request, ano_censo) except ValueError: messages.error(request, "Ano do censo nao cadastrado.") else: messages.success(request, f"Ano de trabalho alterado para {ano_censo}.") next_url = request.POST.get("next") or reverse("home") if not url_has_allowed_host_and_scheme( next_url, allowed_hosts={request.get_host()}, require_https=request.is_secure(), ): next_url = reverse("home") return redirect(next_url) @login_required def import_alunos_view(request: HttpRequest) -> HttpResponse: result = None form = ImportAlunosForm(request.POST or None, request.FILES or None) if request.method != "POST": form.fields["ano_censo"].initial = get_selected_ano_censo(request) if request.method == "POST" and form.is_valid(): uploaded_file = form.cleaned_data["arquivo"] content = uploaded_file.read().decode("utf-8-sig") result = import_alunos( lines=content.splitlines(), ano_censo=form.cleaned_data["ano_censo"], ) messages.success(request, "Importacao processada.") return render( request, "core/import_alunos.html", { "form": form, "result": result, }, ) def is_task_active(task_id: str) -> bool: if not task_id: return False return AsyncResult(task_id).state in {"PENDING", "STARTED", "PROGRESS", "RETRY"} @login_required def task_page_view(request: HttpRequest, operacao: str) -> HttpResponse: config = TASK_OPERATIONS.get(operacao) if config is None: from django.http import Http404 raise Http404("Operacao desconhecida") return render( request, "core/task_page.html", { "operacao": operacao, "label": config["label"], "task_id": request.session.get(f"task_id:{operacao}", ""), "selected_year": get_selected_ano_censo(request), }, ) @login_required def start_task_view(request: HttpRequest, operacao: str) -> HttpResponse: config = TASK_OPERATIONS.get(operacao) if config is None: from django.http import Http404 raise Http404("Operacao desconhecida") if request.method != "POST": return HttpResponseNotAllowed(["POST"]) session_key = f"task_id:{operacao}" existing_task_id = request.session.get(session_key, "") if is_task_active(existing_task_id): return JsonResponse({"task_id": existing_task_id, "already_running": True}) page_size = int(request.POST.get("page_size") or 100) threads = int(request.POST.get("threads") or 4) if operacao == "pessoas-ingressantes": primeiro_ano = int( request.POST.get("primeiro_ano") or get_selected_ano_censo(request) ) async_result = config["task"].delay(primeiro_ano=primeiro_ano, page_size=page_size) elif operacao == "vinculos-censup": ano = int(request.POST.get("ano") or get_selected_ano_censo(request)) async_result = config["task"].delay(ano=ano, page_size=page_size, threads=threads) else: async_result = config["task"].delay(page_size=page_size, threads=threads) request.session[session_key] = async_result.id return JsonResponse({"task_id": async_result.id, "already_running": False}) @login_required def task_status_view(request: HttpRequest, task_id: str) -> HttpResponse: result = AsyncResult(task_id) data: dict[str, Any] = {"state": result.state} if result.state == "PROGRESS": data["meta"] = result.info elif result.state == "SUCCESS": data["result"] = result.result elif result.state == "FAILURE": data["error"] = str(result.info) return JsonResponse(data) @login_required def processar_situacao_view(request: HttpRequest) -> HttpResponse: selected_year = get_selected_ano_censo(request) result = None form = ProcessarSituacaoForm(request.POST or None) if request.method == "POST" and form.is_valid(): matriculas = form.cleaned_data["matriculas"] situacao = form.cleaned_data["situacao"] vinculos = Vinculo.objects.filter( ano_censo=selected_year, matricula__in=matriculas, ) found_matriculas = set(vinculos.values_list("matricula", flat=True)) updated = vinculos.update(situacao=situacao) result = { "updated": updated, "requested": len(matriculas), "not_found": [ matricula for matricula in matriculas if matricula not in found_matriculas ], "situacao": dict(Vinculo.Situacao.choices).get(int(situacao), situacao), "ano_censo": selected_year, } messages.success(request, "Situacao processada.") return render( request, "core/processar_situacao.html", { "form": form, "result": result, "selected_year": selected_year, }, ) @login_required def crud_list(request: HttpRequest, slug: str) -> HttpResponse: config = get_crud_config(slug) query = request.GET.get("q", "").strip() queryset = config.model.objects.all() selected_year = get_selected_ano_censo(request) if config.model is Vinculo: queryset = queryset.filter(ano_censo=selected_year) if query: queryset = queryset.filter(build_search_query(config, query)) paginator = Paginator(queryset, 20) page_obj = paginator.get_page(request.GET.get("page")) rows = [ { "object": obj, "values": [format_list_field_value(obj, field) for field in config.list_fields], } for obj in page_obj.object_list ] return render( request, "core/crud/list.html", { "config": config, "query": query, "page_obj": page_obj, "rows": rows, }, ) @login_required def crud_detail(request: HttpRequest, slug: str, pk: str) -> HttpResponse: config = get_crud_config(slug) obj = get_object_or_404(config.model, pk=pk) vinculo_navigation = vinculo_person_navigation(obj) if isinstance(obj, Vinculo) else None return render( request, "core/crud/detail.html", { "config": config, "object": obj, "fields": object_fields(obj), "vinculo_navigation": vinculo_navigation, }, ) @login_required def crud_create(request: HttpRequest, slug: str) -> HttpResponse: config = get_crud_config(slug) form_class = build_model_form(config.model) form = form_class(request.POST or None) if request.method == "POST" and form.is_valid(): obj = form.save() messages.success(request, f"{config.title} salvo com sucesso.") return redirect("crud_detail", slug=slug, pk=obj.pk) return render( request, "core/crud/form.html", { "config": config, "form": form, "action": "Criar", }, ) @login_required def crud_update(request: HttpRequest, slug: str, pk: str) -> HttpResponse: config = get_crud_config(slug) obj = get_object_or_404(config.model, pk=pk) form_class = build_model_form(config.model) form = form_class(request.POST or None, instance=obj) if request.method == "POST" and form.is_valid(): obj = form.save() messages.success(request, f"{config.title} atualizado com sucesso.") return redirect("crud_detail", slug=slug, pk=obj.pk) return render( request, "core/crud/form.html", { "config": config, "form": form, "object": obj, "action": "Editar", }, ) @login_required def crud_delete(request: HttpRequest, slug: str, pk: str) -> HttpResponse: config = get_crud_config(slug) obj = get_object_or_404(config.model, pk=pk) if request.method == "POST": obj.delete() messages.success(request, f"{config.title} excluido com sucesso.") return redirect("crud_list", slug=slug) return render( request, "core/crud/confirm_delete.html", { "config": config, "object": obj, }, ) @login_required def pending_required_fields(request: HttpRequest) -> HttpResponse: pending = [] selected_year = get_selected_ano_censo(request) for config in CRUD_LINKS: queryset = config.model.objects.all() if config.model is Vinculo: queryset = queryset.filter(ano_censo=selected_year) for obj in queryset: missing = missing_required_fields(obj) if missing: pending.append( { "config": config, "object": obj, "missing": missing, } ) paginator = Paginator(pending, 20) page_obj = paginator.get_page(request.GET.get("page")) return render( request, "core/pending_required_fields.html", { "page_obj": page_obj, "total_pending": len(pending), }, ) def get_crud_config(slug: str) -> CrudConfig: return get_object_or_404_config(slug) def get_object_or_404_config(slug: str) -> CrudConfig: config = CRUD_CONFIGS.get(slug) if config is None: from django.http import Http404 raise Http404("Cadastro nao encontrado") return config def build_search_query(config: CrudConfig, query: str) -> Q: filters = Q() for field, lookup in config.search_fields: if lookup == "int": if query.isdigit(): filters |= Q(**{field: int(query)}) else: filters |= Q(**{f"{field}__{lookup}": query}) return filters def build_model_form(model: type[Model]) -> type[forms.ModelForm]: form_class = modelform_factory(model, fields="__all__") class StyledForm(form_class): def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) for field in self.fields.values(): css_class = "form-check-input" if isinstance( field.widget, forms.CheckboxInput, ) else "form-control" current_class = field.widget.attrs.get("class", "") field.widget.attrs["class"] = f"{current_class} {css_class}".strip() return StyledForm def format_field_value(obj: Model, field: str) -> Any: value = getattr(obj, field) display_method = getattr(obj, f"get_{field}_display", None) if callable(display_method): return display_method() return value def format_list_field_value(obj: Model, field: str) -> dict[str, Any]: raw_value = getattr(obj, field) is_status = isinstance(obj, Curso) and field == "situacao_funcionamento_curso" return { "value": format_field_value(obj, field), "is_boolean": isinstance(raw_value, bool), "is_status": is_status, "status_class": course_status_class(raw_value) if is_status else "", "raw": raw_value, } def course_status_class(value: str) -> str: normalized = str(value or "").strip().lower() if normalized == "em atividade": return "bg-green-lt text-green" if normalized == "em extinção" or normalized == "em extincao": return "bg-yellow-lt text-yellow" return "bg-secondary-lt text-secondary" def object_fields(obj: Model) -> list[tuple[str, Any]]: fields = [] for field in obj._meta.fields: value = format_field_value(obj, field.name) fields.append((field.verbose_name.title(), value)) return fields def vinculo_person_navigation(vinculo: Vinculo) -> dict[str, Any]: vinculos = list( Vinculo.objects.filter(pessoa=vinculo.pessoa) .select_related("curso", "pessoa") .order_by("-ano_censo", "curso__descricao", "id") ) current_index = next( ( index for index, pessoa_vinculo in enumerate(vinculos) if pessoa_vinculo.pk == vinculo.pk ), None, ) previous_vinculo = None next_vinculo = None if current_index is not None: if current_index > 0: previous_vinculo = vinculos[current_index - 1] if current_index < len(vinculos) - 1: next_vinculo = vinculos[current_index + 1] return { "pessoa": vinculo.pessoa, "vinculos": vinculos, "previous": previous_vinculo, "next": next_vinculo, } def missing_required_fields(obj: Model) -> list[str]: missing = [] for field in obj._meta.fields: if field.primary_key or field.auto_created or field.blank or field.null: continue value = getattr(obj, field.name) if value is None or value == "": missing.append(field.verbose_name.title()) if isinstance(obj, Curso) and obj.carga_horaria_total == 0: missing.append("Carga Horaria Total") return missing def crud_list_url(slug: str) -> str: return reverse("crud_list", kwargs={"slug": slug})