Comercial-Scardua/API-Scardua
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Deploy inicial: API FastAPI + infra Podman/Caddy + docs

Browse source 
- App FastAPI (main.py, app/): rotas /health e /dados_retorno, middleware anti-scanner, logging configurado
- Config via configs.json (Pydantic); segredos fora do codigo e do git (.gitignore)
- Dockerfile (oracledb thin, HEALTHCHECK) + requirements.txt + .dockerignore
- Deploy Podman/Quadlet + Caddy em deploy/
- Docs: CLAUDE.md + docs/ (arquitetura, deploy, known-issues)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Ricardo Leite 2026-07-28 16:37:13 -03:00
commit 15c89c9038
22 changed files with 1397 additions and 0 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

8
.gitignore vendored Normal file
View file

@ -0,0 +1,8 @@
.venv/
.ruff_cache
__pycache__/
*.pyc
.env
logs/
configs.json
certs/

45
CLAUDE.md Normal file
View file

@ -0,0 +1,45 @@
# API-Scardua
API FastAPI de integração da Scardua com serviços externos (Holmes, Apollo) e banco Oracle.
No ar: **https://api-scardua.flstecnologia.tech**
## Visão e roadmap
Ideia, arquitetura e roadmap completos: **[docs/arquitetura.md](docs/arquitetura.md)**.
Resumo: a API roda na **nossa VPS (FLS)**, não na do cliente — recebe dados do Holmes do cliente (Scardua), trata, e devolve pro Oracle dele via um **conector no server do cliente** (o Oracle está em rede privada `10.16.x`, inalcançável direto da VPS).
## Stack
- Python ≥3.10 · FastAPI + Uvicorn (ASGI)
- Pydantic v2 (validação e config)
- httpx (chamadas às APIs externas)
- oracledb — driver Oracle em **thin mode** (sem Instant Client)
- Deps em `pyproject.toml` (extra `dev`: pytest, ruff). `requirements.txt` = só runtime (usado no Dockerfile).
## Estrutura
- `main.py` — app FastAPI, middleware, rotas raiz (`/health`, `/dados_retorno`, `/docs` protegido), config de logging
- `app/config.py` — carrega `configs.json` (Pydantic) **no import** (precisa existir ou a app não sobe)
- `app/middleware.py` — `block_scanners`: denylist + heurística anti-scanner (NÃO é allowlist estrita)
- `app/security.py` — JWT + auth via Oracle (⚠️ incompleto — ver docs/known-issues.md)
- `app/services/holmes.py` — integração com a API do Holmes
- `app/services/controle_api.py` — grava telemetria/contadores no Oracle
- `app/v1/` — router versionado (`/v1/...`)
## Rodar local (Windows)
Precisa do `configs.json` na raiz (NÃO versionado — ver `configs.example.json`).
```
uvicorn main:app --reload --port 8000
```
## Config e segredos
- Tudo vem de `configs.json`, lido em `app/config.py`. Está no `.gitignore` e `.dockerignore` — **nunca commitar**.
- Template seguro e versionável: `configs.example.json`.
## Deploy
Container Podman na VPS (Debian 13, rootless + Quadlet + Caddy). Runbook: **[docs/deploy.md](docs/deploy.md)**.
## Convenções
- Código, nomes e comentários em **português**.
- Lint/format: **ruff** (config no `pyproject.toml` — linha 120, aspas duplas).
- Logging: use `logging.info/error`; o nível é configurado em `main.py` via `settings.log_level`.
## Problemas conhecidos
Lista completa em **[docs/known-issues.md](docs/known-issues.md)**. (A senha do Holmes já saiu do código pro `configs.json` — feito em 2026-07-23.)

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"]

45
app/config.py Normal file
View file

@ -0,0 +1,45 @@
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"
_PATH = Path(__file__).resolve().parent.parent / "configs.json"
settings = Settings.model_validate_json(_PATH.read_text(encoding="utf-8"))

45
app/middleware.py Normal file
View file

