Browse Source

Adiciona appconfig, atualiza job_controller

pull/201/head 4.0.36
Sesóstris Vieira 1 week ago
parent
commit
3ade1ce970
  1. 3
      requirements/requirements.txt
  2. 10
      sigi/apps/convenios/admin.py
  3. 75
      sigi/apps/convenios/appconfig.py
  4. 520
      sigi/apps/convenios/jobs/hourly/importa_gescon.py
  5. 19
      sigi/apps/convenios/migrations/0041_delete_gescon.py
  6. 599
      sigi/apps/convenios/models.py
  7. 11
      sigi/apps/convenios/templates/convenios/report/erros_gescon_report_view/report_pdf.html
  8. 1
      sigi/apps/convenios/urls.py
  9. 33
      sigi/apps/convenios/views.py
  10. 15
      sigi/apps/eventos/appconfig.py
  11. 7
      sigi/apps/eventos/jobs/daily/encerra_inscricao.py
  12. 8
      sigi/apps/utils/admin.py
  13. 10
      sigi/apps/utils/admin_urls.py
  14. 5
      sigi/apps/utils/management/jobs.py
  15. 16
      sigi/apps/utils/migrations/0009_delete_config.py
  16. 41
      sigi/apps/utils/models.py
  17. 18
      sigi/apps/utils/templates3/admin/jobs/job_result.html
  18. 54
      sigi/apps/utils/views.py
  19. 10
      sigi/menu_conf.yaml
  20. 1
      sigi/settings.py
  21. 2
      sigi/urls.py

3
requirements/requirements.txt

@ -37,4 +37,5 @@ weasyprint==68.1
xlsxwriter==3.2.9
djbs-theme @ git+https://github.com/interlegis/djbs-theme.git@v1.0.8
django-dashboard @ git+https://github.com/interlegis/django-dashboard.git
dx-job-controller @ git+https://github.com/interlegis/dx-job-controller.git@v1.0.3
dx-job-controller @ git+https://github.com/interlegis/dx-job-controller.git@v1.0.5
django-appconfig @ git+https://github.com/interlegis/django-appconfig.git@v1.0.0

10
sigi/apps/convenios/admin.py

@ -18,7 +18,6 @@ from sigi.apps.convenios.models import (
Convenio,
EquipamentoPrevisto,
Anexo,
Gescon,
)
from sigi.apps.utils.mixins import AsciifyQParameter
from sigi.apps.casas.admin import GerentesInterlegisFilter
@ -344,14 +343,5 @@ class EquipamentoPrevistoAdmin(admin.ModelAdmin):
)
@admin.register(Gescon)
class GesconAdmin(admin.ModelAdmin):
list_display = (
"url_gescon",
"email",
)
exclude = ["ultima_importacao", "checksums"]
admin.site.register(StatusConvenio)
admin.site.register(TipoSolicitacao)

75
sigi/apps/convenios/appconfig.py

@ -0,0 +1,75 @@
from django import forms
from django.utils.translation import gettext_lazy as _
class AppConfigForm(forms.Form):
form_title = _("Definições para convênios")
form_subtitle = _("Importação do Gescon")
url_gescon = forms.URLField(
required=True,
label=_("Webservice Gescon"),
initial=(
"https://adm.senado.gov.br/gestao-contratos/api/contratos"
"/busca?subespecie={s}"
),
help_text=_(
"Informe o ponto de consulta do webservice do Gescon, "
"inclusive com a querystring. No ponto onde deve ser "
"inserida a sigla da subespecie do contrato, use a "
"marcação {s}.<br/><strong>Por exemplo:</strong> "
"https://adm.senado.gov.br/gestao-contratos/api/contratos"
"/busca?subespecie=<strong>{s}</strong>"
),
)
subespecies = forms.CharField(
required=True,
label=_("Subespécies"),
initial="AC=ACT\nTC=TC",
help_text=_(
"Informe as siglas das subespécies de contratos que "
"devem ser pesquisados no Gescon com a sigla "
"correspondente do projeto no SIGI. Coloque um par de "
"siglas por linha, no formato SIGLA_GESTON=SIGLA_SIGI. "
"As siglas não encontradas serão ignoradas."
),
widget=forms.Textarea(),
)
palavras = forms.CharField(
required=True,
label=_("Palavras de filtro"),
initial="ILB\nINTERLEGIS\nSCCO",
help_text=_(
"Palavras que devem aparecer no campo OBJETO dos dados do "
"Gescon para identificar se o contrato pertence ao ILB. "
"<ul><li>Informe uma palavra por linha.</li>"
"<li>Ocorrendo qualquer uma das palavras, o contrato será "
"importado.</li></ul>"
),
widget=forms.Textarea(),
)
palavras_excluir = forms.CharField(
required=True,
label=_("palavras de exclusão"),
initial="DTCOM",
help_text=_(
"Palavras que não podem aparecer no campo OBJETO dos dados do "
"Gescon."
"<ul><li>Informe uma palavra por linha.</li>"
"<li>Ocorrendo qualquer uma das palavras, o contrato será "
"ignorado.</li></ul>"
),
widget=forms.Textarea(),
)
orgaos_gestores = forms.CharField(
required=True,
label=_("Órgãos gestores"),
initial="SCCO",
help_text=_(
"Siglas de órgãos gestores que devem aparecer no campo"
"ORGAOSGESTORESTITULARES"
"<ul><li>Informe um sigla por linha.</li>"
"<li>Ocorrendo qualquer uma das siglas, o contrato será "
"importado.</li></ul>"
),
widget=forms.Textarea(),
)

520
sigi/apps/convenios/jobs/hourly/importa_gescon.py

