FLS/API-Holmes
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Reestrutura para API multi-cliente com payload cifrado

Browse source 
A API deixa de ser especifica da Scardua e passa a atender varios
clientes pela rota /v1/holmes/{cliente}/...

Camada de clientes:
- app/clientes/base.py: ClienteBase (comportamento) + ClienteConfig (dado),
  separados de proposito — config sai do configs.json hoje e pode sair do
  banco amanha sem o resto mudar
- app/clientes/registry.py: resolve o {cliente} da rota pra instancia
- cada cliente traz a propria conta do Holmes; o cache de token virou um
  arquivo por cliente (antes era constante de modulo e dois clientes
  sobrescreveriam o token um do outro)

Holmes:
- funcoes de modulo viraram a classe HolmesAPI, amarrada a uma conta
- exposta pronta em cliente.holmes (cached_property)
- as funcoes de parsing puro (extrair_*) ficaram fora da classe

Config:
- app/schemas.py: modulo folha, so pydantic. Config e services usam os
  mesmos modelos sem um importar o outro
- app/config.py: so o carregamento do configs.json
- configs.json passa a ter "clientes": {"<nome>": {...}}

Compras:
- schemas aceitam o payload real do Holmes (PEDIDO_LINX vem como numero,
  NUMERO NF ENTRADA so existe depois da fase da nota)
- parcelas viram dict[int, Parcela], pareando parcelaN com o vencimento
  correspondente e pulando os slots vazios
- payload de saida carrega id_processo como chave de deduplicacao: o
  Holmes reentrega webhook e sem isso duplica linha no destino

Criptografia:
- app/services/cripto.py: envelope AES-256-GCM + RSA-OAEP-SHA256. RSA
  sozinho nao serve (cifra ~190 bytes; o payload passa disso)
- ciframos com a publica do conector e nao conseguimos desfazer

Deploy:
- Dockerfile e .dockerignore vieram da VPS pro repo
- docs/deploy.md: runbook, incluindo que --format docker e obrigatorio
  (em OCI o podman ignora o HEALTHCHECK silenciosamente)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ricardo 2026-08-18 06:01:01 -03:00
parent 0f47953453
commit 4d47887c00
15 changed files with 1004 additions and 683 deletions
Download patch file Download diff file Expand all files Collapse all files

19
.dockerignore Normal file
View file

@ -0,0 +1,19 @@
# Ambiente virtual / caches (nunca vao pra imagem)
.venv/
__pycache__/
*.pyc
.ruff_cache/
# Git
.git/
.gitignore
# SEGREDOS — nunca bakear na imagem (montar por volume em runtime)
configs.json
.env
certs/
# Ruido
logs/
tests/
*.md

6
.gitignore vendored
View file

@ -6,3 +6,9 @@ __pycache__/
logs/
configs.json
certs/
# Backups de config e chaves — contem segredo, nunca versionar
configs.json.*
*.bak-*
*.pem
*.key

31
Dockerfile Normal file
View file