@ -0,0 +1,45 @@
import re
from fastapi import Request, status
from fastapi.responses import Response
BLOCKED_PATHS = {
"/",
"/metrics",
"/security.txt",
"/.env",
"/wp-admin",
"/wp-login.php",
"/admin",
"/config",
"/actuator",
"/nice%20ports%2C/Trinity.txt.bak",
}
BLOCKED_UA_PATTERNS = re.compile(
r"(nmap|nikto|masscan|zgrab|censys|shodan|nuclei|httpx|gobuster|dirbuster)",
re.IGNORECASE,
)
async def block_scanners(request: Request, call_next):
path = request.url.path
ua = request.headers.get("user-agent", "")
if (
path.startswith("/v1/")
or path.startswith("/v2/")
or path in ("/health", "/dados_retorno", "/token")
):
return await call_next(request)
if path in BLOCKED_PATHS or path.endswith((".bak", ".env", ".git", ".php")):
return Response(status_code=status.HTTP_403_FORBIDDEN)
if BLOCKED_UA_PATTERNS.search(ua):
return Response(status_code=status.HTTP_403_FORBIDDEN)
if not ua:
return Response(status_code=status.HTTP_403_FORBIDDEN)
return await call_next(request)

57
app/security.py Normal file
View file

@ -0,0 +1,57 @@
from datetime import datetime, timedelta, timezone
import jwt
from app.config import settings
from app.infra.database import db_instance
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
ALGORITHM = "HS256"
TOKEN_EXPIRE_MINUTES = 5
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
def criar_token(data: dict) -> str:
payload = data.copy()
payload["exp"] = datetime.now(timezone.utc) + timedelta(
minutes=TOKEN_EXPIRE_MINUTES
)
return jwt.encode(payload, settings.api.jwt_secret, algorithm=ALGORITHM)
def verificar_credenciais(client_id: str, client_secret: str) -> bool:
db_instance.create_pool()
assert db_instance.pool is not None
conn = db_instance.pool.acquire()
try:
cursor = conn.cursor()
cursor.execute(
"""
SELECT 1
FROM orvel_ti.CAD_USUARIO
WHERE PERFIL_TI = 'operador_ti'
AND ativo = 'S'
AND API = 'S'
AND login = :login
AND senha = :senha
""",
{"login": client_id, "senha": client_secret},
)
return cursor.fetchone() is not None
finally:
db_instance.pool.release(conn)
async def token_valido(token: str = Depends(oauth2_scheme)) -> dict:
try:
payload = jwt.decode(token, settings.api.jwt_secret, algorithms=[ALGORITHM])
return payload
except jwt.ExpiredSignatureError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expirado"
)
except jwt.InvalidTokenError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, detail="Token inválido"
)

33
app/services/controle_api.py Normal file
View file

@ -0,0 +1,33 @@
# app/services/api_tracker.py
import logging
def registrar_contador(
conn, api, endpoint, metodo, status_code, sucesso, duracao_ms, erro=None
):
"""Grava uma linha na tabela contadores_api."""
cursor = conn.cursor()
try:
cursor.execute(
"""
INSERT INTO orvel_ti.contadores_api
(api, endpoint, metodo_http, status_code, sucesso, duracao_ms, mensagem_erro)
VALUES
(:api, :endpoint, :metodo, :status, :sucesso, :duracao, :erro)
""",
{
"api": api,
"endpoint": endpoint,
"metodo": metodo,
"status": status_code,
"sucesso": "S" if sucesso else "N",
"duracao": duracao_ms,
"erro": erro[:500] if erro else None,
},
)
conn.commit()
except Exception as e:
conn.rollback()
logging.error(f"Falha ao gravar contador: {e}")
finally:
cursor.close()

704
app/services/holmes.py Normal file
View file