@ -1,16 +1,522 @@
import re
import sys
from django.forms.models import model_to_dict
import requests
from appconfig import config
from hashlib import md5
from url_normalize import url_normalize
from django.db.models import Q, F
from django.contrib.sites.shortcuts import get_current_site
from django.urls import reverse
from django.utils import timezone
from django.utils.translation import gettext as _
from django_extensions.management.jobs import HourlyJob
from sigi.apps.convenios.models import Gescon
from sigi.apps.utils import mask_cnpj, to_ascii
from sigi.apps.casas.models import Orgao
from sigi.apps.convenios.models import Projeto, Convenio
class Job(HourlyJob):
help = "Carga de dados do Gescon."
gescon = None
dominio = None
subespecies = None
todos_orgaos = None
novos = 0
erros = 0
atualizados = 0
def execute(self):
gescon = Gescon.load()
errors = gescon.importa_contratos()
if errors:
print(gescon.ultima_importacao, file=sys.stderr)
else:
print(gescon.ultima_importacao)
print(
"Importação iniciada em {timezone.localtime():%d/%m/%Y %H:%M:%S}\n"
"==========================================================\n"
)
if not self.setup():
return
for sigla_gescon, sigla_sigi in self.subespecies:
print(
_("\n**Importando subespécie {sigla_gescon}**").format(
sigla_gescon=sigla_gescon
)
)
try:
projeto = Projeto.objects.get(sigla=sigla_sigi)
except Projeto.DoesNotExist:
print(
_(
"Projeto com a sigla {sigla_sigi} não existe no SIGI"
).format(sigla_sigi=sigla_sigi),
file=sys.stderr,
)
continue
except Projeto.MultipleObjectsReturned:
print(
_(
"A sigla {sigla_sigi} foi cadastrada com duplicidade "
"no SIGI"
).format(sigla_sigi=sigla_sigi),
file=sys.stderr,
)
continue
nossos = self.get_contratos_gescon(sigla_gescon)
if not nossos:
print(
_(
"Nenhum contrato do tipo {sigla_gescon} pertence ao "
"ILB / Interlegis"
).format(sigla_gescon=sigla_gescon)
)
continue
# Desmarca erros de importações passadas
Convenio.objects.filter(projeto=projeto).update(erro_gescon=False)
for contrato in nossos:
if contrato["numero"] is None:
print(
_(
"\t* O contrato de id {id} subespécie {sub} foi "
"cadastrado no Gescon sem número e não será "
"importado."
).format(id=contrato["id"], sub=contrato["subEspecie"]),
file=sys.stderr,
)
continue
numero = re.sub(
r"(\d{4})(\d{4})", r"\1/\2", contrato["numero"].zfill(8)
)
sigad = re.sub(
r"(\d{5})(\d{6})(\d{4})(\d{2})",
r"\1.\2/\3-\4",
contrato["processo"].zfill(17),
)
if contrato["cnpjCpfFornecedor"]:
cnpj = contrato["cnpjCpfFornecedor"].zfill(14)
cnpj_masked = mask_cnpj(cnpj)
else:
cnpj = None
cnpj_masked = None
if contrato["nomeFornecedor"]:
nome = to_ascii(
contrato["nomeFornecedor"]
.replace("VEREADORES DE", "")
.replace("DO ESTADO", "")
.split("-")[0]
.split("/")[0]
.strip()
.replace(" ", " ")
)
else:
nome = None
convenio = self.get_convenio(
contrato, projeto, sigad, numero, cnpj_masked
)
if convenio:
# Encontrou um único convênio, basta atualizar seus dados
self.update_convenio(
convenio, contrato, projeto, sigad, numero
)
self.update_orgao(convenio.casa_legislativa, cnpj_masked)
else:
# Não encontrou o convênio. Um novo convênio precisa ser
# criado. Primeiro, é preciso identificar qual órgão
# consta no contrato do Gescon
orgao = self.get_orgao(
contrato, cnpj, cnpj_masked, nome, numero, sigad
)
if orgao is None:
# Não encontrou o órgão... bora reportar o erro
print(
_(
"\t* Órgão não encontrado no SIGI ou mais de "
"um órgão encontrado com o mesmo CNPJ ou nome. "
"Favor regularizar o cadastro: \n"
"\t * CNPJ: {cnpj}\n"
"\t * Nome: {nome}"
).format(
cnpj=contrato["cnpjCpfFornecedor"],
nome=contrato["nomeFornecedor"],
),
file=sys.stderr,
)
self.erros += 1
else:
# Bora criar o convênio
self.create_convenio(
orgao, contrato, projeto, sigad, numero
)
self.update_orgao(orgao, cnpj_masked)
self.novos += 1
print("\n\nRESUMO DA IMPORTAÇÃO")
print("--------------------\n\n")
print(
"Novos convenios adicionados ao SIGI: {novos}".format(
novos=self.novos
)
)
print(
"Convênios atualizados..............: {atualizados}".format(
atualizados=self.atualizados
)
)
print(
"Erros encontrados..................: {erros}".format(
erros=self.erros
)
)
def setup(self):
self.novos = 0
self.erros = 0
self.atualizados = 0
self.gescon = config.convenios
self.gescon.ultima_importacao = None
if self.gescon.checksums is None:
self.gescon.checksums = {}
self.dominio = get_current_site(None).domain
if self.gescon.palavras == "":
print(
_("Nenhuma palavra de pesquisa definida - processo abortado."),
file=sys.stderr,
)
return False
if self.gescon.orgaos_gestores == "":
print(
_("Nenhum órgão gestor definido - processo abortado"),
file=sys.stderr,
)
return False
if self.gescon.subespecies == "":
print(
_("Nenhuma subespécie definida - processo abortado."),
file=sys.stderr,
)
return False
if "{s}" not in self.gescon.url_gescon:
print(
_(
"Falta a marcação {s} na URL para indicar o local onde "
"inserir a sigla da subespécia na consulta ao webservice "
"- processo abortado."
),
file=sys.stderr,
)
return False
self.subespecies = {
tuple(s.split("=")) for s in self.gescon.subespecies.split()
}
self.todos_orgaos = [
(o, f"{to_ascii(o.nome)} - {o.uf_sigla}".lower())
for o in Orgao.objects.all()
.order_by()
.annotate(uf_sigla=F("municipio__uf__sigla"))
]
self.palavras = self.gescon.palavras.splitlines()
self.orgaos = self.gescon.orgaos_gestores.splitlines()
self.excludentes = self.gescon.palavras_excluir.splitlines()
return True
def get_contratos_gescon(self, sigla_gescon):
requests.packages.urllib3.disable_warnings()
url = self.gescon.url_gescon.format(s=sigla_gescon)
try:
response = requests.get(url, verify=False)
except Exception as e:
print(
_("\tErro ao acessar {url}: {message}").format(
url=url, message=e.message
),
file=sys.stderr,
)
return None
if response.status_code != 200:
print(
_(
"\tErro na leitura dos dados de {url}: "
"[{status_code}] {reason}"
).format(
url=url,
status_code=response.status_code,
reason=response.reason,
),
file=sys.stderr,
)
return None
if "application/json" not in response.headers.get("Content-Type", ""):
print(
_(
"\tResultado da consulta à {url} não retornou dados "
"em formato json: {content_type}"
).format(
url=url,
content_type=response.headers.get("Content-Type", ""),
),
file=sys.stderr,
)
return None
md5sum = md5(response.text.encode(response.encoding)).hexdigest()
if (
sigla_gescon in self.gescon.checksums
and self.gescon.checksums[sigla_gescon] == md5sum
):
print(
_(
"\tDados da subespécie {sigla_gescon} inalterados no "
"Gescon. Processamento desnecessário."
).format(sigla_gescon=sigla_gescon)
)
return None
self.gescon.checksums[sigla_gescon] = md5sum
contratos = response.json()
# Pegar só os contratos que possuem alguma das palavras-chave
nossos = [
c
for c in contratos
if (
any(palavra in c["objeto"] for palavra in self.palavras)
or any(
orgao in c["orgaosGestoresTitulares"]
for orgao in self.orgaos
if c["orgaosGestoresTitulares"] is not None
)
)
and not any(palavra in c["objeto"] for palavra in self.excludentes)
]
print(
_("\t{qty} contratos encontrados no Gescon").format(qty=len(nossos))
)
return nossos
def get_convenio(self, contrato, projeto, sigad, numero, cnpj_masked):
# Buscar o convenio pelo ID do Gescon
try:
convenio = Convenio.objects.get(id_gescon=contrato["id"])
return convenio
except Convenio.DoesNotExist:
# Não existe, mas vamos buscar com outras chaves candidatas
pass
# Buscar o Convenio pelo NUP #
convenios = Convenio.objects.filter(
projeto=projeto, num_processo_sf=sigad
)
num_convenios = convenios.count()
if num_convenios == 0:
# Nenhum convênio. Vamos continuar buscando em todos convenios
# desse tipo de projeto
convenios = Convenio.objects.filter(projeto=projeto)
elif num_convenios == 1:
# Encontrou um único, então deve ser ele
return convenios.get()
# Se chegou aqui, é porque encontrou vários ou nenhum.
# Vamos tentar diferenciar pelas outras chaves candidatas
convenios = convenios.filter(
Q(Q(num_convenio=numero) | Q(num_processo_sf=numero))
)
num_convenios = convenios.count()
if num_convenios == 0:
# Nenhum encontrado. Já finaliza como não encontrado
return None
elif num_convenios == 1:
# Achou um. Deve ser ele
return convenios.get()
# Encontrou N: Marcamos todos como erro e reportamos
urls = []
for c in convenios:
uri = reverse("admin:convenios_convenio_change", args=[c.id])
url = url_normalize(f"{self.dominio}{uri}")
urls.append(f'<a href="{url}">{c.id}</a>')
urls = ", ".join(urls)
convenios.update(
erro_gescon=True,
observacao_gescon=_(
"Este convênio possui o mesmo número dos convenios {urls}"
).format(urls=urls),
)
print(
_(
"\t* O contrato {numero} no Gescon pode "
"ser relacionado aos seguintes convênios "
"do SIGI: {urls}"
).format(numero=numero, urls=urls),
file=sys.stderr,
)
self.erros += 1
# Porém, talvez seja possível ser desambiguado pelo CNPJ do
# fornecedor
if cnpj_masked is not None:
convenios = convenios.filter(casa_legislativa__cnpj=cnpj_masked)
if convenios.count() == 1:
# Achou exatamente o único que deveria existir.
return convenios.get()
# Continua ambíguo. Não dá pra fazer nada.
return None
def get_orgao(self, contrato, cnpj, cnpj_masked, nome, numero, sigad):
if (cnpj is None) and (nome is None):
print(
_(
"\t* O contrato {numero} no Gescon não informa nem o CNPJ "
"nem o nome do órgão, então não é possível importar "
"para o SIGI."
).format(numero=numero),
file=sys.stderr,
)
self.erros += 1
return None
# Vamos tentar primeiro com o CNPJ
if cnpj is not None:
try:
orgao = Orgao.objects.get(cnpj=cnpj_masked)
return orgao
except Orgao.MultipleObjectsReturned:
# Pode acontecer de uma câmara usar o mesmo CNPJ da prefeitura,
# e ambos terem convênio com o ILB. Podemos tentar desambiguar
# pelo nome mais semelhante.
orgaos = Orgao.get_semelhantes(
to_ascii(contrato["nomeFornecedor"]).lower(),
[
(
o,
f"{to_ascii(o.nome)} - {o.uf_sigla}".lower(),
)
for o in Orgao.objects.filter(cnpj=cnpj_masked)
.order_by()
.annotate(uf_sigla=F("municipio__uf__sigla"))
],
min_ratio=0,
)
if orgaos:
# Retorna o mais semelhante
return orgaos[0][0]
except Orgao.DoesNotExist:
# Não encontrou nenhum. Vamos seguir sem órgao e tentar
# encontrar pelo nome logo abaixo
pass
# Não achou pelo CNPJ. Bora ver se acha por similaridade do nome
if nome is None:
# Também não tem nome... então temos que reportar erro
print(
_(
"\t* O contrato {numero} no Gescon com NUP sigad {sigad}, "
"fornecedor {cnpj_masked} não pode ser imortado porque "
"não é possível identificar o órgão no SIGI. Cadastre um "
"órgão com o CNPJ desse fornecedor, que na próxima "
"importação este contrato será importado."
).format(numero=numero, sigad=sigad, cnpj_masked=cnpj_masked),
file=sys.stderr,
)
self.erros += 1
return None
# Tentar primeiro com o nome igual veio do GESCON
semelhantes = Orgao.get_semelhantes(
to_ascii(contrato["nomeFornecedor"]).lower(),
self.todos_orgaos,
)
if not semelhantes:
# Não achou, então vamos tentar com o nome limpado
semelhantes = Orgao.get_semelhantes(
to_ascii(nome).lower(),
self.todos_orgaos,
)
if len(semelhantes) > 0:
# Encontrou algo semelhante.... bora usar.
return semelhantes[0][0]
# Não encontrou nada parecido. Bora reportar como erro
print(
_(
"\t* O contrato {numero} no Gescon com NUP Sigad {sigad}, "
"indica o fornecedor com CNPJ {cnpj_masked} e com o nome "
"{nome}, que não tem correspondência no SIGI. Este convênio "
"precisa ser cadastrado manualmente no SIGI para este erro "
"parar de acontecer."
).format(
numero=numero,
sigad=sigad,
cnpj_masked=cnpj_masked,
nome=contrato["nomeFornecedor"],
),
file=sys.stderr,
)
self.erros += 1
return None
def update_convenio(self, convenio, contrato, projeto, sigad, numero):
antes = model_to_dict(convenio)
convenio.projeto = projeto
convenio.num_processo_sf = sigad
convenio.num_convenio = numero
convenio.data_sigad = contrato["assinatura"]
convenio.observacao = contrato["objeto"]
convenio.data_retorno_assinatura = contrato["inicioVigencia"]
convenio.data_termino_vigencia = contrato["terminoVigencia"]
convenio.data_pub_diario = contrato["publicacao"]
convenio.atualizacao_gescon = timezone.localtime()
convenio.erro_gescon = False
convenio.observacao_gescon = ""
convenio.id_contrato_gescon = contrato["codTextoContrato"] or ""
convenio.id_gescon = contrato["id"]
if antes != model_to_dict(convenio):
convenio.save()
self.atualizados += 1
def create_convenio(self, orgao, contrato, projeto, sigad, numero):
convenio = Convenio(
casa_legislativa=orgao,
projeto=projeto,
num_processo_sf=sigad,
num_convenio=numero,
data_sigi=timezone.localdate(),
data_sigad=contrato["assinatura"],
observacao=contrato["objeto"],
data_retorno_assinatura=contrato["inicioVigencia"],
data_termino_vigencia=contrato["terminoVigencia"],
data_pub_diario=contrato["publicacao"],
atualizacao_gescon=timezone.localtime(),
observacao_gescon=_("Importado integralmente do Gescon"),
id_contrato_gescon=(contrato["codTextoContrato"] or ""),
id_gescon=contrato["id"],
)
convenio.save()
def update_orgao(self, orgao, cnpj_masked):
# Corrigir o CNPJ do órgão se estiver diferente do
# Gescon. O gescon é um pouquinho mais confiável,
# por enquanto.
if cnpj_masked and orgao.cnpj != cnpj_masked:
orgao.cnpj = cnpj_masked
orgao.save()

