NeoAnd/evolution-client-python
1
0
Fork 0
mirror of https://github.com/EvolutionAPI/evolution-client-python.git synced 2025-12-19 11:52:19 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

style: format codebase with ruff

Browse Source
This commit is contained in:
Jackson Vieira
2025-12-03 13:54:30 -03:00
parent 44539123d9
commit 2421b1cf5b
21 changed files with 458 additions and 441 deletions
Download Patch File Download Diff File Expand all files Collapse all files

82
evolutionapi/client.py
View File

@@ -11,6 +11,7 @@ from .services.profile import ProfileService
from .services.group import GroupService from .services.group import GroupService
from .services.websocket import WebSocketService, WebSocketManager from .services.websocket import WebSocketService, WebSocketManager
class EvolutionClient: class EvolutionClient:
""" """
Cliente para interagir com a API Evolution. Cliente para interagir com a API Evolution.
@@ -21,7 +22,7 @@ class EvolutionClient:
""" """
def __init__(self, base_url: str, api_token: str): def __init__(self, base_url: str, api_token: str):
self.base_url = base_url.rstrip('/') self.base_url = base_url.rstrip("/")
self.api_token = api_token self.api_token = api_token
self.instances = InstanceService(self) self.instances = InstanceService(self)
self.instance_operations = InstanceOperationsService(self) self.instance_operations = InstanceOperationsService(self)
@@ -32,33 +33,30 @@ class EvolutionClient:
self.profile = ProfileService(self) self.profile = ProfileService(self)
self.group = GroupService(self) self.group = GroupService(self)
self.websocket = WebSocketService(self) self.websocket = WebSocketService(self)
def _get_headers(self, instance_token: str = None): def _get_headers(self, instance_token: str = None):
return { return {"apikey": instance_token or self.api_token, "Content-Type": "application/json"}
'apikey': instance_token or self.api_token,
'Content-Type': 'application/json'
}
def _get_full_url(self, endpoint): def _get_full_url(self, endpoint):
return f'{self.base_url}/{endpoint}' return f"{self.base_url}/{endpoint}"
def _handle_response(self, response): def _handle_response(self, response):
if response.status_code == 401: if response.status_code == 401:
raise EvolutionAuthenticationError('Falha na autenticação.') raise EvolutionAuthenticationError("Falha na autenticação.")
elif response.status_code == 404: elif response.status_code == 404:
raise EvolutionNotFoundError('Recurso não encontrado.') raise EvolutionNotFoundError("Recurso não encontrado.")
elif response.ok: elif response.ok:
try: try:
return response.json() return response.json()
except ValueError: except ValueError:
return response.content return response.content
else: else:
error_detail = '' error_detail = ""
try: try:
error_detail = f' - {response.json()}' error_detail = f" - {response.json()}"
except: except:
error_detail = f' - {response.text}' error_detail = f" - {response.text}"
raise EvolutionAPIError(f'Erro na requisição: {response.status_code}{error_detail}') raise EvolutionAPIError(f"Erro na requisição: {response.status_code}{error_detail}")
def get(self, endpoint: str, instance_token: str = None): def get(self, endpoint: str, instance_token: str = None):
"""Faz uma requisição GET.""" """Faz uma requisição GET."""
@@ -66,42 +64,40 @@ class EvolutionClient:
response = requests.get(url, headers=self._get_headers(instance_token)) response = requests.get(url, headers=self._get_headers(instance_token))
return self._handle_response(response) return self._handle_response(response)
def post(self, endpoint: str, data: dict = None, instance_token: str = None, files: dict = None): def post(
url = f'{self.base_url}/{endpoint}' 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) headers = self._get_headers(instance_token)
if files: if files:
# Remove o Content-Type do header quando enviando arquivos # Remove o Content-Type do header quando enviando arquivos
if 'Content-Type' in headers: if "Content-Type" in headers:
del headers['Content-Type'] del headers["Content-Type"]
# Prepara os campos do multipart # Prepara os campos do multipart
fields = {} fields = {}
# Adiciona os campos do data # Adiciona os campos do data
for key, value in data.items(): for key, value in data.items():
fields[key] = str(value) if not isinstance(value, (int, float)) else (None, str(value), 'text/plain') fields[key] = (
str(value)
if not isinstance(value, (int, float))
else (None, str(value), "text/plain")
)
# Adiciona o arquivo # Adiciona o arquivo
file_tuple = files['file'] file_tuple = files["file"]
fields['file'] = (file_tuple[0], file_tuple[1], file_tuple[2]) fields["file"] = (file_tuple[0], file_tuple[1], file_tuple[2])
# Cria o multipart encoder # Cria o multipart encoder
multipart = MultipartEncoder(fields=fields) multipart = MultipartEncoder(fields=fields)
headers['Content-Type'] = multipart.content_type headers["Content-Type"] = multipart.content_type
response = requests.post( response = requests.post(url, headers=headers, data=multipart)
url,
headers=headers,
data=multipart
)
else: else:
response = requests.post( response = requests.post(url, headers=headers, json=data)
url,
headers=headers,
json=data
)
return response.json() return response.json()
def put(self, endpoint, data=None): def put(self, endpoint, data=None):
@@ -116,16 +112,18 @@ class EvolutionClient:
response = requests.delete(url, headers=self._get_headers(instance_token)) response = requests.delete(url, headers=self._get_headers(instance_token))
return self._handle_response(response) 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: 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. Create a WebSocket manager for the specified instance.
Args: Args:
instance_id (str): The instance ID instance_id (str): The instance ID
api_token (str): The API token api_token (str): The API token
max_retries (int): Maximum number of reconnection attempts max_retries (int): Maximum number of reconnection attempts
retry_delay (float): Initial delay between attempts in seconds retry_delay (float): Initial delay between attempts in seconds
Returns: Returns:
WebSocketManager: The WebSocket manager instance WebSocketManager: The WebSocket manager instance
""" """
@@ -134,5 +132,5 @@ class EvolutionClient:
instance_id=instance_id, instance_id=instance_id,
api_token=api_token, api_token=api_token,
max_retries=max_retries, max_retries=max_retries,
retry_delay=retry_delay retry_delay=retry_delay,
) )

5
evolutionapi/exceptions.py
View File

@@ -1,11 +1,16 @@
class EvolutionAPIError(Exception): class EvolutionAPIError(Exception):
"""Erro genérico da API Evolution.""" """Erro genérico da API Evolution."""
pass pass
class EvolutionAuthenticationError(EvolutionAPIError): class EvolutionAuthenticationError(EvolutionAPIError):
"""Erro de autenticação com a API Evolution.""" """Erro de autenticação com a API Evolution."""
pass pass
class EvolutionNotFoundError(EvolutionAPIError): class EvolutionNotFoundError(EvolutionAPIError):
"""Recurso não encontrado na API Evolution.""" """Recurso não encontrado na API Evolution."""
pass pass

14
evolutionapi/models/call.py
View File

@@ -2,15 +2,7 @@ class BaseCall:
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None}) self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class FakeCall(BaseCall): class FakeCall(BaseCall):
def __init__( def __init__(self, number: str, isVideo: bool, callDuration: int):
self, super().__init__(number=number, isVideo=isVideo, callDuration=callDuration)
number: str,
isVideo: bool,
callDuration: int
):
super().__init__(
number=number,
isVideo=isVideo,
callDuration=callDuration
)

69
evolutionapi/models/chat.py
View File