@ -0,0 +1,704 @@
import asyncio
import json
import logging
import os
import random
import re
import tempfile
import time
import httpx
from app.config import settings
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")
def extrair_origem(origem: str):
"""Extrair origem da string do Holmes
Args:
origem (str): 1548 Entrada de Freio
Returns:
str: 1548
"""
return str(origem[:4])
def extrair_empresa(unidade: str):
"""Extrai a empresa da string inteira do holmes
Args:
unidade (str): Ex 10.1 Hyundai Teix. Freitas
Returns:
empresa: 10 | None
revenda: 1 | None
"""
match = re.search(r"^(\d+)\.(\d+)", unidade)
if match:
return match.group(1), match.group(2)
return None, None
def extrair_transacao(transacao: str) -> str:
"""Extrai apenas a parte transação do texto do Holmes
Args:
transacao (str): D15 Entrada de Nota
Returns:
str: D15
"""
return transacao[:3]
async def get_holmes_process(id_processo: str, conn: Connection):
"""
Busca um processo no Holmes. Centralizado para Peças e Despesas.
"""
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
return response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
return None
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return None
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}",
metodo="GET",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def get_holmes_process_details(id_processo: str, conn: Connection):
"""
Busca os detalhes (properties) de um processo no Holmes.
"""
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/details"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
return response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes details (ID {id_processo}): {e}")
return None
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes details: {e}")
return None
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/details",
metodo="GET",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def get_holmes_history(id_processo: str, conn: Connection):
"""
Busca o historico no holmes (importante para pegar o id da ultima task, para avançar futuramente)
Args:
id_processo (str): id do processo no holmes
"""
url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/history"
payload = {
"filters": [],
"page": 1,
"per_page": 100,
"sortBy": ["created_at", "desc"],
}
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
return response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
return None
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return None
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/history",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def get_holmes_rateio(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
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("get", url)
response.raise_for_status()
status = response.status_code
sucesso = True
return response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
return None
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return None
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/table/(rateio)",
metodo="GET",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def task_id_recente(id_processo: str, conn: Connection):
dados_tasks = await get_holmes_history(id_processo, conn)
if (
not dados_tasks
or "histories" not in dados_tasks
or not dados_tasks["histories"]
):
return None
mais_recente = max(dados_tasks["histories"], key=lambda x: x["created_at"])
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)
if not dados_tasks or "histories" not in dados_tasks:
return None # Tratamento se a API falhar
# print(dados_tasks)
mais_recente = max(dados_tasks["histories"], key=lambda x: x["created_at"])
return mais_recente
async def historicos_task(id_processo: str, conn: Connection) -> dict | None:
"""_summary_
Args:
id_processo (str): Id do processo no holmes
Returns:
dict | None : dicionario do historico | None
"""
dados_tasks = await 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(
conn: Connection,
chave: str | None = None,
fluxos: list[str] | bool = False,
ativos: bool = True,
payload: dict | bool = False,
) -> dict:
"""Obtem os processos que existem com a sua chave
Args:
chave (str): Chave principal a ser procurada, preferencialmente unica pfvr, ajuda ae po.
ativos (bool) Defaults to True
fluxos (list[str] | bool, optional): _description_. Defaults to False. se quer pegar de um fluxo específico ou geral. Padrão: Geral
Returns:
dict: _description_
"""
if not payload and chave is None:
raise ValueError(
"Chave é obrigatória caso o payload não seja enviado chefia, fica alerta ae rapa"
)
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
url = "https://app-api.holmesdoc.io/v2/search"
if not payload:
payload = {
"query": {
"from": 0,
"size": 200,
"context": "process",
"sort": "updated_at",
"order": "desc",
"groups": [
{
"match_all": True,
"terms": [
{
"value": f"{chave}",
"type": "match_phrase",
"field": "_content",
}
],
}
],
},
"trash": False,
"deleted_by_me": False,
}
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
dados = response.json()
docs = dados.get("docs", [])
if ativos:
docs = [d for d in docs if d.get("status") != "canceled"]
if fluxos:
docs = [d for d in docs if d.get("name") in fluxos]
return {"status": True, "dados": {**dados, "docs": docs, "total": len(docs)}}
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {chave}): {e}")
return {"status": False, "error": e}
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return {"status": False, "error": e}
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/search/por-chave",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def buscar_processo_por_chaves(
conn: Connection,
combinacoes: list[list[str]],
fluxos: list[str] | bool = False,
ativos: bool = True,
) -> dict:
"""Busca processos no Holmes usando combinações de termos.
Cada item de combinacoes é uma lista de valores que juntos identificam
um processo único (ex: [cnpj, numero_nf]). Cada combinação vira um group
separado na query.
Args:
combinacoes: Ex: [["03657256000164", "19"], ["698cd1c570fd0f8f5f8436a4"]]
ativos: Ignora processos cancelados. Padrão: True.
fluxos: Filtra por nome de fluxo. Padrão: False (todos).
"""
url = "https://app-api.holmesdoc.io/v2/search"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
payload = {
"query": {
"from": 0,
"size": 200,
"context": "process",
"sort": "updated_at",
"order": "desc",
"groups": [
{
"match_all": True,
"terms": [
{"value": termo, "type": "match_phrase", "field": "_content"}
for termo in combinacao
],
}
for combinacao in combinacoes
],
},
"trash": False,
"deleted_by_me": False,
}
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
dados = response.json()
docs = dados.get("docs", [])
if ativos:
docs = [d for d in docs if d.get("status") != "canceled"]
if fluxos:
docs = [d for d in docs if d.get("name") in fluxos]
return {"status": True, "dados": {**dados, "docs": docs, "total": len(docs)}}
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (combinacoes {combinacoes}): {e}")
return {"status": False, "error": e}
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return {"status": False, "error": e}
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/search/por-chaves",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def action(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
sucesso = False
erro = None
api_nome = "holmes"
try:
async with httpx.AsyncClient() as client:
response = await client.post(
url, headers={"api_token": settings.holmes.token_api}, json=payload
)
response.raise_for_status()
status = response.status_code
sucesso = True
return True, response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
return False, e
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao consultar Holmes: {e}")
return False, e
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/action",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def cria_processo(
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
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("post", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
return True, response.json()
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro na API Holmes - Criar Processo ({payload}): {e}")
return False, str(e)
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao criar processo no Holmes: {e}")
return False, str(e)
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/workflows/{id}/start",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def enviar_documento(
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)
if not task_id:
return False, "Não foi possível obter a task mais recente do Holmes"
url = f"https://app-api.holmesdoc.io/v1/tasks/{task_id}/documents/{id_documento}"
inicio = time.perf_counter()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
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
)
response.raise_for_status()
status = response.status_code
sucesso = True
return True, {"task_id": task_id}
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro ao enviar documento Holmes (ID {id_processo}): {e}")
return False, str(e)
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao enviar documento Holmes: {e}")
return False, str(e)
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/tasks/{id}/documents/{id_documento}",
metodo="POST",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
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]:
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()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await _holmes_request("put", url, json=payload)
response.raise_for_status()
status = response.status_code
sucesso = True
return True, response.json() if response.content else {
"mensagem": "processo cancelado"
}
except httpx.HTTPStatusError as e:
status = e.response.status_code
erro = str(e)
logging.error(f"Erro ao cancelar processo Holmes (ID {id_processo}): {e}")
return False, str(e)
except Exception as e:
erro = str(e)
logging.error(f"Erro inesperado ao cancelar processo Holmes: {e}")
return False, str(e)
finally:
duracao = (time.perf_counter() - inicio) * 1000
registrar_contador(
conn=conn,
api=api_nome,
endpoint="/v1/processes/{id}/cancel",
metodo="PUT",
status_code=status,
sucesso=sucesso,
duracao_ms=duracao,
erro=erro,
)
async def main():
# print(aaaa())
print(await token_usuario())
if __name__ == "__main__":
import asyncio
asyncio.run(main())