19
sigi/apps/convenios/migrations/0041_delete_gescon.py

@ -0,0 +1,19 @@
# Generated by Django 6.0.5 on 2026-09-22 12:23
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
(
"convenios",
"0040_alter_convenio_options_convenio_id_gescon_and_more",
),
]
operations = [
migrations.DeleteModel(
name="Gescon",
),
]

599
sigi/apps/convenios/models.py

@ -565,602 +565,3 @@ class Tramitacao(models.Model):
if self.observacao:
result = f"{result} ({self.observacao})"
return result
class Gescon(models.Model):
_report_user = False
_dominio = ""
_novos = 0
_erros = 0
_alertas = 0
_atualizados = 0
_todos_orgaos = None
url_gescon = models.URLField(
_("Webservice Gescon"),
default=(
"https://adm.senado.gov.br/gestao-contratos/api/contratos"
"/busca?especie={s}"
),
help_text=_(
"Informe o ponto de consulta do webservice do Gescon, "
"inclusive com a querystring. No ponto onde deve ser "
"inserida a sigla da subespecie do contrato, use a "
"marcação {s}.<br/><strong>Por exemplo:</strong> "
"https://adm.senado.gov.br/gestao-contratos/api/contratos"
"/busca?especie=<strong>{s}</strong>"
),
)
subespecies = models.TextField(
_("Subespécies"),
default="AC=ACT\nPI=PI\nCN=PML\nTA=PML",
help_text=_(
"Informe as siglas das subespécies de contratos que "
"devem ser pesquisados no Gescon com a sigla "
"correspondente do projeto no SIGI. Coloque um par de "
"siglas por linha, no formato SIGLA_GESTON=SIGLA_SIGI. "
"As siglas não encontradas serão ignoradas."
),
)
palavras = models.TextField(
_("Palavras de filtro"),
default="ILB\nINTERLEGIS",
help_text=_(
"Palavras que devem aparecer no campo OBJETO dos dados do "
"Gescon para identificar se o contrato pertence ao ILB. "
"<ul><li>Informe uma palavra por linha.</li>"
"<li>Ocorrendo qualquer uma das palavras, o contrato será "
"importado.</li></ul>"
),
)
palavras_excluir = models.TextField(
_("palavras de exclusão"),
default="DTCOM",
help_text=_(
"Palavras que não podem aparecer no campo OBJETO dos dados do "
"Gescon."
"<ul><li>Informe uma palavra por linha.</li>"
"<li>Ocorrendo qualquer uma das palavras, o contrato será "
"ignorado.</li></ul>"
),
)
orgaos_gestores = models.TextField(
_("Órgãos gestores"),
default="SCCO",
help_text=_(
"Siglas de órgãos gestores que devem aparecer no campo"
"ORGAOSGESTORESTITULARES"
"<ul><li>Informe um sigla por linha.</li>"
"<li>Ocorrendo qualquer uma das siglas, o contrato será "
"importado.</li></ul>"
),
)
email = models.EmailField(
_("E-mail"),
help_text=_(
"Caixa de e-mail para onde o relatório diário de "
"importação será enviado."
),
)
ultima_importacao = models.TextField(
_("Resultado da última importação"), blank=True
)
checksums = models.JSONField(null=True)
class Meta:
verbose_name = _("Configuração do Gescon")
verbose_name_plural = _("Configurações do Gescon")
def __str__(self):
return self.url_gescon
def save(self, *args, **kwargs):
self.pk = 1 # Highlander (singleton pattern)
return super(Gescon, self).save(*args, **kwargs)
def delete(self, *args, **kwargs):
pass # Highlander is immortal
def add_message(self, msg, save=False):
self.ultima_importacao += msg + "\n\n"
if save:
self.save()
self.email_report()
def email_report(self):
if self.email:
send_mail(
subject=_("Relatório de importação GESCON"),
message=self.ultima_importacao,
recipient_list=self.email,
fail_silently=True,
)
else:
self.ultima_importacao += _(
"\n\n*Não foi definida uma caixa de e-mail nas configurações "
"do Gescon*"
)
self.save()
def importa_contratos(self):
"""Importa dados de convênios do GESCON
Returns:
boolean: Indica se há o que reportar ao usuário (erro ou dados
importados/atualizados)
"""
self._dominio = get_current_site(None).domain
self.ultima_importacao = ""
if self.checksums is None:
self.checksums = {}
self.add_message(
_(
f"Importação iniciada em {timezone.localtime():%d/%m/%Y %H:%M:%S}\n"
"==========================================================\n"
)
)
if self.palavras == "" or self.orgaos_gestores == "":
self.add_message(
_(
"Nenhuma palavra de pesquisa ou orgãos "
"gestores definidos - processo abortado."
),
True,
)
return True
if self.subespecies == "":
self.add_message(
_("Nenhuma subespécie definida - processo abortado."), True
)
return True
if "{s}" not in self.url_gescon:
self.add_message(
_(
"Falta a marcação {s} na URL para indicar o local onde "
"inserir a sigla da subespécia na consulta ao webservice "
"- processo abortado."
),
True,
)
return True
subespecies = {tuple(s.split("=")) for s in self.subespecies.split()}
self._todos_orgaos = [
(o, f"{to_ascii(o.nome)} - {o.uf_sigla}".lower())
for o in Orgao.objects.all()
.order_by()
.annotate(uf_sigla=F("municipio__uf__sigla"))
]
self._report_user = False
for sigla_gescon, sigla_sigi in subespecies:
self.add_message(
_("\n**Importando subespécie {sigla_gescon}**").format(
sigla_gescon=sigla_gescon
)
)
try:
projeto = Projeto.objects.get(sigla=sigla_sigi)
except Projeto.DoesNotExist:
self.add_message(
_(
"Projeto com a sigla {sigla_sigi} não existe no SIGI"
).format(sigla_sigi=sigla_sigi)
)
continue
except Projeto.MultipleObjectsReturned:
self.add_message(
_(
"A sigla {sigla_sigi} foi cadastrada com duplicidade "
"no SIGI"
).format(sigla_sigi=sigla_sigi)
)
continue
nossos = self._get_contratos_gescon(sigla_gescon)
if not nossos:
self.add_message(
_(
"Nenhum contrato do tipo {sigla_gescon} pertence ao "
"ILB / Interlegis"
).format(sigla_gescon=sigla_gescon)
)
continue
# Desmarca erros de importações passadas
Convenio.objects.filter(projeto=projeto).update(erro_gescon=False)
self._novos = 0
self._erros = 0
self._alertas = 0
self._atualizados = 0
for contrato in nossos:
if contrato["numero"] is None:
self.add_message(
_(
"\t* O contrato de id {id} subespécie {sub} foi "
"cadastrado no Gescon sem número e não será "
"importado."
).format(id=contrato["id"], sub=contrato["subEspecie"])
)
self._report_user = True
continue
numero = re.sub(
r"(\d{4})(\d{4})", r"\1/\2", contrato["numero"].zfill(8)
)
sigad = re.sub(
r"(\d{5})(\d{6})(\d{4})(\d{2})",
r"\1.\2/\3-\4",
contrato["processo"].zfill(17),
)
if contrato["cnpjCpfFornecedor"]:
cnpj = contrato["cnpjCpfFornecedor"].zfill(14)
cnpj_masked = mask_cnpj(cnpj)
else:
cnpj = None
cnpj_masked = None
if contrato["nomeFornecedor"]:
nome = to_ascii(
contrato["nomeFornecedor"]
.replace("VEREADORES DE", "")
.replace("DO ESTADO", "")
.split("-")[0]
.split("/")[0]
.strip()
.replace(" ", " ")
)
else:
nome = None
convenio = self._get_convenio(
contrato, projeto, sigad, numero, cnpj_masked
)
if convenio:
# Encontrou um único convênio, basta atualizar seus dados
self._update_convenio(
convenio, contrato, projeto, sigad, numero
)
self._update_orgao(convenio.casa_legislativa, cnpj_masked)
self._atualizados += 1
else:
# Não encontrou o convênio. Um novo convênio precisa ser
# criado. Primeiro, é preciso identificar qual órgão
# consta no contrato do Gescon
orgao = self._get_orgao(
contrato, cnpj, cnpj_masked, nome, numero, sigad
)
if orgao is None:
# Não encontrou o órgão... bora reportar o erro
self.add_message(
_(
"\t* Órgão não encontrado no SIGI ou mais de "
"um órgão encontrado com o mesmo CNPJ ou nome. "
"Favor regularizar o cadastro: \n"
"\t * CNPJ: {cnpj}\n"
"\t * Nome: {nome}"
).format(
cnpj=contrato["cnpjCpfFornecedor"],
nome=contrato["nomeFornecedor"],
)
)
self._erros += 1
else:
# Bora criar o convênio
self._create_convenio(
orgao, contrato, projeto, sigad, numero
)
self._update_orgao(orgao, cnpj_masked)
self._novos += 1
if self._novos or self._erros or self._alertas or self._atualizados:
self._report_user = True
self.add_message(
_(
"\n\n\t{novos} novos convenios adicionados ao SIGI, "
"{atualizados} atualizados, sendo {alertas} com alertas, e "
"{erros} reportados com erro."
).format(
novos=self._novos,
atualizados=self._atualizados,
alertas=self._alertas,
erros=self._erros,
)
)
self.save()
return self._report_user
def _get_contratos_gescon(self, sigla_gescon):
requests.packages.urllib3.disable_warnings()
url = self.url_gescon.format(s=sigla_gescon)
palavras = self.palavras.splitlines()
orgaos = self.orgaos_gestores.splitlines()
excludentes = self.palavras_excluir.splitlines()
try:
response = requests.get(url, verify=False)
except Exception as e:
self.add_message(_(f"\tErro ao acessar {url}: {e.message}"))
self._report_user = True
return None
if response.status_code != 200:
self.add_message(
_(
"\tErro na leitura dos dados de {url}: "
"[{status_code}] {reason}"
).format(
url=url,
status_code=response.status_code,
reason=response.reason,
)
)
self._report_user = True
return None
if "application/json" not in response.headers.get("Content-Type", ""):
self.add_message(
_(
"\tResultado da consulta à {url} não retornou dados "
"em formato json: {content_type}"
).format(
url=url,
content_type=response.headers.get("Content-Type", ""),
)
)
self._report_user = True
return None
md5sum = md5(response.text.encode(response.encoding)).hexdigest()
if (
sigla_gescon in self.checksums
and self.checksums[sigla_gescon] == md5sum
):
self.add_message(
_(
"\tDados da subespécie {sigla_gescon} inalterados no "
"Gescon. Processamento desnecessário."
).format(sigla_gescon=sigla_gescon)
)
return None
self.checksums[sigla_gescon] = md5sum
contratos = response.json()
# Pegar só os contratos que possuem alguma das palavras-chave
nossos = [
c
for c in contratos
if (
any(palavra in c["objeto"] for palavra in palavras)
or any(
orgao in c["orgaosGestoresTitulares"]
for orgao in orgaos
if c["orgaosGestoresTitulares"] is not None
)
)
and not any(palavra in c["objeto"] for palavra in excludentes)
]
self.add_message(
_("\t{qty} contratos encontrados no Gescon").format(qty=len(nossos))
)
return nossos
def _get_convenio(self, contrato, projeto, sigad, numero, cnpj_masked):
# Buscar o convenio pelo ID do Gescon
try:
convenio = Convenio.objects.get(id_gescon=contrato["id"])
return convenio
except Convenio.DoesNotExist:
# Não existe, mas vamos buscar com outras chaves candidatas
pass
# Buscar o Convenio pelo NUP #
convenios = Convenio.objects.filter(
projeto=projeto, num_processo_sf=sigad
)
num_convenios = convenios.count()
if num_convenios == 0:
# Nenhum convênio. Vamos continuar buscando em todos convenios
# desse tipo de projeto
convenios = Convenio.objects.filter(projeto=projeto)
elif num_convenios == 1:
# Encontrou um único, então deve ser ele
return convenios.get()
# Se chegou aqui, é porque encontrou vários ou nenhum.
# Vamos tentar diferenciar pelas outras chaves candidatas
convenios = convenios.filter(
Q(Q(num_convenio=numero) | Q(num_processo_sf=numero))
)
num_convenios = convenios.count()
if num_convenios == 0:
# Nenhum encontrado. Já finaliza como não encontrado
return None
elif num_convenios == 1:
# Achou um. Deve ser ele
return convenios.get()
# Encontrou N: Marcamos todos como erro e reportamos
urls = []
for c in convenios:
uri = reverse("admin:convenios_convenio_change", args=[c.id])
url = url_normalize(f"{self._dominio}{uri}")
urls.append(f'<a href="{url}">{c.id}</a>')
urls = ", ".join(urls)
convenios.update(
erro_gescon=True,
observacao_gescon=_(
"Este convênio possui o mesmo número dos convenios {urls}"
).format(urls=urls),
)
self.add_message(
_(
"\t* O contrato {numero} no Gescon pode "
"ser relacionado aos seguintes convênios "
"do SIGI: {urls}"
).format(numero=numero, urls=urls)
)
self._erros += 1
# Porém, talvez seja possível ser desambiguado pelo CNPJ do
# fornecedor
if cnpj_masked is not None:
convenios = convenios.filter(casa_legislativa__cnpj=cnpj_masked)
if convenios.count() == 1:
# Achou exatamente o único que deveria existir.
return convenios.get()
# Continua ambíguo. Não dá pra fazer nada.
return None
def _get_orgao(self, contrato, cnpj, cnpj_masked, nome, numero, sigad):
if (cnpj is None) and (nome is None):
self.add_message(
_(
"\t* O contrato {numero} no Gescon não informa nem o CNPJ "
"nem o nome do órgão, então não é possível importar "
"para o SIGI."
).format(numero=numero)
)
self._erros += 1
return None
# Vamos tentar primeiro com o CNPJ
if cnpj is not None:
try:
orgao = Orgao.objects.get(cnpj=cnpj_masked)
return orgao
except Orgao.MultipleObjectsReturned:
# Pode acontecer de uma câmara usar o mesmo CNPJ da prefeitura,
# e ambos terem convênio com o ILB. Podemos tentar desambiguar
# pelo nome mais semelhante.
orgaos = Orgao.get_semelhantes(
to_ascii(contrato["nomeFornecedor"]).lower(),
[
(
o,
f"{to_ascii(o.nome)} - {o.uf_sigla}".lower(),
)
for o in Orgao.objects.filter(cnpj=cnpj_masked)
.order_by()
.annotate(uf_sigla=F("municipio__uf__sigla"))
],
min_ratio=0,
)
if orgaos:
# Retorna o mais semelhante
return orgaos[0][0]
except Orgao.DoesNotExist:
# Não encontrou nenhum. Vamos seguir sem órgao e tentar
# encontrar pelo nome logo abaixo
pass
# Não achou pelo CNPJ. Bora ver se acha por similaridade do nome
if nome is None:
# Também não tem nome... então temos que reportar erro
self.add_message(
_(
"\t* O contrato {numero} no Gescon com NUP sigad {sigad}, "
"fornecedor {cnpj_masked} não pode ser imortado porque "
"não é possível identificar o órgão no SIGI. Cadastre um "
"órgão com o CNPJ desse fornecedor, que na próxima "
"importação este contrato será importado."
).format(numero=numero, sigad=sigad, cnpj_masked=cnpj_masked)
)
self._erros += 1
return None
# Tentar primeiro com o nome igual veio do GESCON
semelhantes = Orgao.get_semelhantes(
to_ascii(contrato["nomeFornecedor"]).lower(),
self._todos_orgaos,
)
if not semelhantes:
# Não achou, então vamos tentar com o nome limpado
semelhantes = Orgao.get_semelhantes(
to_ascii(nome).lower(),
self._todos_orgaos,
)
if len(semelhantes) > 0:
# Encontrou algo semelhante.... bora usar.
return semelhantes[0][0]
# Não encontrou nada parecido. Bora reportar como erro
self.add_message(
_(
"\t* O contrato {numero} no Gescon com NUP Sigad {sigad}, "
"indica o fornecedor com CNPJ {cnpj_masked} e com o nome "
"{nome}, que não tem correspondência no SIGI. Este convênio "
"precisa ser cadastrado manualmente no SIGI para este erro "
"parar de acontecer."
).format(
numero=numero,
sigad=sigad,
cnpj_masked=cnpj_masked,
nome=contrato["nomeFornecedor"],
)
)
self._erros += 1
return None
def _update_convenio(self, convenio, contrato, projeto, sigad, numero):
convenio.projeto = projeto
convenio.num_processo_sf = sigad
convenio.num_convenio = numero
convenio.data_sigad = contrato["assinatura"]
convenio.observacao = contrato["objeto"]
convenio.data_retorno_assinatura = contrato["inicioVigencia"]
convenio.data_termino_vigencia = contrato["terminoVigencia"]
convenio.data_pub_diario = contrato["publicacao"]
convenio.atualizacao_gescon = timezone.localtime()
convenio.erro_gescon = False
convenio.observacao_gescon = ""
convenio.id_contrato_gescon = contrato["codTextoContrato"] or ""
convenio.id_gescon = contrato["id"]
convenio.save()
def _create_convenio(self, orgao, contrato, projeto, sigad, numero):
convenio = Convenio(
casa_legislativa=orgao,
projeto=projeto,
num_processo_sf=sigad,
num_convenio=numero,
data_sigi=timezone.localdate(),
data_sigad=contrato["assinatura"],
observacao=contrato["objeto"],
data_retorno_assinatura=contrato["inicioVigencia"],
data_termino_vigencia=contrato["terminoVigencia"],
data_pub_diario=contrato["publicacao"],
atualizacao_gescon=timezone.localtime(),
observacao_gescon=_("Importado integralmente do Gescon"),
id_contrato_gescon=(contrato["codTextoContrato"] or ""),
id_gescon=contrato["id"],
)
convenio.save()
def _update_orgao(self, orgao, cnpj_masked):
# Corrigir o CNPJ do órgão se estiver diferente do
# Gescon. O gescon é um pouquinho mais confiável,
# por enquanto.
if cnpj_masked and orgao.cnpj != cnpj_masked:
orgao.cnpj = cnpj_masked
orgao.save()
@classmethod
def load(cls):
obj, created = cls.objects.get_or_create(pk=1)
return obj