@@ -1,93 +1,64 @@
from typing import List, Optional, Dict, Any from typing import List, Optional, Dict, Any
class BaseChat: class BaseChat:
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None}) self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class CheckIsWhatsappNumber(BaseChat): class CheckIsWhatsappNumber(BaseChat):
def __init__( def __init__(self, numbers: List[str]):
self, super().__init__(numbers=numbers)
numbers: List[str]
):
super().__init__(
numbers=numbers
)
class MessageKey: class MessageKey:
def __init__( def __init__(self, remote_jid: str, from_me: bool, id: str, participant: Optional[str] = None):
self,
remote_jid: str,
from_me: bool,
id: str,
participant: Optional[str] = None
):
self.remoteJid = remote_jid self.remoteJid = remote_jid
self.fromMe = from_me self.fromMe = from_me
self.id = id self.id = id
self.participant = participant self.participant = participant
class ReadMessage: class ReadMessage:
def __init__( def __init__(self, remote_jid: str, from_me: bool, id: str):
self,
remote_jid: str,
from_me: bool,
id: str
):
self.remoteJid = remote_jid self.remoteJid = remote_jid
self.fromMe = from_me self.fromMe = from_me
self.id = id self.id = id
class ArchiveChat: class ArchiveChat:
def __init__( def __init__(self, last_message: Dict[str, Any], chat: str, archive: bool):
self,
last_message: Dict[str, Any],
chat: str,
archive: bool
):
self.lastMessage = last_message self.lastMessage = last_message
self.chat = chat self.chat = chat
self.archive = archive self.archive = archive
class UnreadChat: class UnreadChat:
def __init__( def __init__(self, last_message: Dict[str, Any], chat: str):
self,
last_message: Dict[str, Any],
chat: str
):
self.lastMessage = last_message self.lastMessage = last_message
self.chat = chat self.chat = chat
class ProfilePicture: class ProfilePicture:
def __init__(self, number: str): def __init__(self, number: str):
self.number = number self.number = number
class MediaMessage: class MediaMessage:
def __init__( def __init__(self, message: Dict[str, Any], convert_to_mp4: bool = False):
self,
message: Dict[str, Any],
convert_to_mp4: bool = False
):
self.message = message self.message = message
self.convertToMp4 = convert_to_mp4 self.convertToMp4 = convert_to_mp4
class UpdateMessage: class UpdateMessage:
def __init__( def __init__(self, number: str, key: Dict[str, Any], text: str):
self,
number: str,
key: Dict[str, Any],
text: str
):
self.number = number self.number = number
self.key = key self.key = key
self.text = text self.text = text
class Presence: class Presence:
def __init__( def __init__(self, number: str, delay: int, presence: str):
self,
number: str,
delay: int,
presence: str
):
self.number = number self.number = number
self.delay = delay self.delay = delay
self.presence = presence self.presence = presence

10
evolutionapi/models/group.py
View File

@@ -1,39 +1,47 @@
from typing import List, Optional, Literal from typing import List, Optional, Literal
from dataclasses import dataclass from dataclasses import dataclass
@dataclass @dataclass
class CreateGroup: class CreateGroup:
subject: str subject: str
participants: List[str] participants: List[str]
description: Optional[str] = None description: Optional[str] = None
@dataclass @dataclass
class GroupPicture: class GroupPicture:
image: str image: str
@dataclass @dataclass
class GroupSubject: class GroupSubject:
subject: str subject: str
@dataclass @dataclass
class GroupDescription: class GroupDescription:
description: str description: str
@dataclass @dataclass
class GroupInvite: class GroupInvite:
groupJid: str groupJid: str
description: str description: str
numbers: List[str] numbers: List[str]
@dataclass @dataclass
class UpdateParticipant: class UpdateParticipant:
action: Literal["add", "remove", "promote", "demote"] action: Literal["add", "remove", "promote", "demote"]
participants: List[str] participants: List[str]
@dataclass @dataclass
class UpdateSetting: class UpdateSetting:
action: Literal["announcement", "not_announcement", "locked", "unlocked"] action: Literal["announcement", "not_announcement", "locked", "unlocked"]
@dataclass @dataclass
class ToggleEphemeral: class ToggleEphemeral:
expiration: int expiration: int

47
evolutionapi/models/instance.py
View File

@@ -1,27 +1,45 @@
from typing import Optional, List, Dict from typing import Optional, List, Dict
class WebhookConfig: class WebhookConfig:
def __init__(self, url: str = None, byEvents: bool = False, base64: bool = True, def __init__(
headers: Dict = None, events: List[str] = None): self,
url: str = None,
byEvents: bool = False,
base64: bool = True,
headers: Dict = None,
events: List[str] = None,
):
self.url = url self.url = url
self.byEvents = byEvents self.byEvents = byEvents
self.base64 = base64 self.base64 = base64
self.headers = headers self.headers = headers
self.events = events self.events = events
class EventsConfig: class EventsConfig:
def __init__(self, enabled: bool = True, events: List[str] = None): def __init__(self, enabled: bool = True, events: List[str] = None):
self.enabled = enabled self.enabled = enabled
self.events = events self.events = events
class ChatwootConfig: class ChatwootConfig:
def __init__(self, accountId: str = None, token: str = None, url: str = None, def __init__(
signMsg: bool = True, reopenConversation: bool = True, self,
conversationPending: bool = False, importContacts: bool = True, accountId: str = None,
nameInbox: str = "evolution", mergeBrazilContacts: bool = True, token: str = None,
importMessages: bool = True, daysLimitImportMessages: int = 3, url: str = None,
organization: str = "Evolution Bot", signMsg: bool = True,
logo: str = "https://evolution-api.com/files/evolution-api-favicon.png"): reopenConversation: bool = True,
conversationPending: bool = False,
importContacts: bool = True,
nameInbox: str = "evolution",
mergeBrazilContacts: bool = True,
importMessages: bool = True,
daysLimitImportMessages: int = 3,
organization: str = "Evolution Bot",
logo: str = "https://evolution-api.com/files/evolution-api-favicon.png",
):
self.chatwootAccountId = accountId self.chatwootAccountId = accountId
self.chatwootToken = token self.chatwootToken = token
self.chatwootUrl = url self.chatwootUrl = url
@@ -36,6 +54,7 @@ class ChatwootConfig:
self.chatwootOrganization = organization self.chatwootOrganization = organization
self.chatwootLogo = logo self.chatwootLogo = logo
class InstanceConfig: class InstanceConfig:
def __init__( def __init__(
self, self,
@@ -50,10 +69,10 @@ class InstanceConfig:
alwaysOnline: bool = None, alwaysOnline: bool = None,
readMessages: bool = None, readMessages: bool = None,
readStatus: bool = None, readStatus: bool = None,
syncFullHistory: bool = None syncFullHistory: bool = None,
): ):
self.__dict__['instanceName'] = instanceName self.__dict__["instanceName"] = instanceName
for key, value in locals().items(): for key, value in locals().items():
if key != 'self' and key != 'instanceName' and value is not None: if key != "self" and key != "instanceName" and value is not None:
self.__dict__[key] = value self.__dict__[key] = value

17
evolutionapi/models/label.py
View File

@@ -1,21 +1,14 @@
from typing import Literal from typing import Literal
class BaseLabel: class BaseLabel:
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None}) self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class HandleLabel(BaseLabel): class HandleLabel(BaseLabel):
def __init__( def __init__(self, number: str, label_id: str, action: Literal["add", "remove"]):
self,
number: str,
label_id: str,
action: Literal["add", "remove"]
):
if action not in ["add", "remove"]: if action not in ["add", "remove"]:
raise ValueError("action deve ser 'add' ou 'remove'") raise ValueError("action deve ser 'add' ou 'remove'")
super().__init__( super().__init__(number=number, labelId=label_id, action=action)
number=number,
labelId=label_id,
action=action
)

96
evolutionapi/models/message.py
View File

