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

alterações gerais

Browse Source
This commit is contained in:
Antonio Roberto
2025-06-16 13:11:57 -04:00
parent 9dca0d6022
commit f196773705
45 changed files with 6774 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
diarios/admin.py
View File

@ -128,7 +128,7 @@ class PageDiarioOficialAdmin(admin.ModelAdmin):
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")
search_fields = ("diario__numero",)
readonly_fields = (
"diario_link",
)

179
diarios/crud_service.py Normal file
View File

@ -0,0 +1,179 @@
from typing import Optional, List
from datetime import date
from diarios.models import DiarioOficial, TipoDiarioOficial
from django.core.exceptions import ObjectDoesNotExist
from ninja.errors import HttpError
class DiarioOficialService:
@staticmethod
def criar_diario(
data: date,
numero: str,
tipo_id: Optional[int] = None,
arquivo=None,
link: Optional[str] = None
) -> DiarioOficial:
"""
Cria um novo diário oficial
Args:
data: Data do diário
numero: Número do diário
tipo_id: ID do tipo de diário
arquivo: Arquivo PDF (opcional)
link: URL do PDF (opcional)
Returns:
DiarioOficial: O diário criado
Raises:
HttpError: Se ocorrer algum erro na criação
"""
try:
tipo = TipoDiarioOficial.objects.get(pk=tipo_id) if tipo_id else None
diario = DiarioOficial(
data=data,
numero=numero,
tipo=tipo,
link=link
)
if arquivo:
diario.arquivo = arquivo
diario.full_clean()
diario.save()
return diario
except ObjectDoesNotExist:
raise HttpError(404, "Tipo de diário não encontrado")
except Exception as e:
raise HttpError(400, f"Erro ao criar diário: {str(e)}")
@staticmethod
def obter_diario_por_id(id: int) -> DiarioOficial:
"""
Obtém um diário pelo ID
Args:
id: ID do diário
Returns:
DiarioOficial: O diário encontrado
Raises:
HttpError: Se o diário não for encontrado
"""
try:
return DiarioOficial.objects.get(pk=id)
except ObjectDoesNotExist:
raise HttpError(404, "Diário não encontrado")
@staticmethod
def listar_diarios(
tipo_id: Optional[int] = None,
data_inicio: Optional[date] = None,
data_fim: Optional[date] = None,
numero: Optional[str] = None
) -> List[DiarioOficial]:
"""
Lista diários com filtros opcionais
Args:
tipo_id: ID do tipo de diário para filtrar
data_inicio: Data inicial para filtrar
data_fim: Data final para filtrar
numero: Número do diário para filtrar
Returns:
List[DiarioOficial]: Lista de diários filtrados
"""
queryset = DiarioOficial.objects.all().order_by('-data')
if tipo_id:
queryset = queryset.filter(tipo_id=tipo_id)
if data_inicio and data_fim:
queryset = queryset.filter(data__range=[data_inicio, data_fim])
elif data_inicio:
queryset = queryset.filter(data__gte=data_inicio)
elif data_fim:
queryset = queryset.filter(data__lte=data_fim)
if numero:
queryset = queryset.filter(numero__icontains=numero)
return list(queryset)
@staticmethod
def atualizar_diario(
id: int,
data: Optional[date] = None,
numero: Optional[str] = None,
tipo_id: Optional[int] = None,
arquivo=None,
link: Optional[str] = None
) -> DiarioOficial:
"""
Atualiza um diário existente
Args:
id: ID do diário a ser atualizado
data: Nova data (opcional)
numero: Novo número (opcional)
tipo_id: Novo tipo (opcional)
arquivo: Novo arquivo (opcional)
link: Novo link (opcional)
Returns:
DiarioOficial: O diário atualizado
Raises:
HttpError: Se ocorrer algum erro na atualização
"""
try:
diario = DiarioOficial.objects.get(pk=id)
if data is not None:
diario.data = data
if numero is not None:
diario.numero = numero
if tipo_id is not None:
tipo = TipoDiarioOficial.objects.get(pk=tipo_id) if tipo_id else None
diario.tipo = tipo
if arquivo is not None:
diario.arquivo = arquivo
if link is not None:
diario.link = link
diario.full_clean()
diario.save()
return diario
except ObjectDoesNotExist:
raise HttpError(404, "Diário ou tipo não encontrado")
except Exception as e:
raise HttpError(400, f"Erro ao atualizar diário: {str(e)}")
@staticmethod
def deletar_diario(id: int) -> None:
"""
Remove um diário
Args:
id: ID do diário a ser removido
Raises:
HttpError: Se o diário não for encontrado
"""
try:
diario = DiarioOficial.objects.get(pk=id)
diario.delete()
except ObjectDoesNotExist:
raise HttpError(404, "Diário não encontrado")