11
sigi/apps/convenios/templates/convenios/report/erros_gescon_report_view/report_pdf.html

@ -1,11 +0,0 @@
{% extends "utils/report/report_pdf.html" %}
{% load i18n %}
{% block main_content %}
{{ block.super }}
<div class="new-page"></div>
<div style="padding: 24px;">
<h1>{% translate "Resumo da última importação de dados do Gescon" %}</h1>
{{ ultima_importacao }}
</div>
{% endblock %}

1
sigi/apps/convenios/urls.py

@ -17,5 +17,4 @@ urlpatterns = [
views.report_regiao,
name="convenios-report_regiao_pdf",
),
path("importar/", views.importar_gescon, name="importar-gescon"),
]

33
sigi/apps/convenios/views.py

@ -20,7 +20,7 @@ from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
from django_weasyprint.views import WeasyTemplateResponse
from sigi.apps.casas.models import Orgao
from sigi.apps.contatos.models import UnidadeFederativa
from sigi.apps.convenios.models import Convenio, Gescon, Projeto
from sigi.apps.convenios.models import Convenio, Projeto
from sigi.apps.utils import get_sigad_url
from sigi.apps.utils.views import ReportListView
@ -69,20 +69,6 @@ class ErrosGesconReportView(
count,
)
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
rst = Gescon.load().ultima_importacao
parts = publish_parts(
rst,
writer_name="html5",
settings_overrides={
"input_encoding": "unicode",
"output_encoding": "unicode",
},
)
context["ultima_importacao"] = mark_safe(parts["html_body"])
return context
@login_required
@staff_member_required
@ -297,23 +283,6 @@ def casas_estado_to_tabela(casas, convenios, regiao):
}
@login_required
@staff_member_required
def importar_gescon(request):
if not request.user.is_superuser:
return HttpResponseForbidden()
action = request.GET.get("action", "")
gescon = Gescon.load()
if action == "importar":
gescon.importa_contratos()
context = {"gescon": gescon}
return render(request, "convenios/importar_gescon.html", context)
def normaliza_data(get, nome_param):
if nome_param in get:
value = get.get(nome_param, "")

