OriusTecnologia/saas_api
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(OnlyOffice): Ajusta o Processador de Documentos

Browse source
This commit is contained in:
Keven 2025-12-30 12:49:17 -03:00
parent 63f0a81970
commit b173bffcf1
9 changed files with 468 additions and 139 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
abstracts/repository.py
View file

@ -40,6 +40,36 @@ class BaseRepository:
self, sql: str, params: Optional[dict[str, Any]], fetch: Literal["result"]
) -> CursorResult[Any]: ...
# ===============================
# Helpers internos (BLOB)
# ===============================
def _materialize_blob(self, value: Any) -> Any:
"""
Converte BLOBs do Firebird em bytes.
Retorna o valor original caso não seja BLOB.
"""
if value is None:
return None
if hasattr(value, "read") and callable(value.read):
try:
return value.read()
except Exception:
return None
return value
def _normalize_row(self, row: Mapping[str, Any]) -> dict[str, Any]:
"""
Converte RowMapping em dict mutável e materializa BLOBs.
"""
data = dict(row)
for key, value in data.items():
data[key] = self._materialize_blob(value)
return data
# Implementação concreta que atende às quatro sobrecargas por meio de um retorno em união.
def _execute(
self,
@ -64,11 +94,11 @@ class BaseRepository:
result = conn.execute(text(sql), params or {})
# Quando for solicitado "all", converte o resultado em lista de mapeamentos (coluna->valor).
if fetch == "all":
# retorna Sequence[RowMapping], compatível com List[Mapping[str, Any]]
return list(result.mappings().all())
return [self._normalize_row(row) for row in result.mappings().all()]
# Quando for solicitado "one", retorna apenas o primeiro registro (ou None).
elif fetch == "one":
return result.mappings().first()
row = result.mappings().first()
return self._normalize_row(row) if row else None
# Quando for solicitado "none", não retorna dados (apenas executa o comando).
elif fetch == "none":
return None

221
actions/data/process_document_action.py Normal file
View file

@ -0,0 +1,221 @@
import os
import subprocess
from pathlib import Path
import uuid
from fastapi import HTTPException, status
from actions.data.text import Text
from schemas.process_document_schema import ProcessDocumentSchema
# Caminho absoluto do executável do LibreOffice no Windows.
# Usar caminho fixo evita problemas de PATH em serviços, venv e produção.
SOFFICE_PATH = r"C:\Program Files\LibreOffice\program\soffice.exe"
class ProcessDocumentAction:
"""
Action responsável por processar documentos editáveis.
REGRAS IMPORTANTES:
- TODO texto recebido vem COMPRIMIDO (RTF ou DOCX)
- A descompressão acontece SEMPRE
- Após descompressão:
- DOCX → salvo diretamente
- RTF → convertido para DOCX via LibreOffice
A saída final é SEMPRE um arquivo DOCX em disco.
"""
# ======================================================
# MÉTODO PÚBLICO (ENTRYPOINT DA ACTION)
# ======================================================
def execute(self, data: ProcessDocumentSchema) -> str:
"""
Orquestra todo o fluxo de processamento do documento.
"""
# 1) Prepara diretório e caminhos únicos
diretorio = self._prepare_directory()
rtf_path, docx_path = self._generate_paths(diretorio, data.id)
# 2) Descompacta obrigatoriamente o conteúdo
conteudo = self._decompress_content(data.texto)
# Remove apenas espaços iniciais para detecção segura
conteudo_strip = conteudo.lstrip()
# 3) Se após descompressão for DOCX, salva direto
if self._is_docx(conteudo_strip):
self._save_docx(docx_path, conteudo)
return docx_path.name
# 4) Caso contrário, valida se é RTF
self._validate_rtf(conteudo_strip)
# 5) Normaliza e salva RTF (simples ou complexo)
texto_rtf = conteudo_strip.strip()
self._save_rtf(rtf_path, texto_rtf)
# 6) Converte RTF → DOCX via LibreOffice
self._convert_rtf_to_docx(diretorio, rtf_path)
# 7) Valida resultado final
self._validate_docx(docx_path)
# 8) Retorno final
return docx_path.name
# ======================================================
# MÉTODOS PRIVADOS — PREPARAÇÃO
# ======================================================
def _prepare_directory(self) -> Path:
"""
Garante que o diretório temporário exista.
"""
diretorio = Path("./storage/temp").resolve()
diretorio.mkdir(parents=True, exist_ok=True)
return diretorio
def _generate_paths(self, diretorio: Path, document_id) -> tuple[Path, Path]:
"""
Gera nomes de arquivos únicos por execução,
evitando conflitos e lock de arquivos no Windows.
"""
base_name = f"{document_id}_{uuid.uuid4().hex}"
return (
diretorio / f"{base_name}.rtf",
diretorio / f"{base_name}.docx",
)
# ======================================================
# MÉTODOS PRIVADOS — CONTEÚDO
# ======================================================
def _decompress_content(self, texto: bytes | None) -> str:
"""
Descompacta obrigatoriamente o conteúdo recebido.
Caso não exista texto, cria um RTF vazio válido.
"""
if not texto:
return r"{\rtf1\ansi\deff0\pard\par}"
return Text.decompress(texto) or ""
def _is_docx(self, conteudo: str) -> bool:
"""
DOCX é um arquivo ZIP e sempre inicia com PK\\x03\\x04.
"""
return conteudo.startswith("PK\x03\x04")
def _validate_rtf(self, conteudo: str) -> None:
"""
Garante que o conteúdo seja um RTF válido.
"""
if not conteudo.startswith("{\\rtf"):
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Conteúdo descompactado não é RTF nem DOCX válido",
)
# ======================================================
# MÉTODOS PRIVADOS — ESCRITA DE ARQUIVOS
# ======================================================
def _save_docx(self, path: Path, conteudo: str) -> None:
"""
Salva DOCX diretamente em disco.
"""
with open(path, "wb") as f:
# DOCX descompactado ainda está em latin1
f.write(conteudo.encode("latin1"))
def _save_rtf(self, path: Path, texto_rtf: str) -> None:
"""
Salva RTF em disco.
Detecta automaticamente RTFs complexos (WPTools, tabelas, bookmarks, unicode)
para escolher a forma correta de escrita.
"""
rtf_complexo = self._is_complex_rtf(texto_rtf)
if rtf_complexo:
# Escrita binária preserva estruturas complexas
rtf_bytes = texto_rtf.encode("cp1252", errors="ignore")
with open(path, "wb") as f:
f.write(rtf_bytes)
else:
# Escrita textual para RTFs simples
with open(path, "w", encoding="cp1252", errors="replace") as f:
f.write(texto_rtf)
def _is_complex_rtf(self, texto_rtf: str) -> bool:
"""
Detecta padrões comuns de RTFs complexos gerados por WPTools,
OnlyOffice, Word, etc.
"""
return any(
token in texto_rtf
for token in (
"\\u",
"{\\*\\bkmkstart",
"\\trowd",
"\\cellx",
"\\wppr",
)
)
# ======================================================
# MÉTODOS PRIVADOS — CONVERSÃO
# ======================================================
def _convert_rtf_to_docx(self, diretorio: Path, rtf_path: Path) -> None:
"""
Executa o LibreOffice em modo headless para conversão RTF → DOCX.
"""
cmd = [
SOFFICE_PATH,
"--headless",
"--nologo",
"--nolockcheck",
"--nodefault",
"--nofirststartwizard",
"--convert-to",
"docx",
"--outdir",
str(diretorio),
str(rtf_path),
]
clean_env = {
"SYSTEMROOT": os.environ.get("SYSTEMROOT", ""),
"WINDIR": os.environ.get("WINDIR", ""),
"PATH": os.environ.get("PATH", ""),
"TEMP": os.environ.get("TEMP", ""),
"TMP": os.environ.get("TMP", ""),
}
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
timeout=180,
env=clean_env,
)
if result.returncode != 0:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Erro ao converter RTF para DOCX via LibreOffice",
)
# ======================================================
# MÉTODOS PRIVADOS — VALIDAÇÃO FINAL
# ======================================================
def _validate_docx(self, docx_path: Path) -> None:
"""
Garante que o DOCX final realmente foi gerado.
"""
if not docx_path.exists():
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="DOCX não foi gerado pelo LibreOffice",
)

