FLS/API-Holmes
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
API-Holmes/app/services/holmes.py
Ricardo 4d47887c00 Reestrutura para API multi-cliente com payload cifrado 
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>
2026-08-18 06:01:01 -03:00

704 lines
24 KiB
Python
Raw Permalink Blame History

import asyncio
import json
import logging
import os
import random
import re
import time
import httpx
from app.schemas import HolmesCliente
from app.services.controle_api import registrar_contador
from oracledb import Connection
# 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):
"""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]
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.
"""
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 self._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(self, 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 self._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(self, 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 self._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(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
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await self._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(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
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(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
# print(dados_tasks)
mais_recente = max(dados_tasks["histories"], key=lambda x: x["created_at"])
return mais_recente
async def historicos_task(self, 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 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(
self,
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 self._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(
self,
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 self._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(
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
sucesso = False
erro = None
api_nome = "holmes"
try:
async with httpx.AsyncClient() as client:
response = await client.post(
url, headers={"api_token": self.cfg.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(
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
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await self._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(
self,
id_processo: str,
arquivo: bytes,
nome_arquivo: str,
id_documento: str,
conn: Connection,
) -> 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"
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": self.cfg.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,
)
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()
status = None
sucesso = False
erro = None
api_nome = "holmes"
try:
response, api_nome = await self._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,
)
# 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():
from app.clientes.registry import obter
print(await obter("scardua").holmes.token_usuario())
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Reference in a new issue View git blame Copy permalink