@@ -2,17 +2,20 @@ from enum import Enum
from typing import List, Optional, Union from typing import List, Optional, Union
from dataclasses import dataclass from dataclasses import dataclass
class MediaType(Enum): class MediaType(Enum):
IMAGE = "image" IMAGE = "image"
VIDEO = "video" VIDEO = "video"
DOCUMENT = "document" DOCUMENT = "document"
class StatusType(Enum): class StatusType(Enum):
TEXT = "text" TEXT = "text"
IMAGE = "image" IMAGE = "image"
VIDEO = "video" VIDEO = "video"
AUDIO = "audio" AUDIO = "audio"
class FontType(Enum): class FontType(Enum):
SERIF = 1 SERIF = 1
NORICAN_REGULAR = 2 NORICAN_REGULAR = 2
@@ -20,14 +23,17 @@ class FontType(Enum):
BEBASNEUE_REGULAR = 4 BEBASNEUE_REGULAR = 4
OSWALD_HEAVY = 5 OSWALD_HEAVY = 5
class BaseMessage: class BaseMessage:
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None}) self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class QuotedMessage(BaseMessage): class QuotedMessage(BaseMessage):
def __init__(self, key: dict, message: Optional[dict] = None): def __init__(self, key: dict, message: Optional[dict] = None):
super().__init__(key=key, message=message) super().__init__(key=key, message=message)
class TextMessage(BaseMessage): class TextMessage(BaseMessage):
def __init__( def __init__(
self, self,
@@ -37,7 +43,7 @@ class TextMessage(BaseMessage):
quoted: Optional[QuotedMessage] = None, quoted: Optional[QuotedMessage] = None,
linkPreview: Optional[bool] = None, linkPreview: Optional[bool] = None,
mentionsEveryOne: Optional[bool] = None, mentionsEveryOne: Optional[bool] = None,
mentioned: Optional[List[str]] = None mentioned: Optional[List[str]] = None,
): ):
super().__init__( super().__init__(
number=number, number=number,
@@ -46,9 +52,10 @@ class TextMessage(BaseMessage):
quoted=quoted.__dict__ if quoted else None, quoted=quoted.__dict__ if quoted else None,
linkPreview=linkPreview, linkPreview=linkPreview,
mentionsEveryOne=mentionsEveryOne, mentionsEveryOne=mentionsEveryOne,
mentioned=mentioned mentioned=mentioned,
) )
class MediaMessage(BaseMessage): class MediaMessage(BaseMessage):
def __init__( def __init__(
self, self,
@@ -61,27 +68,28 @@ class MediaMessage(BaseMessage):
delay: Optional[Union[int, float, str]] = None, delay: Optional[Union[int, float, str]] = None,
quoted: Optional[QuotedMessage] = None, quoted: Optional[QuotedMessage] = None,
mentionsEveryOne: Optional[bool] = None, mentionsEveryOne: Optional[bool] = None,
mentioned: Optional[List[str]] = None mentioned: Optional[List[str]] = None,
): ):
data = { data = {
'number': number, "number": number,
'mediatype': mediatype, "mediatype": mediatype,
'caption': caption, "caption": caption,
'mimetype': mimetype, "mimetype": mimetype,
'fileName': fileName, "fileName": fileName,
'quoted': quoted.__dict__ if quoted else None, "quoted": quoted.__dict__ if quoted else None,
'mentionsEveryOne': mentionsEveryOne, "mentionsEveryOne": mentionsEveryOne,
'mentioned': mentioned "mentioned": mentioned,
} }
if delay is not None: if delay is not None:
data['delay'] = delay data["delay"] = delay
if media and media != {}: if media and media != {}:
data['media'] = media data["media"] = media
super().__init__(**{k: v for k, v in data.items() if v is not None}) super().__init__(**{k: v for k, v in data.items() if v is not None})
class StatusMessage(BaseMessage): class StatusMessage(BaseMessage):
def __init__( def __init__(
self, self,
@@ -91,7 +99,7 @@ class StatusMessage(BaseMessage):
backgroundColor: Optional[str] = None, backgroundColor: Optional[str] = None,
font: Optional[FontType] = None, font: Optional[FontType] = None,
allContacts: bool = False, allContacts: bool = False,
statusJidList: Optional[List[str]] = None statusJidList: Optional[List[str]] = None,
): ):
super().__init__( super().__init__(
type=type.value, type=type.value,
@@ -100,9 +108,10 @@ class StatusMessage(BaseMessage):
backgroundColor=backgroundColor, backgroundColor=backgroundColor,
font=font.value if font else None, font=font.value if font else None,
allContacts=allContacts, allContacts=allContacts,
statusJidList=statusJidList statusJidList=statusJidList,
) )
class LocationMessage(BaseMessage): class LocationMessage(BaseMessage):
def __init__( def __init__(
self, self,
@@ -112,7 +121,7 @@ class LocationMessage(BaseMessage):
latitude: float, latitude: float,
longitude: float, longitude: float,
delay: Optional[int] = None, delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None quoted: Optional[QuotedMessage] = None,
): ):
super().__init__( super().__init__(
number=number, number=number,
@@ -121,9 +130,10 @@ class LocationMessage(BaseMessage):
latitude=latitude, latitude=latitude,
longitude=longitude, longitude=longitude,
delay=delay, delay=delay,
quoted=quoted.__dict__ if quoted else None quoted=quoted.__dict__ if quoted else None,
) )
class Contact(BaseMessage): class Contact(BaseMessage):
def __init__( def __init__(
self, self,
@@ -132,7 +142,7 @@ class Contact(BaseMessage):
phoneNumber: str, phoneNumber: str,
organization: Optional[str] = None, organization: Optional[str] = None,
email: Optional[str] = None, email: Optional[str] = None,
url: Optional[str] = None url: Optional[str] = None,
): ):
super().__init__( super().__init__(
fullName=fullName, fullName=fullName,
@@ -140,20 +150,20 @@ class Contact(BaseMessage):
phoneNumber=phoneNumber, phoneNumber=phoneNumber,
organization=organization, organization=organization,
email=email, email=email,
url=url url=url,
) )
class ContactMessage(BaseMessage): class ContactMessage(BaseMessage):
def __init__(self, number: str, contact: List[Contact]): def __init__(self, number: str, contact: List[Contact]):
super().__init__( super().__init__(number=number, contact=[c.__dict__ for c in contact])
number=number,
contact=[c.__dict__ for c in contact]
)
class ReactionMessage(BaseMessage): class ReactionMessage(BaseMessage):
def __init__(self, key: dict, reaction: str): def __init__(self, key: dict, reaction: str):
super().__init__(key=key, reaction=reaction) super().__init__(key=key, reaction=reaction)
class PollMessage(BaseMessage): class PollMessage(BaseMessage):
def __init__( def __init__(
self, self,
@@ -162,7 +172,7 @@ class PollMessage(BaseMessage):
selectableCount: int, selectableCount: int,
values: List[str], values: List[str],
delay: Optional[int] = None, delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None quoted: Optional[QuotedMessage] = None,
): ):
super().__init__( super().__init__(
number=number, number=number,
@@ -170,23 +180,19 @@ class PollMessage(BaseMessage):
selectableCount=selectableCount, selectableCount=selectableCount,
values=values, values=values,
delay=delay, delay=delay,
quoted=quoted.__dict__ if quoted else None quoted=quoted.__dict__ if quoted else None,
) )
class ListRow(BaseMessage): class ListRow(BaseMessage):
def __init__(self, title: str, description: str, rowId: str): def __init__(self, title: str, description: str, rowId: str):
super().__init__( super().__init__(title=title, description=description, rowId=rowId)
title=title,
description=description,
rowId=rowId
)
class ListSection(BaseMessage): class ListSection(BaseMessage):
def __init__(self, title: str, rows: List[ListRow]): def __init__(self, title: str, rows: List[ListRow]):
super().__init__( super().__init__(title=title, rows=[r.__dict__ for r in rows])
title=title,
rows=[r.__dict__ for r in rows]
)
class ListMessage(BaseMessage): class ListMessage(BaseMessage):
def __init__( def __init__(
@@ -198,7 +204,7 @@ class ListMessage(BaseMessage):
footerText: str, footerText: str,
sections: List[ListSection], sections: List[ListSection],
delay: Optional[int] = None, delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None quoted: Optional[QuotedMessage] = None,
): ):
super().__init__( super().__init__(
number=number, number=number,
@@ -208,9 +214,10 @@ class ListMessage(BaseMessage):
footerText=footerText, footerText=footerText,
sections=[s.__dict__ for s in sections], sections=[s.__dict__ for s in sections],
delay=delay, delay=delay,
quoted=quoted.__dict__ if quoted else None quoted=quoted.__dict__ if quoted else None,
) )
class Button(BaseMessage): class Button(BaseMessage):
def __init__( def __init__(
self, self,
@@ -223,7 +230,7 @@ class Button(BaseMessage):
currency: Optional[str] = None, currency: Optional[str] = None,
name: Optional[str] = None, name: Optional[str] = None,
keyType: Optional[str] = None, keyType: Optional[str] = None,
key: Optional[str] = None key: Optional[str] = None,
): ):
super().__init__( super().__init__(
type=type, type=type,
@@ -235,9 +242,10 @@ class Button(BaseMessage):
currency=currency, currency=currency,
name=name, name=name,
keyType=keyType, keyType=keyType,
key=key key=key,
) )
class ButtonMessage(BaseMessage): class ButtonMessage(BaseMessage):
def __init__( def __init__(
self, self,
@@ -247,7 +255,7 @@ class ButtonMessage(BaseMessage):
footer: str, footer: str,
buttons: List[Button], buttons: List[Button],
delay: Optional[int] = None, delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None quoted: Optional[QuotedMessage] = None,
): ):
super().__init__( super().__init__(
number=number, number=number,
@@ -256,5 +264,5 @@ class ButtonMessage(BaseMessage):
footer=footer, footer=footer,
buttons=[b.__dict__ for b in buttons], buttons=[b.__dict__ for b in buttons],
delay=delay, delay=delay,
quoted=quoted.__dict__ if quoted else None quoted=quoted.__dict__ if quoted else None,
) )

4
evolutionapi/models/presence.py
View File

@@ -1,9 +1,11 @@
from enum import Enum from enum import Enum
class PresenceStatus(Enum): class PresenceStatus(Enum):
AVAILABLE = "available" AVAILABLE = "available"
UNAVAILABLE = "unavailable" UNAVAILABLE = "unavailable"
class PresenceConfig: class PresenceConfig:
def __init__(self, presence: PresenceStatus): def __init__(self, presence: PresenceStatus):
self.presence = presence.value self.presence = presence.value