9
app/v1/api.py Normal file
View file

@ -0,0 +1,9 @@
from app.v1.holmes.holmes import holmes_router
from fastapi import APIRouter
api_router = APIRouter()
api_router.include_router(
holmes_router,
prefix='/holmes'
)

12
app/v1/holmes/holmes.py Normal file
View file

@ -0,0 +1,12 @@
from app.services.holmes import buscar_processo
from fastapi import APIRouter
holmes_router = APIRouter()
@holmes_router.post(
path="/dados_compras",
)
def obter_dados_compras(
dados_front
):
pass

0
app/v1/holmes/models.py Normal file
View file

26
configs.example.json Normal file
View file

@ -0,0 +1,26 @@
{
"banco": {
"user": "SEU_USUARIO_ORACLE",
"password": "SUA_SENHA_ORACLE",
"dns": "host:1521/service_name",
"instant_client": ""
},
"holmes": {
"token_api": "SEU_TOKEN_HOLMES",
"usuario": "usuario_ou_email_do_holmes",
"senha": "SENHA_DO_HOLMES"
},
"apollo": {
"subscription_key": "",
"ambiente": ""
},
"api": {
"port": 7168,
"ambiente": "homolog",
"workers": 4
},
"docs": {
"user": "usuario_docs",
"password": "senha_docs"
}
}

