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

feat(OnlyOffice): Ajusta para trabalhar com rtf

Browse source
This commit is contained in:
Keven 2025-12-29 17:13:18 -03:00
parent ae1cbca20c
commit 659ec645a2
5 changed files with 178 additions and 123 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

55
actions/data/text.py
View file

@ -13,45 +13,62 @@ class Text:
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
Versão segura contra BLOB inválido (Firebird).
"""
# Verifica se o valor recebido é nulo, vazio ou None.
if vf_string is None:
return ""
# ===============================
# 1) Tentativa segura de leitura
# ===============================
if hasattr(vf_string, "read"):
try:
vf_string = vf_string.read()
except Exception:
# BLOB inválido, conexão fechada ou handle perdido
return ""
if not vf_string:
return ""
# Caso seja um objeto tipo stream (ex: campo BLOB do Firebird)
if hasattr(vf_string, "read"):
vf_string = vf_string.read()
# Garante que o valor trabalhado é uma sequência de bytes
# ===============================
# 2) Garantir bytes
# ===============================
if isinstance(vf_string, str):
vf_bytes = vf_string.encode("latin1", errors="ignore")
else:
vf_bytes = vf_string
try:
vf_bytes = bytes(vf_string)
except Exception:
return ""
# Detecta assinatura zlib (0x78 0x9C ou 0x78 0xDA)
# ===============================
# 3) Detectar zlib
# ===============================
is_zlib = (
len(vf_bytes) > 2 and vf_bytes[0] == 0x78 and vf_bytes[1] in (0x9C, 0xDA)
len(vf_bytes) > 2
and vf_bytes[0] == 0x78
and vf_bytes[1] in (0x01, 0x9C, 0xDA)
)
# Se for zlib, tenta descompactar
# ===============================
# 4) Descompactar se necessário
# ===============================
if is_zlib:
try:
return zlib.decompress(vf_bytes).decode("iso-8859-1", errors="ignore")
except Exception:
# Se falhar, tenta como texto puro
pass
# Caso não seja zlib, trata como texto puro
# ===============================
# 5) Texto puro
# ===============================
try:
return vf_bytes.decode("iso-8859-1", errors="ignore")
except Exception:
return str(vf_string)
return ""
@staticmethod
def compress(text, *, encoding: str = "iso-8859-1"):

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

@ -1,68 +1,81 @@
import os
from docx import Document
import pypandoc
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.
Gera arquivo DOCX a partir de texto RTF utilizando pypandoc.
"""
def execute(self, data: TServicoItemPedidoCreateCertidaoSchema):
texto = None
# Verifica se existe conteudo a ser escrito
# ===============================
# 1) Texto puro vindo do sistema
# ===============================
if data.certidao_texto:
# 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
# ===============================
# 2) Diretório e nomes
# ===============================
diretorio = "./storage/temp"
os.makedirs(diretorio, exist_ok=True)
nome_base = str(data.servico_itempedido_id)
caminho_rtf = os.path.join(diretorio, f"{nome_base}.rtf")
caminho_docx = os.path.join(diretorio, f"{nome_base}.docx")
# ===============================
# 3) Escrita do RTF (entrada)
# ===============================
if texto:
# 4. Escreve o texto em disco
with open(caminho_completo, "wb") as f:
f.write(texto.encode("utf-8"))
with open(caminho_rtf, "w", encoding="utf-8") as f:
f.write(texto)
else:
# RTF mínimo vazio (evita falha no pandoc)
with open(caminho_rtf, "w", encoding="utf-8") as f:
f.write(r"{\rtf1\utf8\deff0\pard\par}")
# 1. Instancia o objeto (cria a estrutura XML em memória)
document = Document()
# 2. Salva o arquivo fisicamente
document.save(caminho_completo)
# 5. VALIDAÇÃO: Verifica se o arquivo existe e se tem conteúdo
if not os.path.exists(caminho_completo):
# ===============================
# 4) Conversão RTF → DOCX
# ===============================
try:
pypandoc.convert_file(
caminho_rtf,
to="docx",
outputfile=caminho_docx,
extra_args=["--standalone"],
)
except Exception as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Falha crítica: O arquivo não foi encontrado em {caminho_completo}",
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Erro ao converter RTF para DOCX: {str(e)}",
)
if os.path.getsize(caminho_completo) == 0:
# ===============================
# 5) Validações
# ===============================
if not os.path.exists(caminho_docx):
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Falha crítica: O arquivo foi criado, mas está vazio (0 bytes).",
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Falha crítica: arquivo DOCX não foi criado.",
)
# RETORNO DE SUCESSO
return name
if os.path.getsize(caminho_docx) == 0:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Falha crítica: arquivo DOCX foi criado, mas está vazio.",
)
# ===============================
# 6) Retorno
# ===============================
return f"{nome_base}.docx"

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