122
diarios/models.py
View File

@ -3,19 +3,16 @@ from urllib.parse import urlparse
import requests
from babel.dates import format_date
from django.core.files.base import ContentFile
from django.db import models
from django.db import models, transaction
from django.core.exceptions import ValidationError
import PyPDF2
import fitz # PyMuPDF
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:
@ -23,16 +20,6 @@ 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(
@ -46,34 +33,30 @@ class DiarioOficial(models.Model):
link = models.URLField(blank=True, null=True, unique=True)
def save(self, *args, **kwargs):
"""Salva o Diário Oficial, baixa o PDF (se houver link) e extrai páginas."""
updated = False
super().save(*args, **kwargs)
if self.link and not self.arquivo:
self._download_pdf_from_link()
updated = True
if self.arquivo and not self.paginas:
if self.arquivo and not self.paginas.exists():
self._extract_pdf_pages()
updated = True
super().save(*args, **kwargs)
if updated:
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()
@ -86,55 +69,73 @@ class DiarioOficial(models.Model):
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:
# Salvar temporariamente o PDF para abrir com o PyMuPDF
with self.arquivo.open("rb") as pdf_file:
pdf = PyPDF2.PdfReader(pdf_file)
self._process_pdf_pages(pdf)
temp_pdf_path = f"/tmp/diario_{self.id}.pdf"
with open(temp_pdf_path, "wb") as temp_file:
temp_file.write(pdf_file.read())
# Abrir e processar com fitz
doc = fitz.open(temp_pdf_path)
self._process_pdf_pages(doc)
doc.close()
# Remover arquivo temporário
os.remove(temp_pdf_path)
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.
def _process_pdf_pages(self, doc):
with transaction.atomic():
self.paginas.all().delete()
Args:
pdf (PdfReader): Objeto PDF carregado.
"""
self.paginas.all().delete()
for i, page in enumerate(doc):
try:
blocks = page.get_text("blocks")
# Ordenar os blocos por coordenadas (y, x) para manter a ordem de leitura
blocks.sort(key=lambda b: (b[1], b[0]))
page_text = ""
for block in blocks:
text = block[4].strip()
if text:
page_text += text + "\n"
for i, pagina in enumerate(pdf.pages):
try:
page_text = pagina.extract_text()
if page_text and page_text.strip():
# Crucial: Remove NULL bytes from the extracted text
# PostgreSQL text fields cannot contain NUL (0x00) bytes
cleaned_text = page_text.strip().replace('\x00', '')
if cleaned_text:
PageDiarioOficial.objects.create(
diario=self,
numero=i + 1,
conteudo=cleaned_text,
)
else:
PageDiarioOficial.objects.create(
diario=self,
numero=i + 1,
conteudo="[Conteúdo não extraído ou vazio]",
)
except Exception as page_error:
PageDiarioOficial.objects.create(
diario=self,
numero=i + 1,
conteudo=page_text.strip(),
conteudo=f"[Erro na extração do texto: {str(page_error)}]",
)
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
print(f"Erro ao processar a página {i+1} no Diario ID {self.id}: {page_error}")
@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):
"""Verifica se o Diário possui um link (online)."""
return bool(self.link)
def __str__(self):
"""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}"
@ -144,15 +145,6 @@ class DiarioOficial(models.Model):
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"
)
@ -165,7 +157,11 @@ class PageDiarioOficial(models.Model):
verbose_name = "Página de Diário Oficial"
verbose_name_plural = "Páginas de Diários Oficiais"
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
if self.diario:
self.diario.save()
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}"

