feat(Certidao): Adiciona recursos para manipular certidões
This commit is contained in:
Keven
parent
433d889060
commit
a6c88e5056
20 changed files with 693 additions and 57 deletions
|
|
@ -277,3 +277,5 @@ gunicorn main:app \
|
|||
* Automatize o serviço em produção via **systemd** (ex: `/etc/systemd/system/saas_api.service`) para iniciar junto com o servidor.
|
||||
|
||||
---
|
||||
|
||||
uvicorn main:app --host 0.0.0.0 --port 8000
|
||||
|
|
|
|||
31
actions/data/microtime.py
Normal file
31
actions/data/microtime.py
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
import time
|
||||
|
||||
|
||||
class Microtime:
|
||||
"""
|
||||
Utilitário para manipulação de tempo com alta precisão (microssegundos).
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def get() -> float:
|
||||
"""
|
||||
Retorna o timestamp Unix atual com precisão de microssegundos.
|
||||
Equivalente ao microtime(true) do PHP.
|
||||
"""
|
||||
return time.time()
|
||||
|
||||
@staticmethod
|
||||
def as_int() -> int:
|
||||
"""
|
||||
Retorna o tempo atual puramente em microssegundos (Inteiro).
|
||||
Útil para gerar IDs únicos ou ordenação precisa.
|
||||
"""
|
||||
# Pega em nanosegundos e converte para microssegundos
|
||||
return time.time_ns() // 1000
|
||||
|
||||
@staticmethod
|
||||
def diff(start_time: float) -> float:
|
||||
"""
|
||||
Calcula a diferença (duração) em segundos com precisão.
|
||||
"""
|
||||
return time.time() - start_time
|
||||
141
actions/data/text.py
Normal file
141
actions/data/text.py
Normal file
|
|
@ -0,0 +1,141 @@
|
|||
# Importa a biblioteca nativa 'zlib' usada para descompressão de dados binários.
|
||||
import base64
|
||||
import zlib
|
||||
|
||||
# Importa a função 'rtf_to_text' da biblioteca 'striprtf',
|
||||
# responsável por converter documentos RTF em texto plano legível.
|
||||
from striprtf.striprtf import rtf_to_text
|
||||
|
||||
|
||||
# Define uma classe utilitária chamada 'String', contendo apenas métodos estáticos.
|
||||
# Essa abordagem permite o uso direto sem necessidade de instanciar a classe.
|
||||
class Text:
|
||||
@staticmethod
|
||||
def decompress(vf_string):
|
||||
"""
|
||||
Descomprime e decodifica texto de origem WPTools/Firebird.
|
||||
|
||||
Finalidade:
|
||||
Converter o conteúdo de campos BLOB ou strings compactadas (como no Delphi)
|
||||
em texto legível, detectando automaticamente se o conteúdo está:
|
||||
- Compactado com zlib
|
||||
- Codificado em ISO-8859-1 (padrão ANSI)
|
||||
- Representado como bytes puros
|
||||
"""
|
||||
# Verifica se o valor recebido é nulo, vazio ou None.
|
||||
# Se for, retorna string vazia para evitar erros de processamento.
|
||||
if not vf_string:
|
||||
return ""
|
||||
|
||||
# Caso seja um objeto tipo stream (ex: campo BLOB do Firebird)
|
||||
# Campos BLOB geralmente possuem o método `.read()` para leitura de bytes.
|
||||
if hasattr(vf_string, "read"):
|
||||
vf_string = vf_string.read() # Lê o conteúdo completo do stream
|
||||
|
||||
# Garante que o valor trabalhado é uma sequência de bytes (não string)
|
||||
# Se o dado já for texto (str), converte para bytes em codificação Latin-1,
|
||||
# que é compatível com ISO-8859-1 usado por sistemas Delphi/Firebird.
|
||||
if isinstance(vf_string, str):
|
||||
vf_bytes = vf_string.encode("latin1", errors="ignore")
|
||||
else:
|
||||
vf_bytes = vf_string # Já está em bytes, então apenas reaproveita
|
||||
|
||||
# Detecta se o conteúdo foi compactado com zlib.
|
||||
# A assinatura padrão do formato zlib começa com bytes: 0x78 0x9C ou 0x78 0xDA.
|
||||
is_zlib = (
|
||||
len(vf_bytes) > 2 and vf_bytes[0] == 0x78 and vf_bytes[1] in (0x9C, 0xDA)
|
||||
)
|
||||
|
||||
# Se a detecção confirmar que o conteúdo é zlib, tenta descompactar.
|
||||
if is_zlib:
|
||||
try:
|
||||
# Descompacta os bytes e decodifica o texto usando ISO-8859-1 (ANSI),
|
||||
# que preserva corretamente acentuação e caracteres especiais.
|
||||
text = zlib.decompress(vf_bytes).decode("iso-8859-1", errors="ignore")
|
||||
return text
|
||||
except Exception:
|
||||
# Caso falhe (por dados corrompidos ou não comprimidos de fato),
|
||||
# o fluxo continua normalmente sem interromper a execução.
|
||||
pass
|
||||
|
||||
# Se não for zlib, tenta tratar o conteúdo como texto puro (não compactado)
|
||||
try:
|
||||
# Decodifica os bytes diretamente de ISO-8859-1 (padrão usado pelo Delphi)
|
||||
return vf_bytes.decode("iso-8859-1", errors="ignore")
|
||||
except Exception:
|
||||
# Como fallback, converte para string bruta para evitar falhas.
|
||||
return str(vf_string)
|
||||
|
||||
# >>> NOVO MÉTODO <<<
|
||||
@staticmethod
|
||||
def compress(text, *, encoding: str = "iso-8859-1", as_base64: bool = True):
|
||||
"""
|
||||
Comprime texto/dados com zlib.
|
||||
|
||||
Parâmetros:
|
||||
text: str | bytes | stream (com .read())
|
||||
encoding: encoding usado quando 'text' for str (padrão: ISO-8859-1)
|
||||
as_base64: se True, retorna string Base64 do conteúdo comprimido;
|
||||
caso False, retorna bytes comprimidos.
|
||||
|
||||
Retorno:
|
||||
- bytes (zlib) quando as_base64=False
|
||||
- str (Base64) quando as_base64=True
|
||||
|
||||
Observações:
|
||||
- Use o mesmo 'encoding' ao descomprimir para simetria.
|
||||
- Ideal para armazenar em BLOB ou trafegar seguro (Base64).
|
||||
"""
|
||||
if text is None or text == "":
|
||||
return "" if as_base64 else b""
|
||||
|
||||
# Se for stream (ex.: BLOB do Firebird)
|
||||
if hasattr(text, "read"):
|
||||
raw = text.read()
|
||||
else:
|
||||
raw = text
|
||||
|
||||
# Garante bytes
|
||||
if isinstance(raw, str):
|
||||
raw_bytes = raw.encode(encoding, errors="ignore")
|
||||
else:
|
||||
raw_bytes = bytes(raw)
|
||||
|
||||
# Comprime com zlib
|
||||
comp = zlib.compress(raw_bytes)
|
||||
|
||||
# Opcional: codifica em Base64 para transporte/JSON
|
||||
if as_base64:
|
||||
return base64.b64encode(comp).decode("ascii")
|
||||
|
||||
return comp
|
||||
|
||||
@staticmethod
|
||||
def to_text(raw_text: str) -> str:
|
||||
"""
|
||||
Converte o conteúdo RTF em texto simples e retorna como string.
|
||||
|
||||
Finalidade:
|
||||
- Detectar automaticamente se o conteúdo está em formato RTF.
|
||||
- Converter para texto plano usando a função 'rtf_to_text' (da striprtf).
|
||||
- Retornar uma string limpa e pronta para ser usada em APIs, logs, etc.
|
||||
"""
|
||||
# Verifica se o texto recebido está vazio ou None — retorna vazio se sim.
|
||||
if not raw_text:
|
||||
return ""
|
||||
|
||||
# Verifica se o texto começa com o cabeçalho padrão de arquivos RTF.
|
||||
# Exemplo: "{\\rtf1\\ansi..." indica conteúdo em formato RTF.
|
||||
if raw_text.strip().startswith("{\\rtf"):
|
||||
try:
|
||||
# Converte o RTF em texto simples, preservando acentuação e quebras de linha.
|
||||
text = rtf_to_text(raw_text)
|
||||
# Remove espaços em branco extras nas extremidades.
|
||||
return text.strip()
|
||||
except Exception:
|
||||
# Se ocorrer erro na conversão (ex: RTF inválido),
|
||||
# retorna o conteúdo original sem alterações.
|
||||
return raw_text
|
||||
|
||||
# Caso o texto não seja RTF, apenas remove espaços em branco extras e retorna.
|
||||
return raw_text.strip()
|
||||
|
|
@ -12,74 +12,58 @@ from actions.log.log import Log
|
|||
|
||||
def register_exception_handlers(app):
|
||||
|
||||
def __init__ (self):
|
||||
def __init__(self):
|
||||
log = Log()
|
||||
|
||||
@app.exception_handler(BusinessRuleException)
|
||||
async def business_rule_exception_handler(request: Request, exc: BusinessRuleException):
|
||||
async def business_rule_exception_handler(
|
||||
request: Request, exc: BusinessRuleException
|
||||
):
|
||||
|
||||
response = {
|
||||
"status": "422",
|
||||
"error": "Regra de negócio",
|
||||
"detail": exc.message
|
||||
}
|
||||
response = {"status": "422", "error": "Regra de negócio", "detail": exc.message}
|
||||
|
||||
# Salva o log em disco
|
||||
Log.register(response, 'storage/temp/business_rule_exception_handler.json')
|
||||
Log.register(response, "storage/temp/business_rule_exception_handler.json")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=422,
|
||||
content=response
|
||||
)
|
||||
return JSONResponse(status_code=422, content=response)
|
||||
|
||||
@app.exception_handler(StarletteHTTPException)
|
||||
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
|
||||
response = {
|
||||
"status": exc.status_code,
|
||||
"error": "HTTP Error",
|
||||
"detail": exc.detail
|
||||
}
|
||||
"status": exc.status_code,
|
||||
"error": "HTTP Error",
|
||||
"detail": exc.detail,
|
||||
}
|
||||
|
||||
# Salva o log em disco
|
||||
Log.register(response, 'storage/temp/http_exception_handler.json')
|
||||
Log.register(response, "storage/temp/http_exception_handler.json")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content=response
|
||||
)
|
||||
return JSONResponse(status_code=exc.status_code, content=response)
|
||||
|
||||
@app.exception_handler(RequestValidationError)
|
||||
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
||||
async def validation_exception_handler(
|
||||
request: Request, exc: RequestValidationError
|
||||
):
|
||||
|
||||
response = {
|
||||
"status": 400,
|
||||
"error": "Erro de validação",
|
||||
"detail": exc.errors()
|
||||
}
|
||||
response = {"status": 400, "error": "Erro de validação", "detail": exc.errors()}
|
||||
|
||||
# Salva o log em disco
|
||||
Log.register(response, 'storage/temp/validation_exception_handler.json')
|
||||
Log.register(response, "storage/temp/validation_exception_handler.json")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=400,
|
||||
content=response
|
||||
)
|
||||
return JSONResponse(status_code=400, content=response)
|
||||
|
||||
@app.exception_handler(Exception)
|
||||
async def global_exception_handler(request: Request, exc: Exception):
|
||||
|
||||
response = {
|
||||
"status": 500,
|
||||
"error": "Erro Interno do Servidor",
|
||||
"type": type(exc).__name__,
|
||||
"message": str(exc),
|
||||
"trace": traceback.format_exc()
|
||||
}
|
||||
"status": 500,
|
||||
"error": "Erro Interno do Servidor",
|
||||
"type": type(exc).__name__,
|
||||
"message": str(exc),
|
||||
"trace": traceback.format_exc(),
|
||||
}
|
||||
|
||||
# Salva o log em disco
|
||||
Log.register(response, 'storage/temp/validation_exception_handler.json')
|
||||
Log.register(response, "storage/temp/validation_exception_handler.json")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=500,
|
||||
content=response
|
||||
)
|
||||
return JSONResponse(status_code=500, content=response)
|
||||
|
|
|
|||
3
main.py
3
main.py
|
|
@ -16,6 +16,7 @@ from pathlib import Path
|
|||
|
||||
# Importa o middleware de CORS
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
# Importa o roteador principal da API versão 1
|
||||
from packages.v1.api import api_router
|
||||
|
|
@ -62,6 +63,8 @@ app.add_middleware(
|
|||
max_age=60 * 60 * 8,
|
||||
)
|
||||
|
||||
app.mount(path="/temp", app=StaticFiles(directory="./storage/temp"), name="temp")
|
||||
|
||||
|
||||
@app.on_event("startup")
|
||||
async def on_startup():
|
||||
|
|
|
|||
|
|
@ -0,0 +1,27 @@
|
|||
from packages.v1.servicos.balcao.repositories.t_ato.t_ato_show_repository import (
|
||||
TAtoShowRepository,
|
||||
)
|
||||
from packages.v1.servicos.balcao.schemas.t_ato_schema import TAtoIdSchema
|
||||
|
||||
|
||||
class TAtoShowAction:
|
||||
"""
|
||||
Serviço responsável por encapsular a lógica de negócio para a operação
|
||||
de listagem de todos os registros na tabela g_tb_regimebens.
|
||||
"""
|
||||
|
||||
def execute(self, data: TAtoIdSchema):
|
||||
"""
|
||||
Executa a operação de listagem no banco de dados.
|
||||
|
||||
Returns:
|
||||
A lista de todos os registros.
|
||||
"""
|
||||
# Instanciamento do repositório
|
||||
show_repository = TAtoShowRepository()
|
||||
|
||||
# Execução do repositório
|
||||
response = show_repository.execute(data)
|
||||
|
||||
# Retorno da informação
|
||||
return response
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
import os
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
from abstracts.action import BaseAction
|
||||
from packages.v1.servicos.balcao.schemas.t_servico_itempedido_schema import (
|
||||
TServicoItemPedidoCreateCertidaoSchema,
|
||||
)
|
||||
from actions.data.text import Text
|
||||
from actions.data.microtime import Microtime
|
||||
|
||||
|
||||
class TServicoItemPedidoCreateCertidaoAction(BaseAction):
|
||||
"""
|
||||
Serviço responsável por gerar o arquivo físico da certidão (.rtf)
|
||||
a partir do texto descomprimido e validar sua integridade.
|
||||
Retorna um dicionário (JSON) indicando sucesso ou falha.
|
||||
"""
|
||||
|
||||
def execute(self, data: TServicoItemPedidoCreateCertidaoSchema):
|
||||
|
||||
# 1. Decodifica o texto
|
||||
texto = Text.decompress(data.certidao_texto)
|
||||
|
||||
# 2. Configuração do caminho e nome
|
||||
diretorio = "./storage/temp/"
|
||||
name = f"{data.servico_itempedido_id}_{Microtime.as_int()}.rtf"
|
||||
|
||||
# Define a variável caminho_completo
|
||||
caminho_completo = os.path.join(diretorio, name)
|
||||
|
||||
# 3. Garante que a pasta existe
|
||||
os.makedirs(diretorio, exist_ok=True)
|
||||
|
||||
# 4. Escreve o texto em disco
|
||||
with open(caminho_completo, "wb") as f:
|
||||
f.write(texto.encode("utf-8"))
|
||||
|
||||
# 5. VALIDAÇÃO: Verifica se o arquivo existe e se tem conteúdo
|
||||
if not os.path.exists(caminho_completo):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Falha crítica: O arquivo não foi encontrado em {caminho_completo}",
|
||||
)
|
||||
|
||||
if os.path.getsize(caminho_completo) == 0:
|
||||
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Falha crítica: O arquivo foi criado, mas está vazio (0 bytes).",
|
||||
)
|
||||
|
||||
# RETORNO DE SUCESSO
|
||||
return name
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
from actions.dynamic_import.service_factory import ServiceFactory
|
||||
from interfaces.service_protocols import ServiceProtocolsInterface
|
||||
from packages.v1.servicos.balcao.schemas.t_ato_schema import TAtoIdSchema
|
||||
|
||||
|
||||
class TAtoController:
|
||||
|
|
@ -18,3 +19,13 @@ class TAtoController:
|
|||
result = service.execute(schema)
|
||||
|
||||
return {"message": "Sucesso", "data": result}
|
||||
|
||||
def show(self, data: TAtoIdSchema):
|
||||
|
||||
# Instânciamento da classe
|
||||
service = self.factory.make("TAtoShowService", ServiceProtocolsInterface)
|
||||
|
||||
# O VS Code sabe que .execute() existe por causa do IService!
|
||||
result = service.execute(data)
|
||||
|
||||
return {"message": "Sucesso", "data": result}
|
||||
|
|
|
|||
|
|
@ -61,6 +61,44 @@ class TServicoItemPedidoController:
|
|||
"data": self.show_service.execute(t_servico_itempedido_id_schema),
|
||||
}
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Busca um registro específico de T_SERVICO_ITEMPEDIDO pelo ID
|
||||
# ----------------------------------------------------
|
||||
def certidao_show(self, t_servico_itempedido_id_schema: TServicoItemPedidoIdSchema):
|
||||
# Importação da classe desejada
|
||||
show_service = self.dynamic_import.service(
|
||||
"t_servico_itempedido_certidao_show_service",
|
||||
"TServicoItemPedidoCertidaoShowService",
|
||||
)
|
||||
|
||||
# Instância da classe service
|
||||
self.show_service = show_service()
|
||||
|
||||
# Execução da busca
|
||||
return {
|
||||
"message": "Registro de T_SERVICO_ITEMPEDIDO localizado com sucesso.",
|
||||
"data": self.show_service.execute(t_servico_itempedido_id_schema),
|
||||
}
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Busca um registro específico de T_SERVICO_ITEMPEDIDO pelo ID
|
||||
# ----------------------------------------------------
|
||||
def certidao_edit(self, t_servico_itempedido_id_schema: TServicoItemPedidoIdSchema):
|
||||
# Importação da classe desejada
|
||||
edit_service = self.dynamic_import.service(
|
||||
"t_servico_itempedido_certidao_edit_service",
|
||||
"TServicoItemPedidoCertidaoEditService",
|
||||
)
|
||||
|
||||
# Instância da classe service
|
||||
self.show_service = edit_service()
|
||||
|
||||
# Execução da busca
|
||||
return {
|
||||
"message": "Registro de T_SERVICO_ITEMPEDIDO localizado com sucesso.",
|
||||
"data": self.show_service.execute(t_servico_itempedido_id_schema),
|
||||
}
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Cadastra um novo registro em T_SERVICO_ITEMPEDIDO
|
||||
# ----------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -3,7 +3,10 @@ from fastapi import APIRouter, Depends, status
|
|||
from actions.data.get_url_params import get_url_params
|
||||
from actions.jwt.get_current_user import get_current_user
|
||||
from packages.v1.servicos.balcao.controllers.t_ato_controller import TAtoController
|
||||
from packages.v1.servicos.balcao.schemas.t_ato_schema import TAtoIndexSchema
|
||||
from packages.v1.servicos.balcao.schemas.t_ato_schema import (
|
||||
TAtoIdSchema,
|
||||
TAtoIndexSchema,
|
||||
)
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Inicializa o roteador para as rotas da tabela T_ATO
|
||||
|
|
@ -32,3 +35,20 @@ async def index(
|
|||
|
||||
# Retorna os dados localizados
|
||||
return response
|
||||
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Lista um registro especifico
|
||||
# ----------------------------------------------------
|
||||
@router.get(
|
||||
"/{ato_id}",
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Lista um registro em especifico",
|
||||
response_description="Lista um registro em especifico",
|
||||
)
|
||||
async def show(ato_id: int, current_user: dict = Depends(get_current_user)):
|
||||
"""
|
||||
Retorna um registro específico de G_GRAMATICA com base no ID informado.
|
||||
"""
|
||||
response = t_ato_controller.show(TAtoIdSchema(ato_id=ato_id))
|
||||
return response
|
||||
|
|
|
|||
|
|
@ -64,6 +64,54 @@ async def show(
|
|||
return response
|
||||
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Busca um registro específico de T_SERVICO_ITEMPEDIDO pelo ID
|
||||
# ----------------------------------------------------
|
||||
@router.get(
|
||||
"/{servico_itempedido_id}/certidao/exibir",
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Busca o texto da certidao",
|
||||
response_description="Busca o texto da certidao",
|
||||
)
|
||||
async def certidao_show(
|
||||
servico_itempedido_id: int, current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Retorna um registro específico de T_SERVICO_ITEMPEDIDO com base no ID informado.
|
||||
"""
|
||||
t_servico_itempedido_id_schema = TServicoItemPedidoIdSchema(
|
||||
servico_itempedido_id=servico_itempedido_id
|
||||
)
|
||||
response = t_servico_itempedido_controller.certidao_show(
|
||||
t_servico_itempedido_id_schema
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Gera o arquivo em disco para Manipulação
|
||||
# ----------------------------------------------------
|
||||
@router.get(
|
||||
"/{servico_itempedido_id}/certidao/editar",
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Gera o arquivo em disco da certidão",
|
||||
response_description="Gera o arquivo em disco da certidão",
|
||||
)
|
||||
async def certidao_edit(
|
||||
servico_itempedido_id: int, current_user: dict = Depends(get_current_user)
|
||||
):
|
||||
"""
|
||||
Retorna um registro específico de T_SERVICO_ITEMPEDIDO com base no ID informado.
|
||||
"""
|
||||
t_servico_itempedido_id_schema = TServicoItemPedidoIdSchema(
|
||||
servico_itempedido_id=servico_itempedido_id
|
||||
)
|
||||
response = t_servico_itempedido_controller.certidao_edit(
|
||||
t_servico_itempedido_id_schema
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Cadastra um novo registro em T_SERVICO_ITEMPEDIDO
|
||||
# ----------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -17,14 +17,17 @@ class TAtoIndexRepository(BaseRepository):
|
|||
"""
|
||||
# Montagem do SQL
|
||||
sql = """ SELECT
|
||||
ta.ato_id,
|
||||
ta.PROTOCOLO ,
|
||||
ta.DATA_LAVRATURA,
|
||||
ta.FOLHA_INICIAL ,
|
||||
ta.FOLHA_FINAL
|
||||
TA.ATO_ID,
|
||||
TA.PROTOCOLO ,
|
||||
TA.DATA_LAVRATURA,
|
||||
TLA.NUMERO_LIVRO,
|
||||
TA.FOLHA_INICIAL ,
|
||||
TA.FOLHA_FINAL
|
||||
FROM
|
||||
t_ato ta
|
||||
ORDER BY TA.ATO_ID ASC """
|
||||
T_ATO TA
|
||||
JOIN T_LIVRO_ANDAMENTO tla ON TA.LIVRO_ANDAMENTO_ID = TLA.LIVRO_ANDAMENTO_ID
|
||||
ORDER BY
|
||||
TA.ATO_ID ASC """
|
||||
|
||||
# Execução do sql
|
||||
response = self.fetch_all(sql)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,32 @@
|
|||
from typing import Any
|
||||
from abstracts.repository import BaseRepository
|
||||
from packages.v1.servicos.balcao.schemas.t_ato_schema import (
|
||||
TAtoIdSchema,
|
||||
)
|
||||
|
||||
|
||||
class TAtoShowRepository(BaseRepository):
|
||||
"""
|
||||
Repositório para a operação de listagem de todos os registros
|
||||
na tabela T_ATO.
|
||||
"""
|
||||
|
||||
def execute(self, data: TAtoIdSchema):
|
||||
"""
|
||||
Executa a consulta SQL para buscar todos os registros.
|
||||
|
||||
Returns:
|
||||
Uma lista de dicionários contendo os dados dos registros.
|
||||
"""
|
||||
# Montagem do SQL
|
||||
sql = """ SELECT TA.* FROM T_ATO TA
|
||||
WHERE TA.ATO_ID = :ato_id """
|
||||
|
||||
# Preenchimento dos parâmetros
|
||||
params: dict[str, Any] = data.model_dump(exclude_unset=True)
|
||||
|
||||
# Execução do sql
|
||||
response = self.fetch_one(sql, params)
|
||||
|
||||
# Retorna os dados localizados
|
||||
return response
|
||||
|
|
@ -30,12 +30,16 @@ class TServicoItemPedidoIndexRepository(BaseRepository):
|
|||
TSP.VALOR_ISS,
|
||||
TSP.FUNDESP,
|
||||
TSP.VALOR,
|
||||
TSP.CERTIDAO_TEXTO,
|
||||
TSP.CERTIDAO_ATO_ID,
|
||||
TST.DESCRICAO,
|
||||
GE.DESCRICAO AS EMOLUMENTO_DESCRICAO,
|
||||
TSP.SITUACAO,
|
||||
TSE.ETIQUETA_MODELO_ID,
|
||||
GMT.TEXTO AS ETIQUETA_TEXTO,
|
||||
GMT.GRUPO,
|
||||
TE.NOME,
|
||||
TE.CPF_CNPJ,
|
||||
GMT.GRUPO
|
||||
GE.DESCRICAO AS EMOLUMENTO_DESCRICAO
|
||||
FROM T_SERVICO_ITEMPEDIDO TSP
|
||||
JOIN T_SERVICO_TIPO TST ON TSP.SERVICO_TIPO_ID = TST.SERVICO_TIPO_ID
|
||||
LEFT JOIN T_SERVICO_ETIQUETA TSE ON TST.SERVICO_TIPO_ID = TSE.SERVICO_TIPO_ID
|
||||
|
|
|
|||
|
|
@ -12,6 +12,14 @@ class TAtoIndexSchema(BaseModel):
|
|||
extra = "allow"
|
||||
|
||||
|
||||
class TAtoIdSchema(BaseModel):
|
||||
|
||||
ato_id: Decimal = None
|
||||
|
||||
class Config:
|
||||
extra = "allow"
|
||||
|
||||
|
||||
class TAtoSchema(BaseModel):
|
||||
ato_id: Optional[Decimal] = None
|
||||
ato_tipo_id: Optional[Decimal] = None
|
||||
|
|
|
|||
|
|
@ -137,6 +137,15 @@ class TServicoItemPedidoSaveSchema_(TServicoItemPedidoSchema):
|
|||
from_attributes = True
|
||||
|
||||
|
||||
class TServicoItemPedidoCreateCertidaoSchema(BaseModel):
|
||||
|
||||
servico_itempedido_id: int
|
||||
certidao_texto: bytes
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class TServicoItemPedidoUpdateSchema(TServicoItemPedidoSchema):
|
||||
|
||||
class Config:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,45 @@
|
|||
from fastapi import HTTPException, status
|
||||
from packages.v1.servicos.balcao.actions.t_ato.t_ato_show_action import TAtoShowAction
|
||||
from packages.v1.servicos.balcao.schemas.t_ato_schema import TAtoIdSchema
|
||||
|
||||
|
||||
class TAtoShowService:
|
||||
"""
|
||||
Serviço responsável por encapsular a lógica de negócio para a operação
|
||||
de listagem de registros na tabela T_PESSOA_CARTAO.
|
||||
"""
|
||||
|
||||
def execute(self, data: TAtoIdSchema):
|
||||
"""
|
||||
Executa a operação de busca de todos os registros no banco de dados.
|
||||
|
||||
Args:
|
||||
t_ato_index_schema (TAtoIndexSchema):
|
||||
Esquema que pode conter filtros ou parâmetros de busca.
|
||||
|
||||
Returns:
|
||||
A lista de registros encontrados.
|
||||
"""
|
||||
# ----------------------------------------------------
|
||||
# Instanciamento da ação
|
||||
# ----------------------------------------------------
|
||||
show_action = TAtoShowAction()
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Execução da ação
|
||||
# ----------------------------------------------------
|
||||
data = show_action.execute(data)
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Verificação de retorno
|
||||
# ----------------------------------------------------
|
||||
if not data:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Não foi possível localizar registros de T_PESSOA_CARTAO.",
|
||||
)
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Retorno da informação
|
||||
# ----------------------------------------------------
|
||||
return data
|
||||
|
|
@ -0,0 +1,70 @@
|
|||
from types import SimpleNamespace
|
||||
from packages.v1.servicos.balcao.actions.t_servico_itempedido.t_servico_itempedido_show_action import (
|
||||
TServicoItemPedidoShowAction,
|
||||
)
|
||||
from packages.v1.servicos.balcao.schemas.t_servico_itempedido_schema import (
|
||||
TServicoItemPedidoIdSchema,
|
||||
)
|
||||
from actions.data.text import Text
|
||||
from actions.data.microtime import Microtime
|
||||
from fastapi import HTTPException, status
|
||||
|
||||
|
||||
class TServicoItemPedidoCertidaoEditService:
|
||||
"""
|
||||
Serviço responsável por encapsular a lógica de negócio para a operação
|
||||
de busca de um registro na tabela T_SERVICO_ITEMPEDIDO.
|
||||
"""
|
||||
|
||||
def execute(self, t_servico_itempedido_id_schema: TServicoItemPedidoIdSchema):
|
||||
"""
|
||||
Executa a operação de busca no banco de dados.
|
||||
|
||||
Args:
|
||||
t_servico_itempedido_id_schema (TServicoItemPedidoIdSchema):
|
||||
O esquema com o ID do registro a ser buscado.
|
||||
|
||||
Returns:
|
||||
O resultado da busca.
|
||||
"""
|
||||
# ----------------------------------------------------
|
||||
# Instanciamento da ação
|
||||
# ----------------------------------------------------
|
||||
t_servico_itempedido_show_action = TServicoItemPedidoShowAction()
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Execução da ação
|
||||
# ----------------------------------------------------
|
||||
row = t_servico_itempedido_show_action.execute(t_servico_itempedido_id_schema)
|
||||
|
||||
data = SimpleNamespace(**row)
|
||||
|
||||
# Decodifica o texto
|
||||
texto = Text.decompress(data.certidao_texto)
|
||||
|
||||
# Nome do arquivo a ser gerado
|
||||
name = f"{data.servico_itempedido_id}_{Microtime.as_int()}.rtf"
|
||||
|
||||
# Escreve o texto em disco
|
||||
with open(
|
||||
"./storage/temp/" + str(name),
|
||||
"wb",
|
||||
) as f:
|
||||
f.write(texto.encode("utf-8"))
|
||||
|
||||
# Atualiza com o caminho do arquivo
|
||||
data.certidao_texto = name
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Verificação de resultado
|
||||
# ----------------------------------------------------
|
||||
if not data:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Não foi possível localizar o registro de T_SERVICO_ITEMPEDIDO.",
|
||||
)
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Retorno da informação
|
||||
# ----------------------------------------------------
|
||||
return data
|
||||
|
|
@ -0,0 +1,49 @@
|
|||
from packages.v1.servicos.balcao.actions.t_servico_itempedido.t_servico_itempedido_show_action import (
|
||||
TServicoItemPedidoShowAction,
|
||||
)
|
||||
from packages.v1.servicos.balcao.schemas.t_servico_itempedido_schema import (
|
||||
TServicoItemPedidoIdSchema,
|
||||
)
|
||||
from fastapi import HTTPException, status
|
||||
|
||||
|
||||
class TServicoItemPedidoCertidaoShowService:
|
||||
"""
|
||||
Serviço responsável por encapsular a lógica de negócio para a operação
|
||||
de busca de um registro na tabela T_SERVICO_ITEMPEDIDO.
|
||||
"""
|
||||
|
||||
def execute(self, t_servico_itempedido_id_schema: TServicoItemPedidoIdSchema):
|
||||
"""
|
||||
Executa a operação de busca no banco de dados.
|
||||
|
||||
Args:
|
||||
t_servico_itempedido_id_schema (TServicoItemPedidoIdSchema):
|
||||
O esquema com o ID do registro a ser buscado.
|
||||
|
||||
Returns:
|
||||
O resultado da busca.
|
||||
"""
|
||||
# ----------------------------------------------------
|
||||
# Instanciamento da ação
|
||||
# ----------------------------------------------------
|
||||
t_servico_itempedido_show_action = TServicoItemPedidoShowAction()
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Execução da ação
|
||||
# ----------------------------------------------------
|
||||
data = t_servico_itempedido_show_action.execute(t_servico_itempedido_id_schema)
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Verificação de resultado
|
||||
# ----------------------------------------------------
|
||||
if not data:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="Não foi possível localizar o registro de T_SERVICO_ITEMPEDIDO.",
|
||||
)
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Retorno da informação
|
||||
# ----------------------------------------------------
|
||||
return data
|
||||
|
|
@ -1,10 +1,19 @@
|
|||
from types import SimpleNamespace
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
from packages.v1.servicos.balcao.actions.t_servico_itempedido.t_servico_itempedido_create_certidao_action import (
|
||||
TServicoItemPedidoCreateCertidaoAction,
|
||||
)
|
||||
from packages.v1.servicos.balcao.actions.t_servico_itempedido.t_servico_itempedido_index_action import (
|
||||
TServicoItemPedidoIndexAction,
|
||||
)
|
||||
from fastapi import HTTPException, status
|
||||
|
||||
from packages.v1.servicos.balcao.schemas.t_ato_schema import TAtoIdSchema
|
||||
from packages.v1.servicos.balcao.schemas.t_servico_itempedido_schema import (
|
||||
TServicoItemIndexSchema,
|
||||
TServicoItemPedidoCreateCertidaoSchema,
|
||||
)
|
||||
from packages.v1.servicos.balcao.services.t_ato.go.t_ato_show_service import (
|
||||
TAtoShowService,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -33,10 +42,59 @@ class TServicoItemPedidoIndexService:
|
|||
# ----------------------------------------------------
|
||||
# Execução da ação
|
||||
# ----------------------------------------------------
|
||||
data = t_servico_itempedido_index_action.execute(
|
||||
rows = t_servico_itempedido_index_action.execute(
|
||||
t_servico_itempedido_index_schema
|
||||
)
|
||||
|
||||
# Lista para armazenar os objetos convertidos e modificados
|
||||
data = []
|
||||
|
||||
for row in rows:
|
||||
|
||||
# Cria o objeto mutável localmente
|
||||
item = SimpleNamespace(**row)
|
||||
|
||||
create_certidao_action = TServicoItemPedidoCreateCertidaoAction()
|
||||
|
||||
if item.etiqueta_texto:
|
||||
|
||||
# Cria o documento editavel em disco
|
||||
item.etiqueta_texto = create_certidao_action.execute(
|
||||
TServicoItemPedidoCreateCertidaoSchema(
|
||||
servico_itempedido_id=item.servico_itempedido_id,
|
||||
certidao_texto=item.etiqueta_texto,
|
||||
)
|
||||
)
|
||||
|
||||
if item.certidao_texto:
|
||||
|
||||
# Cria o documento editavel em disco
|
||||
item.certidao_texto = create_certidao_action.execute(
|
||||
TServicoItemPedidoCreateCertidaoSchema(
|
||||
servico_itempedido_id=item.servico_itempedido_id,
|
||||
certidao_texto=item.certidao_texto,
|
||||
)
|
||||
)
|
||||
|
||||
elif not item.certidao_texto and item.certidao_ato_id:
|
||||
|
||||
show_service = TAtoShowService()
|
||||
|
||||
ato_show = show_service.execute(
|
||||
TAtoIdSchema(ato_id=item.certidao_ato_id)
|
||||
)
|
||||
|
||||
# Cria o documento editavel em disco
|
||||
item.certidao_texto = create_certidao_action.execute(
|
||||
TServicoItemPedidoCreateCertidaoSchema(
|
||||
servico_itempedido_id=item.servico_itempedido_id,
|
||||
certidao_texto=ato_show.texto,
|
||||
)
|
||||
)
|
||||
|
||||
# Adiciona o objeto processado na lista final
|
||||
data.append(item)
|
||||
|
||||
# ----------------------------------------------------
|
||||
# Verificação de retorno
|
||||
# ----------------------------------------------------
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue