-
Notifications
You must be signed in to change notification settings - Fork 22
Expand file tree
/
Copy pathclient.py
More file actions
138 lines (118 loc) · 5.17 KB
/
client.py
File metadata and controls
138 lines (118 loc) · 5.17 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import requests
from requests_toolbelt import MultipartEncoder
from .exceptions import EvolutionAuthenticationError, EvolutionNotFoundError, EvolutionAPIError
from .services.instance import InstanceService
from .services.instance_operations import InstanceOperationsService
from .services.message import MessageService
from .services.call import CallService
from .services.chat import ChatService
from .services.label import LabelService
from .services.profile import ProfileService
from .services.group import GroupService
from .services.websocket import WebSocketService, WebSocketManager
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:
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
)