Antonio/Diarios-Oficiais-ALEMS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: várias melhorias e evoluções no projeto

Browse Source
This commit is contained in:
root
2025-05-26 14:22:19 +02:00
parent 78e994eb6a
commit 9dca0d6022
108 changed files with 2601 additions and 2131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
diarios/admin.py
View File

@ -1,13 +1,150 @@
from django.contrib import admin
from django.db import models
from django.utils.html import format_html
from django.urls import reverse
from django.utils.safestring import mark_safe
from .models import TipoDiarioOficial, DiarioOficial, PageDiarioOficial
from .forms import PageDiarioOficialInlineForm
from .models import DiarioOficial, TipoDiarioOficial
@admin.register(TipoDiarioOficial)
class TipoDiarioOficialAdmin(admin.ModelAdmin):
list_display = ("nome", "quantidade_diarios")
search_fields = ("nome",)
ordering = ("nome",)
def quantidade_diarios(self, obj):
return obj.diarios.count()
quantidade_diarios.short_description = "Nº de Diários"
class PageDiarioOficialInline(admin.TabularInline):
form = PageDiarioOficialInlineForm
model = PageDiarioOficial
extra = 0
fields = ("id", "numero", "conteudo")
can_delete = False
def numero_link(self, instance):
if instance.id:
url = reverse("admin:diarios_pagediariooficial_change", args=[instance.id])
return mark_safe(f'<a href="{url}">{instance.numero}</a>')
return instance.numero
def conteudo_resumido(self, instance):
return (
(instance.conteudo[:100] + "...")
if len(instance.conteudo) > 100
else instance.conteudo
)
conteudo_resumido.short_description = "Conteúdo (resumo)"
@admin.register(DiarioOficial)
class DiarioOficialAdmin(admin.ModelAdmin):
pass
list_display = (
"numero",
"tipo_nome",
"data_formatada_admin",
"arquivo_link",
"link_externo",
"paginas_count",
)
list_filter = ("tipo", "data")
search_fields = ("numero", "tipo__nome", "data")
date_hierarchy = "data"
ordering = ("-data", "-numero")
readonly_fields = (
"data_formatada_admin",
"arquivo_preview",
"paginas_count",
"link_externo",
)
fieldsets = (
(
"Informações Básicas",
{"fields": ("numero", "tipo", "data", "data_formatada_admin")},
),
(
"Arquivos e Links",
{"fields": ("arquivo", "arquivo_preview", "link", "link_externo")},
),
("Estatísticas", {"fields": ("paginas_count",), "classes": ("collapse",)}),
)
# inlines = (PageDiarioOficialInline,)
@admin.register(TipoDiarioOficial)
class TipoDiarioOficialAdmin(admin.ModelAdmin):
pass
def tipo_nome(self, obj):
return obj.tipo.nome if obj.tipo else "-"
tipo_nome.short_description = "Tipo"
tipo_nome.admin_order_field = "tipo__nome"
def data_formatada_admin(self, obj):
return obj.data_formatada
data_formatada_admin.short_description = "Data"
def arquivo_link(self, obj):
if obj.arquivo:
return mark_safe(
f'<a href="{obj.arquivo.url}" target="_blank">Download PDF</a>'
)
return "-"
arquivo_link.short_description = "Arquivo"
arquivo_link.allow_tags = True
def link_externo(self, obj):
if obj.link:
return mark_safe(f'<a href="{obj.link}" target="_blank">Acessar Online</a>')
return "-"
link_externo.short_description = "Link Externo"
link_externo.allow_tags = True
def arquivo_preview(self, obj):
if obj.arquivo:
return mark_safe(
f'<a href="{obj.arquivo.url}" target="_blank">Visualizar PDF</a>'
)
return "-"
arquivo_preview.short_description = "Pré-visualização"
arquivo_preview.allow_tags = True
def paginas_count(self, obj):
return obj.paginas.count()
paginas_count.short_description = "Nº de Páginas"
def get_queryset(self, request):
return super().get_queryset(request).prefetch_related("paginas")
@admin.register(PageDiarioOficial)
class PageDiarioOficialAdmin(admin.ModelAdmin):
autocomplete_fields = ("diario",)
list_display = ("id", "diario_link", "numero", "conteudo_resumido")
list_display_links = ('id', 'numero')
list_filter = ("diario__tipo", "diario__data", "layout_duas_colunas")
search_fields = ("conteudo", "diario__numero")
readonly_fields = (
"diario_link",
)
def diario_link(self, obj):
url = reverse("admin:diarios_diariooficial_change", args=[obj.diario.id])
return mark_safe(f'<a href="{url}">{obj.diario}</a>')
diario_link.short_description = "Diário Oficial"
diario_link.allow_tags = True
def conteudo_resumido(self, obj):
return (obj.conteudo[:100] + "...") if len(obj.conteudo) > 100 else obj.conteudo
conteudo_resumido.short_description = "Conteúdo"
def get_queryset(self, request):
return super().get_queryset(request).select_related("diario")

5
diarios/api.py Normal file
View File

@ -0,0 +1,5 @@
from ninja import NinjaAPI
api = NinjaAPI()
api.add_router("/diarios/", route)

4
diarios/apps.py
View File

@ -4,3 +4,7 @@ from django.apps import AppConfig
class DiariosConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "diarios"
def ready(self):
import diarios.documents

8
diarios/custom_filters.py
View File

@ -1,8 +0,0 @@
from django import template
register = template.Library()
@register.filter
def get_range(value):
return range(value)

123
diarios/documents.py
View File