15
sigi/apps/eventos/appconfig.py

@ -0,0 +1,15 @@
from django import forms
from django.utils.translation import gettext_lazy as _
class AppConfigForm(forms.Form):
encerra_inscricao = forms.IntegerField(
min_value=0,
required=True,
label=_("Encerra inscrições de oficinas no Portal"),
initial=30,
help_text=_(
"Número de dias a esperar após o término da oficina antes de "
"a remover do Portal Interlegis"
),
)

7
sigi/apps/eventos/jobs/daily/encerra_inscricao.py

@ -1,7 +1,7 @@
from appconfig import config
from django_extensions.management.jobs import DailyJob
from django.utils import timezone
from django.utils.translation import gettext as _
from sigi.apps.utils.models import Config
from sigi.apps.eventos.models import Evento
INSCRICOES_ENCERRADAS = _("INSCRIÇÕES ENCERRADAS")
@ -14,10 +14,11 @@ class Job(DailyJob):
report_data = []
def execute(self):
dias_a_retroagir = int(Config.get_param("ENCERRA_INSCRICAO")[0])
self.report_data = []
hoje = timezone.localtime().replace(hour=23, minute=59, second=59)
retroagir = hoje - timezone.timedelta(days=dias_a_retroagir)
retroagir = hoje - timezone.timedelta(
days=config.eventos.encerra_inscricao
)
total_encerrar = 0
total_despublicar = 0