87
actions/data/text.py
View file

@ -1,124 +1,93 @@
# Importa a biblioteca nativa 'zlib' usada para compressão/descompressão de dados binários.
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 'Text', 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):
def decompress_bytes(vf_string):
"""
Descomprime e decodifica texto de origem WPTools/Firebird.
Versão segura contra BLOB inválido (Firebird).
Descomprime e retorna BYTES (sem decode/ignore), ideal para RTF/DOCX.
REGRA:
- Se estiver compactado (zlib) → retorna bytes descompactados
- Se NÃO estiver compactado → retorna os bytes originais
"""
if vf_string is None:
return ""
return b""
# ===============================
# 1) Tentativa segura de leitura
# ===============================
# 1) Se for stream (BLOB)
if hasattr(vf_string, "read"):
try:
vf_string = vf_string.read()
except Exception:
# BLOB inválido, conexão fechada ou handle perdido
return ""
return b""
if not vf_string:
return ""
return b""
# ===============================
# 2) Garantir bytes
# ===============================
if isinstance(vf_string, str):
vf_bytes = vf_string.encode("latin1", errors="ignore")
else:
try:
vf_bytes = bytes(vf_string)
except Exception:
return ""
return b""
# ===============================
# 3) Detectar zlib
# ===============================
# 3) Detectar zlib (header 0x78 0x01/0x9C/0xDA)
is_zlib = (
len(vf_bytes) > 2
and vf_bytes[0] == 0x78
and vf_bytes[1] in (0x01, 0x9C, 0xDA)
)
# ===============================
# 4) Descompactar se necessário
# ===============================
# 4) Descompactar se necessário (RETORNA BYTES)
if is_zlib:
try:
return zlib.decompress(vf_bytes).decode("iso-8859-1", errors="ignore")
return zlib.decompress(vf_bytes)
except Exception:
# Se falhar, tenta como texto puro
pass
# fallback: retorna bytes originais
return vf_bytes
# ===============================
# 5) Texto puro
# ===============================
return vf_bytes
@staticmethod
def decompress(vf_string):
"""
Mantido para compatibilidade: retorna STR (uso geral).
ATENÇÃO: para RTF/DOCX use decompress_bytes().
"""
raw = Text.decompress_bytes(vf_string)
if not raw:
return ""
try:
return vf_bytes.decode("iso-8859-1", errors="ignore")
return raw.decode("iso-8859-1", errors="ignore")
except Exception:
return ""
@staticmethod
def compress(text, *, encoding: str = "iso-8859-1"):
"""
Comprime texto/dados com zlib SEM Base64.
Parâmetros:
text: str | bytes | stream (com .read())
encoding: encoding usado quando 'text' for str (padrão: ISO-8859-1)
Retorno:
- bytes comprimidos (zlib)
Observações:
- Ideal para armazenamento direto em BLOB (Firebird, PostgreSQL, etc.)
- Evita overhead e custo do Base64
- Totalmente compatível com 'decompress'
"""
if text is None or text == "":
return 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 e retorna bytes
return zlib.compress(raw_bytes)
@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'.
- Retornar uma string limpa e pronta para uso.
"""
if not raw_text:
return ""
# Detecta cabeçalho RTF
if raw_text.strip().startswith("{\\rtf"):
try:
return rtf_to_text(raw_text).strip()

23
packages/v1/administrativo/controllers/g_gramatica_controller.py
View file

@ -2,7 +2,7 @@ from actions.dynamic_import.dynamic_import import DynamicImport
from packages.v1.administrativo.schemas.g_gramatica_schema import (
GGramaticaSaveSchema,
GGramaticaUpdateSchema,
GGramaticaIdSchema
GGramaticaIdSchema,
)
@ -24,8 +24,11 @@ class GGramaticaController:
# Lista todos os registros de G_GRAMATICA
# ----------------------------------------------------
def index(self):
# Importação da classe desejada
index_service = self.dynamic_import.service("g_gramatica_index_service", "GGramaticaIndexService")
index_service = self.dynamic_import.service(
"g_gramatica_index_service", "GGramaticaIndexService"
)
# Instância da classe service
self.index_service = index_service()
@ -41,7 +44,9 @@ class GGramaticaController:
# ----------------------------------------------------
def show(self, g_gramatica_id_schema: GGramaticaIdSchema):
# Importação da classe desejada
show_service = self.dynamic_import.service("g_gramatica_show_service", "GGramaticaShowService")
show_service = self.dynamic_import.service(
"g_gramatica_show_service", "GGramaticaShowService"
)
# Instância da classe service
self.show_service = show_service()
@ -57,7 +62,9 @@ class GGramaticaController:
# ----------------------------------------------------
def save(self, g_gramatica_save_schema: GGramaticaSaveSchema):
# Importação da classe desejada
save_service = self.dynamic_import.service("g_gramatica_save_service", "GGramaticaSaveService")
save_service = self.dynamic_import.service(
"g_gramatica_save_service", "GGramaticaSaveService"
)
# Instância da classe service
self.save_service = save_service()
@ -73,7 +80,9 @@ class GGramaticaController:
# ----------------------------------------------------
def update(self, g_gramatica_update_schema: GGramaticaUpdateSchema):
# Importação da classe desejada
update_service = self.dynamic_import.service("g_gramatica_update_service", "GGramaticaUpdateService")
update_service = self.dynamic_import.service(
"g_gramatica_update_service", "GGramaticaUpdateService"
)
# Instância da classe service
self.update_service = update_service()
@ -89,7 +98,9 @@ class GGramaticaController:
# ----------------------------------------------------
def delete(self, g_gramatica_id_schema: GGramaticaIdSchema):
# Importação da classe desejada
delete_service = self.dynamic_import.service("g_gramatica_delete_service", "GGramaticaDeleteService")
delete_service = self.dynamic_import.service(
"g_gramatica_delete_service", "GGramaticaDeleteService"
)
# Instância da classe service
self.delete_service = delete_service()

124
packages/v1/servicos/balcao/actions/t_servico_itempedido/t_servico_itempedido_create_certidao_action.py
View file

@ -1,38 +1,134 @@
import os
import subprocess
from pathlib import Path
from fastapi import HTTPException, status
from packages.v1.servicos.balcao.schemas.t_servico_itempedido_schema import (
TServicoItemPedidoCreateCertidaoSchema,
)
from actions.data.text import Text
# Caminho do LibreOffice (Windows)
SOFFICE_PATH = r"C:\Program Files\LibreOffice\program\soffice.exe"
class TServicoItemPedidoCreateCertidaoAction:
class ProcessDocumentAction:
"""
Gera RTF 100% compatível com OnlyOffice (sem Spire, sem licença).
Trata certidão em RTF ou DOCX.
- RTF -> normaliza e converte para DOCX via LibreOffice
- DOCX -> salva direto, sem conversão
"""
def execute(self, data: TServicoItemPedidoCreateCertidaoSchema):
# ===============================
# 1) Descompacta o texto
# 1) Descompacta o conteúdo
# ===============================
texto = ""
conteudo = ""
# Verifica se existe texto de certidão
if not data.certidao_texto:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Texto do documento vázio",
)
# Verifica se existe o texto
if data.certidao_texto is not None:
texto = Text.decompress(data.certidao_texto) or ""
conteudo = Text.decompress(data.certidao_texto) or ""
# ===============================
# 2) Diretório e nome
# 2) Diretório
# ===============================
diretorio = "./storage/temp"
os.makedirs(diretorio, exist_ok=True)
diretorio = Path("./storage/temp").resolve()
diretorio.mkdir(parents=True, exist_ok=True)
name = f"{data.servico_itempedido_id}.rtf"
caminho_completo = os.path.join(diretorio, name)
base_name = str(data.servico_itempedido_id)
rtf_path = diretorio / f"{base_name}.rtf"
docx_path = diretorio / f"{base_name}.docx"
conteudo_strip = conteudo.lstrip()
# ===============================
# 5) Escrita em disco
# 3) CASO 1 — DOCX (ZIP)
# ===============================
with open(caminho_completo, "wb") as f:
f.write(texto.encode("iso-8859-1", errors="ignore"))
# DOCX é um ZIP e começa com PK\x03\x04
if conteudo_strip.startswith("PK\x03\x04"):
with open(docx_path, "wb") as f:
f.write(conteudo.encode("latin1"))
return name
return docx_path.name
# ===============================
# 4) CASO 2 — RTF
# ===============================
if not conteudo_strip.startswith("{\\rtf"):
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Conteúdo não é um RTF nem um DOCX válido",
)
# Remove partes brancas do texto no início e fim
texto_rtf = conteudo_strip.strip()
# Salva RTF
with open(rtf_path, "w", encoding="cp1252", errors="replace") as f:
f.write(texto_rtf)
# ===============================
# 5) Converte RTF → DOCX (LibreOffice)
# ===============================
cmd = [
SOFFICE_PATH,
"--headless",
"--nologo",
"--nolockcheck",
"--nodefault",
"--nofirststartwizard",
"--convert-to",
"docx",
"--outdir",
str(diretorio),
str(rtf_path),
]
# Ambiente limpo (evita conflito com Python/venv)
clean_env = {
"SYSTEMROOT": os.environ.get("SYSTEMROOT", ""),
"WINDIR": os.environ.get("WINDIR", ""),
"PATH": os.environ.get("PATH", ""),
"TEMP": os.environ.get("TEMP", ""),
"TMP": os.environ.get("TMP", ""),
}
result = subprocess.run(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
timeout=180,
env=clean_env,
)
if result.returncode != 0:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Erro ao converter RTF para DOCX via LibreOffice",
)
# ===============================
# 6) Validação
# ===============================
if not docx_path.exists():
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="DOCX não foi gerado pelo LibreOffice",
)
# ===============================
# 7) Retorno
# ===============================
return docx_path.name

22
packages/v1/servicos/balcao/repositories/t_servico_itempedido/t_servico_itempedido_index_repository.py
View file

@ -66,27 +66,7 @@ class TServicoItemPedidoIndexRepository(BaseRepository):
# ===============================
# 1) Executa a query
# ===============================
raw_rows = self.fetch_all(sql, params)
# ===============================
# 2) Materializa os dados
# ===============================
response = []
for row in raw_rows:
data = dict(row) # converte RowMapping -> dict mutável
blob = data.get("CERTIDAO_TEXTO")
if blob is not None and hasattr(blob, "read"):
try:
# LÊ O BLOB AINDA NA CONEXÃO
data["CERTIDAO_TEXTO"] = blob.read()
except Exception:
# BLOB inválido / handle perdido
data["CERTIDAO_TEXTO"] = None
response.append(data)
response = self.fetch_all(sql, params)
# ===============================
# 3) Retorno seguro

44
packages/v1/servicos/balcao/services/t_servico_itempedido/go/t_servico_itempedido_certidao_save_service.py
View file

@ -2,7 +2,9 @@ from pathlib import Path
import tempfile
import os
import requests
from fastapi import HTTPException, status
from actions.data.text import Text
from packages.v1.servicos.balcao.actions.t_servico_itempedido.t_servico_itempedido_certidao_save_action import (
TServicoItemPedidoCertidaSaveAction,
@ -13,15 +15,27 @@ from packages.v1.servicos.balcao.schemas.t_servico_itempedido_schema import (
class TServicoItemPedidoCertidaoSaveService:
"""
Serviço responsável por receber o callback do OnlyOffice,
persistir o documento no banco e ATUALIZAR o arquivo em disco
de forma segura (atomic write).
O nome do arquivo em disco é baseado em:
data.servico_itempedido_id
"""
def execute(self, data: TServicoItemPedidoCertidaoSaveSchema):
# ----------------------------------------------------
# 1. Apenas status FINAL COM SAVE
# 1. Processa apenas STATUS FINAL (save concluído)
# ----------------------------------------------------
# Status 2 = documento salvo definitivamente no OnlyOffice
if data.data.get("status") != 2:
return {"error": 0}
# ----------------------------------------------------
# 2. Obtém a URL do arquivo final
# ----------------------------------------------------
file_url = data.data.get("url")
if not file_url:
raise HTTPException(
@ -30,7 +44,7 @@ class TServicoItemPedidoCertidaoSaveService:
)
# ----------------------------------------------------
# 2. Download binário
# 3. Download do arquivo (binário)
# ----------------------------------------------------
response = requests.get(file_url, timeout=30)
response.raise_for_status()
@ -44,17 +58,9 @@ class TServicoItemPedidoCertidaoSaveService:
)
# ----------------------------------------------------
# 3. Validação RTF mínima
# ----------------------------------------------------
if not arquivo_bytes.lstrip().startswith(b"{\\rtf"):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Conteúdo retornado não é um RTF válido.",
)
# ----------------------------------------------------
# 4. Persistência no banco
# 4. Persistência no banco (conteúdo comprimido)
# ----------------------------------------------------
# Salva o conteúdo final do DOCX no banco
data.certidao_texto = Text.compress(arquivo_bytes)
save_action = TServicoItemPedidoCertidaSaveAction()
@ -66,21 +72,29 @@ class TServicoItemPedidoCertidaoSaveService:
)
# ----------------------------------------------------
# 5. Escrita em disco (atomic)
# 5. Escrita em disco (ATUALIZAÇÃO ATÔMICA)
# ----------------------------------------------------
destino = Path("./storage/temp")
# Diretório onde ficam os documentos editáveis
destino = Path("./storage/temp").resolve()
destino.mkdir(parents=True, exist_ok=True)
arquivo_final = destino / data.data["key"]
# Nome FINAL do arquivo baseado no ID do domínio
# (sempre o mesmo → sobrescreve o existente)
arquivo_final = destino / f"{data.servico_itempedido_id}.docx"
# Escrita atômica:
# 1) grava em arquivo temporário
# 2) substitui o arquivo final
with tempfile.NamedTemporaryFile(delete=False, dir=destino) as tmp:
tmp.write(arquivo_bytes)
tmp.flush()
temp_name = tmp.name
# Substitui o arquivo antigo pelo novo (atomic)
os.replace(temp_name, arquivo_final)
# ----------------------------------------------------
# 6. Retorno esperado pelo OnlyOffice
# ----------------------------------------------------
# error = 0 indica sucesso no callback
return {"error": 0}

40
packages/v1/servicos/balcao/services/t_servico_itempedido/go/t_servico_itempedido_index_service.py
View file

@ -1,20 +1,18 @@
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 actions.data.process_document_action import ProcessDocumentAction
from packages.v1.servicos.balcao.actions.t_servico_itempedido.t_servico_itempedido_index_action import (
TServicoItemPedidoIndexAction,
)
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,
)
from schemas.process_document_schema import ProcessDocumentSchema
class TServicoItemPedidoIndexService:
@ -54,25 +52,25 @@ class TServicoItemPedidoIndexService:
# Cria o objeto mutável localmente
item = SimpleNamespace(**row)
create_certidao_action = TServicoItemPedidoCreateCertidaoAction()
process_document_action = ProcessDocumentAction()
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,
item.etiqueta_texto = process_document_action.execute(
ProcessDocumentSchema(
id=str(item.servico_itempedido_id),
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,
item.certidao_texto = process_document_action.execute(
ProcessDocumentSchema(
id=str(item.servico_itempedido_id),
texto=item.certidao_texto,
)
)
@ -85,20 +83,20 @@ class TServicoItemPedidoIndexService:
)
# 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,
item.certidao_texto = process_document_action.execute(
ProcessDocumentSchema(
id=str(item.servico_itempedido_id),
texto=ato_show.texto,
)
)
else:
# Cria o documento editavel em disco
item.certidao_texto = create_certidao_action.execute(
TServicoItemPedidoCreateCertidaoSchema(
servico_itempedido_id=item.servico_itempedido_id,
certidao_texto=None,
item.certidao_texto = process_document_action.execute(
ProcessDocumentSchema(
id=str(item.servico_itempedido_id),
texto=None,
)
)

10
schemas/process_document_schema.py Normal file
View file

@ -0,0 +1,10 @@
from typing import Optional
from pydantic.main import BaseModel
class ProcessDocumentSchema(BaseModel):
id: str = None
texto: Optional[bytes] = None
class Config:
from_attributes = True