Instalacion
Copy
pip install requests
Cliente Completo
Copy
import os
import json
import time
import requests
from typing import Optional, Dict, Any, List
class NeuracallClient:
"""Cliente para la API de Neuracall"""
def __init__(
self,
client_id: Optional[str] = None,
client_secret: Optional[str] = None,
base_url: str = "https://api.neuracall.com"
):
self.base_url = base_url
self.client_id = client_id or os.environ["NEURACALL_CLIENT_ID"]
self.client_secret = client_secret or os.environ["NEURACALL_CLIENT_SECRET"]
self._token = None
self._expires_at = 0
def _get_token(self) -> str:
"""Obtiene o renueva el token de acceso"""
if time.time() > self._expires_at - 300:
response = requests.post(
f"{self.base_url}/v1/auth",
json={
"client_id": self.client_id,
"client_secret": self.client_secret
}
)
response.raise_for_status()
data = response.json()
self._token = data["access_token"]
self._expires_at = data["expires_at"]
return self._token
def _headers(self) -> Dict[str, str]:
"""Headers con autorizacion"""
return {"Authorization": f"Bearer {self._get_token()}"}
# ==================== CALL ANALYSIS ====================
def create_analysis(
self,
audio_path: str,
external_id: str,
agent_id: str,
agent_name: str,
model_name: str,
additional_parameters: Optional[Dict] = None
) -> Dict[str, Any]:
"""
Crea un nuevo análisis de llamada.
Args:
audio_path: Ruta al archivo de audio
external_id: ID externo de la llamada
agent_id: ID del agente
agent_name: Nombre del agente
model_name: Nombre del modelo de análisis
additional_parameters: Metadatos adicionales opciónales
Returns:
Datos del análisis creado
"""
with open(audio_path, "rb") as audio_file:
files = {"audio_file": audio_file}
data = {
"external_id": external_id,
"agent_id": agent_id,
"agent_name": agent_name,
"model_name": model_name
}
if additional_parameters:
data["additional_parameters"] = json.dumps(additional_parameters)
response = requests.post(
f"{self.base_url}/v1/call-analysis",
headers=self._headers(),
files=files,
data=data
)
response.raise_for_status()
return response.json()
def get_analysis(self, analysis_id: str) -> Dict[str, Any]:
"""Obtiene un análisis por ID"""
response = requests.get(
f"{self.base_url}/v1/call-analysis/{analysis_id}",
headers=self._headers()
)
response.raise_for_status()
return response.json()
def list_analyses(
self,
model_id: Optional[int] = None,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
page: int = 0,
size: int = 20
) -> Dict[str, Any]:
"""Lista análisis con filtros y paginación"""
params = {"page": page, "size": size}
if model_id:
params["model_id"] = model_id
if start_date:
params["start_date"] = start_date
if end_date:
params["end_date"] = end_date
response = requests.get(
f"{self.base_url}/v1/call-analysis",
headers=self._headers(),
params=params
)
response.raise_for_status()
return response.json()
def get_transcription(self, analysis_id: str) -> Dict[str, Any]:
"""Obtiene la transcripción de un análisis"""
response = requests.get(
f"{self.base_url}/v1/call-analysis/{analysis_id}/transcription",
headers=self._headers()
)
response.raise_for_status()
return response.json()
def wait_for_completion(
self,
analysis_id: str,
timeout: int = 600,
poll_interval: int = 5
) -> Dict[str, Any]:
"""
Espera a que un análisis se complete.
Args:
analysis_id: ID del análisis
timeout: Tiempo máximo de espera en segúndos
poll_interval: Intervalo entre consultas en segúndos
Returns:
Datos del análisis completado
"""
start = time.time()
while time.time() - start < timeout:
result = self.get_analysis(analysis_id)
status = result["status"]
if status == "PROCESS_COMPLETED":
return result
elif status == "ERROR":
raise Exception(f"Analysis failed: {result.get('error_message')}")
time.sleep(poll_interval)
raise TimeoutError(f"Analysis {analysis_id} did not complete within {timeout}s")
# ==================== MODELS ====================
def list_models(self, order_by_name: bool = False) -> List[Dict[str, Any]]:
"""Lista todos los modelos de análisis"""
response = requests.get(
f"{self.base_url}/v1/models",
headers=self._headers(),
params={"order_by_name": order_by_name}
)
response.raise_for_status()
return response.json()
def get_model(self, model_id: int) -> Dict[str, Any]:
"""Obtiene un modelo por ID"""
response = requests.get(
f"{self.base_url}/v1/models/{model_id}",
headers=self._headers()
)
response.raise_for_status()
return response.json()
# ==================== STATISTICS ====================
def get_basic_stats(
self,
start_date: str,
end_date: str,
model_name: str
) -> Dict[str, Any]:
"""Obtiene estadísticas básicas"""
response = requests.get(
f"{self.base_url}/v1/stats/basic",
headers=self._headers(),
params={
"start_date": start_date,
"end_date": end_date,
"model_name": model_name
}
)
response.raise_for_status()
return response.json()
# ==================== EJEMPLO DE USO ====================
if __name__ == "__main__":
# Inicializar cliente (usa variables de entorno)
client = NeuracallClient()
# Listar modelos disponibles
print("Modelos disponibles:")
models = client.list_models()
for model in models:
print(f" - {model['name']} (ID: {model['id']})")
# Crear análisis
print("\nCreando análisis...")
analysis = client.create_analysis(
audio_path="llamada.mp3",
external_id="CALL-2024-001",
agent_id="AGT-001",
agent_name="Juan Perez",
model_name=models[0]["name"]
)
print(f"Análisis creado: {analysis['id']}")
print(f"Estado: {analysis['status']}")
# Esperar a que complete
print("\nEsperando resultado...")
result = client.wait_for_completion(analysis["id"])
# Mostrar resultados
print(f"\n=== RESULTADO ===")
print(f"NeuraScore: {result['score_percentage']}%")
print(f"Resumen: {result['summary']}")
if result.get("insights"):
print("\nInsights:")
for i, insight in enumerate(result["insights"], 1):
print(f" {i}. {insight}")
if result.get("keywords"):
print(f"\nKeywords: {', '.join(result['keywords'])}")
Uso Rápido
Copy
from neuracall_client import NeuracallClient
client = NeuracallClient()
# Analizar una llamada
result = client.create_analysis(
audio_path="llamada.mp3",
external_id="CALL-001",
agent_id="AGT-001",
agent_name="Juan Perez",
model_name="Evaluación Servicio"
)
# Esperar y obtener resultados
completed = client.wait_for_completion(result["id"])
print(f"Score: {completed['score_percentage']}%")
Manejo de Errores
Copy
import requests
try:
result = client.create_analysis(...)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 401:
print("Error de autenticación")
elif e.response.status_code == 400:
print(f"Datos invalidos: {e.response.json()}")
else:
print(f"Error HTTP: {e}")
except Exception as e:
print(f"Error: {e}")