@ -0,0 +1,31 @@
# ---------------------------------------------------------------------------
# API - Scardua
# Imagem enxuta: thin mode do oracledb (sem Instant Client / sem libs nativas)
# ---------------------------------------------------------------------------
FROM python:3.12-slim
# Nao gera .pyc e nao bufferiza o log (aparece na hora no docker logs)
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1
WORKDIR /app
# Dependencias primeiro para aproveitar o cache de camada do Docker
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Codigo da aplicacao
COPY main.py .
COPY app/ ./app/
EXPOSE 8000
# Healthcheck em exec-form (JSON, sem shell -> sem problema de aspas).
# O podman auto-update usa isto pra decidir rollback.
HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
CMD ["python3", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health').read()"]
# ATENCAO: config/segredos NAO entram na imagem (.dockerignore).
# No deploy sao injetados via Quadlet (Volume do configs.json) -> ver deploy/api-scardua.container.
# Roda com 1 worker (VPS tem 2 vCPU; evitar paralelismo pesado).
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

73
app/clientes/base.py
View file

@ -1,55 +1,31 @@
import os
import tempfile
from abc import ABC, abstractmethod
from typing import Any
from abc import ABC
from functools import cached_property
from fastapi import HTTPException
from app.schemas import HolmesCliente
from app.services.holmes import HolmesAPI
from pydantic import BaseModel
class HolmesCliente(BaseModel):
"""Credenciais do Holmes. Cada cliente tem a propria conta de integracao."""
token_api: str
usuario: str
senha: str
class ClienteConfig(BaseModel):
"""
Os DADOS do cliente.
Fica separado da classe de proposito: dado da pra carregar do banco,
classe nao. No dia que cliente novo deixar de exigir deploy, e daqui
que a config vai sair — o resto do codigo nao muda.
"""
nome: str
razao_social: str
cnpj: str
holmes: HolmesCliente
# Chave publica (PEM) da API do cliente. Ciframos com ela; so a privada,
# que fica com eles, abre.
chave_publica: str
class ClienteBase(ABC):
"""
Base de todo cliente atendido pela API.
# A subclasse atribui direto: config = ClienteConfig(...)
config: ClienteConfig
Regra de ouro: aqui so entra o que muda de COMPORTAMENTO entre clientes.
Valor (CNPJ, token, id de fluxo) vai em `config`, nao em atributo de classe
solto — senao a heranca vira dicionario com passos extras.
A subclasse satisfaz `config` e `modulos` como atributo de classe mesmo,
nao precisa escrever @property.
"""
@property
@abstractmethod
def config(self) -> ClienteConfig: ...
@property
@abstractmethod
def modulos(self) -> dict[str, Any]:
"""Processos habilitados: {"compras": ScarduaCompras(), ...}"""
def __init_subclass__(cls, **kwargs):
super().__init_subclass__(**kwargs)
if not hasattr(cls, "config"):
raise TypeError(f"{cls.__name__} precisa definir `config`")
# --- comum a todos: ninguem sobrescreve ---
@ -57,13 +33,18 @@ class ClienteBase(ABC):
def nome(self) -> str:
return self.config.nome
def modulo(self, nome: str) -> Any:
"""Pega um processo habilitado do cliente, ou 404 se ele nao tem."""
if modulo := self.modulos.get(nome):
return modulo
raise HTTPException(
status_code=404,
detail=f"modulo '{nome}' nao habilitado para {self.nome}",
@cached_property
def holmes(self) -> HolmesAPI:
"""
API do Holmes ja amarrada nas credenciais deste cliente.
cached_property e nao property: uma instancia por cliente, criada no
primeiro uso. Com @property comum sairia uma HolmesAPI nova a cada
acesso, e qualquer estado em memoria (token) se perderia.
"""
return HolmesAPI(
credenciais=self.config.holmes,
cache_token=self.cache_token_holmes,
)
@property
@ -71,8 +52,8 @@ class ClienteBase(ABC):
"""
Arquivo de cache do token do Holmes, um por cliente.
Sem isso dois clientes com contas diferentes sobrescrevem o token
um do outro (ver _TOKEN_FILE em app/services/holmes.py).
Sem isso dois clientes com contas diferentes sobrescreveriam o token
um do outro — era o caso quando isso era uma constante de modulo.
"""
return os.path.join(
tempfile.gettempdir(), f"holmes_token_{self.nome}.json"

41
app/config.py
View file

@ -1,45 +1,6 @@
from pathlib import Path
from pydantic import BaseModel
class BancoConfig(BaseModel):
user: str
password: str
dns: str
instant_client: str
class HolmesConfig(BaseModel):
token_api: str
usuario: str
senha: str
class ApolloConfig(BaseModel):
subscription_key: str
ambiente: str
class ApiConfig(BaseModel):
port: int
ambiente: str
workers: int
class DocsConfig(BaseModel):
user: str
password: str
class Settings(BaseModel):
banco: BancoConfig
holmes: HolmesConfig
api: ApiConfig
apollo: ApolloConfig
docs: DocsConfig
log_level: str = "INFO"
from app.schemas import Settings
_PATH = Path(__file__).resolve().parent.parent / "configs.json"
settings = Settings.model_validate_json(_PATH.read_text(encoding="utf-8"))

32
app/schemas.py Normal file
View file

@ -0,0 +1,32 @@
from pydantic import BaseModel
class HolmesCliente(BaseModel):
"""Credenciais do Holmes. Cada cliente tem a propria conta de integracao."""
token_api: str
usuario: str
senha: str
class ApiConfig(BaseModel):
port: int
ambiente: str
workers: int
class DocsConfig(BaseModel):
user: str
password: str
class ClienteSettings(BaseModel):
holmes: HolmesCliente
chave_publica: str
class Settings(BaseModel):
api: ApiConfig
clientes: dict[str, ClienteSettings]
docs: DocsConfig
log_level: str = "INFO"

99
app/services/cripto.py Normal file
View file

@ -0,0 +1,99 @@
"""
Criptografia do payload enviado pra API do cliente.
Envelope (hibrida): AES-256-GCM nos dados, RSA-OAEP na chave AES.
Por que nao RSA puro: RSA-2048 com OAEP-SHA256 cifra no maximo 190 bytes.
Um payload de compras passa disso facil. O padrao — mesmo do TLS, PGP e JWE —
e sortear uma chave simetrica por mensagem, cifrar os dados com ela e mandar
essa chave cifrada com a publica do destinatario.
So o dono da chave privada abre. Nos ciframos e nao conseguimos desfazer,
que era o objetivo.
"""
import base64
import json
import os
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
ALGORITMO = "RSA-OAEP-256+A256GCM"
_OAEP = padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None,
)
def carregar_chave_publica(pem: str):
"""Le a chave publica do cliente (formato PEM, vinda do configs.json)."""
return serialization.load_pem_public_key(pem.encode())
def carregar_chave_privada(pem: str, senha: bytes | None = None):
"""So usada em teste. Em producao a privada fica com o cliente."""
return serialization.load_pem_private_key(pem.encode(), password=senha)
def criptografar(dados: dict, chave_publica) -> dict:
"""
Cifra o dict e devolve o envelope pronto pra mandar no corpo do POST.
Sai assim:
{
"alg": "RSA-OAEP-256+A256GCM",
"chave": "<chave AES cifrada com a publica, base64>",
"nonce": "<nonce do GCM, base64>",
"dados": "<json cifrado + tag de autenticacao, base64>"
}
"""
chave_aes = AESGCM.generate_key(bit_length=256)
nonce = os.urandom(12)
corpo = json.dumps(dados, ensure_ascii=False, separators=(",", ":")).encode()
cifrado = AESGCM(chave_aes).encrypt(nonce, corpo, None)
chave_cifrada = chave_publica.encrypt(chave_aes, _OAEP)
return {
"alg": ALGORITMO,
"chave": base64.b64encode(chave_cifrada).decode(),
"nonce": base64.b64encode(nonce).decode(),
"dados": base64.b64encode(cifrado).decode(),
}
def descriptografar(envelope: dict, chave_privada) -> dict:
"""
O lado do cliente. Fica aqui pra testar o round-trip e pra servir de
referencia de implementacao pra quem for escrever a API que recebe.
"""
chave_aes = chave_privada.decrypt(base64.b64decode(envelope["chave"]), _OAEP)
corpo = AESGCM(chave_aes).decrypt(
base64.b64decode(envelope["nonce"]),
base64.b64decode(envelope["dados"]),
None,
)
return json.loads(corpo)
def gerar_par_de_chaves() -> tuple[str, str]:
"""Gera um par RSA-2048 em PEM. Util pra homologacao: (privada, publica)."""
chave = rsa.generate_private_key(public_exponent=65537, key_size=2048)
privada = chave.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
).decode()
publica = (
chave.public_key()
.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
)
.decode()
)
return privada, publica