@ -6,56 +6,89 @@ from packages.v1.servicos.balcao.schemas.t_servico_itempedido_schema import (
class TServicoItemPedidoIndexRepository(BaseRepository):
"""
Repositório para a operação de listagem de todos os registros
na tabela t_censec_qualidade.
Repositório para listagem de itens do pedido de serviço.
IMPORTANTE:
- Materializa BLOBs (Firebird) ainda na camada de banco
- Nunca retorna objetos BlobReader
- Retorna apenas tipos seguros (dict, bytes, str, int, etc.)
"""
def execute(self, t_servico_itempedido_index_schema: TServicoItemIndexSchema):
"""
Executa a consulta SQL para buscar todos os registros.
Executa a consulta SQL para buscar os registros do pedido.
Returns:
Uma lista de dicionários contendo os dados dos registros.
Retorno:
List[dict] com dados já materializados
"""
sql = """
SELECT
TSP.SERVICO_ITEMPEDIDO_ID,
TSP.EMOLUMENTO_ID,
TSP.EMOLUMENTO_ITEM_ID,
TSP.SERVICO_TIPO_ID,
TSP.TIPO_ITEM,
TSP.QTD,
TSP.EMOLUMENTO,
TSP.TAXA_JUDICIARIA,
TSP.VALOR_ISS,
TSP.FUNDESP,
TSP.VALOR,
TSP.CERTIDAO_TEXTO,
TSP.CERTIDAO_ATO_ID,
TST.DESCRICAO,
TSP.SITUACAO,
TSE.ETIQUETA_MODELO_ID,
GMT.TEXTO AS ETIQUETA_TEXTO,
GMT.GRUPO,
TE.NOME,
TE.CPF_CNPJ,
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
LEFT JOIN G_MARCACAO_TIPO GMT
ON TSE.ETIQUETA_MODELO_ID = GMT.MARCACAO_TIPO_ID
JOIN G_EMOLUMENTO GE
ON TSP.EMOLUMENTO_ID = GE.EMOLUMENTO_ID
LEFT JOIN T_PESSOA TE
ON TSP.PESSOA_ID = TE.PESSOA_ID
WHERE TSP.SERVICO_PEDIDO_ID = :servico_pedido_id
"""
# Montagem do SQL
sql = """ SELECT
TSP.SERVICO_ITEMPEDIDO_ID,
TSP.EMOLUMENTO_ID,
TSP.EMOLUMENTO_ITEM_ID,
TSP.SERVICO_TIPO_ID,
TSP.TIPO_ITEM,
TSP.QTD,
TSP.EMOLUMENTO,
TSP.TAXA_JUDICIARIA,
TSP.VALOR_ISS,
TSP.FUNDESP,
TSP.VALOR,
TSP.CERTIDAO_TEXTO,
TSP.CERTIDAO_ATO_ID,
TST.DESCRICAO,
TSP.SITUACAO,
TSE.ETIQUETA_MODELO_ID,
GMT.TEXTO AS ETIQUETA_TEXTO,
GMT.GRUPO,
TE.NOME,
TE.CPF_CNPJ,
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
LEFT JOIN G_MARCACAO_TIPO GMT ON TSE.ETIQUETA_MODELO_ID = GMT.MARCACAO_TIPO_ID
JOIN G_EMOLUMENTO GE ON TSP.EMOLUMENTO_ID = GE.EMOLUMENTO_ID
LEFT JOIN T_PESSOA TE ON TSP.PESSOA_ID = TE.PESSOA_ID
WHERE SERVICO_PEDIDO_ID = :servico_pedido_id
"""
# preenchimento de parâmetros
params = {
"servico_pedido_id": t_servico_itempedido_index_schema.servico_pedido_id
}
# Execução do sql
response = self.fetch_all(sql, params)
# ===============================
# 1) Executa a query
# ===============================
raw_rows = self.fetch_all(sql, params)
# Retorna os dados localizados
# ===============================
# 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)
# ===============================
# 3) Retorno seguro
# ===============================
return response

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

@ -1,6 +1,3 @@
from pathlib import Path
import tempfile
import os
import requests
from fastapi import HTTPException, status
from actions.data.text import Text
@ -13,11 +10,15 @@ from packages.v1.servicos.balcao.schemas.t_servico_itempedido_schema import (
class TServicoItemPedidoCertidaoSaveService:
"""
Serviço responsável por receber o callback do OnlyOffice,
baixar o DOCX final, validar, salvar no banco e persistir em disco.
"""
def execute(self, data: TServicoItemPedidoCertidaoSaveSchema):
# ----------------------------------------------------
# 1. Apenas status FINAL COM SAVE
# 1. Apenas STATUS FINAL (2 = save concluído)
# ----------------------------------------------------
if data.data.get("status") != 2:
return {"error": 0}
@ -30,10 +31,16 @@ class TServicoItemPedidoCertidaoSaveService:
)
# ----------------------------------------------------
# 2. Download binário
# 2. Download binário (DOCX)
# ----------------------------------------------------
response = requests.get(file_url, timeout=30)
response.raise_for_status()
try:
response = requests.get(file_url, timeout=30)
response.raise_for_status()
except Exception as e:
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Erro ao baixar o arquivo do OnlyOffice: {str(e)}",
)
arquivo_bytes = response.content
@ -44,16 +51,17 @@ class TServicoItemPedidoCertidaoSaveService:
)
# ----------------------------------------------------
# 3. Validação RTF mínima
# 3. Validação mínima de DOCX
# DOCX é um ZIP → começa com PK\x03\x04
# ----------------------------------------------------
if not arquivo_bytes.lstrip().startswith(b"{\\rtf"):
if not arquivo_bytes.startswith(b"PK"):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="Conteúdo retornado não é um RTF válido.",
detail="Conteúdo retornado não é um DOCX válido.",
)
# ----------------------------------------------------
# 4. Persistência no banco
# 4. Persistência no banco (binário comprimido)
# ----------------------------------------------------
data.certidao_texto = Text.compress(arquivo_bytes)
@ -65,21 +73,6 @@ class TServicoItemPedidoCertidaoSaveService:
detail="Erro ao salvar certidão no banco.",
)
# ----------------------------------------------------
# 5. Escrita em disco (atomic)
# ----------------------------------------------------
destino = Path("./storage/temp")
destino.mkdir(parents=True, exist_ok=True)
arquivo_final = destino / data.data["key"]
with tempfile.NamedTemporaryFile(delete=False, dir=destino) as tmp:
tmp.write(arquivo_bytes)
tmp.flush()
temp_name = tmp.name
os.replace(temp_name, arquivo_final)
# ----------------------------------------------------
# 6. Retorno esperado pelo OnlyOffice
# ----------------------------------------------------

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

@ -1,6 +1,5 @@
from types import SimpleNamespace
from docx import Document
from fastapi import HTTPException, status
from packages.v1.servicos.balcao.actions.t_servico_itempedido.t_servico_itempedido_create_certidao_action import (
TServicoItemPedidoCreateCertidaoAction,