9
deploy/Caddyfile.scardua Normal file
View file

@ -0,0 +1,9 @@
# Adicione este bloco ao Caddyfile da VPS e recarregue o Caddy.
# O Caddy termina o TLS (Let's Encrypt) e faz proxy pro container na rede fls.
#
# PRE-REQUISITO: registro DNS A do subdominio -> IP da VPS,
# senao o Caddy nao consegue emitir o certificado.
api-scardua.flstecnologia.tech {
reverse_proxy api-scardua:8000
}

29
deploy/api-scardua.container Normal file
View file

@ -0,0 +1,29 @@
# Quadlet do Podman -> o systemd (usuario) converte isto em api-scardua.service
# Instalar em: ~/.config/containers/systemd/api-scardua.container
# Aplicar com: systemctl --user daemon-reload && systemctl --user start api-scardua
[Unit]
Description=API Scardua
[Container]
# AJUSTE <owner> para o teu usuario/org no Forgejo
Image=git.flstecnologia.tech/<owner>/api-scardua:latest
AutoUpdate=registry
ContainerName=api-scardua
# Somente a rede do proxy (Caddy alcanca por "api-scardua:8000").
# NAO entra na db-net: esta API usa Oracle EXTERNO, nao o Postgres da VPS.
Network=fls.network
# configs.json montado read-only. :Z reetiqueta o arquivo pro rootless/SELinux.
# Coloque o arquivo real em ~/.config/api-scardua/configs.json no servidor.
Volume=%h/.config/api-scardua/configs.json:/app/configs.json:ro,Z
# Healthcheck vem embutido na imagem (HEALTHCHECK no Dockerfile) -> o
# podman auto-update usa pra decidir rollback.
[Service]
Restart=always
[Install]
WantedBy=default.target

64
docs/arquitetura.md Normal file
View file