65
diarios/schemas.py
View File

@ -1,5 +1,7 @@
from ninja import Schema
from ninja import Schema, ModelSchema, UploadedFile
from typing import List, Optional
from .models import TipoDiarioOficial
from datetime import date
class PaginaSchema(Schema):
@ -61,3 +63,64 @@ class SugestaoResponse(Schema):
"""
sugestao: Optional[str]
class TipoDiarioSchema(ModelSchema):
class Config:
model = TipoDiarioOficial
model_fields = ["id", "nome"]
class DiarioOficialIn(Schema):
"""Schema para criação de Diário Oficial"""
data: date
numero: str
tipo_id: Optional[int] = None
link: Optional[str] = None
arquivo: Optional[UploadedFile] = None # Adicionando o campo de arquivo
class Config:
# Configuração adicional para o Swagger UI
json_schema_extra = {
"example": {
"data": "2023-12-01",
"numero": "1234/2023",
}
}
class DiarioOficialOut(Schema):
"""Schema para retorno de Diário Oficial (com detalhes completos)"""
id: int
data: str # Será formatado como ISO (YYYY-MM-DD)
numero: str
link: Optional[str]
tipo: Optional[TipoDiarioSchema]
total_paginas: int
@staticmethod
def resolve_data(obj):
return obj.data.isoformat()
@staticmethod
def resolve_total_paginas(obj):
return obj.paginas.count()
class DiarioOficialUpdate(Schema):
"""Schema para atualização de Diário Oficial"""
data: Optional[date] = None
numero: Optional[str] = None
tipo_id: Optional[int] = None
link: Optional[str] = None
class DiarioListagem(Schema):
"""Schema simplificado para listagem de Diários"""
id: int
data: str
numero: str
tipo_nome: Optional[str]
@staticmethod
def resolve_data(obj):
return obj.data.isoformat()
@staticmethod
def resolve_tipo_nome(obj):
return obj.tipo.nome if obj.tipo else None

131
diarios/search_service.py
View File

@ -2,7 +2,7 @@ import re
from datetime import datetime
from typing import Optional, Dict, Any, List
from elasticsearch import Elasticsearch, AsyncElasticsearch
from .schemas import ResultadoSchema, PaginaSchema
from .schemas import BuscaDiariosResponseSchema, ResultadoSchema, PaginaSchema
import unicodedata
import asyncio
from django.conf import settings
@ -588,16 +588,19 @@ async def processar_paginas_encontradas(
Returns:
List[PaginaSchema]: Lista de páginas encontradas
"""
if not query:
return []
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", "")
)
)
#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"]
@ -731,3 +734,115 @@ async def buscar_diarios_simples(
# Processar e retornar os resultados
return await processar_resultados(response, query, page, page_size)
async def list_diarios(
es_client: AsyncElasticsearch,
page: int = 1,
page_size: int = 10,
data_inicio: Optional[str] = None,
data_fim: Optional[str] = None,
tipo_diario: Optional[str] = None,
numero_diario: Optional[str] = None,
ordenar_por: str = "data_desc"
) -> Dict[str, any]:
"""
Lista diários com paginação e filtros opcionais
Args:
es_client: Cliente do Elasticsearch
page: Número da página
page_size: Itens por página
data_inicio: Data inicial (YYYY-MM-DD)
data_fim: Data final (YYYY-MM-DD)
tipo_diario: Filtro por tipo de diário
numero_diario: Filtro por número do diário
ordenar_por: Campo para ordenação (data_desc, data_asc)
Returns:
Dict com total de itens e lista de diários
"""
# Construir query de filtros
filters = []
# Filtro por data
if data_inicio or data_fim:
date_range = {}
if data_inicio:
try:
datetime.strptime(data_inicio, "%Y-%m-%d")
date_range["gte"] = data_inicio
except ValueError:
pass
if data_fim:
try:
datetime.strptime(data_fim, "%Y-%m-%d")
date_range["lte"] = data_fim
except ValueError:
pass
if date_range:
filters.append({"range": {"data": date_range}})
# Filtro por tipo
if tipo_diario:
filters.append({"term": {"tipo.nome": tipo_diario}})
# Filtro por número
if numero_diario:
filters.append({"wildcard": {"numero": f"*{numero_diario}*"}})
# Construir query completa
query = {
"bool": {
"must": [{"match_all": {}}],
"filter": filters
}
}
# Definir ordenação
sort = [{"data": {"order": "asc" if ordenar_por == "data_asc" else "desc"}}]
# Executar consulta
try:
response = await es_client.search(
index="diario_oficial",
body={
"query": query,
"sort": sort,
"from": (page - 1) * page_size,
"size": page_size,
"_source": ["numero", "data", "tipo", "link"]
}
)
hits = response["hits"]["hits"]
total = response["hits"]["total"]["value"]
# Processar resultados
diarios = []
for hit in hits:
source = hit["_source"]
diarios.append(BuscaDiariosResponseSchema(
id=hit["_id"],
numero=source.get("numero"),
data=source.get("data"),
tipo=source.get("tipo", {}).get("nome"),
link=source.get("link")
))
return {
"total": total,
"page": page,
"page_size": page_size,
"results": diarios
}
except Exception as e:
print(f"Erro ao listar diários: {e}")
return {
"total": 0,
"page": page,
"page_size": page_size,
"results": []
}

2
diarios/urls.py
View File

@ -2,5 +2,5 @@ from django.urls import path
from . import views
urlpatterns = [
path('busca/', views.index, name='index')
path("", views.home, name='home')
]

37
diarios/views.py
View File

@ -1,19 +1,19 @@
from ninja import Router
from ninja import Router, File, Form
from ninja.files import UploadedFile
from typing import Optional
from django.http import HttpRequest
from django.http import HttpRequest, HttpResponse
from .search_service import (
buscar_diarios,
sugestao_termo,
buscar_diarios_simples,
)
from .schemas import BuscaDiariosResponseSchema, SugestaoResponse
from .schemas import BuscaDiariosResponseSchema, SugestaoResponse, DiarioOficialIn, DiarioOficialOut
from django.shortcuts import render
router = Router(tags=["Diários Oficiais"])
async def index(request):
return render(request, 'diarios/busca.html')
async def home(request):
return render(request, 'diarios/index.html')
@router.get(
"/sugestao",
@ -86,3 +86,28 @@ async def busca_diarios_oficiais_simples(
)
return resultado
@router.post("/", response=DiarioOficialOut, summary="Criar novo diário oficial")
def criar_diario(
request: HttpRequest,
# Usamos Form para os dados normais e File para o upload
payload: DiarioOficialIn = Form(...),
arquivo: UploadedFile = File(None)
):
"""
Cria um novo diário oficial.
Observações:
- Aceita tanto upload de arquivo PDF quanto link para o diário
- Se ambos (arquivo e link) forem fornecidos, o arquivo terá prioridade
- O arquivo deve ser um PDF válido
"""
# Prioriza o arquivo se ambos existirem
arquivo_final = arquivo if arquivo else payload.arquivo
return DiarioOficialService.criar_diario(
data=payload.data,
numero=payload.numero,
tipo_id=payload.tipo_id,
arquivo=arquivo_final,
link=payload.link
)