mirror of
https://github.com/EvolutionAPI/evolution-client-python.git
synced 2025-12-12 19:39:33 -06:00
138 lines
5.0 KiB
Python
138 lines
5.0 KiB
Python
import requests
|
|
from requests_toolbelt import MultipartEncoder
|
|
|
|
from .exceptions import EvolutionAPIError, EvolutionAuthenticationError, EvolutionNotFoundError
|
|
from .services.call import CallService
|
|
from .services.chat import ChatService
|
|
from .services.group import GroupService
|
|
from .services.instance import InstanceService
|
|
from .services.instance_operations import InstanceOperationsService
|
|
from .services.label import LabelService
|
|
from .services.message import MessageService
|
|
from .services.profile import ProfileService
|
|
from .services.websocket import WebSocketManager, WebSocketService
|
|
|
|
|
|
class EvolutionClient:
|
|
"""
|
|
Cliente para interagir com a API Evolution.
|
|
|
|
Args:
|
|
base_url (str): A URL base do servidor da API Evolution.
|
|
api_token (str): O token de autenticação para acessar a API.
|
|
"""
|
|
|
|
def __init__(self, base_url: str, api_token: str):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.api_token = api_token
|
|
self.instances = InstanceService(self)
|
|
self.instance_operations = InstanceOperationsService(self)
|
|
self.messages = MessageService(self)
|
|
self.calls = CallService(self)
|
|
self.chat = ChatService(self)
|
|
self.label = LabelService(self)
|
|
self.profile = ProfileService(self)
|
|
self.group = GroupService(self)
|
|
self.websocket = WebSocketService(self)
|
|
|
|
def _get_headers(self, instance_token: str = None):
|
|
return {"apikey": instance_token or self.api_token, "Content-Type": "application/json"}
|
|
|
|
def _get_full_url(self, endpoint):
|
|
return f"{self.base_url}/{endpoint}"
|
|
|
|
def _handle_response(self, response):
|
|
if response.status_code == 401:
|
|
raise EvolutionAuthenticationError("Falha na autenticação.")
|
|
elif response.status_code == 404:
|
|
raise EvolutionNotFoundError("Recurso não encontrado.")
|
|
elif response.ok:
|
|
try:
|
|
return response.json()
|
|
except ValueError:
|
|
return response.content
|
|
else:
|
|
error_detail = ""
|
|
try:
|
|
error_detail = f" - {response.json()}"
|
|
except Exception:
|
|
error_detail = f" - {response.text}"
|
|
raise EvolutionAPIError(f"Erro na requisição: {response.status_code}{error_detail}")
|
|
|
|
def get(self, endpoint: str, instance_token: str = None):
|
|
"""Faz uma requisição GET."""
|
|
url = self._get_full_url(endpoint)
|
|
response = requests.get(url, headers=self._get_headers(instance_token))
|
|
return self._handle_response(response)
|
|
|
|
def post(
|
|
self, endpoint: str, data: dict = None, instance_token: str = None, files: dict = None
|
|
):
|
|
url = f"{self.base_url}/{endpoint}"
|
|
headers = self._get_headers(instance_token)
|
|
|
|
if files:
|
|
# Remove o Content-Type do header quando enviando arquivos
|
|
if "Content-Type" in headers:
|
|
del headers["Content-Type"]
|
|
|
|
# Prepara os campos do multipart
|
|
fields = {}
|
|
|
|
# Adiciona os campos do data
|
|
for key, value in data.items():
|
|
fields[key] = (
|
|
str(value)
|
|
if not isinstance(value, (int, float))
|
|
else (None, str(value), "text/plain")
|
|
)
|
|
|
|
# Adiciona o arquivo
|
|
file_tuple = files["file"]
|
|
fields["file"] = (file_tuple[0], file_tuple[1], file_tuple[2])
|
|
|
|
# Cria o multipart encoder
|
|
multipart = MultipartEncoder(fields=fields)
|
|
headers["Content-Type"] = multipart.content_type
|
|
|
|
response = requests.post(url, headers=headers, data=multipart)
|
|
else:
|
|
response = requests.post(url, headers=headers, json=data)
|
|
|
|
return response.json()
|
|
|
|
def put(self, endpoint, data=None):
|
|
"""Faz uma requisição PUT."""
|
|
url = self._get_full_url(endpoint)
|
|
response = requests.put(url, headers=self.headers, json=data)
|
|
return self._handle_response(response)
|
|
|
|
def delete(self, endpoint: str, instance_token: str = None):
|
|
"""Faz uma requisição DELETE."""
|
|
url = self._get_full_url(endpoint)
|
|
response = requests.delete(url, headers=self._get_headers(instance_token))
|
|
return self._handle_response(response)
|
|
|
|
def create_websocket(
|
|
self, instance_id: str, api_token: str, max_retries: int = 5, retry_delay: float = 1.0
|
|
) -> WebSocketManager:
|
|
"""
|
|
Create a WebSocket manager for the specified instance.
|
|
|
|
Args:
|
|
instance_id (str): The instance ID
|
|
api_token (str): The API token
|
|
max_retries (int): Maximum number of reconnection attempts
|
|
retry_delay (float): Initial delay between attempts in seconds
|
|
|
|
Returns:
|
|
WebSocketManager: The WebSocket manager instance
|
|
"""
|
|
return WebSocketManager(
|
|
base_url=self.base_url,
|
|
instance_id=instance_id,
|
|
api_token=api_token,
|
|
max_retries=max_retries,
|
|
retry_delay=retry_delay,
|
|
)
|