@ -0,0 +1,64 @@
# Arquitetura e Visão — API-Scardua
> Documento da **ideia** do projeto: o quê, o porquê e o roadmap. Salvo pra não
> reexplicar toda sessão. Atualizar quando a visão evoluir.
## Em uma frase
Uma API que **recebe** dados do Holmes do cliente, **trata**, e **envia** de volta
pro banco do cliente — hospedada na infra do dev (FLS), não na do cliente.
## Os atores
- **FLS Tecnologia (dev / nós)** — constrói e hospeda a API na **própria VPS**
(`api-scardua.flstecnologia.tech`). A infra e o código são nossos.
- **Comercial Scardua (cliente)** — tem a conta no **Holmes** e um banco **Oracle**
(IP privado, `10.16.x` — não acessível pela internet).
- **Holmes** — SaaS externo de workflow de documentos; dispara os dados (webhook)
pra nossa API.
## O fluxo
```
Holmes --webhook--> [API na VPS da FLS] --trata--> [Conector no server do cliente] --> [Oracle do cliente]
```
1. O Holmes envia os dados pra nossa API (URL pública, TLS).
2. A API recebe e **trata** os dados.
3. A API fala com o **banco do cliente** através de um **conector que roda no
servidor do cliente** (que tem acesso à rede privada do Oracle).
## As decisões e o PORQUÊ (o mais importante)
A API fica na **nossa VPS**, e não no servidor do cliente, por 2 motivos:
1. **Não depender de o cliente ter domínio / rota pública.** O Holmes precisa de
um endpoint público e estável pra mandar os webhooks — damos isso nós, com
domínio + TLS próprios.
2. **Não deixar nosso código Python no servidor do cliente.** A lógica fica sob
nosso controle, na nossa infra.
**Consequência técnica:** como o Oracle do cliente está numa rede privada
(`10.16.x`), a VPS **não alcança o banco direto**. Por isso existe o **conector no
server do cliente** — ele é a ponte entre a API (pública) e o Oracle (privado).
## Roadmap
| # | Fase | Estado |
|---|---|---|
| 1 | **Receber dados** — esqueleto da API, rota de teste, deploy na VPS | ✅ feito |
| 2 | **Lógica do Holmes** — ligar os endpoints `/v1/holmes/...` ao serviço `app/services/holmes.py` (hoje `obter_dados_compras` é stub) | ⏳ em andamento (dev) |
| 3 | **Conector no cliente** — a ponte API (VPS) ↔ Oracle do cliente (rede privada) | ⬜ a fazer |
| 4 | **CI/CD** — Forgejo Actions (valida no PR/main) + registry + auto-update pull-based (ver [deploy.md](deploy.md)) | ⬜ desenhado |
## Pontos em aberto
- **Como** o conector no cliente vai funcionar — duas abordagens possíveis:
- **Túnel/VPN reverso**: o server do cliente abre um túnel e a VPS passa a
alcançar o `10.16.x` → a API conecta no Oracle direto (é o que o código de
hoje assume: `oracledb` com o DSN privado em `configs.json`).
- **Agente HTTP no cliente**: um serviço no server do cliente que a API chama;
ele consulta o Oracle **localmente** e devolve o resultado (aí a API nunca
toca no Oracle direto).
- Essa escolha decide o destino do `app/infra/database.py` (hoje faltando):
conexão direta via túnel, ou um cliente HTTP pro agente.

54
docs/deploy.md Normal file
View file