@ -1,92 +1,67 @@
from django_elasticsearch_dsl import Document, fields
from django_elasticsearch_dsl.registries import registry
from .models import DiarioOficial
@registry.register_document
class DiarioOficialDocument(Document):
tipo = fields.ObjectField(properties={
'nome': fields.TextField()
})
numero = fields.TextField()
numero = fields.KeywordField()
data = fields.DateField()
link = fields.TextField()
# Campo para armazenar todas as páginas para busca
content = fields.TextField(
analyzer='pt_analyzer',
link = fields.KeywordField()
tipo = fields.ObjectField(properties={"nome": fields.KeywordField()})
# Campo para páginas
paginas = fields.NestedField(
properties={
"id": fields.IntegerField(),
"numero": fields.IntegerField(),
"conteudo": fields.TextField(
analyzer="custom_portuguese",
fields={
"keyword": fields.KeywordField(),
"search": fields.TextField(analyzer="custom_portuguese"),
},
),
}
)
# Campo para armazenar páginas individualmente
pages = fields.NestedField(properties={
'number': fields.IntegerField(),
'content': fields.TextField(
analyzer='pt_analyzer',
)
})
class Index:
name = 'diarios_oficiais'
name = "diario_oficial"
settings = {
'number_of_shards': 1,
'number_of_replicas': 0,
'analysis': {
'filter': {
'portuguese_stop': {
'type': 'stop',
'stopwords': '_portuguese_'
},
'portuguese_stemmer': {
'type': 'stemmer',
'language': 'portuguese'
},
'synonym_filter': {
'type': 'synonym',
'synonyms': [
'lei, legislação, norma',
'processo, procedimento, autos',
'contrato, acordo, convênio',
]
"number_of_shards": 1,
"number_of_replicas": 0,
"analysis": {
"analyzer": {
"custom_portuguese": {
"type": "custom",
"tokenizer": "standard",
"filter": [
"lowercase",
"asciifolding",
"portuguese_stop",
],
}
},
'analyzer': {
'pt_analyzer': {
'tokenizer': 'standard',
'filter': [
'lowercase',
'portuguese_stop',
'portuguese_stemmer',
'synonym_filter'
]
}
}
}
"filter": {
"portuguese_stop": {"type": "stop", "stopwords": "_portuguese_"},
},
},
}
class Django:
model = DiarioOficial
fields = [
'id'
]
def prepare_tipo(self, instance):
if instance.tipo:
return {
'nome': instance.tipo.nome
}
return {}
def prepare_content(self, instance):
"""Concatena todo o conteúdo de todas as páginas em um único campo para busca"""
if instance.page_content:
return " ".join([page.get('content', '') for page in instance.page_content])
return ""
def prepare_pages(self, instance):
"""Prepara o campo de páginas individuais para exibição e destaque"""
if instance.page_content:
return instance.page_content
return []
fields = ["id"]
def prepare_tipo(self, instance):
return {"nome": instance.tipo.nome if instance.tipo else "Sem Tipo"}
def prepare_link(self, instance):
return instance.link or ""
def prepare_paginas(self, instance):
# Preparar páginas ordenadas
paginas = instance.paginas.all().order_by("numero")
return [
{"id": pagina.id, "numero": pagina.numero, "conteudo": pagina.conteudo}
for pagina in paginas
]

21
diarios/forms.py Normal file
View File

@ -0,0 +1,21 @@
from django import forms
from .models import PageDiarioOficial
class PageDiarioOficialInlineForm(forms.ModelForm):
class Meta:
model = PageDiarioOficial
fields = "__all__"
def clean(self):
cleaned_data = super().clean()
if "numero" in cleaned_data and self.instance.diario:
if (
PageDiarioOficial.objects.filter(
diario=self.instance.diario, numero=cleaned_data["numero"]
)
.exclude(pk=self.instance.pk)
.exists()
):
self.add_error("numero", "Já existe uma página com este número")
return cleaned_data

105
diarios/management/commands/importar_diarios_com_ocr.py Normal file
View File

@ -0,0 +1,105 @@
import os
import time
from django.core.files import File
from django.core.management.base import BaseCommand
from diarios.models import DiarioOficial
from diarios.signals import update_document, delete_document
from django.db.models.signals import post_save, post_delete
class Command(BaseCommand):
help = "Importa arquivos PDF de diários oficiais e associa aos objetos existentes no banco."
def add_arguments(self, parser):
parser.add_argument(
"pasta",
type=str,
help="Caminho para a pasta contendo os arquivos PDF (nomes devem conter o número do diário)",
)
def handle(self, *args, **options):
post_save.disconnect(update_document, sender=DiarioOficial)
post_delete.disconnect(delete_document, sender=DiarioOficial)
pasta = options["pasta"]
if not os.path.isdir(pasta):
self.stderr.write(
self.style.ERROR(f"A pasta fornecida não existe: {pasta}")
)
return
arquivos_pdf = [
f
for f in os.listdir(pasta)
if f.lower().endswith(".pdf") and "Diário_Oficial_Eletrônico_nº_" in f
]
if not arquivos_pdf:
self.stdout.write(
self.style.WARNING("Nenhum arquivo PDF válido encontrado na pasta.")
)
return
total = len(arquivos_pdf)
erros = []
atualizados = []
start_time = time.time()
# Mapeia os números aos nomes de arquivos
numero_para_arquivo = {
f.replace("Diário_Oficial_Eletrônico_nº_", "")
.replace(".pdf", "")
.strip("_-"): f
for f in arquivos_pdf
}
# Busca todos os objetos de uma vez
diarios_existentes = DiarioOficial.objects.in_bulk(
numero_para_arquivo.keys(), field_name="numero"
)
for idx, (numero, nome_arquivo) in enumerate(
numero_para_arquivo.items(), start=1
):
try:
diario = diarios_existentes.get(numero)
if not diario:
raise DiarioOficial.DoesNotExist()
caminho_pdf = os.path.join(pasta, nome_arquivo)
with open(caminho_pdf, "rb") as f:
diario.arquivo.save(nome_arquivo, File(f), save=True)
atualizados.append(diario.pk)
elapsed = time.time() - start_time
remaining = (elapsed / idx) * (total - idx)
self.stdout.write(
f"[{idx}/{total}] Atualizado: {nome_arquivo} | Estimativa restante: {remaining:.1f}s"
)
except DiarioOficial.DoesNotExist:
msg = f"Não encontrado no banco: {nome_arquivo}"
erros.append(msg)
self.stderr.write(self.style.WARNING(msg))
except Exception as e:
msg = f"Erro ao processar {nome_arquivo}: {str(e)}"
erros.append(msg)
self.stderr.write(self.style.ERROR(msg))
self.stdout.write(
self.style.SUCCESS(f"{len(atualizados)} arquivos importados com sucesso.")
)
self.stdout.write(self.style.WARNING(f"{len(erros)} arquivos com erro."))
if erros:
caminho_log = os.path.join(pasta, "erros_importacao.txt")
with open(caminho_log, "w", encoding="utf-8") as erro_file:
for linha in erros:
erro_file.write(f"{linha}\n")
self.stdout.write(
self.style.WARNING(f"Erros registrados em: {caminho_log}")
)
post_save.connect(update_document, sender=DiarioOficial)
post_delete.connect(delete_document, sender=DiarioOficial)

24
diarios/management/commands/reindex_diarios.py
View File

@ -1,24 +0,0 @@
from django.core.management.base import BaseCommand
from django_elasticsearch_dsl.registries import registry
class Command(BaseCommand):
help = 'Reindexar todos os Diários Oficiais no Elasticsearch'
def handle(self, *args, **options):
self.stdout.write('Iniciando reindexação...')
# Recria os índices
registry.delete_indices()
registry.create_indices()
# Reindexar documentos
for index in registry.get_indices():
self.stdout.write(f'Reindexando {index}...')
documents = []
for doc in registry.get_documents():
if index == doc._index._name:
self.stdout.write(f' + {doc.__name__}')
doc().update()
self.stdout.write(self.style.SUCCESS('Reindexação concluída!'))

47
diarios/migrations/0006_remove_diariooficial_page_content_pagediariooficial.py Normal file
View File

@ -0,0 +1,47 @@
# Generated by Django 5.0.12 on 2025-03-27 12:30
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("diarios", "0005_delete_pdfdocument"),
]
operations = [
migrations.RemoveField(
model_name="diariooficial",
name="page_content",
),
migrations.CreateModel(
name="PageDiarioOficial",
fields=[
(
"id",
models.BigAutoField(
auto_created=True,
primary_key=True,
serialize=False,
verbose_name="ID",
),
),
("numero", models.PositiveIntegerField()),
("conteudo", models.TextField()),
(
"diario",
models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE,
related_name="paginas",
to="diarios.diariooficial",
),
),
],
options={
"verbose_name": "Página de Diário Oficial",
"verbose_name_plural": "Páginas de Diários Oficiais",
"unique_together": {("diario", "numero")},
},
),
]

18
diarios/migrations/0007_diariooficial_layout_duas_colunas.py Normal file
View File

@ -0,0 +1,18 @@
# Generated by Django 5.0.12 on 2025-04-28 13:35
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("diarios", "0006_remove_diariooficial_page_content_pagediariooficial"),
]
operations = [
migrations.AddField(
model_name="diariooficial",
name="layout_duas_colunas",
field=models.BooleanField(default=False),
),
]

22
diarios/migrations/0008_remove_diariooficial_layout_duas_colunas_and_more.py Normal file
View File

@ -0,0 +1,22 @@
# Generated by Django 5.0.12 on 2025-04-28 13:43
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("diarios", "0007_diariooficial_layout_duas_colunas"),
]
operations = [
migrations.RemoveField(
model_name="diariooficial",
name="layout_duas_colunas",
),
migrations.AddField(
model_name="pagediariooficial",
name="layout_duas_colunas",
field=models.BooleanField(default=False),
),
]

173
diarios/models.py
View File

