Skip to main content

Instalacion

pip install requests

Cliente Completo

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

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

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}")