8
evolutionapi/models/profile.py
View File

@@ -1,9 +1,11 @@
from typing import Literal from typing import Literal
class BaseProfile: class BaseProfile:
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None}) self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class FetchProfile(BaseProfile): class FetchProfile(BaseProfile):
def __init__( def __init__(
self, self,
@@ -13,6 +15,7 @@ class FetchProfile(BaseProfile):
number=number, number=number,
) )
class ProfileName(BaseProfile): class ProfileName(BaseProfile):
def __init__( def __init__(
self, self,
@@ -22,6 +25,7 @@ class ProfileName(BaseProfile):
name=name, name=name,
) )
class ProfileStatus(BaseProfile): class ProfileStatus(BaseProfile):
def __init__( def __init__(
self, self,
@@ -31,6 +35,7 @@ class ProfileStatus(BaseProfile):
status=status, status=status,
) )
class ProfilePicture(BaseProfile): class ProfilePicture(BaseProfile):
def __init__( def __init__(
self, self,
@@ -40,6 +45,7 @@ class ProfilePicture(BaseProfile):
picture=picture, picture=picture,
) )
class PrivacySettings(BaseProfile): class PrivacySettings(BaseProfile):
def __init__( def __init__(
self, self,
@@ -57,4 +63,4 @@ class PrivacySettings(BaseProfile):
online=online, online=online,
last=last, last=last,
groupadd=groupadd, groupadd=groupadd,
) )

6
evolutionapi/models/websocket.py
View File

@@ -1,6 +1,7 @@
from typing import List, Optional from typing import List, Optional
from dataclasses import dataclass from dataclasses import dataclass
@dataclass @dataclass
class WebSocketConfig: class WebSocketConfig:
enabled: bool enabled: bool
@@ -10,11 +11,12 @@ class WebSocketConfig:
self.enabled = enabled self.enabled = enabled
self.events = events self.events = events
@dataclass @dataclass
class WebSocketInfo: class WebSocketInfo:
enabled: bool enabled: bool
events: List[str] events: List[str]
def __init__(self, **kwargs): def __init__(self, **kwargs):
self.enabled = kwargs.get('enabled', False) self.enabled = kwargs.get("enabled", False)
self.events = kwargs.get('events', []) self.events = kwargs.get("events", [])

7
evolutionapi/services/call.py
View File

@@ -1,13 +1,12 @@
from typing import Union, BinaryIO from typing import Union, BinaryIO
from ..models.call import * from ..models.call import *
class CallService: class CallService:
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
def fake_call(self, instance_id: str, data: FakeCall, instance_token: str): def fake_call(self, instance_id: str, data: FakeCall, instance_token: str):
return self.client.post( return self.client.post(
f'call/offer/{instance_id}', f"call/offer/{instance_id}", data=data.__dict__, instance_token=instance_token
data=data.__dict__, )
instance_token=instance_token
)

8
evolutionapi/services/chat.py
View File