274
app/services/holmes.py
View file

@ -4,83 +4,14 @@ import logging
import os
import random
import re
import tempfile
import time
import httpx
from app.config import settings
from app.schemas import HolmesCliente
from app.services.controle_api import registrar_contador
from oracledb import Connection
_TOKEN_FILE = os.path.join(tempfile.gettempdir(), "holmes_token_cache.json")
def _ler_token_arquivo() -> dict | None:
try:
with open(_TOKEN_FILE) as f:
return json.load(f)
except Exception:
return None
def _salvar_token_arquivo(token: dict):
try:
with open(_TOKEN_FILE, "w") as f:
json.dump(token, f)
except Exception:
pass
def _invalidar_token_usuario():
try:
os.remove(_TOKEN_FILE)
except Exception:
pass
async def token_usuario():
cached = _ler_token_arquivo()
if cached:
return cached
await asyncio.sleep(random.uniform(0, 0.5))
cached = _ler_token_arquivo()
if cached:
return cached
url = "https://app-api.holmesdoc.io/v1/session"
body = {"email": settings.holmes.usuario , "password": settings.holmes.senha}
headers = {"Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36"}
async with httpx.AsyncClient() as client:
response = await client.post(url, json=body, headers=headers)
response.raise_for_status()
token = {"authorization": response.json()["token"]}
_salvar_token_arquivo(token)
return token
async def get_header() -> tuple[dict, str]:
if random.randint(1, 1) >= 2:
return {"api_token": settings.holmes.token_api}, "holmes"
header = await token_usuario()
return header, "holmes_user"
async def _holmes_request(
method: str, url: str, **kwargs
) -> tuple[httpx.Response, str]:
api_nome: str = "holmes"
for tentativa in range(2):
header, api_nome = await get_header()
async with httpx.AsyncClient() as client:
response = await getattr(client, method)(url, headers=header, **kwargs)
if (
response.status_code == 401
and api_nome == "holmes_user"
and tentativa == 0
):
_invalidar_token_usuario()
continue
return response, api_nome
raise RuntimeError("Holmes: falha de autenticação após retry")
# Ricardinho aqui, como esse projeto vai ser para varios clientes tem que alterar a logica desse arquivo, colocar em clase que um cliente herda
def extrair_origem(origem: str):
@ -123,7 +54,82 @@ def extrair_transacao(transacao: str) -> str:
return transacao[:3]
async def get_holmes_process(id_processo: str, conn: Connection):
class HolmesAPI:
"""
Cliente da API do Holmes, amarrado a UMA conta de integracao.
Nao importa nada de app.clientes de proposito: quem conhece o cliente e a
camada de cima. Aqui so entram as credenciais e o caminho do cache.
Instanciar por `cliente.holmes` (ver app/clientes/base.py).
"""
def __init__(self, credenciais: HolmesCliente, cache_token: str):
self.cfg = credenciais
self.cache_token = cache_token
def _ler_token_arquivo(self) -> dict | None:
try:
with open(self.cache_token) as f:
return json.load(f)
except Exception:
return None
def _salvar_token_arquivo(self, token: dict):
try:
with open(self.cache_token, "w") as f:
json.dump(token, f)
except Exception:
pass
def _invalidar_token_usuario(self):
try:
os.remove(self.cache_token)
except Exception:
pass
async def token_usuario(self):
cached = self._ler_token_arquivo()
if cached:
return cached
await asyncio.sleep(random.uniform(0, 0.5))
cached = self._ler_token_arquivo()
if cached:
return cached
url = "https://app-api.holmesdoc.io/v1/session"
body = {"email": self.cfg.usuario , "password": self.cfg.senha}
headers = {"Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36"}
async with httpx.AsyncClient() as client:
response = await client.post(url, json=body, headers=headers)
response.raise_for_status()
token = {"authorization": response.json()["token"]}
self._salvar_token_arquivo(token)
return token
async def get_header(self) -> tuple[dict, str]:
if random.randint(1, 1) >= 2:
return {"api_token": self.cfg.token_api}, "holmes"
header = await self.token_usuario()
return header, "holmes_user"
async def _holmes_request(
self, method: str, url: str, **kwargs
) -> tuple[httpx.Response, str]:
api_nome: str = "holmes"
for tentativa in range(2):
header, api_nome = await self.get_header()
async with httpx.AsyncClient() as client:
response = await getattr(client, method)(url, headers=header, **kwargs)
if (
response.status_code == 401
and api_nome == "holmes_user"
and tentativa == 0
):
self._invalidar_token_usuario()
continue
return response, api_nome
raise RuntimeError("Holmes: falha de autenticação após retry")
async def get_holmes_process(self, id_processo: str, conn: Connection):
"""
Busca um processo no Holmes. Centralizado para Peças e Despesas.
"""
@ -135,7 +141,7 @@ async def get_holmes_process(id_processo: str, conn: Connection):
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response, api_nome = await self._holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
@ -162,8 +168,7 @@ async def get_holmes_process(id_processo: str, conn: Connection):
erro=erro,
)
async def get_holmes_process_details(id_processo: str, conn: Connection):
async def get_holmes_process_details(self, id_processo: str, conn: Connection):
"""
Busca os detalhes (properties) de um processo no Holmes.
"""
@ -175,7 +180,7 @@ async def get_holmes_process_details(id_processo: str, conn: Connection):
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response, api_nome = await self._holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
@ -202,8 +207,7 @@ async def get_holmes_process_details(id_processo: str, conn: Connection):
erro=erro,
)
async def get_holmes_history(id_processo: str, conn: Connection):
async def get_holmes_history(self, id_processo: str, conn: Connection):
"""
Busca o historico no holmes (importante para pegar o id da ultima task, para avançar futuramente)
Args:
@ -223,7 +227,7 @@ async def get_holmes_history(id_processo: str, conn: Connection):
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response, api_nome = await self._holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
@ -250,8 +254,7 @@ async def get_holmes_history(id_processo: str, conn: Connection):
erro=erro,
)
async def get_holmes_rateio(id_processo: str, conn: Connection):
async def get_holmes_rateio(self, id_processo: str, conn: Connection):
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/tables/e124b2d0-ee14-11ef-95b4-25dee32fe73f/table_items?page=1&per_page=800"
inicio = time.perf_counter()
status = None
@ -260,7 +263,7 @@ async def get_holmes_rateio(id_processo: str, conn: Connection):
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response, api_nome = await self._holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
@ -287,9 +290,8 @@ async def get_holmes_rateio(id_processo: str, conn: Connection):
erro=erro,
)
async def task_id_recente(id_processo: str, conn: Connection):
dados_tasks = await get_holmes_history(id_processo, conn)
async def task_id_recente(self, id_processo: str, conn: Connection):
dados_tasks = await self.get_holmes_history(id_processo, conn)
if (
not dados_tasks
@ -302,9 +304,8 @@ async def task_id_recente(id_processo: str, conn: Connection):
return mais_recente["properties"]["task_id"]
async def task_mais_recente(id_processo: str, conn: Connection):
dados_tasks = await get_holmes_history(id_processo, conn)
async def task_mais_recente(self, id_processo: str, conn: Connection):
dados_tasks = await self.get_holmes_history(id_processo, conn)
if not dados_tasks or "histories" not in dados_tasks:
return None # Tratamento se a API falhar
@ -315,8 +316,7 @@ async def task_mais_recente(id_processo: str, conn: Connection):
return mais_recente
async def historicos_task(id_processo: str, conn: Connection) -> dict | None:
async def historicos_task(self, id_processo: str, conn: Connection) -> dict | None:
"""_summary_
Args:
@ -325,20 +325,20 @@ async def historicos_task(id_processo: str, conn: Connection) -> dict | None:
Returns:
dict | None : dicionario do historico | None
"""
dados_tasks = await get_holmes_history(id_processo, conn)
dados_tasks = await self.get_holmes_history(id_processo, conn)
if not dados_tasks or "histories" not in dados_tasks:
return None # Tratamento se a API falhar
return dados_tasks
async def buscar_processo(
async def buscar_processo(
self,
conn: Connection,
chave: str | None = None,
fluxos: list[str] | bool = False,
ativos: bool = True,
payload: dict | bool = False,
) -> dict:
) -> dict:
"""Obtem os processos que existem com a sua chave
Args:
@ -387,7 +387,7 @@ async def buscar_processo(
"deleted_by_me": False,
}
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response, api_nome = await self._holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
@ -420,13 +420,13 @@ async def buscar_processo(
erro=erro,
)
async def buscar_processo_por_chaves(
async def buscar_processo_por_chaves(
self,
conn: Connection,
combinacoes: list[list[str]],
fluxos: list[str] | bool = False,
ativos: bool = True,
) -> dict:
) -> dict:
"""Busca processos no Holmes usando combinações de termos.
Cada item de combinacoes é uma lista de valores que juntos identificam
@ -468,7 +468,7 @@ async def buscar_processo_por_chaves(
}
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response, api_nome = await self._holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
@ -501,8 +501,9 @@ async def buscar_processo_por_chaves(
erro=erro,
)
async def action(payload: dict, id_task: str, id_processo: str, conn: Connection):
async def action(
self, payload: dict, id_task: str, id_processo: str, conn: Connection
):
url = f"https://app-api.holmesdoc.io/v1/tasks/{id_task}/action"
inicio = time.perf_counter()
status = None
@ -513,7 +514,7 @@ async def action(payload: dict, id_task: str, id_processo: str, conn: Connection
try:
async with httpx.AsyncClient() as client:
response = await client.post(
url, headers={"api_token": settings.holmes.token_api}, json=payload
url, headers={"api_token": self.cfg.token_api}, json=payload
)
response.raise_for_status()
status = response.status_code
@ -541,12 +542,9 @@ async def action(payload: dict, id_task: str, id_processo: str, conn: Connection
erro=erro,
)
async def cria_processo(
id_start: str,
payload: dict,
conn: Connection
) -> tuple[bool, dict | str]:
async def cria_processo(
self, id_start: str, payload: dict, conn: Connection
) -> tuple[bool, dict | str]:
url = f"https://app-api.holmesdoc.io/v1/workflows/{id_start}/start"
inicio = time.perf_counter()
status = None
@ -555,7 +553,7 @@ async def cria_processo(
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response, api_nome = await self._holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
@ -582,15 +580,15 @@ async def cria_processo(
erro=erro,
)
async def enviar_documento(
async def enviar_documento(
self,
id_processo: str,
arquivo: bytes,
nome_arquivo: str,
id_documento: str,
conn: Connection,
) -> tuple[bool, str | dict]:
task_id = await task_id_recente(id_processo, conn)
) -> tuple[bool, str | dict]:
task_id = await self.task_id_recente(id_processo, conn)
if not task_id:
return False, "Não foi possível obter a task mais recente do Holmes"
@ -606,7 +604,7 @@ async def enviar_documento(
files = {"file": (nome_arquivo, arquivo, "application/pdf")}
async with httpx.AsyncClient() as client:
response = await client.post(
url, headers={"api_token": settings.holmes.token_api}, files=files
url, headers={"api_token": self.cfg.token_api}, files=files
)
response.raise_for_status()
status = response.status_code
@ -634,26 +632,9 @@ async def enviar_documento(
erro=erro,
)
# Para testar as funcoes
# async def main():
# from app.infra.database import db_instance
# db_instance.create_pool()
# conn = db_instance.pool.acquire()
# try:
# print(await get_holmes_history('69e65ea83fad950fad5715ed', conn))
# finally:
# db_instance.pool.release(conn)
# if __name__ == '__main__':
# import asyncio
# asyncio.run(main())
async def cancela_processo(
id_processo: str, conn: Connection
) -> tuple[bool, str | dict]:
async def cancela_processo(
self, id_processo: str, conn: Connection
) -> tuple[bool, str | dict]:
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/cancel"
payload = {"reason": "Erro na emissão, data de vencimento. Problema na Disal."}
inicio = time.perf_counter()
@ -663,7 +644,7 @@ async def cancela_processo(
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("put", url, json=payload)
response, api_nome = await self._holmes_request("put", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
@ -693,9 +674,28 @@ async def cancela_processo(
)
# Para testar as funcoes
# async def main():
# from app.clientes.registry import obter
# from app.infra.database import db_instance
# db_instance.create_pool()
# conn = db_instance.pool.acquire()
# try:
# holmes = obter('scardua').holmes
# print(await holmes.get_holmes_history('69e65ea83fad950fad5715ed', conn))
# finally:
# db_instance.pool.release(conn)
# if __name__ == '__main__':
# import asyncio
# asyncio.run(main())
async def main():
# print(aaaa())
print(await token_usuario())
from app.clientes.registry import obter
print(await obter("scardua").holmes.token_usuario())
if __name__ == "__main__":

7
app/v1/api.py
View file

@ -1,9 +1,12 @@
from app.v1.holmes.holmes import holmes_router
import app.v1.scardua.cliente # noqa: F401 (registra o Scardua no registry)
from app.v1.scardua.scardua import holmes_router
from fastapi import APIRouter
api_router = APIRouter()
# O {cliente} vira path param e cai no get_cliente por Depends.
# Rota final: POST /v1/holmes/scardua/compras/dados
api_router.include_router(
holmes_router,
prefix='/holmes'
prefix="/holmes/{cliente}",
)

16
app/v1/scardua/cliente.py Normal file
View file

@ -0,0 +1,16 @@
from app.clientes.base import ClienteBase, ClienteConfig
from app.clientes.registry import registrar
from app.config import settings
class Scardua(ClienteBase):
config = ClienteConfig(
nome="scardua",
razao_social="Comercial Scardua",
cnpj="",
holmes=settings.clientes["scardua"].holmes,
chave_publica=settings.clientes["scardua"].chave_publica,
)
registrar(Scardua())

11
app/v1/scardua/modules/compras/router.py
View file

@ -1,4 +1,6 @@
from fastapi import APIRouter
from app.clientes.base import ClienteBase
from app.clientes.registry import get_cliente
from fastapi import APIRouter, Depends
from .schemas import DadosCompras
from .service import processar_compras
@ -6,5 +8,8 @@ from .service import processar_compras
router = APIRouter(prefix="/compras", tags=["compras"])
@router.post("/dados")
async def dados_compras(dados: DadosCompras):
return await processar_compras(dados)
async def dados_compras(
dados: DadosCompras,
cli: ClienteBase = Depends(get_cliente),
):
return await processar_compras(dados, cli)

24
app/v1/scardua/modules/compras/schemas.py
View file

@ -1,6 +1,6 @@
from datetime import datetime
from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field
class Parcela(BaseModel):
@ -8,13 +8,24 @@ class Parcela(BaseModel):
vencimento: datetime | None = None
class Autor(BaseModel):
id: str
name: str
email: str
class ComprasProperties(BaseModel):
# O Holmes manda PEDIDO_LINX como numero (345, sem aspas). O Pydantic v2
# nao converte numero pra string sozinho, entao precisa habilitar.
model_config = ConfigDict(coerce_numbers_to_str=True)
protocolo: str
pedido_linx: str = Field(alias="PEDIDO_LINX")
fornecedor: str = Field(alias="Fornecedores")
tipo_compra: str = Field(alias="Tipo de Compra")
cnpj: str = Field(alias="CNPJ")
nf_entrada: str = Field(alias="NUMERO NF ENTRADA")
# Chega vazio enquanto o processo nao passou pela fase da nota.
nf_entrada: str | None = Field(default=None, alias="NUMERO NF ENTRADA")
num_parcelas: int | None = Field(default=None, alias="num. parcelas")
@ -64,14 +75,21 @@ class ComprasProperties(BaseModel):
class DadosCompras(BaseModel):
id: str
author: Autor
properties: ComprasProperties
class PayloadCompras(BaseModel):
# Chave de deduplicacao: o id do processo no Holmes. Como ele reentrega
# webhook, sem isso todo reenvio vira linha duplicada no banco do cliente.
id_processo: str
protocolo: str
cnpj: str
pedido_linx: str
fornecedor: str
tipo: str
nf_entrada: str
nf_entrada: str | None
aprovador: str
valor_total: float
parcelas: dict[int, Parcela]

41
app/v1/scardua/modules/compras/service.py
View file

@ -1,26 +1,41 @@
from app.clientes.base import ClienteBase
from app.services import cripto
from .schemas import DadosCompras, PayloadCompras
from .service import processar_compras
from app.services.api_cliente import
from app.services.holmes import
async def processar_compras(
dados: DadosCompras
) -> dict:
payload = montar_payload(dados)
ok, retorno = await apollo.enviar_compra(payload.model_dump(mode="json"), conn)
if not ok:
raise HTTPException(502, f"Falha ao enviar para a API externa: {retorno}")
return {"status": "enviado", "id_processo": dados.id, "retorno": retorno}
def montar_payload(dados: DadosCompras) -> PayloadCompras:
"""
Traduz o formato do Holmes pro formato que a API do cliente espera.
Funcao pura, sem I/O: da pra testar jogando o JSON do Holmes e conferindo
a saida, sem rede nem mock.
"""
p = dados.properties
return PayloadCompras(
id_processo=dados.id,
protocolo=p.protocolo,
cnpj=p.cnpj,
pedido_linx=p.pedido_linx,
fornecedor=p.fornecedor,
tipo=p.tipo_compra,
nf_entrada=p.nf_entrada,
aprovador=dados.author.name,
valor_total=p.total_parcelas,
parcelas=p.parcelas,
)
def montar_envelope(dados: DadosCompras, cli: ClienteBase) -> dict:
"""Monta o payload e cifra com a chave publica do cliente."""
payload = montar_payload(dados)
chave = cripto.carregar_chave_publica(cli.config.chave_publica)
return cripto.criptografar(payload.model_dump(mode="json"), chave)
async def processar_compras(dados: DadosCompras, cli: ClienteBase) -> dict:
envelope = montar_envelope(dados, cli)
# TODO: POST do envelope pra API do cliente. Erro tem que subir (nao
# engolir), pra o Holmes reentregar — ele e a nossa fila.
return {"status": "pendente_envio", "id_processo": dados.id, "envelope": envelope}

115
docs/deploy.md Normal file
View file

@ -0,0 +1,115 @@
# Deploy — API (VPS da FLS)
Ambiente: **VPS Hostinger** (Debian 13), **Podman rootless + Quadlet**, **Caddy**
(proxy + TLS), user `admin`.
No ar: **https://api-scardua.flstecnologia.tech**
> ℹ️ O nome `api-scardua` é **legado** — a API hoje atende múltiplos clientes
> (`/v1/holmes/{cliente}/...`). Renomear implica mexer junto em: nome do quadlet,
> `Image=`, `ContainerName=`, pasta do `configs.json`, build dir, bloco do
> Caddyfile e DNS. O `reverse_proxy` do Caddy resolve o container pelo
> `ContainerName` dentro da rede `fls` — se um mudar sem o outro, dá **502**.
## Acesso à VPS
```bash
ssh vps-fls # alias -> 179.197.230.154, porta 3115, user admin
```
- Porta 22 é **fechada** no firewall. Sempre `-p 3115` (ou o alias).
- `systemctl --user` sempre (rootless) — **nunca** `sudo systemctl` pros containers.
- ⚠️ O `~/.ssh/config` tem um **BOM** na primeira linha e o OpenSSH do Git Bash
recusa (`Bad configuration option: \357\273\277host`). Contorno sem editar o
arquivo:
```bash
ssh -F none -i ~/.ssh/id_ed25519 -p 3115 admin@179.197.230.154
```
## Arquitetura
- Imagem: buildada **local na VPS** = `localhost/api-scardua:latest` (sem registry).
- Container na rede **`fls`** — só o Caddy alcança; a porta 8000 **não** é publicada.
- Config: `configs.json` montado read-only de `~/.config/api-scardua/configs.json`.
- Quadlet: `~/.config/containers/systemd/api-scardua.container`.
- Caddy: `api-scardua.flstecnologia.tech` → `api-scardua:8000`, TLS automático.
Caddyfile em `/srv/containers/stacks/caddy/Caddyfile`.
## Atualizar a app
```bash
# 1. marcar rollback ANTES de qualquer coisa
ssh vps-fls 'podman tag localhost/api-scardua:latest localhost/api-scardua:rollback-$(date +%F)'
# 2. mandar o código (não tem rsync no Git Bash do Windows; tar resolve)
tar czf - --exclude='__pycache__' --exclude='*.pyc' --exclude='.ruff_cache' \
main.py app requirements.txt Dockerfile .dockerignore pyproject.toml \
| ssh vps-fls 'rm -rf ~/build/api-scardua && mkdir -p ~/build/api-scardua && tar xzf - -C ~/build/api-scardua'
# 3. buildar e reiniciar
ssh vps-fls 'cd ~/build/api-scardua && podman build --format docker -t api-scardua:latest . \
&& systemctl --user restart api-scardua'
```
> ⚠️ **`--format docker` é obrigatório.** No formato OCI (padrão do Podman) o
> `HEALTHCHECK` do Dockerfile é **silenciosamente ignorado** — e é ele que o
> `podman auto-update` usa pra decidir rollback. Conferir depois do build:
> ```bash
> podman inspect localhost/api-scardua:latest --format '{{json .HealthCheck}}'
> ```
> ⚠️ Diretórios da build antiga podem estar **sem bit de escrita** (`dr-x------`)
> e travar o `rm -rf` no meio. Se acontecer: `chmod -R u+rwX ~/build/api-scardua`.
## O configs.json da VPS é separado
Ele **não** vem do repo (gitignored). Vive em `~/.config/api-scardua/configs.json`
e tem que ser atualizado à mão quando o formato do `Settings` mudar — senão o
container entra em **crash-loop** na validação do Pydantic, que roda no import.
```bash
cat configs.json | ssh vps-fls 'cat > ~/.config/api-scardua/configs.json && chmod 600 $_'
```
## Rollback
```bash
ssh vps-fls '
podman tag localhost/api-scardua:rollback-<data> localhost/api-scardua:latest
systemctl --user restart api-scardua
'
```
## Mexer no Caddy (com segurança)
```bash
cd /srv/containers/stacks/caddy
cp Caddyfile Caddyfile.bak
# editar...
podman exec caddy caddy validate --config /etc/caddy/Caddyfile # valida ANTES
podman exec caddy caddy reload --config /etc/caddy/Caddyfile # sem downtime
```
## Comandos úteis
```bash
systemctl --user status api-scardua
podman logs -f api-scardua
podman ps --filter name=api-scardua # estado + health
podman healthcheck run api-scardua
```
## Verificar depois de subir
```bash
curl -s -o /dev/null -w '%{http_code}\n' https://api-scardua.flstecnologia.tech/health
curl -X POST https://api-scardua.flstecnologia.tech/v1/holmes/scardua/compras/dados \
-H 'Content-Type: application/json' -d @payload-do-holmes.json
```
Cliente desconhecido na rota deve dar **404** (`cliente 'x' nao atendido`).
## Evolução planejada
- **Registry + CI** (Forgejo Actions): publicar em `git.flstecnologia.tech`, trocar
o Quadlet pro alvo com `AutoUpdate=registry` → deploy pull-based.
- Backup do Postgres da VPS (pendência geral do servidor).

34
requirements.txt
View file

@ -1,7 +1,27 @@
# Dependencias de runtime da API (espelham o [project.dependencies] do pyproject.toml)
# Usado pelo Dockerfile para evitar o build do pacote (que dependeria de README.md)
fastapi>=0.115
uvicorn[standard]>=0.34
pydantic>=2.0
httpx>=0.27
oracledb>=2.0
# gerado em 2026-08-16 11:39 | origem: pip freeze | python 3.13.13
annotated-doc==0.0.5
annotated-types==0.8.0
anyio==4.14.2
certifi==2026.7.22
cffi==2.1.1
click==8.4.2
colorama==0.4.6
cryptography==50.0.0
fastapi==0.141.1
h11==0.16.0
httpcore==1.0.9
httptools==0.8.0
httpx==0.28.1
idna==3.18
oracledb==4.0.2
pycparser==3.0
pydantic==2.13.4
pydantic_core==2.46.4
python-dotenv==1.2.2
PyYAML==6.0.3
starlette==1.3.1
typing-inspection==0.4.2
typing_extensions==4.16.0
uvicorn==0.52.1
watchfiles==1.2.0
websockets==17.0.1