8
sigi/apps/utils/admin.py

@ -10,7 +10,7 @@ from django.utils.translation import gettext as _
from django_extensions.management.jobs import get_job, get_jobs
from tinymce.models import HTMLField
from tinymce.widgets import AdminTinyMCE
from sigi.apps.utils.models import SigiAlert, AlertViews, Config
from sigi.apps.utils.models import SigiAlert, AlertViews
class AlertViewsInline(admin.TabularInline):
@ -26,9 +26,3 @@ class SigiAlertAdmin(admin.ModelAdmin):
formfield_overrides = {HTMLField: {"widget": AdminTinyMCE}}
list_filter = ("destinatarios",)
inlines = [AlertViewsInline]
@admin.register(Config)
class ConfigAdmin(admin.ModelAdmin):
list_display = ["parametro", "valor"]
list_filter = ["parametro"]

10
sigi/apps/utils/admin_urls.py

@ -1,10 +0,0 @@
from django.urls import path, include
from sigi.apps.utils import views
urlpatterns = [
path(
"runjob/<str:job_name>/",
views.user_run_job,
name="utils_runjob",
),
]

5
sigi/apps/utils/management/jobs.py

@ -7,11 +7,6 @@ from django.contrib.contenttypes.models import ContentType
from django.core.mail import send_mail
from django.template.loader import render_to_string
from django_extensions.management.jobs import BaseJob
from sigi.apps.utils.models import Config
class MisconfiguredError(Exception):
pass
class QuarterDailyJob(BaseJob):