@@ -38,9 +38,7 @@ class ChatService:
instance_token=instance_token, instance_token=instance_token,
) )
def delete_message_for_everyone( def delete_message_for_everyone(self, instance_id: str, data: MessageKey, instance_token: str):
self, instance_id: str, data: MessageKey, instance_token: str
):
return self.client.delete( return self.client.delete(
f"chat/deleteMessageForEveryone/{instance_id}", f"chat/deleteMessageForEveryone/{instance_id}",
data=data.__dict__, data=data.__dict__,
@@ -65,9 +63,7 @@ class ChatService:
instance_token=instance_token, instance_token=instance_token,
) )
def update_message( def update_message(self, instance_id: str, data: UpdateMessage, instance_token: str):
self, instance_id: str, data: UpdateMessage, instance_token: str
):
return self.client.post( return self.client.post(
f"chat/updateMessage/{instance_id}", f"chat/updateMessage/{instance_id}",
data=data.__dict__, data=data.__dict__,

91
evolutionapi/services/group.py
View File

@@ -1,105 +1,110 @@
from typing import Optional from typing import Optional
from ..models.group import * from ..models.group import *
class GroupService: class GroupService:
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
def create_group(self, instance_id: str, data: CreateGroup, instance_token: str): def create_group(self, instance_id: str, data: CreateGroup, instance_token: str):
return self.client.post( return self.client.post(
f'group/create/{instance_id}', f"group/create/{instance_id}", data=data.__dict__, instance_token=instance_token
data=data.__dict__,
instance_token=instance_token
) )
def update_group_picture(self, instance_id: str, group_jid: str, data: GroupPicture, instance_token: str): def update_group_picture(
self, instance_id: str, group_jid: str, data: GroupPicture, instance_token: str
):
return self.client.post( return self.client.post(
f'group/updateGroupPicture/{instance_id}?groupJid={group_jid}', f"group/updateGroupPicture/{instance_id}?groupJid={group_jid}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def update_group_subject(self, instance_id: str, group_jid: str, data: GroupSubject, instance_token: str): def update_group_subject(
self, instance_id: str, group_jid: str, data: GroupSubject, instance_token: str
):
return self.client.post( return self.client.post(
f'group/updateGroupSubject/{instance_id}?groupJid={group_jid}', f"group/updateGroupSubject/{instance_id}?groupJid={group_jid}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def update_group_description(self, instance_id: str, group_jid: str, data: GroupDescription, instance_token: str): def update_group_description(
self, instance_id: str, group_jid: str, data: GroupDescription, instance_token: str
):
return self.client.post( return self.client.post(
f'group/updateGroupDescription/{instance_id}?groupJid={group_jid}', f"group/updateGroupDescription/{instance_id}?groupJid={group_jid}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def get_invite_code(self, instance_id: str, group_jid: str, instance_token: str): def get_invite_code(self, instance_id: str, group_jid: str, instance_token: str):
return self.client.get( return self.client.get(
f'group/inviteCode/{instance_id}?groupJid={group_jid}', f"group/inviteCode/{instance_id}?groupJid={group_jid}", instance_token=instance_token
instance_token=instance_token
) )
def revoke_invite_code(self, instance_id: str, group_jid: str, instance_token: str): def revoke_invite_code(self, instance_id: str, group_jid: str, instance_token: str):
return self.client.post( return self.client.post(
f'group/revokeInviteCode/{instance_id}?groupJid={group_jid}', f"group/revokeInviteCode/{instance_id}?groupJid={group_jid}",
instance_token=instance_token instance_token=instance_token,
) )
def send_invite(self, instance_id: str, data: GroupInvite, instance_token: str): def send_invite(self, instance_id: str, data: GroupInvite, instance_token: str):
return self.client.post( return self.client.post(
f'group/sendInvite/{instance_id}', f"group/sendInvite/{instance_id}", data=data.__dict__, instance_token=instance_token
data=data.__dict__,
instance_token=instance_token
) )
def get_invite_info(self, instance_id: str, invite_code: str, instance_token: str): def get_invite_info(self, instance_id: str, invite_code: str, instance_token: str):
return self.client.get( return self.client.get(
f'group/inviteInfo/{instance_id}?inviteCode={invite_code}', f"group/inviteInfo/{instance_id}?inviteCode={invite_code}",
instance_token=instance_token instance_token=instance_token,
) )
def get_group_info(self, instance_id: str, group_jid: str, instance_token: str): def get_group_info(self, instance_id: str, group_jid: str, instance_token: str):
return self.client.get( return self.client.get(
f'group/findGroupInfos/{instance_id}?groupJid={group_jid}', f"group/findGroupInfos/{instance_id}?groupJid={group_jid}",
instance_token=instance_token instance_token=instance_token,
) )
def fetch_all_groups(self, instance_id: str, instance_token: str, get_participants: bool = False): def fetch_all_groups(
url = f'group/fetchAllGroups/{instance_id}?getParticipants={str(get_participants).lower()}' self, instance_id: str, instance_token: str, get_participants: bool = False
return self.client.get( ):
url, url = f"group/fetchAllGroups/{instance_id}?getParticipants={str(get_participants).lower()}"
instance_token=instance_token return self.client.get(url, instance_token=instance_token)
)
def get_participants(self, instance_id: str, group_jid: str, instance_token: str): def get_participants(self, instance_id: str, group_jid: str, instance_token: str):
return self.client.get( return self.client.get(
f'group/participants/{instance_id}?groupJid={group_jid}', f"group/participants/{instance_id}?groupJid={group_jid}", instance_token=instance_token
instance_token=instance_token
) )
def update_participant(self, instance_id: str, group_jid: str, data: UpdateParticipant, instance_token: str): def update_participant(
self, instance_id: str, group_jid: str, data: UpdateParticipant, instance_token: str
):
return self.client.post( return self.client.post(
f'group/updateParticipant/{instance_id}?groupJid={group_jid}', f"group/updateParticipant/{instance_id}?groupJid={group_jid}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def update_setting(self, instance_id: str, group_jid: str, data: UpdateSetting, instance_token: str): def update_setting(
self, instance_id: str, group_jid: str, data: UpdateSetting, instance_token: str
):
return self.client.post( return self.client.post(
f'group/updateSetting/{instance_id}?groupJid={group_jid}', f"group/updateSetting/{instance_id}?groupJid={group_jid}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def toggle_ephemeral(self, instance_id: str, group_jid: str, data: ToggleEphemeral, instance_token: str): def toggle_ephemeral(
self, instance_id: str, group_jid: str, data: ToggleEphemeral, instance_token: str
):
return self.client.post( return self.client.post(
f'group/toggleEphemeral/{instance_id}?groupJid={group_jid}', f"group/toggleEphemeral/{instance_id}?groupJid={group_jid}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def leave_group(self, instance_id: str, group_jid: str, instance_token: str): def leave_group(self, instance_id: str, group_jid: str, instance_token: str):
return self.client.delete( return self.client.delete(
f'group/leaveGroup/{instance_id}?groupJid={group_jid}', f"group/leaveGroup/{instance_id}?groupJid={group_jid}", instance_token=instance_token
instance_token=instance_token )
)

4
evolutionapi/services/instance.py
View File

@@ -3,7 +3,7 @@ class InstanceService:
self.client = client self.client = client
def fetch_instances(self): def fetch_instances(self):
return self.client.get('instance/fetchInstances') return self.client.get("instance/fetchInstances")
def create_instance(self, config): def create_instance(self, config):
return self.client.post('instance/create', data=config.__dict__) return self.client.post("instance/create", data=config.__dict__)

15
evolutionapi/services/instance_operations.py
View File

@@ -1,28 +1,29 @@
from ..models.presence import PresenceStatus, PresenceConfig from ..models.presence import PresenceStatus, PresenceConfig
class InstanceOperationsService: class InstanceOperationsService:
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
def connect(self, instance_id: str, instance_token: str): def connect(self, instance_id: str, instance_token: str):
return self.client.get(f'instance/connect/{instance_id}', instance_token) return self.client.get(f"instance/connect/{instance_id}", instance_token)
def restart(self, instance_id: str, instance_token: str): def restart(self, instance_id: str, instance_token: str):
return self.client.post(f'instance/restart/{instance_id}', instance_token=instance_token) return self.client.post(f"instance/restart/{instance_id}", instance_token=instance_token)
def set_presence(self, instance_id: str, presence: PresenceStatus, instance_token: str): def set_presence(self, instance_id: str, presence: PresenceStatus, instance_token: str):
config = PresenceConfig(presence) config = PresenceConfig(presence)
return self.client.post( return self.client.post(
f'instance/setPresence/{instance_id}', f"instance/setPresence/{instance_id}",
data=config.__dict__, data=config.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def get_connection_state(self, instance_id: str, instance_token: str): def get_connection_state(self, instance_id: str, instance_token: str):
return self.client.get(f'instance/connectionState/{instance_id}', instance_token) return self.client.get(f"instance/connectionState/{instance_id}", instance_token)
def logout(self, instance_id: str, instance_token: str): def logout(self, instance_id: str, instance_token: str):
return self.client.delete(f'instance/logout/{instance_id}', instance_token) return self.client.delete(f"instance/logout/{instance_id}", instance_token)
def delete(self, instance_id: str, instance_token: str): def delete(self, instance_id: str, instance_token: str):
return self.client.delete(f'instance/delete/{instance_id}', instance_token) return self.client.delete(f"instance/delete/{instance_id}", instance_token)

12
evolutionapi/services/label.py
View File

@@ -1,19 +1,15 @@
from typing import Union, BinaryIO from typing import Union, BinaryIO
from ..models.label import * from ..models.label import *
class LabelService: class LabelService:
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
def find_labels(self, instance_id: str, instance_token: str): def find_labels(self, instance_id: str, instance_token: str):
return self.client.get( return self.client.get(f"label/findLabels/{instance_id}", instance_token=instance_token)
f'label/findLabels/{instance_id}',
instance_token=instance_token
)
def handle_label(self, instance_id: str, data: HandleLabel, instance_token: str): def handle_label(self, instance_id: str, data: HandleLabel, instance_token: str):
return self.client.post( return self.client.post(
f'label/handleLabel/{instance_id}', f"label/handleLabel/{instance_id}", data=data.__dict__, instance_token=instance_token
data=data.__dict__, )
instance_token=instance_token
)

218
evolutionapi/services/message.py
View File

@@ -4,202 +4,210 @@ from requests_toolbelt import MultipartEncoder
import mimetypes import mimetypes
import requests import requests
class MessageService: class MessageService:
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
def send_text(self, instance_id: str, message: TextMessage, instance_token: str): def send_text(self, instance_id: str, message: TextMessage, instance_token: str):
# Preparar os dados como JSON # Preparar os dados como JSON
data = { data = {"number": message.number, "text": message.text}
'number': message.number,
'text': message.text if hasattr(message, "delay") and message.delay is not None:
} data["delay"] = message.delay
if hasattr(message, 'delay') and message.delay is not None:
data['delay'] = message.delay
# Usar o método post do cliente que já trata JSON corretamente # Usar o método post do cliente que já trata JSON corretamente
return self.client.post( return self.client.post(
f'message/sendText/{instance_id}', f"message/sendText/{instance_id}", data=data, instance_token=instance_token
data=data,
instance_token=instance_token
) )
def send_media(self, instance_id: str, message: MediaMessage, instance_token: str, file: Union[BinaryIO, str] = None): def send_media(
self,
instance_id: str,
message: MediaMessage,
instance_token: str,
file: Union[BinaryIO, str] = None,
):
# Preparar os dados do formulário # Preparar os dados do formulário
fields = { fields = {
'number': (None, message.number, 'text/plain'), "number": (None, message.number, "text/plain"),
'mediatype': (None, message.mediatype, 'text/plain'), "mediatype": (None, message.mediatype, "text/plain"),
'mimetype': (None, message.mimetype, 'text/plain'), "mimetype": (None, message.mimetype, "text/plain"),
'caption': (None, message.caption, 'text/plain'), "caption": (None, message.caption, "text/plain"),
'fileName': (None, message.fileName, 'text/plain'), "fileName": (None, message.fileName, "text/plain"),
} }
# Adicionar delay apenas se existir # Adicionar delay apenas se existir
if hasattr(message, 'delay') and message.delay is not None: if hasattr(message, "delay") and message.delay is not None:
fields['delay'] = (None, str(message.delay), 'text/plain; type=number') fields["delay"] = (None, str(message.delay), "text/plain; type=number")
# Adicionar o arquivo se fornecido # Adicionar o arquivo se fornecido
if file: if file:
if isinstance(file, str): if isinstance(file, str):
mime_type = mimetypes.guess_type(file)[0] or 'application/octet-stream' mime_type = mimetypes.guess_type(file)[0] or "application/octet-stream"
fields['file'] = ('file', open(file, 'rb'), mime_type) fields["file"] = ("file", open(file, "rb"), mime_type)
else: else:
fields['file'] = ('file', file, 'application/octet-stream') fields["file"] = ("file", file, "application/octet-stream")
# Criar o multipart encoder # Criar o multipart encoder
multipart = MultipartEncoder(fields=fields) multipart = MultipartEncoder(fields=fields)
# Preparar os headers # Preparar os headers
headers = self.client._get_headers(instance_token) headers = self.client._get_headers(instance_token)
headers['Content-Type'] = multipart.content_type headers["Content-Type"] = multipart.content_type
# Fazer a requisição diretamente # Fazer a requisição diretamente
url = f'{self.client.base_url}/message/sendMedia/{instance_id}' url = f"{self.client.base_url}/message/sendMedia/{instance_id}"
response = requests.post( response = requests.post(url, headers=headers, data=multipart)
url,
headers=headers,
data=multipart
)
return response.json() return response.json()
def send_ptv(self, instance_id: str, message: dict, instance_token: str, file: Union[BinaryIO, str] = None): def send_ptv(
self,
instance_id: str,
message: dict,
instance_token: str,
file: Union[BinaryIO, str] = None,
):
fields = {} fields = {}
# Adiciona todos os campos do message como text/plain # Adiciona todos os campos do message como text/plain
for key, value in message.items(): for key, value in message.items():
if key == 'delay' and value is not None: if key == "delay" and value is not None:
fields[key] = (None, str(value), 'text/plain; type=number') fields[key] = (None, str(value), "text/plain; type=number")
else: else:
fields[key] = (None, str(value), 'text/plain') fields[key] = (None, str(value), "text/plain")
if file: if file:
if isinstance(file, str): if isinstance(file, str):
mime_type = mimetypes.guess_type(file)[0] or 'application/octet-stream' mime_type = mimetypes.guess_type(file)[0] or "application/octet-stream"
fields['file'] = ('file', open(file, 'rb'), mime_type) fields["file"] = ("file", open(file, "rb"), mime_type)
else: else:
fields['file'] = ('file', file, 'application/octet-stream') fields["file"] = ("file", file, "application/octet-stream")
multipart = MultipartEncoder(fields=fields) multipart = MultipartEncoder(fields=fields)
headers = self.client._get_headers(instance_token) headers = self.client._get_headers(instance_token)
headers['Content-Type'] = multipart.content_type headers["Content-Type"] = multipart.content_type
url = f'{self.client.base_url}/message/sendPtv/{instance_id}' url = f"{self.client.base_url}/message/sendPtv/{instance_id}"
response = requests.post(url, headers=headers, data=multipart) response = requests.post(url, headers=headers, data=multipart)
return response.json() return response.json()
def send_whatsapp_audio(self, instance_id: str, message: dict, instance_token: str, file: Union[BinaryIO, str] = None): def send_whatsapp_audio(
self,
instance_id: str,
message: dict,
instance_token: str,
file: Union[BinaryIO, str] = None,
):
fields = {} fields = {}
# Adiciona todos os campos do message como text/plain # Adiciona todos os campos do message como text/plain
for key, value in message.items(): for key, value in message.items():
if key == 'delay' and value is not None: if key == "delay" and value is not None:
fields[key] = (None, str(value), 'text/plain; type=number') fields[key] = (None, str(value), "text/plain; type=number")
else: else:
fields[key] = (None, str(value), 'text/plain') fields[key] = (None, str(value), "text/plain")
if file: if file:
if isinstance(file, str): if isinstance(file, str):
mime_type = mimetypes.guess_type(file)[0] or 'application/octet-stream' mime_type = mimetypes.guess_type(file)[0] or "application/octet-stream"
fields['file'] = ('file', open(file, 'rb'), mime_type) fields["file"] = ("file", open(file, "rb"), mime_type)
else: else:
fields['file'] = ('file', file, 'application/octet-stream') fields["file"] = ("file", file, "application/octet-stream")
multipart = MultipartEncoder(fields=fields) multipart = MultipartEncoder(fields=fields)
headers = self.client._get_headers(instance_token) headers = self.client._get_headers(instance_token)
headers['Content-Type'] = multipart.content_type headers["Content-Type"] = multipart.content_type
url = f'{self.client.base_url}/message/sendWhatsAppAudio/{instance_id}' url = f"{self.client.base_url}/message/sendWhatsAppAudio/{instance_id}"
response = requests.post(url, headers=headers, data=multipart) response = requests.post(url, headers=headers, data=multipart)
return response.json() return response.json()
def send_status(self, instance_id: str, message: StatusMessage, instance_token: str): def send_status(self, instance_id: str, message: StatusMessage, instance_token: str):
return self.client.post( return self.client.post(
f'message/sendStatus/{instance_id}', f"message/sendStatus/{instance_id}",
data=message.__dict__, data=message.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def send_sticker(self, instance_id: str, message: dict, instance_token: str, file: Union[BinaryIO, str] = None): def send_sticker(
self,
instance_id: str,
message: dict,
instance_token: str,
file: Union[BinaryIO, str] = None,
):
fields = {} fields = {}
# Adiciona todos os campos do message como text/plain # Adiciona todos os campos do message como text/plain
for key, value in message.items(): for key, value in message.items():
if key == 'delay' and value is not None: if key == "delay" and value is not None:
fields[key] = (None, str(value), 'text/plain; type=number') fields[key] = (None, str(value), "text/plain; type=number")
else: else:
fields[key] = (None, str(value), 'text/plain') fields[key] = (None, str(value), "text/plain")
if file: if file:
if isinstance(file, str): if isinstance(file, str):
mime_type = mimetypes.guess_type(file)[0] or 'application/octet-stream' mime_type = mimetypes.guess_type(file)[0] or "application/octet-stream"
fields['file'] = ('file', open(file, 'rb'), mime_type) fields["file"] = ("file", open(file, "rb"), mime_type)
else: else:
fields['file'] = ('file', file, 'application/octet-stream') fields["file"] = ("file", file, "application/octet-stream")
multipart = MultipartEncoder(fields=fields) multipart = MultipartEncoder(fields=fields)
headers = self.client._get_headers(instance_token) headers = self.client._get_headers(instance_token)
headers['Content-Type'] = multipart.content_type headers["Content-Type"] = multipart.content_type
url = f'{self.client.base_url}/message/sendSticker/{instance_id}' url = f"{self.client.base_url}/message/sendSticker/{instance_id}"
response = requests.post(url, headers=headers, data=multipart) response = requests.post(url, headers=headers, data=multipart)
return response.json() return response.json()
def send_location(self, instance_id: str, message: LocationMessage, instance_token: str): def send_location(self, instance_id: str, message: LocationMessage, instance_token: str):
data = message.__dict__.copy() data = message.__dict__.copy()
if 'delay' in data and data['delay'] is not None: if "delay" in data and data["delay"] is not None:
data['delay'] = int(data['delay']) data["delay"] = int(data["delay"])
return self.client.post( return self.client.post(
f'message/sendLocation/{instance_id}', f"message/sendLocation/{instance_id}", data=data, instance_token=instance_token
data=data,
instance_token=instance_token
) )
def send_contact(self, instance_id: str, message: ContactMessage, instance_token: str): def send_contact(self, instance_id: str, message: ContactMessage, instance_token: str):
return self.client.post( return self.client.post(
f'message/sendContact/{instance_id}', f"message/sendContact/{instance_id}",
data=message.__dict__, data=message.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def send_reaction(self, instance_id: str, message: ReactionMessage, instance_token: str): def send_reaction(self, instance_id: str, message: ReactionMessage, instance_token: str):
return self.client.post( return self.client.post(
f'message/sendReaction/{instance_id}', f"message/sendReaction/{instance_id}",
data=message.__dict__, data=message.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def send_poll(self, instance_id: str, message: PollMessage, instance_token: str): def send_poll(self, instance_id: str, message: PollMessage, instance_token: str):
data = message.__dict__.copy() data = message.__dict__.copy()
if 'delay' in data and data['delay'] is not None: if "delay" in data and data["delay"] is not None:
data['delay'] = int(data['delay']) data["delay"] = int(data["delay"])
return self.client.post( return self.client.post(
f'message/sendPoll/{instance_id}', f"message/sendPoll/{instance_id}", data=data, instance_token=instance_token
data=data,
instance_token=instance_token
) )
def send_list(self, instance_id: str, message: ListMessage, instance_token: str): def send_list(self, instance_id: str, message: ListMessage, instance_token: str):
data = message.__dict__.copy() data = message.__dict__.copy()
if 'delay' in data and data['delay'] is not None: if "delay" in data and data["delay"] is not None:
data['delay'] = int(data['delay']) data["delay"] = int(data["delay"])
return self.client.post( return self.client.post(
f'message/sendList/{instance_id}', f"message/sendList/{instance_id}", data=data, instance_token=instance_token
data=data,
instance_token=instance_token
) )
def send_buttons(self, instance_id: str, message: ButtonMessage, instance_token: str): def send_buttons(self, instance_id: str, message: ButtonMessage, instance_token: str):
data = message.__dict__.copy() data = message.__dict__.copy()
if 'delay' in data and data['delay'] is not None: if "delay" in data and data["delay"] is not None:
data['delay'] = int(data['delay']) data["delay"] = int(data["delay"])
return self.client.post( return self.client.post(
f'message/sendButtons/{instance_id}', f"message/sendButtons/{instance_id}", data=data, instance_token=instance_token
data=data, )
instance_token=instance_token
)

35
evolutionapi/services/profile.py
View File

@@ -1,60 +1,57 @@
from typing import Union, BinaryIO from typing import Union, BinaryIO
from ..models.profile import * from ..models.profile import *
class ProfileService: class ProfileService:
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
def fetch_business_profile(self, instance_id: str, data: FetchProfile, instance_token: str): def fetch_business_profile(self, instance_id: str, data: FetchProfile, instance_token: str):
return self.client.post( return self.client.post(
f'chat/fetchBusinessProfile/{instance_id}', f"chat/fetchBusinessProfile/{instance_id}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def fetch_profile(self, instance_id: str, data: FetchProfile, instance_token: str): def fetch_profile(self, instance_id: str, data: FetchProfile, instance_token: str):
return self.client.post( return self.client.post(
f'chat/fetchProfile/{instance_id}', f"chat/fetchProfile/{instance_id}", data=data.__dict__, instance_token=instance_token
data=data.__dict__,
instance_token=instance_token
) )
def update_profile_name(self, instance_id: str, data: ProfileName, instance_token: str): def update_profile_name(self, instance_id: str, data: ProfileName, instance_token: str):
return self.client.post( return self.client.post(
f'chat/updateProfileName/{instance_id}', f"chat/updateProfileName/{instance_id}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def update_profile_status(self, instance_id: str, data: ProfileStatus, instance_token: str): def update_profile_status(self, instance_id: str, data: ProfileStatus, instance_token: str):
return self.client.post( return self.client.post(
f'chat/updateProfileStatus/{instance_id}', f"chat/updateProfileStatus/{instance_id}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def update_profile_picture(self, instance_id: str, data: ProfilePicture, instance_token: str): def update_profile_picture(self, instance_id: str, data: ProfilePicture, instance_token: str):
return self.client.post( return self.client.post(
f'chat/updateProfilePicture/{instance_id}', f"chat/updateProfilePicture/{instance_id}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )
def remove_profile_picture(self, instance_id: str, instance_token: str): def remove_profile_picture(self, instance_id: str, instance_token: str):
return self.client.delete( return self.client.delete(
f'chat/removeProfilePicture/{instance_id}', f"chat/removeProfilePicture/{instance_id}", instance_token=instance_token
instance_token=instance_token
) )
def fetch_privacy_settings(self, instance_id: str, instance_token: str): def fetch_privacy_settings(self, instance_id: str, instance_token: str):
return self.client.get( return self.client.get(
f'chat/fetchPrivacySettings/{instance_id}', f"chat/fetchPrivacySettings/{instance_id}", instance_token=instance_token
instance_token=instance_token
) )
def update_privacy_settings(self, instance_id: str, data: PrivacySettings, instance_token: str): def update_privacy_settings(self, instance_id: str, data: PrivacySettings, instance_token: str):
return self.client.post( return self.client.post(
f'chat/updatePrivacySettings/{instance_id}', f"chat/updatePrivacySettings/{instance_id}",
data=data.__dict__, data=data.__dict__,
instance_token=instance_token instance_token=instance_token,
) )

96
evolutionapi/services/websocket.py
View File

@@ -5,6 +5,7 @@ import time
from typing import Optional from typing import Optional
from ..models.websocket import WebSocketConfig, WebSocketInfo from ..models.websocket import WebSocketConfig, WebSocketInfo
class WebSocketService: class WebSocketService:
def __init__(self, client): def __init__(self, client):
self.client = client self.client = client
@@ -12,43 +13,46 @@ class WebSocketService:
def set_websocket(self, instance_id: str, config: WebSocketConfig, instance_token: str): def set_websocket(self, instance_id: str, config: WebSocketConfig, instance_token: str):
""" """
Configure WebSocket settings for an instance Configure WebSocket settings for an instance
Args: Args:
instance_id (str): The instance ID instance_id (str): The instance ID
config (WebSocketConfig): The WebSocket configuration config (WebSocketConfig): The WebSocket configuration
instance_token (str): The instance token instance_token (str): The instance token
Returns: Returns:
dict: The response from the API dict: The response from the API
""" """
return self.client.post( return self.client.post(
f'websocket/set/{instance_id}', f"websocket/set/{instance_id}", data=config.__dict__, instance_token=instance_token
data=config.__dict__,
instance_token=instance_token
) )
def find_websocket(self, instance_id: str, instance_token: str) -> WebSocketInfo: def find_websocket(self, instance_id: str, instance_token: str) -> WebSocketInfo:
""" """
Get WebSocket settings for an instance Get WebSocket settings for an instance
Args: Args:
instance_id (str): The instance ID instance_id (str): The instance ID
instance_token (str): The instance token instance_token (str): The instance token
Returns: Returns:
WebSocketInfo: The WebSocket information WebSocketInfo: The WebSocket information
""" """
response = self.client.get( response = self.client.get(f"websocket/find/{instance_id}", instance_token=instance_token)
f'websocket/find/{instance_id}',
instance_token=instance_token
)
return WebSocketInfo(**response) return WebSocketInfo(**response)
class WebSocketManager: class WebSocketManager:
def __init__(self, base_url: str, instance_id: str, api_token: str, max_retries: int = 5, retry_delay: float = 1.0): def __init__(
self,
base_url: str,
instance_id: str,
api_token: str,
max_retries: int = 5,
retry_delay: float = 1.0,
):
""" """
Initialize the WebSocket manager Initialize the WebSocket manager
Args: Args:
base_url (str): Base URL of the API base_url (str): Base URL of the API
instance_id (str): Instance ID instance_id (str): Instance ID
@@ -56,58 +60,60 @@ class WebSocketManager:
max_retries (int): Maximum number of reconnection attempts max_retries (int): Maximum number of reconnection attempts
retry_delay (float): Initial delay between attempts in seconds retry_delay (float): Initial delay between attempts in seconds
""" """
self.base_url = base_url.rstrip('/') self.base_url = base_url.rstrip("/")
self.instance_id = instance_id self.instance_id = instance_id
self.api_token = api_token self.api_token = api_token
self.max_retries = max_retries self.max_retries = max_retries
self.retry_delay = retry_delay self.retry_delay = retry_delay
self.retry_count = 0 self.retry_count = 0
self.should_reconnect = True self.should_reconnect = True
# Socket.IO configuration # Socket.IO configuration
self.sio = socketio.Client( self.sio = socketio.Client(
ssl_verify=False, # For local development ssl_verify=False, # For local development
logger=False, logger=False,
engineio_logger=False, engineio_logger=False,
request_timeout=30 request_timeout=30,
) )
# Configure class logger to INFO # Configure class logger to INFO
self.logger = logging.getLogger(__name__) self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.INFO) self.logger.setLevel(logging.INFO)
# Dictionary to store registered handlers # Dictionary to store registered handlers
self._handlers = {} self._handlers = {}
# Configure event handlers # Configure event handlers
self.sio.on('connect', self._on_connect) self.sio.on("connect", self._on_connect)
self.sio.on('disconnect', self._on_disconnect) self.sio.on("disconnect", self._on_disconnect)
self.sio.on('error', self._on_error) self.sio.on("error", self._on_error)
# Register global handler in instance-specific namespace # Register global handler in instance-specific namespace
self.sio.on('*', self._handle_event, namespace=f'/{self.instance_id}') self.sio.on("*", self._handle_event, namespace=f"/{self.instance_id}")
def _on_connect(self): def _on_connect(self):
"""Handler for connection event""" """Handler for connection event"""
self.logger.info("Socket.IO connected") self.logger.info("Socket.IO connected")
self.retry_count = 0 # Reset retry counter after successful connection self.retry_count = 0 # Reset retry counter after successful connection
def _on_disconnect(self): def _on_disconnect(self):
"""Handler for disconnection event""" """Handler for disconnection event"""
self.logger.warning(f"Socket.IO disconnected. Attempt {self.retry_count + 1}/{self.max_retries}") self.logger.warning(
f"Socket.IO disconnected. Attempt {self.retry_count + 1}/{self.max_retries}"
)
if self.should_reconnect and self.retry_count < self.max_retries: if self.should_reconnect and self.retry_count < self.max_retries:
self._attempt_reconnect() self._attempt_reconnect()
else: else:
self.logger.error("Maximum number of reconnection attempts reached") self.logger.error("Maximum number of reconnection attempts reached")
def _on_error(self, error): def _on_error(self, error):
"""Handler for error events""" """Handler for error events"""
self.logger.error(f"Socket.IO error: {str(error)}", exc_info=True) self.logger.error(f"Socket.IO error: {str(error)}", exc_info=True)
def _attempt_reconnect(self): def _attempt_reconnect(self):
"""Attempt to reconnect with exponential backoff""" """Attempt to reconnect with exponential backoff"""
try: try:
delay = self.retry_delay * (2 ** self.retry_count) # Exponential backoff delay = self.retry_delay * (2**self.retry_count) # Exponential backoff
self.logger.info(f"Attempting to reconnect in {delay:.2f} seconds...") self.logger.info(f"Attempting to reconnect in {delay:.2f} seconds...")
time.sleep(delay) time.sleep(delay)
self.connect() self.connect()
@@ -118,18 +124,18 @@ class WebSocketManager:
self._attempt_reconnect() self._attempt_reconnect()
else: else:
self.logger.error("All reconnection attempts failed") self.logger.error("All reconnection attempts failed")
def _handle_event(self, event, *args): def _handle_event(self, event, *args):
"""Global handler for all events""" """Global handler for all events"""
# Only process registered events # Only process registered events
if event in self._handlers: if event in self._handlers:
self.logger.debug(f"Event received in namespace /{self.instance_id}: {event}") self.logger.debug(f"Event received in namespace /{self.instance_id}: {event}")
self.logger.debug(f"Event data: {args}") self.logger.debug(f"Event data: {args}")
try: try:
# Extract event data # Extract event data
raw_data = args[0] if args else {} raw_data = args[0] if args else {}
# Ensure we're passing the correct object to the callback # Ensure we're passing the correct object to the callback
if isinstance(raw_data, dict): if isinstance(raw_data, dict):
self.logger.debug(f"Calling handler for {event} with data: {raw_data}") self.logger.debug(f"Calling handler for {event} with data: {raw_data}")
@@ -138,37 +144,39 @@ class WebSocketManager:
self.logger.error(f"Invalid data received for event {event}: {raw_data}") self.logger.error(f"Invalid data received for event {event}: {raw_data}")
except Exception as e: except Exception as e:
self.logger.error(f"Error processing event {event}: {str(e)}", exc_info=True) self.logger.error(f"Error processing event {event}: {str(e)}", exc_info=True)
def connect(self): def connect(self):
"""Connect to Socket.IO server""" """Connect to Socket.IO server"""
try: try:
# Connect only to instance namespace with authentication header # Connect only to instance namespace with authentication header
self.sio.connect( self.sio.connect(
f"{self.base_url}?apikey={self.api_token}", f"{self.base_url}?apikey={self.api_token}",
transports=['websocket'], transports=["websocket"],
namespaces=[f'/{self.instance_id}'], namespaces=[f"/{self.instance_id}"],
wait_timeout=30 wait_timeout=30,
) )
# Join instance-specific room # Join instance-specific room
self.sio.emit('subscribe', {'instance': self.instance_id}, namespace=f'/{self.instance_id}') self.sio.emit(
"subscribe", {"instance": self.instance_id}, namespace=f"/{self.instance_id}"
)
except Exception as e: except Exception as e:
self.logger.error(f"Error connecting to Socket.IO: {str(e)}", exc_info=True) self.logger.error(f"Error connecting to Socket.IO: {str(e)}", exc_info=True)
raise raise
def disconnect(self): def disconnect(self):
"""Disconnect from Socket.IO server""" """Disconnect from Socket.IO server"""
self.should_reconnect = False # Prevent reconnection attempts self.should_reconnect = False # Prevent reconnection attempts
if self.sio.connected: if self.sio.connected:
self.sio.disconnect() self.sio.disconnect()
def on(self, event: str, callback: Callable): def on(self, event: str, callback: Callable):
""" """
Register a callback for a specific event Register a callback for a specific event
Args: Args:
event (str): Event name event (str): Event name
callback (Callable): Function to be called when the event occurs callback (Callable): Function to be called when the event occurs
""" """
self._handlers[event] = callback self._handlers[event] = callback

55
test_evolution.py
View File

@@ -7,16 +7,14 @@ import logging
# Configuração do logging # Configuração do logging
logging.basicConfig( logging.basicConfig(
level=logging.DEBUG, level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
) )
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
print("Iniciando cliente") print("Iniciando cliente")
client = EvolutionClient( client = EvolutionClient(
base_url='http://localhost:8081', base_url="http://localhost:8081", api_token="429683C4C977415CAAFCCE10F7D57E11"
api_token='429683C4C977415CAAFCCE10F7D57E11'
) )
instance_token = "82D55E57CBBC-48A5-98FB-E99655AE7148" instance_token = "82D55E57CBBC-48A5-98FB-E99655AE7148"
@@ -49,8 +47,8 @@ websocket_config = WebSocketConfig(
"LABELS_ASSOCIATION", "LABELS_ASSOCIATION",
"CALL", "CALL",
"TYPEBOT_START", "TYPEBOT_START",
"TYPEBOT_CHANGE_STATUS" "TYPEBOT_CHANGE_STATUS",
] ],
) )
# Configurando WebSocket para a instância # Configurando WebSocket para a instância
@@ -66,60 +64,65 @@ logger.info(f"Eventos configurados: {websocket_info.events}")
# Criando gerenciador WebSocket usando o cliente # Criando gerenciador WebSocket usando o cliente
logger.info("Criando gerenciador WebSocket...") logger.info("Criando gerenciador WebSocket...")
websocket_manager = client.create_websocket( websocket_manager = client.create_websocket(
instance_id=instance_id, instance_id=instance_id, api_token=instance_token, max_retries=5, retry_delay=1.0
api_token=instance_token,
max_retries=5,
retry_delay=1.0
) )
def on_message(data): def on_message(data):
"""Handler para evento de mensagens""" """Handler para evento de mensagens"""
try: try:
if 'data' in data: if "data" in data:
message_data = data['data'] message_data = data["data"]
logger.info("=== Mensagem Recebida ===") logger.info("=== Mensagem Recebida ===")
logger.info(f"De: {message_data['key']['remoteJid']}") logger.info(f"De: {message_data['key']['remoteJid']}")
logger.info(f"Tipo: {message_data['messageType']}") logger.info(f"Tipo: {message_data['messageType']}")
# Extrai o conteúdo baseado no tipo da mensagem # Extrai o conteúdo baseado no tipo da mensagem
if 'message' in message_data: if "message" in message_data:
if 'conversation' in message_data['message']: if "conversation" in message_data["message"]:
logger.info(f"Conteúdo: {message_data['message']['conversation']}") logger.info(f"Conteúdo: {message_data['message']['conversation']}")
elif 'extendedTextMessage' in message_data['message']: elif "extendedTextMessage" in message_data["message"]:
logger.info(f"Conteúdo: {message_data['message']['extendedTextMessage']['text']}") logger.info(
elif 'imageMessage' in message_data['message']: f"Conteúdo: {message_data['message']['extendedTextMessage']['text']}"
logger.info(f"Conteúdo: [Imagem] {message_data['message']['imageMessage'].get('caption', '')}") )
elif "imageMessage" in message_data["message"]:
logger.info(
f"Conteúdo: [Imagem] {message_data['message']['imageMessage'].get('caption', '')}"
)
else: else:
logger.info(f"Conteúdo: {message_data['message']}") logger.info(f"Conteúdo: {message_data['message']}")
logger.info("=======================") logger.info("=======================")
except Exception as e: except Exception as e:
logger.error(f"Erro ao processar mensagem: {e}", exc_info=True) logger.error(f"Erro ao processar mensagem: {e}", exc_info=True)
def on_qrcode(data): def on_qrcode(data):
"""Handler para evento de QR Code""" """Handler para evento de QR Code"""
logger.info("=== QR Code Atualizado ===") logger.info("=== QR Code Atualizado ===")
logger.info(f"QR Code: {data}") logger.info(f"QR Code: {data}")
logger.info("=======================") logger.info("=======================")
def on_connection(data): def on_connection(data):
"""Handler para evento de conexão""" """Handler para evento de conexão"""
logger.info("=== Status de Conexão ===") logger.info("=== Status de Conexão ===")
logger.info(f"Status: {data}") logger.info(f"Status: {data}")
logger.info("=======================") logger.info("=======================")
logger.info("Registrando handlers de eventos...") logger.info("Registrando handlers de eventos...")
# Registrando handlers de eventos # Registrando handlers de eventos
websocket_manager.on('messages.upsert', on_message) websocket_manager.on("messages.upsert", on_message)
websocket_manager.on('qrcode.updated', on_qrcode) websocket_manager.on("qrcode.updated", on_qrcode)
websocket_manager.on('connection.update', on_connection) websocket_manager.on("connection.update", on_connection)
try: try:
logger.info("Iniciando conexão WebSocket...") logger.info("Iniciando conexão WebSocket...")
# Conectando ao WebSocket # Conectando ao WebSocket
websocket_manager.connect() websocket_manager.connect()
# Mantendo o programa rodando para receber eventos # Mantendo o programa rodando para receber eventos
logger.info("Aguardando eventos...") logger.info("Aguardando eventos...")
while True: while True:
@@ -153,4 +156,4 @@ finally:
# response = client.messages.send_media(instance_id, media_message, instance_token, "arquivo.pdf") # response = client.messages.send_media(instance_id, media_message, instance_token, "arquivo.pdf")
# print("Mensagem de mídia enviada") # print("Mensagem de mídia enviada")
# print(response) # print(response)