@ -0,0 +1,54 @@
# Deploy — API-Scardua
Ambiente: **VPS Hostinger** (Debian 13), **Podman rootless + Quadlet**, **Caddy** (proxy + TLS), user `admin`.
No ar: **https://api-scardua.flstecnologia.tech**
## Acesso à VPS
- SSH: `ssh vps-fls` (alias já configurado → 179.197.230.154, porta **3115**, user `admin`).
- ⚠️ No Windows use o **OpenSSH nativo (PowerShell/cmd)** — o Git Bash não enxerga o `known_hosts`/config e falha.
- Porta 22 é **fechada** no firewall. Sempre `-p 3115` (ou o alias).
- `systemctl --user` sempre (rootless) — **nunca** `sudo systemctl` pros containers.
## Arquitetura
- Imagem: hoje buildada **local na VPS** = `localhost/api-scardua:latest` (sem registry ainda).
- Container `api-scardua` na rede **`fls`** — só o Caddy alcança; a porta 8000 **não** é publicada no host.
- 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 (Let's Encrypt).
Caddyfile em `/srv/containers/stacks/caddy/Caddyfile`.
## Atualizar a app (hoje é manual)
```bash
# 1. copiar o código atualizado pro build dir da VPS (scp de main.py / app/ / etc.)
# 2. na VPS:
cd ~/build/api-scardua
podman build --format docker -t localhost/api-scardua:latest . # --format docker é OBRIGATÓRIO
systemctl --user restart api-scardua
```
> ⚠️ **`--format docker`**: em formato OCI (padrão do Podman) o `HEALTHCHECK` do Dockerfile é **ignorado**.
## Mexer no Caddy (com segurança)
```bash
cd /srv/containers/stacks/caddy
cp Caddyfile Caddyfile.bak # backup
# editar Caddyfile...
podman exec caddy caddy validate --config /etc/caddy/Caddyfile # valida ANTES
podman exec caddy caddy reload --config /etc/caddy/Caddyfile # recarrega sem downtime
```
## Comandos úteis
```bash
systemctl --user status api-scardua # estado do serviço
podman logs -f api-scardua # logs ao vivo
podman ps --filter name=api-scardua # estado + health
podman healthcheck run api-scardua # roda o healthcheck na hora
```
## Healthcheck
Embutido na imagem (`HEALTHCHECK` no Dockerfile, exec-form). Bate em `/health` a cada 30s.
É o sinal que o `podman auto-update` vai usar pra **rollback** quando o registry/CI estiver pronto.
## Evolução planejada
- **Registry + CI** (Forgejo Actions): buildar e publicar em `git.flstecnologia.tech/<owner>/api-scardua`,
trocar o Quadlet pro alvo em `deploy/api-scardua.container` (imagem do registry + `AutoUpdate=registry`) → deploy pull-based.
- Backup do Postgres da VPS (pendência geral do servidor).

17
docs/known-issues.md Normal file
View file

@ -0,0 +1,17 @@
# Problemas conhecidos / TODO — API-Scardua
## Segurança
- ✅ **Senha do Holmes movida pro config** (2026-07-23): agora é `settings.holmes.usuario` / `settings.holmes.senha`, lidos do `configs.json` (gitignored). Saiu do código. **Recomendado ainda trocar a senha**, já que ficou em texto claro antes.
- `/dados_retorno` está **público e sem autenticação** — qualquer um pode postar (enche o log). OK pra teste; proteger com token na API real.
- `/docs`, `/redoc`, `/openapi.json` ficam expostos quando `ambiente != prod` (atrás de basic auth). Fechar de vez com `ambiente = prod` no config, se não precisar deles.
- Log mostra o IP do **Caddy** (10.89.0.x), não o real. Pra ver o IP de origem: rodar o uvicorn com `--proxy-headers --forwarded-allow-ips=*` (seguro aqui, porque a porta 8000 não é publicada no host).
## Código incompleto
- **`app/infra/database.py` não existe** — `app/security.py` importa `db_instance` dele e quebraria. Não há camada de conexão Oracle funcional ainda (nada chama `create_pool` no caminho que roda).
- `app/security.py` usa `settings.api.jwt_secret`, que **não existe** em `app/config.py` (`ApiConfig` só tem `port`/`ambiente`/`workers`).
- `PyJWT` é importado em `security.py` mas **não está** nas dependências.
- `app/v1/holmes/holmes.py` → `obter_dados_compras` é um **stub** (`pass`), sem tipo no parâmetro.
- `pyproject.toml` aponta `readme = "README.md"`, mas o arquivo não existe (quebra `pip install .`; por isso o Docker usa `requirements.txt`).
## Middleware
- `block_scanners` é **denylist + heurística**, não allowlist estrita: qualquer path fora da blocklist, com User-Agent normal, **passa** e vira 404 na app (ex.: `/robots.txt`). Não vaza nada, mas é mais permissivo do que parece. Dá pra inverter pra allowlist estrita (403 em tudo fora da lista permitida) — lembrar de incluir `/docs`,`/redoc`,`/openapi.json` na lista.

108
main.py Normal file
View file

@ -0,0 +1,108 @@
import json
import logging
import secrets
from app.config import settings
from app.middleware import block_scanners
from app.v1.api import api_router
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
from fastapi.security import HTTPBasic, HTTPBasicCredentials
# --- Logging da aplicacao ---
# Sem isto, os logging.info(...) das rotas sao engolidos: o root logger vem em
# WARNING por padrao, entao INFO nao aparece no `podman logs`.
logging.basicConfig(
level=settings.log_level,
format="%(asctime)s %(levelname)s %(message)s",
)
# Filtra o acesso do healthcheck (/health a cada 30s) pra nao poluir o log.
class _FiltraHealthAccess(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
return "/health" not in record.getMessage()
logging.getLogger("uvicorn.access").addFilter(_FiltraHealthAccess())
is_prod = settings.api.ambiente.lower() == "prod"
app = FastAPI(
title='API - Scardua',
version='0.0.1',
docs_url=None,
redoc_url=None,
openapi_url=None,
)
app.middleware("http")(block_scanners)
app.include_router(api_router, prefix="/v1")
# --- Documentação protegida por login/senha (somente fora de produção) ---
if not is_prod:
_docs_security = HTTPBasic()
def _verificar_docs(
credenciais: HTTPBasicCredentials = Depends(_docs_security),
) -> str:
usuario_ok = secrets.compare_digest(
credenciais.username.encode("utf-8"),
settings.docs.user.encode("utf-8"),
)
senha_ok = secrets.compare_digest(
credenciais.password.encode("utf-8"),
settings.docs.password.encode("utf-8"),
)
if not (usuario_ok and senha_ok):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Credenciais inválidas",
headers={"WWW-Authenticate": "Basic"},
)
return credenciais.username
@app.get("/openapi.json", include_in_schema=False)
def _openapi_protegido(_: str = Depends(_verificar_docs)):
return app.openapi()
@app.get("/docs", include_in_schema=False)
def _swagger_protegido(_: str = Depends(_verificar_docs)):
return get_swagger_ui_html(openapi_url="/openapi.json", title=app.title)
@app.get("/redoc", include_in_schema=False)
def _redoc_protegido(_: str = Depends(_verificar_docs)):
return get_redoc_html(openapi_url="/openapi.json", title=app.title)
@app.get("/health", include_in_schema=False)
def health():
"""Healthcheck usado pelo Podman (auto-update / rollback)."""
return {"status": "ok"}
@app.post('/dados_retorno')
def dados_retorno(dados_retorno: dict):
"""
Rota para vizualizar qualquer retorno do holmes
"""
logging.info("Dados retornados do teste")
logging.info(json.dumps(dados_retorno, ensure_ascii=False))
return dados_retorno
if __name__ == "__main__":
import uvicorn
port = settings.api.port
ambiente = settings.api.ambiente
workers = settings.api.workers
is_reload = ambiente.lower() != "prod"
logging.info("API ligada.")
uvicorn.run(
"main:app",
host="0.0.0.0",
port=port,
log_level="info",
workers=workers,
reload=is_reload,
)

75
pyproject.toml Normal file
View file

@ -0,0 +1,75 @@
[project]
name = "api-scardua"
version = "0.1.0"
description = "API de integracao com servicos externos (Holmes, Apollo) e banco Oracle"
requires-python = ">=3.10"
readme = "README.md"
# ---------------------------------------------------------------------------
# Dependencias principais — necessarias para a aplicacao rodar
# (todas sao importadas no carregamento dos modulos, portanto obrigatorias)
# ---------------------------------------------------------------------------
dependencies = [
"fastapi>=0.115", # Framework da API
"uvicorn[standard]>=0.34", # Servidor ASGI
"pydantic>=2.0", # Validacao / configs (app/config.py)
"httpx>=0.27", # Chamadas HTTP as APIs externas (Holmes etc.)
"oracledb>=2.0", # Driver Oracle moderno (substitui cx_Oracle)
]
# ---------------------------------------------------------------------------
# Dependencias opcionais
# pip install .[dev] → ferramentas de desenvolvimento
# ---------------------------------------------------------------------------
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-cov>=6.0",
"pytest-asyncio>=0.24",
"ruff>=0.8",
"httpx>=0.27", # Necessario para o TestClient do FastAPI
]
# ---------------------------------------------------------------------------
# Configuracao do pytest
# ---------------------------------------------------------------------------
[tool.pytest.ini_options]
testpaths = ["tests"]
pythonpath = ["."] # Permite "from app...." a partir da raiz
addopts = "-v --tb=short"
asyncio_mode = "auto"
# ---------------------------------------------------------------------------
# Configuracao do Ruff (linter + formatter)
# ---------------------------------------------------------------------------
[tool.ruff]
target-version = "py310"
line-length = 120
src = ["app", "tests"]
[tool.ruff.lint]
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # pyflakes
"I", # isort
"N", # pep8-naming
"UP", # pyupgrade
]
ignore = [
"E501", # line too long (controlado pelo formatter)
]
[tool.ruff.format]
quote-style = "double"
# ---------------------------------------------------------------------------
# Build system
# ---------------------------------------------------------------------------
[build-system]
requires = ["setuptools>=75.0", "wheel"]
build-backend = "setuptools.build_meta"
[tool.setuptools.packages.find]
where = ["."]
include = ["app*"]

7
requirements.txt Normal file
View file

@ -0,0 +1,7 @@
# 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

0
tests/tests_holmes.py Normal file
View file