@ -1,19 +1,21 @@
import json
import os
from urllib.parse import urlparse
import PyPDF2
import requests
from babel.dates import format_date
from django.core.files.base import ContentFile
from django.core.serializers.json import DjangoJSONEncoder
from django.db import models
from django.core.exceptions import ValidationError
import PyPDF2
from asgiref.sync import async_to_sync
class TipoDiarioOficial(models.Model):
"""Representa um tipo de Diário Oficial (e.g., Municipal, Estadual, Federal)."""
nome = models.CharField(max_length=100, unique=True)
def __str__(self):
"""Retorna o nome do tipo de Diário Oficial."""
return self.nome
class Meta:
@ -21,6 +23,16 @@ class TipoDiarioOficial(models.Model):
class DiarioOficial(models.Model):
"""Modelo que representa um Diário Oficial, contendo data, arquivo PDF, tipo e link.
Attributes:
data (DateField): Data de publicação do Diário Oficial.
arquivo (FileField): Arquivo PDF do Diário Oficial (opcional).
tipo (ForeignKey): Tipo do Diário Oficial (Municipal, Estadual, etc.).
numero (CharField): Número de identificação único do Diário.
link (URLField): URL para o Diário Oficial (opcional).
"""
data = models.DateField()
arquivo = models.FileField(upload_to="diarios_oficiais/", blank=True, null=True)
tipo = models.ForeignKey(
@ -32,63 +44,128 @@ class DiarioOficial(models.Model):
)
numero = models.CharField(max_length=20, unique=True)
link = models.URLField(blank=True, null=True, unique=True)
page_content = models.JSONField(encoder=DjangoJSONEncoder, blank=True, null=True)
def save(self, *args, **kwargs):
# Se houver um link, baixa o PDF e extrai o conteúdo
if self.link and not self.arquivo:
try:
# Faz o download do PDF
response = requests.get(self.link)
response.raise_for_status() # Verifica se o download foi bem-sucedido
# Define o nome do arquivo a partir do link
parsed_url = urlparse(self.link)
file_name = (
os.path.basename(parsed_url.path) or f"diario_{self.numero}.pdf"
)
# Salva o arquivo no campo `arquivo`
self.arquivo.save(file_name, ContentFile(response.content), save=False)
# Extrai o conteúdo do PDF
pdf = PyPDF2.PdfReader(self.arquivo)
pages_data = []
for i, pagina in enumerate(pdf.pages):
page_text = pagina.extract_text()
if page_text: # Ignora páginas sem conteúdo
pages_data.append(
{
"number": i + 1,
"content": page_text,
}
)
# Salva o conteúdo das páginas no campo `page_content`
self.page_content = pages_data
except requests.RequestException as e:
print(f"Erro ao baixar o PDF: {e}")
except PyPDF2.PdfReadError as e:
print(f"Erro ao ler o PDF: {e}")
except Exception as e:
print(f"Erro inesperado: {e}")
# Salva o modelo
"""Salva o Diário Oficial, baixa o PDF (se houver link) e extrai páginas."""
super().save(*args, **kwargs)
if self.link and not self.arquivo:
self._download_pdf_from_link()
if self.arquivo and not self.paginas:
self._extract_pdf_pages()
super().save(*args, **kwargs)
def clean(self):
"""Valida o modelo antes de salvar (chamado automaticamente no admin/form)."""
super().clean()
if not self.arquivo and not self.link:
raise ValidationError("Informe um arquivo ou um link para o Diário.")
def _validar_link(self):
"""Verifica se o link é um PDF válido."""
if not self.link.lower().endswith(".pdf"):
raise ValidationError("O link deve apontar para um arquivo PDF.")
def _download_pdf_from_link(self):
"""Faz download do PDF a partir do link e salva no campo `arquivo`.
Raises:
ValidationError: Se o download falhar.
"""
try:
response = requests.get(self.link)
response.raise_for_status()
parsed_url = urlparse(self.link)
file_name = os.path.basename(parsed_url.path) or f"diario_{self.numero}.pdf"
self.arquivo.save(file_name, ContentFile(response.content), save=True)
except requests.RequestException as e:
raise ValidationError(f"Não foi possível baixar o PDF: {e}")
def _extract_pdf_pages(self):
"""Extrai o texto de cada página do PDF e salva no modelo `PageDiarioOficial`.
Raises:
ValidationError: Se a extração falhar.
"""
try:
with self.arquivo.open("rb") as pdf_file:
pdf = PyPDF2.PdfReader(pdf_file)
self._process_pdf_pages(pdf)
except Exception as pdf_error:
raise ValidationError(f"Não foi possível processar o PDF: {pdf_error}")
def _process_pdf_pages(self, pdf):
"""Processa cada página do PDF e salva seu conteúdo.
Args:
pdf (PdfReader): Objeto PDF carregado.
"""
self.paginas.all().delete()
for i, pagina in enumerate(pdf.pages):
try:
page_text = pagina.extract_text()
if page_text and page_text.strip():
PageDiarioOficial.objects.create(
diario=self,
numero=i + 1,
conteudo=page_text.strip(),
)
except Exception as page_error:
PageDiarioOficial.objects.create(
diario=self,
numero=i + 1,
conteudo=f"[Erro na extração do texto: {str(page_error)}]",
)
continue
@property
def data_formatada(self):
"""Retorna a data formatada em português (e.g., '1 de Janeiro de 2023')."""
return format_date(self.data, format="long", locale="pt_BR")
@property
def is_online(self):
return True if self.link else False
"""Verifica se o Diário possui um link (online)."""
return bool(self.link)
def __str__(self):
return f"Diário {self.tipo.nome} nº {self.numero}, {self.data_formatada}"
"""Representação em string do Diário Oficial."""
tipo_nome = self.tipo.nome if self.tipo else "Sem Tipo"
return f"Diário {tipo_nome} nº {self.numero}, {self.data_formatada}"
class Meta:
constraints = [models.UniqueConstraint(fields=["numero"], name="unique_numero")]
verbose_name_plural = "Diários Oficiais"
class PageDiarioOficial(models.Model):
"""Representa uma página de um Diário Oficial com seu conteúdo textual.
Attributes:
diario (ForeignKey): Diário Oficial associado.
layout_duas_colunas (BooleanField): Indica se a página tem duas colunas.
numero (PositiveIntegerField): Número da página no Diário.
conteudo (TextField): Texto extraído da página.
"""
diario = models.ForeignKey(
DiarioOficial, on_delete=models.CASCADE, related_name="paginas"
)
layout_duas_colunas = models.BooleanField(default=False)
numero = models.PositiveIntegerField()
conteudo = models.TextField()
class Meta:
unique_together = ("diario", "numero")
verbose_name = "Página de Diário Oficial"
verbose_name_plural = "Páginas de Diários Oficiais"
def __str__(self):
"""Representação em string da página (e.g., 'Página 1 do Diário 123')."""
return f"Página {self.numero} do Diário {self.diario.numero}"

63
diarios/schemas.py Normal file
View File

@ -0,0 +1,63 @@
from ninja import Schema
from typing import List, Optional
class PaginaSchema(Schema):
"""Schema que representa uma página de um Diário Oficial.
Attributes:
numero (int): Número ordinal da página no Diário (começa em 1).
conteudo (str): Texto extraído da página.
"""
numero: int
conteudo: str
class ResultadoSchema(Schema):
"""Schema de resposta para um Diário Oficial em resultados de busca.
Attributes:
id (int): ID único do Diário no banco de dados.
numero (str): Número de identificação oficial do Diário (e.g., '123-A').
data (str): Data de publicação no formato ISO (YYYY-MM-DD).
link (str): URL para acessar o Diário Oficial online.
tipo (str): Nome do tipo de Diário (e.g., 'Municipal', 'Federal').
paginas (List[PaginaSchema]): Lista de páginas com conteúdo extraído.
score (Optional[float]): Relevância do resultado (0 a 1), se aplicável.
"""
id: int
numero: str
data: str
link: str
tipo: str
paginas: List[PaginaSchema]
score: Optional[float] = None
class BuscaDiariosResponseSchema(Schema):
"""Schema de resposta para buscas paginadas em Diários Oficiais.
Attributes:
total (int): Total de resultados disponíveis (ignorando paginação).
resultados (List[ResultadoSchema]): Lista de Diários encontrados.
pagina (int): Número da página atual (começa em 1).
por_pagina (int): Quantidade de resultados por página.
"""
total: int
resultados: List[ResultadoSchema]
pagina: int
por_pagina: int
class SugestaoResponse(Schema):
"""Schema para sugestões de correção de busca (e.g., 'Voc quis dizer...?').
Attributes:
sugestao (Optional[str]): Termo sugerido para refinar a busca.
None se nenhuma sugestão for relevante.
"""
sugestao: Optional[str]

780
diarios/search_service.py
View File