16
sigi/apps/utils/migrations/0009_delete_config.py

@ -0,0 +1,16 @@
# Generated by Django 6.0.5 on 2026-09-22 11:22
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
("utils", "0008_sigialert_repeticao_alertviews"),
]
operations = [
migrations.DeleteModel(
name="Config",
),
]

41
sigi/apps/utils/models.py

@ -71,44 +71,3 @@ class AlertViews(models.Model):
alert=str(self.alert),
visualizacoes=self.visualizacoes,
)
class Config(models.Model):
PARAMETRO_CHOICES = (
("ENCERRA_INSCRICAO", _("Encerra inscrições de oficinas no Portal")),
("EMAIL_JOBS", _("E-mail de jobs")),
)
DEFAULTS = {
"ENCERRA_INSCRICAO": "30",
"EMAIL_JOBS": "sigi@interlegis.leg.br",
}
parametro = models.CharField(
_("parâmetro"), max_length=100, choices=PARAMETRO_CHOICES
)
valor = models.CharField(_("valor do parâmettro"), max_length=200)
class Meta:
ordering = ("parametro",)
verbose_name = _("Parâmetro de configuração")
verbose_name_plural = _("Parâmetros de configuração")
def __str__(self):
return f"{self.get_parametro_display()}: {self.valor}"
@classmethod
def get_param(cls, parametro):
if parametro not in cls.DEFAULTS:
raise cls.DoesNotExist(
_(
f"Não existe o parâmetro '{parametro}'. "
f"As opções são {', '.join(cls.DEFAULTS.keys())}."
)
)
valores = list(
cls.objects.filter(parametro=parametro).values_list(
"valor", flat=True
)
)
if not valores:
valores.append(cls.DEFAULTS[parametro])
return valores

