Rogério Lima
3eca55578d
Skip re-fetching a CPF from the per-pessoa Gennera loop when the bulk ingressantes step already resolved it for the same ano, cutting redundant HTTP fan-out. Add bounded retry with backoff on transient connection errors (never on HTTP status errors) across the four Gennera fetch functions, backed by a shared requests.Session for connection reuse instead of one-off urllib calls. Also stop syncing carga_horaria_integralizada from the API once a vinculo is FORMADO, matching the existing sticky-situacao rule, in both the vinculo import and the standalone carga-horaria command. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from functools import partial
|
|
from typing import Any
|
|
|
|
from celery import shared_task
|
|
from django.conf import settings
|
|
|
|
from .importers import (
|
|
fetch_gennera_vinculos_censup,
|
|
import_carga_horaria_from_gennera,
|
|
import_pessoas_ingressantes_from_gennera,
|
|
import_vinculos_from_gennera,
|
|
)
|
|
|
|
|
|
@shared_task(bind=True)
|
|
def import_pessoas_ingressantes_task(
|
|
self, primeiro_ano: int, page_size: int = 100
|
|
) -> dict[str, Any]:
|
|
def report_progress(page: int, total_pages: int, processed: int, total_count: int) -> None:
|
|
self.update_state(
|
|
state="PROGRESS",
|
|
meta={
|
|
"processed": processed,
|
|
"total": total_count,
|
|
"message": f"Pagina {page}/{total_pages}",
|
|
},
|
|
)
|
|
|
|
result = import_pessoas_ingressantes_from_gennera(
|
|
primeiro_ano=primeiro_ano,
|
|
base_url=settings.GENNERA_LOCAL_BASE_URL,
|
|
token=settings.GENNERA_LOCAL_AUTH_TOKEN,
|
|
page_size=page_size,
|
|
progress_callback=report_progress,
|
|
)
|
|
return {
|
|
"created": result.created,
|
|
"skipped": result.skipped,
|
|
"ignorados_sem_cpf": result.ignorados_sem_cpf,
|
|
"errors": len(result.errors),
|
|
"total": result.total,
|
|
}
|
|
|
|
|
|
@shared_task(bind=True)
|
|
def import_vinculos_censup_task(
|
|
self, ano: int, page_size: int = 100, threads: int = 4
|
|
) -> dict[str, Any]:
|
|
def report_progress(index: int, total: int, cpf: str, vinculos: list) -> None:
|
|
self.update_state(
|
|
state="PROGRESS",
|
|
meta={
|
|
"processed": index,
|
|
"total": total,
|
|
"message": f"CPF {cpf}",
|
|
},
|
|
)
|
|
|
|
result = import_vinculos_from_gennera(
|
|
ano=ano,
|
|
base_url=settings.GENNERA_LOCAL_BASE_URL,
|
|
token=settings.GENNERA_LOCAL_AUTH_TOKEN,
|
|
fetcher=partial(fetch_gennera_vinculos_censup, page_size=page_size),
|
|
progress_callback=report_progress,
|
|
max_workers=threads,
|
|
ingressantes_page_size=page_size,
|
|
)
|
|
return {
|
|
"pessoas_processed": result.pessoas_processed,
|
|
"pessoas_without_vinculo": result.pessoas_without_vinculo,
|
|
"created": result.created,
|
|
"updated": result.updated,
|
|
"situacao_nao_informada_corrigidos": result.situacao_nao_informada_corrigidos,
|
|
"errors": len(result.errors),
|
|
}
|
|
|
|
|
|
@shared_task(bind=True)
|
|
def import_carga_horaria_censup_task(
|
|
self, page_size: int = 100, threads: int = 4
|
|
) -> dict[str, Any]:
|
|
def report_progress(index: int, total: int, matricula: str, vinculos: list) -> None:
|
|
self.update_state(
|
|
state="PROGRESS",
|
|
meta={
|
|
"processed": index,
|
|
"total": total,
|
|
"message": f"Matricula {matricula}",
|
|
},
|
|
)
|
|
|
|
result = import_carga_horaria_from_gennera(
|
|
base_url=settings.GENNERA_LOCAL_BASE_URL,
|
|
token=settings.GENNERA_LOCAL_AUTH_TOKEN,
|
|
page_size=page_size,
|
|
max_workers=threads,
|
|
progress_callback=report_progress,
|
|
)
|
|
return {
|
|
"matriculas_processadas": result.matriculas_processadas,
|
|
"vinculos_atualizados": result.vinculos_atualizados,
|
|
"sem_dados": result.sem_dados,
|
|
"errors": len(result.errors),
|
|
}
|