@ -1,67 +1,733 @@
from elasticsearch_dsl import Q, Search
from .documents import DiarioOficialDocument
import re
from datetime import datetime
from typing import Optional, Dict, Any, List
from elasticsearch import Elasticsearch, AsyncElasticsearch
from .schemas import ResultadoSchema, PaginaSchema
import unicodedata
import asyncio
from django.conf import settings
class DiarioOficialSearchService:
@staticmethod
def search(query, highlight=True, fuzziness=1, page=1, page_size=10, tipos=None, data_inicio=None, data_fim=None):
# Configura a busca básica
s = DiarioOficialDocument.search().source(excludes=['page_content.content'])
# Filtros
if tipos:
s = s.filter('terms', tipo_nome=tipos)
if data_inicio and data_fim:
s = s.filter('range', data={'gte': data_inicio, 'lte': data_fim})
# Query principal com fuzziness e sinônimos
main_query = Q(
'multi_match',
query=query,
fields=[
'numero^3', # Maior peso para o número
'tipo_nome^2', # Peso médio para o tipo
'page_content.content' # Peso padrão para o conteúdo
],
fuzziness=fuzziness,
analyzer='portuguese_synonyms'
async def is_fuzzy_appropriate(term: str) -> bool:
"""
Determina se a fuzziness é apropriada para o termo de busca.
Args:
term: Termo de busca a ser avaliado
Returns:
bool: True se fuzziness é apropriada, False caso contrário
"""
return not re.match(r"^\d+/\d+$", term.strip())
async def parse_date(date_str: Optional[str]) -> Optional[str]:
"""
Converte string de data para formato ISO para ElasticSearch.
Args:
date_str: String de data no formato YYYY-MM-DD
Returns:
Optional[str]: Data formatada ou None se inválida
"""
if not date_str:
return None
try:
dt = datetime.strptime(date_str, "%Y-%m-%d")
return dt.strftime("%Y-%m-%d")
except ValueError:
print(f"Alerta: Formato de data inválido recebido: {date_str}")
return None
async def buscar_diarios(
query: Optional[str] = None,
data_inicio: Optional[str] = None,
data_fim: Optional[str] = None,
tipo_diario: Optional[str] = None,
page: int = 1,
page_size: int = 10,
) -> Dict[str, Any]:
"""
Realiza busca nos diários oficiais com os parâmetros fornecidos.
Args:
query: Termo de busca
data_inicio: Data inicial no formato YYYY-MM-DD
data_fim: Data final no formato YYYY-MM-DD
tipo_diario: Tipo de diário a ser filtrado
page: Número da página de resultados
page_size: Quantidade de resultados por página
Returns:
Dict[str, Any]: Dicionário com resultados da busca
"""
try:
es = AsyncElasticsearch(
"http://elasticsearch:9200",
request_timeout=30,
basic_auth=(
settings.ELASTICSEARCH_USER,
settings.ELASTICSEARCH_PASSWORD,
),
)
s = s.query(main_query)
if not await es.ping():
raise ConnectionError("Não foi possível conectar ao Elasticsearch")
except Exception as e:
print(f"Erro ao conectar com Elasticsearch: {e}")
return {"total": 0, "resultados": [], "pagina": page, "por_pagina": page_size}
# Highlighting
if highlight:
s = s.highlight(
'page_content.content',
fragment_size=150,
number_of_fragments=3,
pre_tags=['<mark>'],
post_tags=['</mark>']
)
es_body = {
"query": {"bool": {"must": [], "filter": []}},
"size": page_size,
"from": (page - 1) * page_size,
"_source": [
"numero",
"data",
"link",
"tipo",
"paginas.numero",
"paginas.conteudo",
],
}
# Paginação
start = (page - 1) * page_size
end = start + page_size
s = s[start:end]
parsed_dt_inicio = await parse_date(data_inicio)
parsed_dt_fim = await parse_date(data_fim)
if parsed_dt_inicio or parsed_dt_fim:
date_range_filter = {}
if parsed_dt_inicio:
date_range_filter["gte"] = parsed_dt_inicio
if parsed_dt_fim:
date_range_filter["lte"] = parsed_dt_fim
es_body["query"]["bool"]["filter"].append(
{"range": {"data": date_range_filter}}
)
# Executa a busca
response = s.execute()
if tipo_diario:
es_body["query"]["bool"]["filter"].append({"term": {"tipo.nome": tipo_diario}})
# Formata os resultados
results = []
for hit in response:
result = {
'id': hit.id,
'numero': hit.numero,
'data': hit.data,
'link': hit.link,
'tipo_nome': hit.tipo_nome,
'score': hit.meta.score
if query:
aplicar_fuzziness = await is_fuzzy_appropriate(query)
text_query_bool = {
"bool": {
"should": [
{
"match_phrase": {
"paginas.conteudo.search": {
"query": query,
"slop": 4,
"boost": 5.0,
}
}
},
{
"match": {
"paginas.conteudo.search": {
"query": query,
"fuzziness": "AUTO",
"operator": "and",
"prefix_length": 3,
}
}
},
],
"minimum_should_match": "75%",
}
if highlight and hasattr(hit.meta, 'highlight'):
result['highlights'] = hit.meta.highlight['page_content.content'].to_dict()
results.append(result)
return {
'total': response.hits.total.value,
'results': results
}
nested_query = {
"nested": {
"path": "paginas",
"query": text_query_bool,
"inner_hits": {
"highlight": {
"fields": {"paginas.conteudo.search": {}},
"fragment_size": 500,
"number_of_fragments": 1,
"pre_tags": ["<mark>"],
"post_tags": ["</mark>"],
},
"_source": ["paginas.numero"],
},
}
}
es_body["query"]["bool"]["must"].append(nested_query)
else:
es_body["query"]["bool"]["must"].append({"match_all": {}})
# Adicionar min_score
if query:
es_body["min_score"] = 2.5 * len(query.split())
try:
response = await es.search(index="diario_oficial", body=es_body)
except Exception as e:
print(f"Erro ao executar busca no Elasticsearch: {e}")
return {"total": 0, "resultados": [], "pagina": page, "por_pagina": page_size}
finally:
await es.close()
hits = response.get("hits", {})
total = hits.get("total", {}).get("value", 0)
resultados_formatados = []
for hit in hits.get("hits", []):
source = hit.get("_source", {})
resultado_data = {
"id": hit.get("_id"),
"numero": source.get("numero", ""),
"data": source.get("data"),
"link": source.get("link", ""),
"tipo": source.get("tipo", {}).get("nome", "Sem Tipo"),
"score": hit.get("_score"),
"paginas": [],
}
pagina_match = None
if query and "inner_hits" in hit and "paginas" in hit["inner_hits"]:
paginas_inner_hits = hit["inner_hits"]["paginas"]["hits"]["hits"]
if paginas_inner_hits:
inner_hit = paginas_inner_hits[0]
inner_source = inner_hit.get("_source", {})
page_num = inner_source.get("numero", "N/A")
highlights = inner_hit.get("highlight", {}).get(
"paginas.conteudo.search", []
)
highlight_content = " ... ".join(highlights) if highlights else ""
if page_num != "N/A" and highlight_content:
pagina_match = PaginaSchema(
numero=page_num, conteudo=highlight_content
)
if pagina_match:
resultado_data["paginas"].append(pagina_match)
elif not query and "paginas" in source and source["paginas"]:
primeira_pagina_source = source["paginas"][0]
conteudo_orig = primeira_pagina_source.get("conteudo", "")
resultado_data["paginas"].append(
PaginaSchema(
numero=primeira_pagina_source.get("numero", 0),
conteudo=conteudo_orig,
)
)
resultados_formatados.append(ResultadoSchema(**resultado_data))
return {
"total": total,
"resultados": resultados_formatados,
"pagina": page,
"por_pagina": page_size,
}
async def remover_acentos(texto: str) -> str:
"""
Remove acentos de uma string para comparações mais neutras.
Args:
texto: Texto a ser normalizado
Returns:
str: Texto sem acentos
"""
return "".join(
c
for c in unicodedata.normalize("NFD", texto)
if unicodedata.category(c) != "Mn"
)
async def processar_query(query: str) -> str:
"""
Faz pré-processamento da query para separar termos como 'processonº404'.
Args:
query: Texto da consulta original
Returns:
str: Consulta processada
"""
return re.sub(r"([a-zA-Z]+)(nº|n°|no)(\d+)", r"\1 \2 \3", query)
async def montar_suggest_body(query_processada: str) -> dict:
"""
Monta o corpo da sugestão para a requisição ao Elasticsearch.
Args:
query_processada: Query processada para sugestão
Returns:
dict: Corpo da requisição para suggest
"""
return {
"suggest": {
"text": query_processada,
"correcao": {
"phrase": {
"field": "paginas.conteudo.search",
"size": 3,
"gram_size": 2,
"confidence": 0.8,
"max_errors": 4,
"highlight": {"pre_tag": "**", "post_tag": "**"},
"collate": {
"query": {
"source": {
"nested": {
"path": "paginas",
"query": {
"bool": {
"should": [
{
"match_phrase": {
"paginas.conteudo.search": {
"query": "{{suggestion}}",
"slop": 1,
}
}
}
]
}
},
}
}
},
"params": {"field_name": "paginas.conteudo.search"},
"prune": True,
},
}
},
},
"size": 0,
}
async def sugestao_termo(query: str) -> Optional[str]:
"""
Oferece sugestões de correção para a query, verificando se a sugestão
realmente retorna resultados antes de apresentá-la ao usuário.
Args:
query: Texto da consulta original
Returns:
Optional[str]: Sugestão para a consulta ou None
"""
es = await conectar_elasticsearch()
if not es:
return None
query_processada = await processar_query(query)
suggest_body = await montar_suggest_body(query_processada)
try:
response = await es.search(index="diario_oficial", body=suggest_body)
suggestions = response.get("suggest", {}).get("correcao", [])
for sug in suggestions:
for option in sug.get("options", []):
if option.get("collate_match", False):
sugestao = option["text"]
# Verifica se a sugestão é idêntica à query (ignorando acentos e case)
if sugestao.lower() == query.lower():
continue
if await remover_acentos(sugestao.lower()) == await remover_acentos(
query.lower()
):
continue
return sugestao
return None
except Exception as e:
print(f"Erro ao buscar sugestão: {e}")
return None
finally:
await es.close()
async def conectar_elasticsearch() -> Optional[AsyncElasticsearch]:
"""
Conecta ao Elasticsearch e retorna o cliente.
Returns:
Optional[AsyncElasticsearch]: Cliente Elasticsearch ou None se falhar
"""
try:
es = AsyncElasticsearch(
"http://elasticsearch:9200",
request_timeout=30,
basic_auth=(
settings.ELASTICSEARCH_USER,
settings.ELASTICSEARCH_PASSWORD,
),
)
if not await es.ping():
raise ConnectionError("Não foi possível conectar ao Elasticsearch")
return es
except Exception as e:
print(f"Erro ao conectar com Elasticsearch: {e}")
return None
async def construir_ordenacao(ordenar_por: str) -> List[Dict[str, Any]]:
"""
Constrói a cláusula de ordenação para a consulta.
Args:
ordenar_por: Critério de ordenação
Returns:
List[Dict[str, Any]]: Lista de ordenação para Elasticsearch
"""
if ordenar_por == "data_asc":
return [{"data": {"order": "asc"}}]
elif ordenar_por == "data_desc":
return [{"data": {"order": "desc"}}]
else: # relevancia
return ["_score"]
async def preencher_numero_do_diario_com_zeros(numero_diario: str) -> str:
"""
Preenche o número do diário com zeros à esquerda.
Args:
numero_diario: Número do diário
Returns:
str: Número formatado com zeros à esquerda
"""
return numero_diario.zfill(4)
async def construir_filtros(
data_inicio: Optional[str],
data_fim: Optional[str],
tipo_diario: Optional[str],
numero_diario: Optional[str],
) -> List[Dict[str, Any]]:
"""
Constrói os filtros de data, tipo e número do diário.
Args:
data_inicio: Data inicial no formato YYYY-MM-DD
data_fim: Data final no formato YYYY-MM-DD
tipo_diario: Tipo de diário a ser filtrado
numero_diario: Número do diário
Returns:
List[Dict[str, Any]]: Lista de filtros para Elasticsearch
"""
filtros = []
parsed_dt_inicio = await parse_date(data_inicio)
parsed_dt_fim = await parse_date(data_fim)
if parsed_dt_inicio or parsed_dt_fim:
date_range = {}
if parsed_dt_inicio:
date_range["gte"] = parsed_dt_inicio
if parsed_dt_fim:
date_range["lte"] = parsed_dt_fim
filtros.append({"range": {"data": date_range}})
if tipo_diario:
filtros.append({"term": {"tipo.nome": tipo_diario}})
if numero_diario:
# numero_diario = await preencher_numero_do_diario_com_zeros(numero_diario)
filtros.append({"wildcard": {"numero": f"*{numero_diario}*"}})
return filtros
async def construir_query_busca(
query: Optional[str], modo_busca: str
) -> Dict[str, Any]:
"""
Constrói a query de busca com base no termo e modo de busca.
Args:
query: Termo de busca
modo_busca: Modo de busca (exata ou qualquer)
Returns:
Dict[str, Any]: Query de busca para Elasticsearch
"""
if not query:
return {
"nested": {
"path": "paginas",
"query": {"match_all": {}},
"inner_hits": {
"_source": ["paginas.numero", "paginas.conteudo"],
"size": 100,
},
}
}
should_queries = []
# Busca exata (boost maior)
should_queries.append(
{
"match_phrase": {
"paginas.conteudo.search": {"query": query, "slop": 3, "boost": 5.0}
}
}
)
# Busca mais leve (separando termos), se modo_busca permitir
if modo_busca == "qualquer":
should_queries.append(
{
"match": {
"paginas.conteudo.search": {
"query": query,
"operator": "or",
"fuzziness": "AUTO",
"prefix_length": 3,
"boost": 1.0,
}
}
}
)
return {
"nested": {
"path": "paginas",
"query": {"bool": {"should": should_queries, "minimum_should_match": 1}},
"inner_hits": {
"highlight": {
"fields": {"paginas.conteudo.search": {}},
"fragment_size": 500,
"number_of_fragments": 1,
"pre_tags": ["<mark>"],
"post_tags": ["</mark>"],
},
"_source": ["paginas.numero"],
"size": 100, # Aumentar para retornar mais páginas correspondentes
},
}
}
async def construir_request_body(
query: Optional[str],
modo_busca: str,
ordenar_por: str,
data_inicio: Optional[str],
data_fim: Optional[str],
tipo_diario: Optional[str],
numero_diario: Optional[str],
page: int,
page_size: int,
) -> Dict[str, Any]:
"""
Constrói o corpo da requisição para o Elasticsearch.
Args:
query: Termo de busca
modo_busca: Modo de busca (exata ou qualquer)
ordenar_por: Critério de ordenação
data_inicio: Data inicial
data_fim: Data final
tipo_diario: Tipo de diário
numero_diario: Número do diário
page: Número da página
page_size: Tamanho da página
Returns:
Dict[str, Any]: Corpo da requisição para Elasticsearch
"""
es_body = {
"query": {"bool": {"must": [], "filter": []}},
"size": page_size,
"from": (page - 1) * page_size,
"_source": [
"numero",
"data",
"link",
"tipo",
"paginas.numero",
"paginas.conteudo",
],
}
# Adicionar ordenação
es_body["sort"] = await construir_ordenacao(ordenar_por)
# Adicionar filtros
es_body["query"]["bool"]["filter"] = await construir_filtros(
data_inicio, data_fim, tipo_diario, numero_diario
)
# Adicionar query principal
query_principal = await construir_query_busca(query, modo_busca)
es_body["query"]["bool"]["must"].append(query_principal)
return es_body
async def processar_paginas_encontradas(
hit: Dict[str, Any], query: Optional[str]
) -> List[PaginaSchema]:
"""
Processa as páginas encontradas em um hit, retornando todas as correspondências.
Args:
hit: Item de resultado do Elasticsearch
query: Termo de busca original
Returns:
List[PaginaSchema]: Lista de páginas encontradas
"""
paginas = []
source = hit.get("_source", {})
if not query and "paginas" in source:
for pagina in source["paginas"]:
paginas.append(
PaginaSchema(
numero=pagina.get("numero", 0), conteudo=pagina.get("conteudo", "")
)
)
# Se temos uma query e há inner_hits, processamos todas as páginas correspondentes
if query and "inner_hits" in hit and "paginas" in hit["inner_hits"]:
paginas_inner_hits = hit["inner_hits"]["paginas"]["hits"]["hits"]
# Processar todas as páginas encontradas, não apenas a primeira
for inner_hit in paginas_inner_hits:
inner_source = inner_hit.get("_source", {})
page_num = inner_source.get("numero", "N/A")
highlights = inner_hit.get("highlight", {}).get(
"paginas.conteudo.search", []
)
highlight_content = " ... ".join(highlights) if highlights else ""
if page_num != "N/A" and highlight_content:
paginas.append(
PaginaSchema(numero=page_num, conteudo=highlight_content)
)
# Se não há query ou não encontramos páginas nos inner_hits, usamos a primeira página do documento
if not paginas and "paginas" in source and source["paginas"]:
primeira_pagina_source = source["paginas"][0]
conteudo_orig = primeira_pagina_source.get("conteudo", "")
paginas.append(
PaginaSchema(
numero=primeira_pagina_source.get("numero", 0),
conteudo=conteudo_orig,
)
)
return paginas
async def processar_resultados(
response: Dict[str, Any], query: Optional[str], page: int, page_size: int
) -> Dict[str, Any]:
"""
Processa os resultados da busca no Elasticsearch.
Args:
response: Resposta do Elasticsearch
query: Termo de busca original
page: Número da página atual
page_size: Tamanho da página
Returns:
Dict[str, Any]: Resultados processados
"""
hits = response.get("hits", {})
total = hits.get("total", {}).get("value", 0)
resultados_formatados = []
for hit in hits.get("hits", []):
source = hit.get("_source", {})
resultado_data = {
"id": hit.get("_id"),
"numero": source.get("numero", ""),
"data": source.get("data"),
"link": source.get("link", ""),
"tipo": source.get("tipo", {}).get("nome", "Sem Tipo"),
"score": hit.get("_score"),
"paginas": [],
}
# Processar todas as páginas encontradas
resultado_data["paginas"] = await processar_paginas_encontradas(hit, query)
resultados_formatados.append(ResultadoSchema(**resultado_data))
return {
"total": total,
"resultados": resultados_formatados,
"pagina": page,
"por_pagina": page_size,
}
async def buscar_diarios_simples(
query: Optional[str] = None,
numero_diario: Optional[str] = None,
modo_busca: str = "exata", # "exata" ou "qualquer"
ordenar_por: str = "data_asc", # "relevancia" ou "data"
data_inicio: Optional[str] = None,
data_fim: Optional[str] = None,
tipo_diario: Optional[str] = None,
page: int = 1,
page_size: int = 10,
) -> Dict[str, Any]:
"""
Função principal para buscar diários oficiais.
Args:
query: Termo de busca
numero_diario: Número do diário oficial
modo_busca: Modo de busca (exata ou qualquer)
ordenar_por: Critério de ordenação
data_inicio: Data inicial
data_fim: Data final
tipo_diario: Tipo de diário
page: Número da página
page_size: Tamanho da página
Returns:
Dict[str, Any]: Dicionário com os resultados da busca
"""
# Conectar ao Elasticsearch
es = await conectar_elasticsearch()
if not es:
return {"total": 0, "resultados": [], "pagina": page, "por_pagina": page_size}
# Construir o corpo da requisição
es_body = await construir_request_body(
query,
modo_busca,
ordenar_por,
data_inicio,
data_fim,
tipo_diario,
numero_diario,
page,
page_size,
)
# Executar a busca
try:
response = await es.search(index="diario_oficial", body=es_body)
except Exception as e:
print(f"Erro ao executar busca no Elasticsearch: {e}")
return {"total": 0, "resultados": [], "pagina": page, "por_pagina": page_size}
finally:
await es.close()
# Processar e retornar os resultados
return await processar_resultados(response, query, page, page_size)

15
diarios/signals.py
View File

@ -1,15 +0,0 @@
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
@receiver(post_save, sender=DiarioOficial)
def update_document(sender, instance, **kwargs):
"""Atualizar documento no Elasticsearch quando o objeto for salvo"""
DiarioOficialDocument.update_document(instance)
@receiver(post_delete, sender=DiarioOficial)
def delete_document(sender, instance, **kwargs):
"""Deletar documento do Elasticsearch quando o objeto for deletado"""
document = DiarioOficialDocument.get(id=instance.id)
document.delete()

226
diarios/templates/diarios/diarios_search.html
View File

@ -1,226 +0,0 @@
{% extends "base.html" %}
{% block content %}
<div class="container">
<h1>Busca de Diários Oficiais</h1>
<form method="GET" action="{% url 'search_diarios' %}" class="mb-4">
<div class="row mb-3">
<div class="col-md-12">
<div class="input-group">
<input type="text" name="q" class="form-control" value="{{ query }}" placeholder="Digite sua busca...">
<button type="submit" class="btn btn-primary">Buscar</button>
</div>
</div>
</div>
<div class="row mb-3">
<div class="col-md-4">
<label for="date_start" class="form-label">Data inicial:</label>
<input type="date" id="date_start" name="date_start" class="form-control" value="{{ date_start }}">
</div>
<div class="col-md-4">
<label for="date_end" class="form-label">Data final:</label>
<input type="date" id="date_end" name="date_end" class="form-control" value="{{ date_end }}">
</div>
<div class="col-md-4">
<label class="form-label">Tipo de correspondência:</label>
<div class="form-check">
<input class="form-check-input" type="radio" name="match_type" id="match_partial" value="partial" {% if match_type == 'partial' or not match_type %}checked{% endif %}>
<label class="form-check-label" for="match_partial">
Qualquer palavra
</label>
</div>
<div class="form-check">
<input class="form-check-input" type="radio" name="match_type" id="match_exact" value="exact" {% if match_type == 'exact' %}checked{% endif %}>
<label class="form-check-label" for="match_exact">
Todas as palavras (frase exata)
</label>
</div>
</div>
</div>
</form>
{% if error %}
<div class="alert alert-danger">
Erro na pesquisa: {{ error }}
</div>
{% endif %}
{% if query %}
<div class="mb-3">
<h2>Resultados para "{{ query }}"</h2>
<p>Encontrados {{ total }} resultados</p>
{% if did_you_mean %}
<div class="alert alert-info">
Você quis dizer: <a href="?q={{ did_you_mean }}&date_start={{ date_start }}&date_end={{ date_end }}&match_type={{ match_type }}">{{ did_you_mean }}</a>?
</div>
{% endif %}
{% if search_suggestions %}
<div class="mt-3 mb-3">
<h5>Pesquisas relacionadas:</h5>
<div class="d-flex flex-wrap gap-2">
{% for suggestion in search_suggestions %}
<a href="?q={{ suggestion }}&date_start={{ date_start }}&date_end={{ date_end }}&match_type={{ match_type }}" class="badge bg-light text-dark p-2 text-decoration-none">{{ suggestion }}</a>
{% endfor %}
</div>
</div>
{% endif %}
</div>
{% if results %}
<div class="search-results">
{% for result in results %}
<div class="card mb-3">
<div class="card-header">
<h5>{{ result.tipo }} nº {{ result.numero }}</h5>
<p class="text-muted">Data: {{ result.data }}</p>
{% if result.occurrences > 0 %}
<span class="badge bg-info">{{ result.occurrences }} ocorrências encontradas</span>
{% endif %}
</div>
<div class="card-body">
{% if result.highlight %}
<div class="highlight-section mb-3">
<h6>Destaques:</h6>
<div class="highlight-content">{{ result.highlight|safe }}</div>
</div>
{% endif %}
{% if result.highlighted_pages %}
<div class="highlighted-pages">
<h6>Páginas com o termo buscado:</h6>
<div class="accordion" id="pagesAccordion{{ result.id }}">
{% for page in result.highlighted_pages %}
<div class="accordion-item">
<h2 class="accordion-header">
<button class="accordion-button collapsed" type="button" data-bs-toggle="collapse"
data-bs-target="#page{{ result.id }}_{{ page.number }}">
Página {{ page.number }}
</button>
</h2>
<div id="page{{ result.id }}_{{ page.number }}" class="accordion-collapse collapse"
data-bs-parent="#pagesAccordion{{ result.id }}">
<div class="accordion-body">
{{ page.content|safe }}
</div>
</div>
</div>
{% endfor %}
</div>
</div>
{% endif %}
<div class="mt-3">
<a href="{{ result.link }}" target="_blank" class="btn btn-sm btn-outline-primary">
Ver Diário Online
</a>
<a href="{% url 'diario_detail' result.id %}" class="btn btn-sm btn-outline-secondary">
Ver Detalhes
</a>
</div>
</div>
</div>
{% endfor %}
</div>
<!-- Paginação aprimorada -->
{% if total_pages > 1 %}
<nav aria-label="Paginação">
<ul class="pagination justify-content-center">
<!-- Botão primeira página -->
<li class="page-item {% if page == 1 %}disabled{% endif %}">
<a class="page-link" href="?q={{ query }}&date_start={{ date_start }}&date_end={{ date_end }}&match_type={{ match_type }}&page=1&size={{ size }}">
<span aria-hidden="true">&laquo;&laquo;</span>
</a>
</li>
<!-- Botão página anterior -->
<li class="page-item {% if page == 1 %}disabled{% endif %}">
<a class="page-link" href="?q={{ query }}&date_start={{ date_start }}&date_end={{ date_end }}&match_type={{ match_type }}&page={{ page|add:'-1' }}&size={{ size }}">
<span aria-hidden="true">&laquo;</span>
</a>
</li>
<!-- Mostrar apenas um conjunto de páginas ao redor da página atual -->
{% with ''|center:total_pages as range %}
{% for _ in range %}
{% with forloop.counter as i %}
{% if i >= page|add:'-2' and i <= page|add:'2' and i > 0 and i <= total_pages %}
<li class="page-item {% if i == page %}active{% endif %}">
<a class="page-link" href="?q={{ query }}&date_start={{ date_start }}&date_end={{ date_end }}&match_type={{ match_type }}&page={{ i }}&size={{ size }}">{{ i }}</a>
</li>
{% endif %}
{% endwith %}
{% endfor %}
{% endwith %}
<!-- Botão próxima página -->
<li class="page-item {% if page >= total_pages %}disabled{% endif %}">
<a class="page-link" href="?q={{ query }}&date_start={{ date_start }}&date_end={{ date_end }}&match_type={{ match_type }}&page={{ page|add:'1' }}&size={{ size }}">
<span aria-hidden="true">&raquo;</span>
</a>
</li>
<!-- Botão última página -->
<li class="page-item {% if page >= total_pages %}disabled{% endif %}">
<a class="page-link" href="?q={{ query }}&date_start={{ date_start }}&date_end={{ date_end }}&match_type={{ match_type }}&page={{ total_pages }}&size={{ size }}">
<span aria-hidden="true">&raquo;&raquo;</span>
</a>
</li>
</ul>
</nav>
{% endif %}
{% else %}
<div class="alert alert-info">
Nenhum resultado encontrado para a sua busca.
{% if date_start or date_end %}
<p class="mt-2">Tente expandir o período de busca ou remover os filtros de data.</p>
{% endif %}
{% if match_type == 'exact' %}
<p class="mt-2">Tente usar a opção "Qualquer palavra" para resultados mais abrangentes.</p>
{% endif %}
</div>
{% endif %}
{% endif %}
</div>
<style>
.highlight-content em {
background-color: #ffeeba;
font-style: normal;
padding: 2px;
border-radius: 2px;
}
.accordion-body em {
background-color: #ffeeba;
font-style: normal;
padding: 2px;
border-radius: 2px;
}
.badge {
font-size: 0.85rem;
}
</style>
<script>
// Validar datas ao enviar o formulário
document.querySelector('form').addEventListener('submit', function(e) {
const dateStart = document.getElementById('date_start').value;
const dateEnd = document.getElementById('date_end').value;
if (dateStart && dateEnd && dateStart > dateEnd) {
e.preventDefault();
alert('A data inicial não pode ser posterior à data final.');
}
});
</script>
{% endblock %}

173
diarios/templates/diarios/search.html
View File

@ -1,173 +0,0 @@
{% extends "base.html" %}
{% load static %}
{% block title %}Busca de Diários Oficiais{% endblock %}
{% block content %}
<div class="container mt-4">
<h1 class="mb-4">Busca de Diários Oficiais</h1>
<div class="card mb-4">
<div class="card-body">
<form method="get" action="{% url 'diario-search' %}">
<div class="row g-3">
<div class="col-md-9">
<label for="q" class="form-label">Buscar por:</label>
<input type="text" id="q" name="q" value="{{ query }}"
class="form-control"
placeholder="Digite palavras-chave, frases ou utilize operadores AND, OR, NOT">
</div>
<div class="col-md-3 d-flex align-items-end">
<button type="submit" class="btn btn-primary w-100">
<i class="bi bi-search"></i> Buscar
</button>
</div>
</div>
<div class="mt-3">
<a class="btn btn-link p-0" data-bs-toggle="collapse" href="#advancedOptions" role="button">
Opções avançadas
</a>
</div>
<div class="collapse" id="advancedOptions">
<div class="row g-3 mt-2">
<div class="col-md-4">
<label class="form-label">Tipos de Diário:</label>
<div class="border rounded p-2" style="max-height: 200px; overflow-y: auto;">
{% for tipo in tipos_disponiveis %}
<div class="form-check">
<input class="form-check-input" type="checkbox"
id="tipo_{{ tipo.id }}" name="tipos" value="{{ tipo.id }}"
{% if tipos_selecionados and tipo.id|stringformat:"i" in tipos_selecionados %}checked{% endif %}>
<label class="form-check-label" for="tipo_{{ tipo.id }}">
{{ tipo.nome }}
</label>
</div>
{% endfor %}
</div>
</div>
<div class="col-md-4">
<label for="data_inicio" class="form-label">Data Inicial:</label>
<input type="date" id="data_inicio" name="data_inicio"
value="{{ data_inicio }}" class="form-control">
</div>
<div class="col-md-4">
<label for="data_fim" class="form-label">Data Final:</label>
<input type="date" id="data_fim" name="data_fim"
value="{{ data_fim }}" class="form-control">
</div>
<div class="col-md-6">
<label for="fuzziness" class="form-label">Tolerância a erros:</label>
<select id="fuzziness" name="fuzziness" class="form-select">
<option value="0" {% if fuzziness == 0 %}selected{% endif %}>Sem tolerância</option>
<option value="1" {% if fuzziness == 1 %}selected{% endif %}>Baixa tolerância</option>
<option value="2" {% if fuzziness == 2 %}selected{% endif %}>Alta tolerância</option>
</select>
</div>
<div class="col-md-6 d-flex align-items-end">
<div class="form-check">
<input class="form-check-input" type="checkbox" id="highlight"
name="highlight" value="true" {% if highlight %}checked{% endif %}>
<label class="form-check-label" for="highlight">
Destacar termos encontrados
</label>
</div>
</div>
</div>
</div>
</form>
</div>
</div>
{% if query %}
<div class="mt-4">
<div class="d-flex justify-content-between align-items-center mb-4">
<h2>Resultados da busca</h2>
<span class="badge bg-primary">{{ total }} resultado(s)</span>
</div>
{% if results %}
{% for result in results %}
<div class="card mb-3">
<div class="card-body">
<h5 class="card-title">
<a href="{% url 'diario-detail' result.id %}?q={{ query|urlencode }}"
class="text-decoration-none">
{{ result.tipo_nome }} nº {{ result.numero }} - {{ result.data|date:"d/m/Y" }}
</a>
</h5>
{% if result.highlights %}
<div class="card-text mt-2">
{% for highlight in result.highlights %}
<p class="mb-1">...{{ highlight|safe }}...</p>
{% endfor %}
</div>
{% endif %}
<div class="mt-3 text-muted small">
<span class="me-3">
<i class="bi bi-star-fill text-warning"></i> Relevância: {{ result.score|floatformat:2 }}
</span>
{% if result.link %}
<a href="{{ result.link }}" target="_blank" class="text-decoration-none">
<i class="bi bi-box-arrow-up-right"></i> Ver original
</a>
{% endif %}
</div>
</div>
</div>
{% endfor %}
{% if pages > 1 %}
<nav aria-label="Page navigation">
<ul class="pagination justify-content-center">
{% if page > 1 %}
<li class="page-item">
<a class="page-link"
href="?q={{ query }}&page={{ page|add:'-1' }}&highlight={{ highlight|lower }}&fuzziness={{ fuzziness }}{% for tipo in tipos_selecionados %}&tipos={{ tipo }}{% endfor %}{% if data_inicio %}&data_inicio={{ data_inicio }}{% endif %}{% if data_fim %}&data_fim={{ data_fim }}{% endif %}">
Anterior
</a>
</li>
{% endif %}
{% for i in page_range %}
<li class="page-item {% if i == page %}active{% endif %}">
<a class="page-link"
href="?q={{ query }}&page={{ i }}&highlight={{ highlight|lower }}&fuzziness={{ fuzziness }}{% for tipo in tipos_selecionados %}&tipos={{ tipo }}{% endfor %}{% if data_inicio %}&data_inicio={{ data_inicio }}{% endif %}{% if data_fim %}&data_fim={{ data_fim }}{% endif %}">
{{ i }}
</a>
</li>
{% endfor %}
{% if page < pages %}
<li class="page-item">
<a class="page-link"
href="?q={{ query }}&page={{ page|add:'1' }}&highlight={{ highlight|lower }}&fuzziness={{ fuzziness }}{% for tipo in tipos_selecionados %}&tipos={{ tipo }}{% endfor %}{% if data_inicio %}&data_inicio={{ data_inicio }}{% endif %}{% if data_fim %}&data_fim={{ data_fim }}{% endif %}">
Próxima
</a>
</li>
{% endif %}
</ul>
</nav>
{% endif %}
{% else %}
<div class="alert alert-warning text-center">
<h4 class="alert-heading">Nenhum resultado encontrado</h4>
<p>Não encontramos resultados para "{{ query }}". Tente ajustar seus termos de busca.</p>
</div>
{% endif %}
</div>
{% else %}
<div class="text-center py-5 bg-light rounded">
<p class="lead text-muted">Digite um termo de busca para encontrar diários oficiais</p>
</div>
{% endif %}
</div>
{% endblock %}

3
diarios/tests.py
View File

@ -1,3 +0,0 @@
from django.test import TestCase
# Create your tests here.

0
diarios/tests/__init__.py Normal file
View File

31
diarios/tests/factories.py Normal file
View File

@ -0,0 +1,31 @@
from factory.django import DjangoModelFactory
from factory import Faker, SubFactory, LazyAttribute, Sequence
import datetime
from diarios.models import PageDiarioOficial, DiarioOficial, TipoDiarioOficial
class TipoDiarioOficialFactory(DjangoModelFactory):
nome = Faker("word")
class Meta:
model = TipoDiarioOficial
django_get_or_create = ["nome"]
class DiarioOficialFactory(DjangoModelFactory):
data = Faker("date_this_decade")
numero = Sequence(lambda n: f"{n:04d}-DO") # ex: 0001-DO, 0002-DO
tipo = SubFactory(TipoDiarioOficialFactory)
link = None
class Meta:
model = DiarioOficial
class PageDiarioOficialFactory(DjangoModelFactory):
diario = SubFactory(DiarioOficialFactory)
numero = Sequence(lambda n: n + 1)
layout_duas_colunas = False
conteudo = Faker("text", max_nb_chars=3000)
class Meta:
model = PageDiarioOficial
django_get_or_create = ["diario", "numero"]

34
diarios/tests/test_models.py Normal file
View File

@ -0,0 +1,34 @@
from django.test import TestCase
from diarios.models import TipoDiarioOficial, DiarioOficial, PageDiarioOficial
from .factories import (
TipoDiarioOficialFactory,
DiarioOficialFactory,
PageDiarioOficialFactory,
)
class TipoDiarioOficialFactoryTest(TestCase):
def test_create_tipo_diario_oficial(self):
tipo = TipoDiarioOficialFactory()
self.assertIsInstance(tipo, TipoDiarioOficial)
self.assertIsNotNone(tipo.pk)
self.assertTrue(tipo.nome)
class DiarioOficialFactoryTest(TestCase):
def test_create_diario_oficial(self):
diario = DiarioOficialFactory()
self.assertIsInstance(diario, DiarioOficial)
self.assertIsNotNone(diario.pk)
self.assertIsNotNone(diario.numero)
self.assertIsInstance(diario.tipo, TipoDiarioOficial)
class PageDiarioOficialFactoryTest(TestCase):
def test_create_page_diario_oficial(self):
page = PageDiarioOficialFactory()
self.assertIsInstance(page, PageDiarioOficial)
self.assertIsNotNone(page.pk)
self.assertIsInstance(page.diario, DiarioOficial)
self.assertIsInstance(page.conteudo, str)
self.assertGreater(len(page.conteudo), 0)

4
diarios/urls.py
View File

@ -2,7 +2,5 @@ from django.urls import path
from . import views
urlpatterns = [
path('diario/<int:pk>/', views.diario_detail, name='diario_detail'),
path('diarios/search/', views.search_diarios, name='search_diarios'),
path('busca/', views.index, name='index')
]

281
diarios/views.py
View File

@ -1,209 +1,88 @@
from ninja import Router
from typing import Optional
from django.http import HttpRequest
from .search_service import (
buscar_diarios,
sugestao_termo,
buscar_diarios_simples,
)
from .schemas import BuscaDiariosResponseSchema, SugestaoResponse
from django.shortcuts import render
from elasticsearch_dsl import Q
from datetime import datetime
from .documents import DiarioOficialDocument
from elasticsearch.exceptions import RequestError
def search_diarios(request):
q = request.GET.get('q', '')
page = int(request.GET.get('page', 1))
size = int(request.GET.get('size', 10))
# Parâmetros de filtro de data
date_start = request.GET.get('date_start', '')
date_end = request.GET.get('date_end', '')
# Tipo de correspondência (exata ou parcial)
match_type = request.GET.get('match_type', 'partial') # 'exact' ou 'partial'
start = (page - 1) * size
end = start + size
router = Router(tags=["Diários Oficiais"])
results = []
total = 0
did_you_mean = None
search_suggestions = []
try:
if q:
# Construir a consulta base
search = DiarioOficialDocument.search()
# Determinar o tipo de consulta com base no match_type
if match_type == 'exact':
# Correspondência exata (frase exata)
query = Q(
'multi_match',
query=q,
fields=['content^3', 'tipo.nome^2', 'numero', 'pages.content'],
type='phrase'
)
else:
# Correspondência parcial (qualquer termo)
query = Q(
'multi_match',
query=q,
fields=['content^3', 'tipo.nome^2', 'numero', 'pages.content'],
fuzziness='AUTO',
operator='or' # Pelo menos um termo deve corresponder
)
# Aplicar a consulta principal
search = search.query(query)
# Aplicar filtros de data se fornecidos
date_filters = []
if date_start:
try:
date_start_obj = datetime.strptime(date_start, '%Y-%m-%d')
date_filters.append(Q('range', data={'gte': date_start_obj}))
except ValueError:
pass # Ignorar datas inválidas
if date_end:
try:
date_end_obj = datetime.strptime(date_end, '%Y-%m-%d')
date_filters.append(Q('range', data={'lte': date_end_obj}))
except ValueError:
pass # Ignorar datas inválidas
if date_filters:
for date_filter in date_filters:
search = search.filter(date_filter)
# Configuração do highlighting
search = search.highlight('content', fragment_size=150, number_of_fragments=3)
search = search.highlight('pages.content', fragment_size=150, number_of_fragments=3)
# Paginação
total_search = search.count()
search = search[start:end]
# Executar a pesquisa
response = search.execute()
total = response.hits.total.value
# "Você quis dizer" - sugestão para termos com erros de digitação
if total < 3 and q: # Se poucos resultados, sugira correções
suggestion_search = DiarioOficialDocument.search()
suggestion_search = suggestion_search.suggest(
'phrase_suggestion',
q,
phrase={
'field': 'content',
'size': 5,
'highlight': {
'pre_tag': '<em>',
'post_tag': '</em>'
}
}
)
suggestion_result = suggestion_search.execute()
# Processe as sugestões
if hasattr(suggestion_result, 'suggest') and 'phrase_suggestion' in suggestion_result.suggest:
suggestions = suggestion_result.suggest['phrase_suggestion'][0]['options']
if suggestions:
for suggestion in suggestions:
if suggestion['text'].lower() != q.lower():
did_you_mean = suggestion['text']
break
# Gerar sugestões de pesquisa relacionadas
if q:
# Use a expansão de termos para sugerir pesquisas relacionadas
related_search = DiarioOficialDocument.search()
related_search = related_search.query(
'more_like_this',
fields=['content'],
like=q,
min_term_freq=1,
max_query_terms=12
)
related_search = related_search[:5] # Limite para 5 sugestões
try:
related_results = related_search.execute()
# Extraia termos relevantes dos resultados relacionados
for hit in related_results:
if hasattr(hit, 'content') and hit.content:
# Extraia alguns termos significativos do conteúdo
content_terms = hit.content.split()[:10] # Primeiros 10 termos
suggestion = ' '.join(content_terms)
if suggestion not in search_suggestions and suggestion != q:
search_suggestions.append(suggestion)
if len(search_suggestions) >= 5: # Limite para 5 sugestões
break
except:
# Ignore erros de sugestões relacionadas
pass
# Processar resultados
for hit in response:
# Adicionar destaque
highlight = ""
if hasattr(hit.meta, 'highlight'):
if 'content' in hit.meta.highlight:
highlight = "...".join(hit.meta.highlight.content)
# Processar páginas com destaque
highlighted_pages = []
total_occurrences = 0
if hasattr(hit.meta, 'highlight') and 'pages.content' in hit.meta.highlight:
# Calcular o número total de ocorrências
for content in hit.meta.highlight['pages.content']:
# Contar o número de <em> tags, que representam termos destacados
total_occurrences += content.count('<em>')
# Processar os destaques por página
for i, content in enumerate(hit.meta.highlight['pages.content']):
# Encontre a página correspondente
page_number = i + 1 # Lógica simplificada, pode precisar de ajuste
highlighted_pages.append({
'number': page_number,
'content': content
})
# Combine dados do documento com os destaques
result = {
'id': hit.id,
'tipo': hit.tipo.nome if hasattr(hit, 'tipo') and hit.tipo else '',
'numero': hit.numero,
'data': hit.data,
'link': hit.link,
'highlight': highlight,
'highlighted_pages': highlighted_pages,
'occurrences': total_occurrences
}
results.append(result)
except RequestError as e:
# Tratar erros de consulta do Elasticsearch
error_message = str(e)
return render(request, 'diarios/diarios_search.html', {
'error': error_message,
'query': q
})
async def index(request):
return render(request, 'diarios/busca.html')
context = {
'query': q,
'date_start': date_start,
'date_end': date_end,
'match_type': match_type,
'results': results,
'total': total,
'page': page,
'size': size,
'total_pages': (total + size - 1) // size if total > 0 else 0,
'did_you_mean': did_you_mean,
'search_suggestions': search_suggestions[:5] # Limite para 5 sugestões
}
@router.get(
"/sugestao",
response=SugestaoResponse,
summary="Sugestão de correção para termo de busca",
)
async def sugestao_busca(request: HttpRequest, q: str) -> SugestaoResponse:
"""
Sugere correção para o termo buscado, se necessário.
return render(request, 'diarios/diarios_search.html', context)
Args:
request (HttpRequest): Requisição HTTP.
q (str): Termo original digitado pelo usuário.
def diario_detail(request, pk):
diario = get_object_or_404(Diario, pk=pk)
return render(request, 'diarios/diario_detail.html', {'diario': diario})
Returns:
SugestaoResponse: Termo corrigido.
"""
sugestao = await sugestao_termo(q)
return {"sugestao": sugestao}
@router.get(
"/busca",
response=BuscaDiariosResponseSchema,
summary="Busca simplificada com modos e ordenação",
)
async def busca_diarios_oficiais_simples(
request: HttpRequest,
q: Optional[str] = None,
numero_diario: Optional[str] = None,
data_inicio: Optional[str] = None,
data_fim: Optional[str] = None,
tipo: Optional[str] = None,
ordenar_por: str = "relevancia", # "relevancia", "data_asc", "data_desc"
modo_busca: str = "exata", # "exata" ou "qualquer"
page: int = 1,
page_size: int = 10,
) -> BuscaDiariosResponseSchema:
"""
Busca com modo de correspondência, ordenação e número do diário.
Args:
request (HttpRequest): Requisição HTTP.
q (Optional[str]): Termo de busca.
numero_diario (Optional[str]): Número exato do diário (ex: 1234/2024).
data_inicio (Optional[str]): Data inicial (YYYY-MM-DD).
data_fim (Optional[str]): Data final (YYYY-MM-DD).
tipo (Optional[str]): Tipo exato do diário.
ordenar_por (str): "relevancia", "data_asc" ou "data_desc".
modo_busca (str): "exata" ou "qualquer".
page (int): Página atual (mínimo: 1).
page_size (int): Itens por página (mínimo: 1, máximo: 50).
Returns:
BuscaDiariosResponseSchema: Resultado paginado da busca.
"""
page_size = min(max(page_size, 1), 50)
page = max(page, 1)
resultado = await buscar_diarios_simples(
query=q,
numero_diario=numero_diario,
data_inicio=data_inicio,
data_fim=data_fim,
tipo_diario=tipo,
ordenar_por=ordenar_por,
modo_busca=modo_busca,
page=page,
page_size=page_size,
)
return resultado