18
sigi/apps/utils/templates3/admin/jobs/job_result.html

@ -1,18 +0,0 @@
{% extends "admin/base_site.html" %}
{% load static i18n %}
{% block extrastyle %}
{{ block.super }}
<style type="text/css">
h1 {
font-size: 2.5rem;
}
h2 {
font-size: 2rem;
}
h3 {
font-size: 1.5rem;
}
</style>
{% endblock %}
{% block content %}{{ content|safe }}{% endblock %}

54
sigi/apps/utils/views.py

@ -154,57 +154,3 @@ class ReportListView(ListView):
def _is_csv(self):
return self.request.GET.get(self.format_param_name, "html") == "csv"
@login_required
@user_passes_test(lambda user: user.is_superuser)
def user_run_job(request, job_name):
job = get_job(None, job_name)()
start_time = timezone.localtime()
if hasattr(job, "do_job"):
job.do_job()
end_time = timezone.localtime()
rst, html = job.prepare_report(start_time, end_time)
else:
with io.StringIO() as so_buf, io.StringIO() as se_buf, redirect_stdout(
so_buf
), redirect_stderr(se_buf):
job.execute()
messages = so_buf.getvalue()
errors = se_buf.getvalue()
report_data = ["", "MENSAGENS", "---------", ""]
if messages:
report_data.extend(messages.splitlines())
else:
report_data.extend(["Nenhuma mensagem gerada", ""])
report_data.extend(["", "ERROS", "-----", ""])
if errors:
report_data.extend(errors.splitlines())
else:
report_data.extend(["Nenhum erro gerado", ""])
end_time = timezone.localtime()
rst = render_to_string(
"emails/base_report.rst",
{
"title": job.help,
"start_time": start_time,
"end_time": end_time,
"report_data": report_data,
},
)
html = docutils.core.publish_string(
rst,
writer_name="html5",
settings_overrides={
"input_encoding": "unicode",
"output_encoding": "unicode",
},
)
dp = Selector(text=html)
return render(
request,
"admin/jobs/job_result.html",
{"content": dp.xpath("//body/*").get()},
)

10
sigi/menu_conf.yaml

@ -6,25 +6,17 @@ admin_menu:
view_name: admin:auth_user_changelist
- title: Grupos
view_name: admin:auth_group_changelist
- title: Importar Gescon
view_name: importar-gescon
- title: Todos os senadores
view_name: admin:parlamentares_senador_changelist
- title: Avisos do sistema
view_name: admin:utils_sigialert_changelist
- title: Sincroniza eventos com Saberes
view_name: utils_runjob
view_param: sincroniza_saberes
- title: Encerra inscrições de eventos
view_name: utils_runjob
view_param: encerra_inscricao
- title: Jobs de cron
view_name: admin:job_controller_cronjob_changelist
- title: Jobs agendados
view_name: admin:job_controller_jobschedule_changelist
querystr: status__exact=A
- title: Configurações
view_name: admin:utils_config_changelist
view_name: appconfig:appconfiglist
main_menu:
- title: Relatórios
icon: pdf

1
sigi/settings.py

@ -65,6 +65,7 @@ INSTALLED_APPS = [
"tinymce",
"rest_framework",
"drf_spectacular",
"appconfig",
"sigi.apps.casas",
"sigi.apps.contatos",
"sigi.apps.convenios",

2
sigi/urls.py

@ -40,7 +40,7 @@ urlpatterns = [
path("admin/eventos/", include("sigi.apps.eventos.admin_urls")),
path("admin/convenios/", include("sigi.apps.convenios.urls")),
path("admin/ocorrencias/", include("sigi.apps.ocorrencias.admin_urls")),
path("admin/utils/", include("sigi.apps.utils.admin_urls")),
path("admin/", include("appconfig.urls")),
path("admin/", admin.site.urls),
path("dash/", dashboard.urls),
path(

Loading…
Cancel
Save