Cliente OTIMUS — implementação de referência¶
Template completo de um cliente OTIMUS reutilizável. Copie e adapte os IDs de clínica.
Arquivo scheduling.py — esqueleto¶
"""Integração OTIMUS para <projeto>.
Responsável por:
- Consultar horários disponíveis
- Buscar/cadastrar paciente
- Confirmar agendamento
IDs específicos da clínica estão no topo do arquivo.
"""
from __future__ import annotations
import datetime
import json
import logging
import re
import unicodedata
import zoneinfo
from datetime import date, timedelta
import requests
from config import OTIMUS_API_URL, OTIMUS_TOKEN
logger = logging.getLogger("<projeto>.scheduling")
_TZ_BR = zoneinfo.ZoneInfo("America/Sao_Paulo")
# =============================================================================
# DADOS DA CLÍNICA (editar ao criar novo projeto)
# =============================================================================
PROCEDIMENTOS = [
# {"id_especialidade": 450, "id_agenda": 3, "id_procedimento": 64639,
# "procedimento": "CONSULTA ORTOPEDISTA", "medico": "NOME COMPLETO DR"},
]
CONVENIOS = [
# {"id": 1, "convenio": "PARTICULAR"},
]
# Lookup estrito por nome — ver docs/padrao/resolver-medico.md
_DOCTOR_KEYWORDS: list[tuple[str, int]] = [
# ("NOME_UNICO", idx_em_PROCEDIMENTOS),
]
# Validação de primeiros nomes únicos (trava ao boot se duplicado)
def _validate_unique_first_names() -> None:
seen: set[str] = set()
for p in PROCEDIMENTOS:
first = _normalize(p["medico"]).split()[0]
if first in seen:
raise RuntimeError(
f"ERRO CRITICO: primeiro nome duplicado '{first}'. "
f"Atualize _DOCTOR_KEYWORDS e _validar_agenda_medico "
f"para diferenciar (use multi-palavra ou sobrenome único)."
)
seen.add(first)
_validate_unique_first_names()
# =============================================================================
# HELPERS
# =============================================================================
def _otimus_headers() -> dict:
return {
"sistema": "AGENTPRO",
"token": OTIMUS_TOKEN,
"Content-Type": "application/json",
"Accept": "*/*",
}
def _normalize(s: str) -> str:
if not s:
return ""
nfkd = unicodedata.normalize("NFKD", s)
return "".join(c for c in nfkd if not unicodedata.combining(c)).upper().strip()
def _only_digits(s: str) -> str:
return "".join(c for c in (s or "") if c.isdigit())
def _yyyymmdd_to_ddmmyyyy(s: str) -> str:
y, m, d = s.split("-")
return f"{d}-{m}-{y}"
def _today_br() -> date:
return datetime.datetime.now(_TZ_BR).date()
def _now_br() -> datetime.datetime:
return datetime.datetime.now(_TZ_BR)
# =============================================================================
# LOOKUPS
# =============================================================================
def buscar_procedimento_por_medico(nome_medico: str) -> dict | None:
"""Busca estrita usando _DOCTOR_KEYWORDS. Sem fuzzy matching."""
if not nome_medico:
return None
nome_norm = _normalize(nome_medico)
for keyword, idx in _DOCTOR_KEYWORDS:
if keyword in nome_norm:
return PROCEDIMENTOS[idx]
return None
def buscar_convenio(nome: str) -> dict | None:
nome_norm = _normalize(nome)
for c in CONVENIOS:
if _normalize(c["convenio"]) == nome_norm:
return c
for c in CONVENIOS:
cn = _normalize(c["convenio"])
if nome_norm in cn or cn in nome_norm:
return c
return None
# =============================================================================
# CHAMADAS OTIMUS
# =============================================================================
def _chamar_api_horarios(
proc: dict, conv: dict, data_inicio: str, data_fim: str, tentativas: int = 2
) -> list | None:
payload = {
"filter": {
"agendas": [{"id": proc["id_agenda"], "tipo": "cs"}],
"convenio_id": conv["id"],
"data": data_inicio,
"endDate": data_fim,
"suceder": 0,
"especialidade_id": proc["id_especialidade"],
"limit": 100,
"ordenacao": "asc",
"procedimentos": [proc["id_procedimento"]],
}
}
for i in range(tentativas):
try:
resp = requests.post(
f"{OTIMUS_API_URL}/api/v1/agendamento/horarios",
headers=_otimus_headers(),
json=payload,
timeout=30,
)
resp.raise_for_status()
return resp.json().get("data", [])
except Exception as e:
logger.warning("OTIMUS /horarios tentativa %d/%d falhou: %s", i + 1, tentativas, e)
return None
def _buscar_paciente_cpf(cpf: str) -> dict | None:
try:
resp = requests.post(
f"{OTIMUS_API_URL}/api/v1/cadastros/pacientes/cpf/{_only_digits(cpf)}",
headers=_otimus_headers(),
timeout=30,
)
resp.raise_for_status()
data = resp.json().get("data")
if isinstance(data, dict) and data.get("cpf"):
return data
return None
except Exception as e:
logger.error("Erro buscar paciente CPF: %s", e)
return None
def _cadastrar_paciente(nome: str, cpf: str, data_nascimento: str, telefone: str) -> dict | None:
payload = {
"pacientes": [
{
"id": 123,
"nome": nome,
"dataNascimento": data_nascimento,
"cpf": _only_digits(cpf),
"celular": _only_digits(telefone),
}
]
}
try:
resp = requests.post(
f"{OTIMUS_API_URL}/api/v1/paciente/multiplos",
headers=_otimus_headers(),
json=payload,
timeout=30,
)
resp.raise_for_status()
except Exception as e:
logger.error("Erro cadastrar paciente: %s", e)
return None
# Sempre re-buscar para confirmar e pegar o paciente_id real
return _buscar_paciente_cpf(cpf)
def _confirmar_agendamento(
paciente_id: int,
cpf: str,
convenio_id: int,
data_consulta_dmy: str,
hora_consulta: str,
agenda_id: int,
procedimento_id: int,
) -> dict:
payload = {
"paciente_id": paciente_id,
"cpf": _only_digits(cpf),
"convenio_id": convenio_id,
"data": data_consulta_dmy,
"hora": hora_consulta,
"agenda_id": agenda_id,
"agenda_tipo": "cs",
"status": "ativo",
"servicos": [{"id": procedimento_id}],
}
resp = requests.post(
f"{OTIMUS_API_URL}/api/v1/agendamento/agendamento",
headers=_otimus_headers(),
json=payload,
timeout=30,
)
resp.raise_for_status()
return resp.json()
Retry strategy¶
Duas tentativas antes de falhar definitivamente. Se falhar → string
FALHA_CONSULTA:. Jamais 3 ou mais tentativas dentro de um único turn do
agente — paciente percebe a latência e o agente entra em loop.
for i in range(tentativas):
try:
resp = requests.post(...)
resp.raise_for_status()
return resp.json().get("data", [])
except Exception as e:
logger.warning("tentativa %d falhou: %s", i + 1, e)
return None
Composição nos fluxos públicos¶
As duas funções expostas ao agente (consultar_horarios e
agendar_consulta) orquestram os helpers acima. Ver: