from __future__ import annotations from concurrent.futures import ThreadPoolExecutor, as_completed import csv from dataclasses import dataclass from datetime import date, datetime from email.utils import parsedate_to_datetime import itertools import json from pathlib import Path import re from typing import Any, Callable, Iterable, Iterator from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen from django.db import IntegrityError, transaction from .models import ( AnoCenso, Curso, Instituicao, Municipio, Pais, Pessoa, Polo, UF, Vinculo, ) GENNERA_USER_AGENT = ( "Mozilla/5.0 (X11; Linux x86_64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0.0.0 Safari/537.36" ) @dataclass class ImportCursosResult: created: int = 0 updated: int = 0 without_carga_horaria: int = 0 total: int = 0 @dataclass class ImportAlunoError: line: int record_type: str identifier: str message: str @dataclass class ImportAlunosResult: pessoas_created: int = 0 pessoas_updated: int = 0 vinculos_created: int = 0 vinculos_updated: int = 0 ignored: int = 0 errors: list[ImportAlunoError] | None = None def __post_init__(self) -> None: if self.errors is None: self.errors = [] @dataclass class ImportPessoaApiError: identifier: str message: str @dataclass class ImportPessoasApiResult: created: int = 0 skipped: int = 0 ignorados_sem_cpf: int = 0 errors: list[ImportPessoaApiError] | None = None def __post_init__(self) -> None: if self.errors is None: self.errors = [] @property def total(self) -> int: return ( self.created + self.skipped + self.ignorados_sem_cpf + len(self.errors or []) ) @dataclass class ImportVinculoApiError: identifier: str message: str @dataclass class ImportVinculosApiResult: pessoas_processed: int = 0 pessoas_without_vinculo: int = 0 created: int = 0 updated: int = 0 situacao_nao_informada_corrigidos: int = 0 errors: list[ImportVinculoApiError] | None = None def __post_init__(self) -> None: if self.errors is None: self.errors = [] @property def total(self) -> int: return self.created + self.updated + len(self.errors or []) @dataclass class ImportCargaHorariaError: identifier: str message: str @dataclass class ImportCargaHorariaResult: matriculas_processadas: int = 0 sem_dados: int = 0 vinculos_atualizados: int = 0 errors: list[ImportCargaHorariaError] | None = None def __post_init__(self) -> None: if self.errors is None: self.errors = [] @dataclass class ImportPolosResult: created: int = 0 updated: int = 0 cursos_vinculados: int = 0 cursos_nao_encontrados: int = 0 total: int = 0 def load_max_carga_horaria_by_curso(alunos_path: Path) -> dict[int, int]: cargas: dict[int, int] = {} with alunos_path.open(encoding="utf-8") as file: for line in file: fields = line.rstrip("\n").split("|") if not fields or fields[0] != "42": continue codigo_curso = parse_int(fields[3]) carga_horaria = parse_int(fields[55]) if codigo_curso is None or carga_horaria is None: continue cargas[codigo_curso] = max(cargas.get(codigo_curso, 0), carga_horaria) return cargas def import_cursos(csv_path: Path, alunos_path: Path) -> ImportCursosResult: cargas = load_max_carga_horaria_by_curso(alunos_path) result = ImportCursosResult() with csv_path.open(encoding="utf-8-sig", newline="") as file: reader = csv.DictReader(file, delimiter=";") for row in reader: codigo_mec = int(row["CODIGO_CURSO"]) carga_horaria_total = cargas.get(codigo_mec, 0) if carga_horaria_total == 0: result.without_carga_horaria += 1 _, created = Curso.objects.update_or_create( codigo_mec=codigo_mec, defaults={ "descricao": row["NOME_CURSO"], "situacao_funcionamento_curso": row[ "SITUACAO_FUNCIONAMENTO_CURSO" ], "licenciatura": row["GRAU_ACADEMICO"] == "Licenciatura", "ead": row["FORMATO_OFERTA"] == "EaD", "carga_horaria_total": carga_horaria_total, }, ) result.total += 1 if created: result.created += 1 else: result.updated += 1 return result def import_polos(csv_path: Path) -> ImportPolosResult: result = ImportPolosResult() with csv_path.open(encoding="utf-8-sig", newline="") as file: reader = csv.DictReader(file, delimiter=";") for row in reader: codigo = int(row["CODIGOLOCALOFERTA"]) polo, created = Polo.objects.update_or_create( codigo=codigo, defaults={"descricao": row["NOMELOCALOFERTA"]}, ) result.total += 1 if created: result.created += 1 else: result.updated += 1 curso = Curso.objects.filter(codigo_mec=row["CODIGOCURSO"]).first() if curso is None: result.cursos_nao_encontrados += 1 continue if curso.ead and curso.polo_id is None: curso.polo = polo curso.save(update_fields=["polo"]) result.cursos_vinculados += 1 return result def fetch_gennera_pessoas_ingressantes_page( base_url: str, token: str, primeiro_ano: int, page: int, page_size: int = 100, order: str = "asc", timeout: int = 30, ) -> tuple[list[dict[str, Any]], int, int]: query = { "primeiroAno": primeiro_ano, "page": page, "page_size": page_size, "order": order, } url = f"{base_url.rstrip('/')}/api/censup_person_ingressantes?{urlencode(query)}" request = Request( url, headers={ "Accept": "application/json", "auth_token": token, "User-Agent": GENNERA_USER_AGENT, }, ) try: with urlopen(request, timeout=timeout) as response: total_count = int(response.headers.get("X-Total-Count", "0")) total_pages = int(response.headers.get("X-Total-Pages", "1")) payload = json.loads(response.read().decode("utf-8")) except HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise ValueError(f"API Gennera retornou HTTP {exc.code}: {detail}") from exc except URLError as exc: raise ValueError(f"Erro ao conectar na API Gennera: {exc.reason}") from exc data = payload.get("data") if isinstance(payload, dict) else payload if not isinstance(data, list): raise ValueError("Resposta da API Gennera sem lista 'data'.") return data, total_count, total_pages def iter_gennera_pessoas_ingressantes( base_url: str, token: str, primeiro_ano: int, page_size: int = 100, fetcher: Callable[ [str, str, int, int, int], tuple[list[dict[str, Any]], int, int] ] = fetch_gennera_pessoas_ingressantes_page, ) -> Iterator[tuple[int, int, int, list[dict[str, Any]]]]: page = 1 total_pages = 1 while page <= total_pages: items, total_count, total_pages = fetcher( base_url, token, primeiro_ano, page, page_size ) yield page, total_pages, total_count, items page += 1 def fetch_gennera_vinculos_censup_page( base_url: str, token: str, cpf: str, page: int, page_size: int = 100, ano: int | None = None, order_by: str = "ano", order: str = "asc", timeout: int = 30, ) -> tuple[list[dict[str, Any]], int, int]: query = { "cpf": cpf, "page": page, "page_size": page_size, "order_by": order_by, "order": order, } if ano is not None: query["ano"] = ano url = f"{base_url.rstrip('/')}/api/censup_vinculos?{urlencode(query)}" request = Request( url, headers={ "Accept": "application/json", "auth_token": token, "User-Agent": GENNERA_USER_AGENT, }, ) try: with urlopen(request, timeout=timeout) as response: total_count = int(response.headers.get("X-Total-Count", "0")) total_pages = int(response.headers.get("X-Total-Pages", "1")) payload = json.loads(response.read().decode("utf-8")) except HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise ValueError(f"API Gennera retornou HTTP {exc.code}: {detail}") from exc except URLError as exc: raise ValueError(f"Erro ao conectar na API Gennera: {exc.reason}") from exc data = payload.get("data") if isinstance(payload, dict) else payload if not isinstance(data, list): raise ValueError("Resposta da API Gennera sem lista 'data'.") return data, total_count, total_pages def fetch_gennera_vinculos_censup( base_url: str, token: str, ano: int, cpf: str, page_size: int = 100, fetcher: Callable[ ..., tuple[list[dict[str, Any]], int, int] ] = fetch_gennera_vinculos_censup_page, ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] page = 1 total_pages = 1 while page <= total_pages: page_items, _total_count, total_pages = fetcher( base_url, token, cpf, page, page_size, ano=ano ) items.extend(page_items) page += 1 return items def fetch_gennera_carga_horaria_page( base_url: str, token: str, matricula: str, page: int, page_size: int = 100, order_by: str = "ano", order: str = "asc", timeout: int = 30, ) -> tuple[list[dict[str, Any]], int, int]: query = { "matricula": matricula, "page": page, "page_size": page_size, "order_by": order_by, "order": order, } url = f"{base_url.rstrip('/')}/api/censup_carga_horaria?{urlencode(query)}" request = Request( url, headers={ "Accept": "application/json", "auth_token": token, "User-Agent": GENNERA_USER_AGENT, }, ) try: with urlopen(request, timeout=timeout) as response: total_count = int(response.headers.get("X-Total-Count", "0")) total_pages = int(response.headers.get("X-Total-Pages", "1")) payload = json.loads(response.read().decode("utf-8")) except HTTPError as exc: detail = exc.read().decode("utf-8", errors="replace") raise ValueError(f"API Gennera retornou HTTP {exc.code}: {detail}") from exc except URLError as exc: raise ValueError(f"Erro ao conectar na API Gennera: {exc.reason}") from exc data = payload.get("data") if isinstance(payload, dict) else payload if not isinstance(data, list): raise ValueError("Resposta da API Gennera sem lista 'data'.") return data, total_count, total_pages def fetch_gennera_carga_horaria( base_url: str, token: str, matricula: str, page_size: int = 100, fetcher: Callable[ ..., tuple[list[dict[str, Any]], int, int] ] = fetch_gennera_carga_horaria_page, ) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] page = 1 total_pages = 1 while page <= total_pages: page_items, _total_count, total_pages = fetcher( base_url, token, matricula, page, page_size ) items.extend(page_items) page += 1 return items def carga_horaria_por_ano(rows: list[dict[str, Any]]) -> dict[int, int]: by_ano: dict[int, int] = {} for row in rows: ano = parse_optional_api_int(row.get("ano")) carga = parse_optional_api_int(row.get("carga_horaria")) if ano is None or carga is None: continue by_ano[ano] = carga return by_ano def carga_horaria_ate_ano(by_ano: dict[int, int], ano: int) -> int | None: if ano in by_ano: return by_ano[ano] anteriores = [carga for row_ano, carga in by_ano.items() if row_ano <= ano] return max(anteriores) if anteriores else None def import_carga_horaria_from_gennera( base_url: str, token: str, page_size: int = 100, max_workers: int = 4, fetcher: Callable[[str, str, str, int], list[dict[str, Any]]] = fetch_gennera_carga_horaria, progress_callback: Callable[[int, int, str, list[Vinculo]], None] | None = None, ) -> ImportCargaHorariaResult: if not token: raise ValueError("Token da API Gennera nao configurado.") result = ImportCargaHorariaResult() matriculas = list( Vinculo.objects.exclude(matricula="") .order_by("matricula") .values_list("matricula", flat=True) .distinct() ) total_matriculas = len(matriculas) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_matricula = { executor.submit(fetcher, base_url, token, matricula, page_size): matricula for matricula in matriculas } index = 0 for future in as_completed(future_to_matricula): matricula = future_to_matricula[future] index += 1 result.matriculas_processadas += 1 try: rows = future.result() except Exception as exc: result.errors.append( ImportCargaHorariaError(identifier=matricula, message=str(exc)) ) if progress_callback is not None: progress_callback(index, total_matriculas, matricula, []) continue by_ano = carga_horaria_por_ano(rows) if not by_ano: result.sem_dados += 1 if progress_callback is not None: progress_callback(index, total_matriculas, matricula, []) continue atualizados = [] for vinculo in Vinculo.objects.filter(matricula=matricula).select_related("curso"): if vinculo.atualizado_manualmente: continue carga = carga_horaria_ate_ano(by_ano, vinculo.ano_censo_id) if carga is None or carga == vinculo.carga_horaria_integralizada: continue vinculo.carga_horaria_integralizada = carga vinculo.save(update_fields=["carga_horaria_integralizada"]) result.vinculos_atualizados += 1 atualizados.append(vinculo) if progress_callback is not None: progress_callback(index, total_matriculas, matricula, atualizados) return result def import_pessoas_ingressantes_from_gennera( primeiro_ano: int, base_url: str, token: str, page_size: int = 100, progress_callback: Callable[[int, int, int, int], None] | None = None, pages: Iterator[tuple[int, int, int, list[dict[str, Any]]]] | None = None, ) -> ImportPessoasApiResult: if not token: raise ValueError("GENNERA_LOCAL_AUTH_TOKEN nao configurado.") result = ImportPessoasApiResult() if pages is None: pages = iter_gennera_pessoas_ingressantes( base_url, token, primeiro_ano, page_size=page_size ) processed = 0 for page, total_pages, total_count, items in pages: for item in items: if not only_digits(str(item.get("cpf") or "")): result.ignorados_sem_cpf += 1 continue identifier = str(item.get("cpf") or item.get("idPerson") or "") try: pessoa, created = import_gennera_pessoa_record(item) attach_pessoa_ano_censo(pessoa, primeiro_ano) except Exception as exc: result.errors.append( ImportPessoaApiError(identifier=identifier, message=str(exc)) ) continue if created: result.created += 1 else: result.skipped += 1 processed += len(items) if progress_callback is not None: progress_callback(page, total_pages, processed, total_count) return result def corrigir_situacao_nao_informada( ano: int, base_url: str, token: str, max_workers: int = 4, ) -> int: vinculos = list( Vinculo.objects.filter( ano_censo=ano, situacao=Vinculo.Situacao.NAO_INFORMADA, atualizado_manualmente=False, ) .exclude(matricula="") .select_related("curso") ) if not vinculos: return 0 matriculas = {vinculo.matricula for vinculo in vinculos} carga_por_matricula: dict[str, dict[int, int]] = {} with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_matricula = { executor.submit(fetch_gennera_carga_horaria, base_url, token, matricula): matricula for matricula in matriculas } for future in as_completed(future_to_matricula): matricula = future_to_matricula[future] try: rows = future.result() except Exception: continue carga_por_matricula[matricula] = carga_horaria_por_ano(rows) corrigidos = 0 for vinculo in vinculos: by_ano = carga_por_matricula.get(vinculo.matricula, {}) carga = carga_horaria_ate_ano(by_ano, vinculo.ano_censo_id) if carga is not None: vinculo.carga_horaria_integralizada = carga vinculo.situacao = Vinculo.Situacao.DESVINCULADO vinculo.save(update_fields=["situacao", "carga_horaria_integralizada"]) corrigidos += 1 return corrigidos def import_vinculos_from_gennera( ano: int, base_url: str, token: str, fetcher: Callable[[str, str, int, str], list[dict[str, Any]]], progress_callback: Callable[[int, int, str, list[Vinculo]], None] | None = None, max_workers: int = 4, ) -> ImportVinculosApiResult: if not token: raise ValueError("Token da API Gennera nao configurado.") AnoCenso.objects.get_or_create(ano=ano) result = ImportVinculosApiResult() pessoas = list(Pessoa.objects.exclude(cpf="").order_by("nome", "cpf")) total_pessoas = len(pessoas) with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_pessoa = { executor.submit(fetcher, base_url, token, ano, pessoa.cpf): pessoa for pessoa in pessoas } index = 0 for future in as_completed(future_to_pessoa): pessoa = future_to_pessoa[future] index += 1 result.pessoas_processed += 1 try: rows = future.result() except Exception as exc: result.errors.append( ImportVinculoApiError(identifier=pessoa.cpf, message=str(exc)) ) if progress_callback is not None: progress_callback(index, total_pessoas, pessoa.cpf, []) continue rows = filter_gennera_vinculos_by_ano(rows, ano) if not rows: result.pessoas_without_vinculo += 1 updated = mark_person_vinculos_as_desvinculado(pessoa=pessoa, ano=ano) if updated: attach_pessoa_ano_censo(pessoa, ano) result.updated += updated if progress_callback is not None: progress_callback(index, total_pessoas, pessoa.cpf, []) continue try: imported = import_gennera_vinculo_records(rows, pessoa=pessoa, ano=ano) attach_pessoa_ano_censo(pessoa, ano) except Exception as exc: result.errors.append( ImportVinculoApiError(identifier=pessoa.cpf, message=str(exc)) ) if progress_callback is not None: progress_callback(index, total_pessoas, pessoa.cpf, []) continue for _, created in imported: if created: result.created += 1 else: result.updated += 1 if progress_callback is not None: progress_callback( index, total_pessoas, pessoa.cpf, [vinculo for vinculo, _ in imported] ) result.situacao_nao_informada_corrigidos = corrigir_situacao_nao_informada( ano=ano, base_url=base_url, token=token, max_workers=max_workers ) return result def import_gennera_pessoa_record(item: dict[str, Any]) -> tuple[Pessoa, bool]: cpf = only_digits(str(item.get("cpf") or "")) if not cpf: raise ValueError("CPF obrigatorio nao informado.") defaults = { "id_aluno_inep": parse_optional_api_int(item.get("idPerson")), "nome": required_api_text(item.get("nome"), "Nome"), "data_nascimento": parse_api_date(item.get("data_nascimento")), "cor_raca": parse_cor_raca(item.get("cor_raca")), "nacionalidade": parse_nacionalidade(item.get("nacionalidade")), "pais_origem": parse_pais_origem(item.get("pais_origem")), "aluno_com_deficiencia": parse_sim_nao_nao_dispoe(item.get("aluno_com_deficiencia")), "tipo_escola_ensino_medio": parse_tipo_escola_ensino_medio( item.get("tipo_escola_ensino_medio") ), } with transaction.atomic(): pessoa, created = Pessoa.objects.get_or_create(cpf=cpf, defaults=defaults) return pessoa, created def attach_pessoa_ano_censo(pessoa: Pessoa, ano: int) -> None: ano_censo, _ = AnoCenso.objects.get_or_create(ano=ano) pessoa.anos_censo.add(ano_censo) def import_gennera_vinculo_record( item: dict[str, Any], pessoa: Pessoa, ano: int, ) -> tuple[Vinculo, bool]: return import_gennera_vinculo_records([item], pessoa=pessoa, ano=ano)[0] def import_gennera_vinculo_records( rows: list[dict[str, Any]], pessoa: Pessoa, ano: int, ) -> list[tuple[Vinculo, bool]]: grouped: dict[int, list[dict[str, Any]]] = {} cursos: dict[int, Curso] = {} for item in rows: item_cpf = only_digits(str(item.get("cpf") or "")) if item_cpf and item_cpf != pessoa.cpf: raise ValueError(f"CPF retornado nao corresponde a pessoa local: {item_cpf}.") curso = get_gennera_curso(item, pessoa=pessoa) grouped.setdefault(curso.codigo_mec, []).append(item) cursos[curso.codigo_mec] = curso imported = [] for codigo_mec, items in grouped.items(): curso = cursos[codigo_mec] vinculo, created = import_gennera_vinculo_group( items, pessoa=pessoa, curso=curso, ano=ano, ) imported.append((vinculo, created)) return imported def import_gennera_vinculo_group( rows: list[dict[str, Any]], pessoa: Pessoa, curso: Curso, ano: int, ) -> tuple[Vinculo, bool]: reference_item = max(rows, key=lambda item: gennera_vinculo_period_key(item, ano)) ingresso_item = min(rows, key=lambda item: gennera_vinculo_period_key(item, ano)) periodo_referencia = parse_gennera_periodo_referencia(reference_item.get("referenceYear")) carga_horaria_integralizada = max( parse_optional_api_int(item.get("carga_horaria_acumulada")) or 0 for item in rows ) existing = Vinculo.objects.filter(pessoa=pessoa, curso=curso, ano_censo=ano).first() if existing is not None and existing.atualizado_manualmente: return existing, False situacao = gennera_vinculo_situacao( rows=rows, curso=curso, carga_horaria_integralizada=carga_horaria_integralizada, existing=existing, ) defaults = { "periodo_referencia": periodo_referencia, "turno": parse_gennera_turno(reference_item.get("turno")), "situacao": situacao, "semestre_ingresso": gennera_semestre_ingresso(ingresso_item, ano), "ingresso_vestibular": False, "ingresso_enem": False, "ingresso_avaliacao_seriada": False, "ingresso_selecao_simplificada": False, "ingresso_egresso_bi_li": False, "ingresso_pec_g": False, "ingresso_transferencia_ex_officio": False, "ingresso_decisao_judicial": False, "ingresso_vagas_remanescentes": False, "ingresso_programas_especiais": False, "apoio_social": False, "atividade_extracurricular": False, "carga_horaria_total_aluno": curso.carga_horaria_total, "carga_horaria_integralizada": carga_horaria_integralizada, "ingressou_acao_afirmativa": False, } with transaction.atomic(): vinculo, created = Vinculo.objects.update_or_create( pessoa=pessoa, curso=curso, ano_censo_id=ano, defaults=defaults, ) return vinculo, created def mark_person_vinculos_as_desvinculado(pessoa: Pessoa, ano: int) -> int: return Vinculo.objects.filter( pessoa=pessoa, ano_censo=ano, atualizado_manualmente=False ).exclude(situacao=Vinculo.Situacao.FORMADO).update( situacao=Vinculo.Situacao.DESVINCULADO ) def import_alunos(lines: Iterable[str], ano_censo: int) -> ImportAlunosResult: result = ImportAlunosResult() current_pessoa: Pessoa | None = None for line_number, raw_line in enumerate(lines, 1): fields = raw_line.rstrip("\n\r").split("|") record_type = fields[0] if fields else "" try: if record_type == "40": result.ignored += 1 continue if record_type == "41": current_pessoa, created = import_pessoa_record(fields) attach_pessoa_ano_censo(current_pessoa, ano_censo) if created: result.pessoas_created += 1 else: result.pessoas_updated += 1 continue if record_type == "42": if current_pessoa is None: raise ValueError("Registro 42 sem pessoa valida anterior.") _, created = import_vinculo_record(fields, current_pessoa, ano_censo) if created: result.vinculos_created += 1 else: result.vinculos_updated += 1 continue result.ignored += 1 result.errors.append( ImportAlunoError(line_number, record_type, "", "Tipo de registro ignorado.") ) except Exception as exc: if record_type == "41": current_pessoa = None result.errors.append( ImportAlunoError( line=line_number, record_type=record_type, identifier=record_identifier(fields), message=str(exc), ) ) return result def import_pessoa_record(fields: list[str]) -> tuple[Pessoa, bool]: require_field_count(fields, 23, "41") defaults = { "id_aluno_inep": parse_int(fields[1]), "nome": fields[2], "documento_estrangeiro": fields[4], "data_nascimento": parse_date(fields[5]), "cor_raca": parse_required_int(fields[6], "Cor/Raca"), "nacionalidade": parse_required_int(fields[7], "Nacionalidade"), "uf_nascimento": get_optional(UF, fields[8], "UF de nascimento"), "municipio_nascimento": get_optional( Municipio, fields[9], "Municipio de nascimento", ), "pais_origem": get_required(Pais, fields[10], "Pais de origem"), "aluno_com_deficiencia": parse_required_int( fields[11], "Aluno com deficiencia", ), "deficiencia_cegueira": parse_optional_bool(fields[12]), "deficiencia_baixa_visao": parse_optional_bool(fields[13]), "deficiencia_visao_monocular": parse_optional_bool(fields[14]), "deficiencia_surdez": parse_optional_bool(fields[15]), "deficiencia_auditiva": parse_optional_bool(fields[16]), "deficiencia_fisica": parse_optional_bool(fields[17]), "deficiencia_surdocegueira": parse_optional_bool(fields[18]), "deficiencia_intelectual": parse_optional_bool(fields[19]), "deficiencia_tea": parse_optional_bool(fields[20]), "deficiencia_altas_habilidades": parse_optional_bool(fields[21]), "tipo_escola_ensino_medio": parse_required_int( fields[22], "Tipo de escola do ensino medio", ), } cpf = fields[3] if not cpf: raise ValueError("CPF obrigatorio nao informado.") with transaction.atomic(): pessoa, created = Pessoa.objects.get_or_create( cpf=cpf, defaults=defaults, ) if not created and not pessoa.atualizado_manualmente: for field, value in defaults.items(): if field in {"nome", "data_nascimento"}: continue setattr(pessoa, field, value) pessoa.save() return pessoa, created def import_vinculo_record( fields: list[str], pessoa: Pessoa, ano_censo: int, ) -> tuple[Vinculo, bool]: require_field_count(fields, 73, "42") curso = get_required_by_field(Curso, "codigo_mec", fields[3], "Curso") defaults = { "matricula": fields[1], "periodo_referencia": parse_optional_int(fields[2]), "polo": get_optional_by_field(Polo, "codigo", fields[4], "Polo"), "turno": parse_optional_int(fields[5]), "situacao": parse_situacao(fields[6]), "curso_origem": get_optional_by_field(Curso, "codigo_mec", fields[7], "Curso origem"), "semestre_conclusao": parse_optional_int(fields[8]), "aluno_parfor": parse_optional_bool(fields[9]), "segunda_licenciatura_formacao_pedagogica": parse_optional_bool(fields[10]), "tipo_segunda_licenciatura_formacao_pedagogica": parse_optional_int(fields[11]), "semestre_ingresso": fields[12], "ingresso_vestibular": parse_required_bool(fields[13], "Vestibular"), "ingresso_enem": parse_required_bool(fields[14], "Enem"), "ingresso_avaliacao_seriada": parse_required_bool(fields[15], "Avaliacao seriada"), "ingresso_selecao_simplificada": parse_required_bool(fields[16], "Selecao simplificada"), "ingresso_egresso_bi_li": parse_required_bool(fields[17], "Egresso BI/LI"), "ingresso_pec_g": parse_required_bool(fields[18], "PEC-G"), "ingresso_transferencia_ex_officio": parse_required_bool(fields[19], "Transferencia ex officio"), "ingresso_decisao_judicial": parse_required_bool(fields[20], "Decisao judicial"), "ingresso_vagas_remanescentes": parse_required_bool(fields[21], "Vagas remanescentes"), "ingresso_programas_especiais": parse_required_bool(fields[22], "Programas especiais"), "mobilidade_academica": parse_optional_bool(fields[23]), "tipo_mobilidade_academica": parse_optional_int(fields[24]), "mobilidade_ies_destino": get_optional(Instituicao, fields[25], "IES destino"), "mobilidade_pais_destino": get_optional(Pais, fields[26], "Pais destino"), "financiamento_estudantil": parse_optional_bool(fields[27]), "financiamento_fies": parse_optional_bool(fields[28]), "financiamento_governo_estadual": parse_optional_bool(fields[29]), "financiamento_governo_municipal": parse_optional_bool(fields[30]), "financiamento_ies": parse_optional_bool(fields[31]), "financiamento_entidades_externas": parse_optional_bool(fields[32]), "financiamento_prouni_integral": parse_optional_bool(fields[33]), "financiamento_prouni_parcial": parse_optional_bool(fields[34]), "financiamento_nao_reembolsavel_entidades_externas": parse_optional_bool(fields[35]), "financiamento_nao_reembolsavel_governo_estadual": parse_optional_bool(fields[36]), "financiamento_nao_reembolsavel_ies": parse_optional_bool(fields[37]), "financiamento_nao_reembolsavel_governo_municipal": parse_optional_bool(fields[38]), "apoio_social": parse_required_bool(fields[39], "Apoio social", default=False), "apoio_alimentacao": parse_optional_bool(fields[40]), "apoio_moradia": parse_optional_bool(fields[41]), "apoio_transporte": parse_optional_bool(fields[42]), "apoio_material_didatico": parse_optional_bool(fields[43]), "apoio_bolsa_trabalho": parse_optional_bool(fields[44]), "apoio_bolsa_permanencia": parse_optional_bool(fields[45]), "atividade_extracurricular": parse_required_bool(fields[46], "Atividade extracurricular", default=False), "atividade_pesquisa": parse_optional_bool(fields[47]), "bolsa_pesquisa": parse_optional_bool(fields[48]), "atividade_extensao": parse_optional_bool(fields[49]), "bolsa_extensao": parse_optional_bool(fields[50]), "atividade_monitoria": parse_optional_bool(fields[51]), "bolsa_monitoria": parse_optional_bool(fields[52]), "atividade_estagio_nao_obrigatorio": parse_optional_bool(fields[53]), "bolsa_estagio_nao_obrigatorio": parse_optional_bool(fields[54]), "carga_horaria_total_aluno": curso.carga_horaria_total, "carga_horaria_integralizada": parse_optional_int(fields[56]) or 0, "ingressou_acao_afirmativa": parse_required_bool(fields[57], "Acao afirmativa", default=False), "tipo_politica_acao_afirmativa": parse_optional_int(fields[58]), "tipo_acao_afirmativa_lei_cotas": parse_optional_int(fields[59]), "reserva_ppi": parse_optional_bool(fields[60]), "reserva_renda": parse_optional_bool(fields[61]), "reserva_pessoa_com_deficiencia": parse_optional_bool(fields[62]), "reserva_escola_publica": parse_optional_bool(fields[63]), "reserva_quilombola": parse_optional_bool(fields[64]), "reserva_transgenero_travesti": parse_optional_bool(fields[65]), "reserva_estudante_internacional": parse_optional_bool(fields[66]), "reserva_refugiado_apatrida_visto_humanitario": parse_optional_bool(fields[67]), "reserva_idoso": parse_optional_bool(fields[68]), "reserva_povos_comunidades_tradicionais": parse_optional_bool(fields[69]), "reserva_olimpiadas_conhecimento": parse_optional_bool(fields[70]), "reserva_outros": parse_optional_bool(fields[71]), "justificativa": parse_optional_int(fields[72]), } existing = Vinculo.objects.filter( pessoa=pessoa, curso=curso, ano_censo_id=ano_censo ).first() if existing is not None and existing.atualizado_manualmente: return existing, False with transaction.atomic(): vinculo, created = Vinculo.objects.update_or_create( pessoa=pessoa, curso=curso, ano_censo_id=ano_censo, defaults=defaults, ) return vinculo, created def _fmt_optional_int(value: int | None) -> str: return "" if value is None else str(value) def _fmt_optional_bool(value: bool | None) -> str: if value is None: return "" return "1" if value else "0" def _fmt_required_bool(value: bool) -> str: return "1" if value else "0" def _fmt_date(value: date | None) -> str: return "" if value is None else value.strftime("%d%m%Y") def _fmt_fk_pk(value: Any) -> str: return "" if value is None else str(value) def export_pessoa_record(pessoa: Pessoa) -> str: fields = [ "41", _fmt_optional_int(pessoa.id_aluno_inep), pessoa.nome, pessoa.cpf, pessoa.documento_estrangeiro, _fmt_date(pessoa.data_nascimento), str(pessoa.cor_raca), str(pessoa.nacionalidade), _fmt_fk_pk(pessoa.uf_nascimento_id), _fmt_fk_pk(pessoa.municipio_nascimento_id), _fmt_fk_pk(pessoa.pais_origem_id), str(pessoa.aluno_com_deficiencia), _fmt_optional_bool(pessoa.deficiencia_cegueira), _fmt_optional_bool(pessoa.deficiencia_baixa_visao), _fmt_optional_bool(pessoa.deficiencia_visao_monocular), _fmt_optional_bool(pessoa.deficiencia_surdez), _fmt_optional_bool(pessoa.deficiencia_auditiva), _fmt_optional_bool(pessoa.deficiencia_fisica), _fmt_optional_bool(pessoa.deficiencia_surdocegueira), _fmt_optional_bool(pessoa.deficiencia_intelectual), _fmt_optional_bool(pessoa.deficiencia_tea), _fmt_optional_bool(pessoa.deficiencia_altas_habilidades), str(pessoa.tipo_escola_ensino_medio), ] return "|".join(fields) def export_vinculo_record(vinculo: Vinculo) -> str: situacao = ( "-1" if vinculo.situacao == Vinculo.Situacao.NAO_INFORMADA else str(vinculo.situacao) ) polo = vinculo.polo or vinculo.curso.polo fields = [ "42", vinculo.matricula, _fmt_optional_int(vinculo.periodo_referencia), _fmt_fk_pk(vinculo.curso_id), _fmt_fk_pk(polo.codigo if polo else None), _fmt_optional_int(vinculo.turno), situacao, _fmt_fk_pk(vinculo.curso_origem_id), _fmt_optional_int(vinculo.semestre_conclusao), _fmt_optional_bool(vinculo.aluno_parfor), _fmt_optional_bool(vinculo.segunda_licenciatura_formacao_pedagogica), _fmt_optional_int(vinculo.tipo_segunda_licenciatura_formacao_pedagogica), vinculo.semestre_ingresso, _fmt_required_bool(vinculo.ingresso_vestibular), _fmt_required_bool(vinculo.ingresso_enem), _fmt_required_bool(vinculo.ingresso_avaliacao_seriada), _fmt_required_bool(vinculo.ingresso_selecao_simplificada), _fmt_required_bool(vinculo.ingresso_egresso_bi_li), _fmt_required_bool(vinculo.ingresso_pec_g), _fmt_required_bool(vinculo.ingresso_transferencia_ex_officio), _fmt_required_bool(vinculo.ingresso_decisao_judicial), _fmt_required_bool(vinculo.ingresso_vagas_remanescentes), _fmt_required_bool(vinculo.ingresso_programas_especiais), _fmt_optional_bool(vinculo.mobilidade_academica), _fmt_optional_int(vinculo.tipo_mobilidade_academica), _fmt_fk_pk(vinculo.mobilidade_ies_destino_id), _fmt_fk_pk(vinculo.mobilidade_pais_destino_id), _fmt_optional_bool(vinculo.financiamento_estudantil), _fmt_optional_bool(vinculo.financiamento_fies), _fmt_optional_bool(vinculo.financiamento_governo_estadual), _fmt_optional_bool(vinculo.financiamento_governo_municipal), _fmt_optional_bool(vinculo.financiamento_ies), _fmt_optional_bool(vinculo.financiamento_entidades_externas), _fmt_optional_bool(vinculo.financiamento_prouni_integral), _fmt_optional_bool(vinculo.financiamento_prouni_parcial), _fmt_optional_bool(vinculo.financiamento_nao_reembolsavel_entidades_externas), _fmt_optional_bool(vinculo.financiamento_nao_reembolsavel_governo_estadual), _fmt_optional_bool(vinculo.financiamento_nao_reembolsavel_ies), _fmt_optional_bool(vinculo.financiamento_nao_reembolsavel_governo_municipal), _fmt_required_bool(vinculo.apoio_social), _fmt_optional_bool(vinculo.apoio_alimentacao), _fmt_optional_bool(vinculo.apoio_moradia), _fmt_optional_bool(vinculo.apoio_transporte), _fmt_optional_bool(vinculo.apoio_material_didatico), _fmt_optional_bool(vinculo.apoio_bolsa_trabalho), _fmt_optional_bool(vinculo.apoio_bolsa_permanencia), _fmt_required_bool(vinculo.atividade_extracurricular), _fmt_optional_bool(vinculo.atividade_pesquisa), _fmt_optional_bool(vinculo.bolsa_pesquisa), _fmt_optional_bool(vinculo.atividade_extensao), _fmt_optional_bool(vinculo.bolsa_extensao), _fmt_optional_bool(vinculo.atividade_monitoria), _fmt_optional_bool(vinculo.bolsa_monitoria), _fmt_optional_bool(vinculo.atividade_estagio_nao_obrigatorio), _fmt_optional_bool(vinculo.bolsa_estagio_nao_obrigatorio), str(vinculo.carga_horaria_total_aluno), str(vinculo.carga_horaria_integralizada), _fmt_required_bool(vinculo.ingressou_acao_afirmativa), _fmt_optional_int(vinculo.tipo_politica_acao_afirmativa), _fmt_optional_int(vinculo.tipo_acao_afirmativa_lei_cotas), _fmt_optional_bool(vinculo.reserva_ppi), _fmt_optional_bool(vinculo.reserva_renda), _fmt_optional_bool(vinculo.reserva_pessoa_com_deficiencia), _fmt_optional_bool(vinculo.reserva_escola_publica), _fmt_optional_bool(vinculo.reserva_quilombola), _fmt_optional_bool(vinculo.reserva_transgenero_travesti), _fmt_optional_bool(vinculo.reserva_estudante_internacional), _fmt_optional_bool(vinculo.reserva_refugiado_apatrida_visto_humanitario), _fmt_optional_bool(vinculo.reserva_idoso), _fmt_optional_bool(vinculo.reserva_povos_comunidades_tradicionais), _fmt_optional_bool(vinculo.reserva_olimpiadas_conhecimento), _fmt_optional_bool(vinculo.reserva_outros), _fmt_optional_int(vinculo.justificativa), ] return "|".join(fields) def build_aluno_export_lines( vinculos: Iterable[Vinculo], ies_codigo: int, ) -> Iterator[str]: yield f"40|{ies_codigo}" for _pessoa_id, group in itertools.groupby(vinculos, key=lambda v: v.pessoa_id): vinculos_grupo = list(group) pessoa = vinculos_grupo[0].pessoa yield export_pessoa_record(pessoa) for vinculo in vinculos_grupo: yield export_vinculo_record(vinculo) def build_aluno_export_file(vinculos: Iterable[Vinculo], ies_codigo: int) -> str: return "\n".join(build_aluno_export_lines(vinculos, ies_codigo)) + "\n" def require_field_count(fields: list[str], expected: int, record_type: str) -> None: if len(fields) != expected: raise ValueError( f"Registro {record_type} com {len(fields)} campos; esperado {expected}." ) def record_identifier(fields: list[str]) -> str: if not fields: return "" if fields[0] == "41" and len(fields) > 3: return fields[3] if fields[0] == "42" and len(fields) > 3: return fields[3] return "" def parse_date(value: str): return datetime.strptime(value, "%d%m%Y").date() def parse_required_int(value: str, field_name: str) -> int: parsed = parse_int(value) if parsed is None: raise ValueError(f"{field_name} obrigatorio nao informado.") return parsed def parse_optional_int(value: str) -> int | None: return parse_int(value) def parse_situacao(value: str) -> int: if value == "-1": return Vinculo.Situacao.NAO_INFORMADA return parse_required_int(value, "Situacao") def parse_optional_bool(value: str) -> bool | None: if value == "": return None return parse_required_bool(value, "Booleano") def parse_required_bool(value: str, field_name: str, default: bool | None = None) -> bool: if value == "" and default is not None: return default if value == "0": return False if value == "1": return True raise ValueError(f"{field_name} deve ser 0 ou 1.") def get_required(model, pk: str, field_name: str): if not pk: raise ValueError(f"{field_name} obrigatorio nao informado.") try: return model.objects.get(pk=pk) except model.DoesNotExist as exc: raise ValueError(f"{field_name} nao encontrado: {pk}.") from exc def get_optional(model, pk: str, field_name: str): if not pk: return None return get_required(model, pk, field_name) def get_optional_by_field(model, field: str, value: str, field_name: str): if not value: return None return get_required_by_field(model, field, value, field_name) def get_required_by_field(model, field: str, value: str, field_name: str): if not value: raise ValueError(f"{field_name} obrigatorio nao informado.") try: return model.objects.get(**{field: value}) except model.DoesNotExist as exc: raise ValueError(f"{field_name} nao encontrado: {value}.") from exc def parse_int(value: str) -> int | None: value = value.strip() if not value: return None return int(value) def parse_optional_api_int(value: Any) -> int | None: if value in {None, ""}: return None return int(value) def required_api_text(value: Any, field_name: str) -> str: text = str(value or "").strip() if not text: raise ValueError(f"{field_name} obrigatorio nao informado.") return text def only_digits(value: str) -> str: return "".join(char for char in value if char.isdigit()) def parse_api_date(value: Any): if not value: return None if hasattr(value, "date"): return value.date() text = str(value).strip() for date_format in ("%Y-%m-%d", "%d/%m/%Y", "%d%m%Y"): try: return datetime.strptime(text, date_format).date() except ValueError: pass return parsedate_to_datetime(text).date() def normalize_label(value: Any) -> str: return str(value or "").strip().casefold() def parse_cor_raca(value: Any) -> int: if value in {None, ""}: return Pessoa.CorRaca.NAO_DISPOE if str(value).strip().isdigit(): return int(value) labels = { "nao declarada": Pessoa.CorRaca.NAO_DECLARADA, "não declarada": Pessoa.CorRaca.NAO_DECLARADA, "branca": Pessoa.CorRaca.BRANCA, "preta": Pessoa.CorRaca.PRETA, "parda": Pessoa.CorRaca.PARDA, "amarela": Pessoa.CorRaca.AMARELA, "indigena": Pessoa.CorRaca.INDIGENA, "indígena": Pessoa.CorRaca.INDIGENA, } return labels.get(normalize_label(value), Pessoa.CorRaca.NAO_DISPOE) def parse_nacionalidade(value: Any) -> int: label = normalize_label(value) if not label or label in {"brasil", "brasileira", "brasileiro"}: return Pessoa.Nacionalidade.BRASILEIRA_NATA if "naturaliza" in label: return Pessoa.Nacionalidade.BRASILEIRA_NATURALIZADA return Pessoa.Nacionalidade.ESTRANGEIRA def parse_pais_origem(value: Any) -> Pais: label = normalize_label(value) if not label or label in {"brasil", "bra"}: return get_required(Pais, "BRA", "Pais de origem") pais = Pais.objects.filter(nome__iexact=str(value).strip()).first() if pais is not None: return pais pais = Pais.objects.filter(codigo=str(value).strip().upper()).first() if pais is not None: return pais return get_required(Pais, "BRA", "Pais de origem") def parse_sim_nao_nao_dispoe(value: Any) -> int: if value in {None, ""}: return Pessoa.SimNaoNaoDispoe.NAO_DISPOE label = normalize_label(value) if label in {"1", "sim", "s"}: return Pessoa.SimNaoNaoDispoe.SIM if label in {"0", "nao", "não", "n"}: return Pessoa.SimNaoNaoDispoe.NAO return Pessoa.SimNaoNaoDispoe.NAO_DISPOE def parse_tipo_escola_ensino_medio(value: Any) -> int: if value in {None, ""}: return Pessoa.TipoEscolaEnsinoMedio.NAO_DISPOE label = normalize_label(value) if label in {"0", "privada", "privado"}: return Pessoa.TipoEscolaEnsinoMedio.PRIVADA if label in {"1", "publica", "pública", "publico", "público"}: return Pessoa.TipoEscolaEnsinoMedio.PUBLICA return Pessoa.TipoEscolaEnsinoMedio.NAO_DISPOE def get_gennera_curso(item: dict[str, Any], pessoa: Pessoa | None = None) -> Curso: course_code = parse_optional_api_int(item.get("course_code")) if course_code is not None: curso = Curso.objects.filter(codigo_mec=course_code).first() if curso is not None: return curso raise ValueError(f"Curso nao encontrado pelo codigo MEC: {course_code}.") if pessoa is not None: course_name = str(item.get("course") or "").strip().lower() if course_name: for vinculo in pessoa.vinculos.select_related("curso").all(): if vinculo.curso.descricao.strip().lower() in course_name: return vinculo.curso raise ValueError( "Codigo MEC ausente e curso nao encontrado pelo nome nos vinculos da pessoa." ) def parse_gennera_periodo_referencia(value: Any) -> int: text = str(value or "").strip() if "2" in text[-2:]: return Vinculo.PeriodoReferencia.SEGUNDO_SEMESTRE return Vinculo.PeriodoReferencia.PRIMEIRO_SEMESTRE def filter_gennera_vinculos_by_ano( rows: list[dict[str, Any]], ano: int, ) -> list[dict[str, Any]]: return [ item for item in rows if gennera_vinculo_ano(item) == ano ] def gennera_vinculo_ano(item: dict[str, Any]) -> int | None: parsed_ano = parse_optional_api_int(item.get("ano")) if parsed_ano is not None: return parsed_ano reference_year = str(item.get("referenceYear") or "").strip() year = reference_year[:4] if year.isdigit(): return int(year) return None def gennera_vinculo_period_key(item: dict[str, Any], fallback_ano: int) -> tuple[int, int]: ano = gennera_vinculo_ano(item) or fallback_ano periodo = parse_gennera_periodo_referencia(item.get("referenceYear")) return ano, int(periodo) def gennera_vinculo_situacao( rows: list[dict[str, Any]], curso: Curso, carga_horaria_integralizada: int, existing: Vinculo | None = None, ) -> int: if existing is not None and existing.situacao == Vinculo.Situacao.FORMADO: return Vinculo.Situacao.FORMADO if carga_horaria_integralizada >= curso.carga_horaria_total: return Vinculo.Situacao.FORMADO periodos = { parse_gennera_periodo_referencia(item.get("referenceYear")) for item in rows } if Vinculo.PeriodoReferencia.SEGUNDO_SEMESTRE in periodos: return Vinculo.Situacao.CURSANDO if Vinculo.PeriodoReferencia.PRIMEIRO_SEMESTRE in periodos: return Vinculo.Situacao.DESVINCULADO return Vinculo.Situacao.DESVINCULADO def gennera_semestre_ingresso(item: dict[str, Any], fallback_ano: int) -> str: ano = gennera_vinculo_ano(item) or fallback_ano periodo = parse_gennera_periodo_referencia(item.get("referenceYear")) prefix = "02" if periodo == Vinculo.PeriodoReferencia.SEGUNDO_SEMESTRE else "01" return f"{prefix}{ano}" def parse_gennera_turno(value: Any) -> int | None: label = normalize_label(value) if not label: return None if label in {"1", "matutino", "manha", "manhã"}: return Vinculo.Turno.MATUTINO if label in {"2", "vespertino", "tarde"}: return Vinculo.Turno.VESPERTINO if label in {"3", "noturno", "noite"}: return Vinculo.Turno.NOTURNO if label in {"4", "integral"}: return Vinculo.Turno.INTEGRAL return None