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

build: remove legacy setup.py and build artifacts

Browse Source
This commit is contained in:
Jackson Vieira 2025-12-03 14:02:00 -03:00
parent 7111cd9969
commit c97d008c1d
45 changed files with 0 additions and 2548 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
build/lib/evolution/__init__.py
View File

81
build/lib/evolution/client.py
View File

@ -1,81 +0,0 @@
import requests
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
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)
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):
"""Faz uma requisição POST."""
url = self._get_full_url(endpoint)
response = requests.post(url, headers=self._get_headers(instance_token), json=data)
return self._handle_response(response)
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)

11
build/lib/evolution/exceptions.py
View File

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

0
build/lib/evolution/models/__init__.py
View File

16
build/lib/evolution/models/call.py
View File

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

93
build/lib/evolution/models/chat.py
View File

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

39
build/lib/evolution/models/group.py
View File

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

59
build/lib/evolution/models/instance.py
View File

@ -1,59 +0,0 @@
from typing import Optional, List, Dict
class WebhookConfig:
def __init__(self, url: str = None, byEvents: bool = False, base64: bool = True,
headers: Dict = None, events: List[str] = None):
self.url = url
self.byEvents = byEvents
self.base64 = base64
self.headers = headers
self.events = events
class EventsConfig:
def __init__(self, enabled: bool = True, events: List[str] = None):
self.enabled = enabled
self.events = events
class ChatwootConfig:
def __init__(self, accountId: str = None, token: str = None, url: str = None,
signMsg: bool = True, 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.chatwootToken = token
self.chatwootUrl = url
self.chatwootSignMsg = signMsg
self.chatwootReopenConversation = reopenConversation
self.chatwootConversationPending = conversationPending
self.chatwootImportContacts = importContacts
self.chatwootNameInbox = nameInbox
self.chatwootMergeBrazilContacts = mergeBrazilContacts
self.chatwootImportMessages = importMessages
self.chatwootDaysLimitImportMessages = daysLimitImportMessages
self.chatwootOrganization = organization
self.chatwootLogo = logo
class InstanceConfig:
def __init__(
self,
instanceName: str,
integration: str = None,
token: str = None,
number: str = None,
qrcode: bool = None,
rejectCall: bool = None,
msgCall: str = None,
groupsIgnore: bool = None,
alwaysOnline: bool = None,
readMessages: bool = None,
readStatus: bool = None,
syncFullHistory: bool = None
):
self.__dict__['instanceName'] = instanceName
for key, value in locals().items():
if key != 'self' and key != 'instanceName' and value is not None:
self.__dict__[key] = value

21
build/lib/evolution/models/label.py
View File

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

254
build/lib/evolution/models/message.py
View File

@ -1,254 +0,0 @@
from enum import Enum
from typing import List, Optional, Union
from dataclasses import dataclass
class MediaType(Enum):
IMAGE = "image"
VIDEO = "video"
DOCUMENT = "document"
class StatusType(Enum):
TEXT = "text"
IMAGE = "image"
VIDEO = "video"
AUDIO = "audio"
class FontType(Enum):
SERIF = 1
NORICAN_REGULAR = 2
BRYNDAN_WRITE = 3
BEBASNEUE_REGULAR = 4
OSWALD_HEAVY = 5
class BaseMessage:
def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class QuotedMessage(BaseMessage):
def __init__(self, key: dict, message: Optional[dict] = None):
super().__init__(key=key, message=message)
class TextMessage(BaseMessage):
def __init__(
self,
number: str,
text: str,
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None,
linkPreview: Optional[bool] = None,
mentionsEveryOne: Optional[bool] = None,
mentioned: Optional[List[str]] = None
):
super().__init__(
number=number,
text=text,
delay=delay,
quoted=quoted.__dict__ if quoted else None,
linkPreview=linkPreview,
mentionsEveryOne=mentionsEveryOne,
mentioned=mentioned
)
class MediaMessage(BaseMessage):
def __init__(
self,
number: str,
mediatype: str,
mimetype: str,
caption: str,
media: str,
fileName: str,
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None,
mentionsEveryOne: Optional[bool] = None,
mentioned: Optional[List[str]] = None
):
super().__init__(
number=number,
mediatype=mediatype,
mimetype=mimetype,
caption=caption,
media=media,
fileName=fileName,
delay=delay,
quoted=quoted.__dict__ if quoted else None,
mentionsEveryOne=mentionsEveryOne,
mentioned=mentioned
)
class StatusMessage(BaseMessage):
def __init__(
self,
type: StatusType,
content: str,
caption: Optional[str] = None,
backgroundColor: Optional[str] = None,
font: Optional[FontType] = None,
allContacts: bool = False,
statusJidList: Optional[List[str]] = None
):
super().__init__(
type=type.value,
content=content,
caption=caption,
backgroundColor=backgroundColor,
font=font.value if font else None,
allContacts=allContacts,
statusJidList=statusJidList
)
class LocationMessage(BaseMessage):
def __init__(
self,
number: str,
name: str,
address: str,
latitude: float,
longitude: float,
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None
):
super().__init__(
number=number,
name=name,
address=address,
latitude=latitude,
longitude=longitude,
delay=delay,
quoted=quoted.__dict__ if quoted else None
)
class Contact(BaseMessage):
def __init__(
self,
fullName: str,
wuid: str,
phoneNumber: str,
organization: Optional[str] = None,
email: Optional[str] = None,
url: Optional[str] = None
):
super().__init__(
fullName=fullName,
wuid=wuid,
phoneNumber=phoneNumber,
organization=organization,
email=email,
url=url
)
class ContactMessage(BaseMessage):
def __init__(self, number: str, contact: List[Contact]):
super().__init__(
number=number,
contact=[c.__dict__ for c in contact]
)
class ReactionMessage(BaseMessage):
def __init__(self, key: dict, reaction: str):
super().__init__(key=key, reaction=reaction)
class PollMessage(BaseMessage):
def __init__(
self,
number: str,
name: str,
selectableCount: int,
values: List[str],
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None
):
super().__init__(
number=number,
name=name,
selectableCount=selectableCount,
values=values,
delay=delay,
quoted=quoted.__dict__ if quoted else None
)
class ListRow(BaseMessage):
def __init__(self, title: str, description: str, rowId: str):
super().__init__(
title=title,
description=description,
rowId=rowId
)
class ListSection(BaseMessage):
def __init__(self, title: str, rows: List[ListRow]):
super().__init__(
title=title,
rows=[r.__dict__ for r in rows]
)
class ListMessage(BaseMessage):
def __init__(
self,
number: str,
title: str,
description: str,
buttonText: str,
footerText: str,
sections: List[ListSection],
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None
):
super().__init__(
number=number,
title=title,
description=description,
buttonText=buttonText,
footerText=footerText,
sections=[s.__dict__ for s in sections],
delay=delay,
quoted=quoted.__dict__ if quoted else None
)
class Button(BaseMessage):
def __init__(
self,
type: str,
displayText: str,
id: Optional[str] = None,
copyCode: Optional[str] = None,
url: Optional[str] = None,
phoneNumber: Optional[str] = None,
currency: Optional[str] = None,
name: Optional[str] = None,
keyType: Optional[str] = None,
key: Optional[str] = None
):
super().__init__(
type=type,
displayText=displayText,
id=id,
copyCode=copyCode,
url=url,
phoneNumber=phoneNumber,
currency=currency,
name=name,
keyType=keyType,
key=key
)
class ButtonMessage(BaseMessage):
def __init__(
self,
number: str,
title: str,
description: str,
footer: str,
buttons: List[Button],
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None
):
super().__init__(
number=number,
title=title,
description=description,
footer=footer,
buttons=[b.__dict__ for b in buttons],
delay=delay,
quoted=quoted.__dict__ if quoted else None
)

9
build/lib/evolution/models/presence.py
View File

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

60
build/lib/evolution/models/profile.py
View File

@ -1,60 +0,0 @@
from typing import Literal
class BaseProfile:
def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class FetchProfile(BaseProfile):
def __init__(
self,
number: str,
):
super().__init__(
number=number,
)
class ProfileName(BaseProfile):
def __init__(
self,
name: str,
):
super().__init__(
name=name,
)
class ProfileStatus(BaseProfile):
def __init__(
self,
status: str,
):
super().__init__(
status=status,
)
class ProfilePicture(BaseProfile):
def __init__(
self,
picture: str,
):
super().__init__(
picture=picture,
)
class PrivacySettings(BaseProfile):
def __init__(
self,
readreceipts: Literal["all", "none"],
profile: Literal["all", "contacts", "contact_blacklist", "none"],
status: Literal["all", "contacts", "contact_blacklist", "none"],
online: Literal["all", "match_last_seen"],
last: Literal["all", "contacts", "contact_blacklist", "none"],
groupadd: Literal["all", "contacts", "contact_blacklist"],
):
super().__init__(
readreceipts=readreceipts,
profile=profile,
status=status,
online=online,
last=last,
groupadd=groupadd,
)

0
build/lib/evolution/services/__init__.py
View File

13
build/lib/evolution/services/call.py
View File

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

69
build/lib/evolution/services/chat.py
View File

@ -1,69 +0,0 @@
from typing import Union, BinaryIO
from ..models.chat import *
class ChatService:
def __init__(self, client):
self.client = client
def check_is_whatsapp_numbers(self, instance_id: str, data: CheckIsWhatsappNumber, instance_token: str):
return self.client.post(
f'chat/checkIsWhatsappNumber/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def mark_message_as_read(self, instance_id: str, messages: List[ReadMessage], instance_token: str):
return self.client.post(
f'chat/markMessageAsRead/{instance_id}',
data={"readMessages": [m.__dict__ for m in messages]},
instance_token=instance_token
)
def archive_chat(self, instance_id: str, data: ArchiveChat, instance_token: str):
return self.client.post(
f'chat/archiveChat/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def mark_chat_unread(self, instance_id: str, data: UnreadChat, instance_token: str):
return self.client.post(
f'chat/markChatUnread/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def delete_message_for_everyone(self, instance_id: str, data: MessageKey, instance_token: str):
return self.client.delete(
f'chat/deleteMessageForEveryone/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def fetch_profile_picture_url(self, instance_id: str, data: ProfilePicture, instance_token: str):
return self.client.post(
f'chat/fetchProfilePictureUrl/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def get_base64_from_media_message(self, instance_id: str, data: MediaMessage, instance_token: str):
return self.client.post(
f'chat/getBase64FromMediaMessage/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def update_message(self, instance_id: str, data: UpdateMessage, instance_token: str):
return self.client.post(
f'chat/updateMessage/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def send_presence(self, instance_id: str, data: Presence, instance_token: str):
return self.client.post(
f'chat/sendPresence/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)

117
build/lib/evolution/services/group.py
View File

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

9
build/lib/evolution/services/instance.py
View File

@ -1,9 +0,0 @@
class InstanceService:
def __init__(self, client):
self.client = client
def fetch_instances(self):
return self.client.get('instance/fetchInstances')
def create_instance(self, config):
return self.client.post('instance/create', data=config.__dict__)

28
build/lib/evolution/services/instance_operations.py
View File

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

19
build/lib/evolution/services/label.py
View File

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

111
build/lib/evolution/services/message.py
View File

@ -1,111 +0,0 @@
from typing import Union, BinaryIO
from ..models.message import *
class MessageService:
def __init__(self, client):
self.client = client
def send_text(self, instance_id: str, message: TextMessage, instance_token: str):
return self.client.post(
f'message/sendText/{instance_id}',
data=message.__dict__,
instance_token=instance_token
)
def send_media(self, instance_id: str, message: MediaMessage, instance_token: str, file: BinaryIO = None):
payload = {
'data': message.__dict__,
'instance_token': instance_token
}
if file:
payload['files'] = {'file': file}
return self.client.post(
f'message/sendMedia/{instance_id}',
**payload
)
def send_ptv(self, instance_id: str, message: dict, instance_token: str, file: BinaryIO = None):
payload = {
'data': message,
'instance_token': instance_token
}
if file:
payload['files'] = {'file': file}
return self.client.post(
f'message/sendPtv/{instance_id}',
**payload
)
def send_whatsapp_audio(self, instance_id: str, message: dict, instance_token: str, file: BinaryIO = None):
payload = {
'data': message,
'instance_token': instance_token
}
if file:
payload['files'] = {'file': file}
return self.client.post(
f'message/sendWhatsAppAudio/{instance_id}',
**payload
)
def send_status(self, instance_id: str, message: StatusMessage, instance_token: str):
return self.client.post(
f'message/sendStatus/{instance_id}',
data=message.__dict__,
instance_token=instance_token
)
def send_sticker(self, instance_id: str, message: dict, instance_token: str):
return self.client.post(
f'message/sendSticker/{instance_id}',
data=message,
instance_token=instance_token
)
def send_location(self, instance_id: str, message: LocationMessage, instance_token: str):
return self.client.post(
f'message/sendLocation/{instance_id}',
data=message.__dict__,
instance_token=instance_token
)
def send_contact(self, instance_id: str, message: ContactMessage, instance_token: str):
return self.client.post(
f'message/sendContact/{instance_id}',
data=message.__dict__,
instance_token=instance_token
)
def send_reaction(self, instance_id: str, message: ReactionMessage, instance_token: str):
return self.client.post(
f'message/sendReaction/{instance_id}',
data=message.__dict__,
instance_token=instance_token
)
def send_poll(self, instance_id: str, message: PollMessage, instance_token: str):
return self.client.post(
f'message/sendPoll/{instance_id}',
data=message.__dict__,
instance_token=instance_token
)
def send_list(self, instance_id: str, message: ListMessage, instance_token: str):
return self.client.post(
f'message/sendList/{instance_id}',
data=message.__dict__,
instance_token=instance_token
)
def send_buttons(self, instance_id: str, message: ButtonMessage, instance_token: str):
return self.client.post(
f'message/sendButtons/{instance_id}',
data=message.__dict__,
instance_token=instance_token
)

60
build/lib/evolution/services/profile.py
View File

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

0
build/lib/evolutionapi/__init__.py
View File

138
build/lib/evolutionapi/client.py
View File

@ -1,138 +0,0 @@
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
)

11
build/lib/evolutionapi/exceptions.py
View File

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

0
build/lib/evolutionapi/models/__init__.py
View File

16
build/lib/evolutionapi/models/call.py
View File

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

93
build/lib/evolutionapi/models/chat.py
View File

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

39
build/lib/evolutionapi/models/group.py
View File

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

59
build/lib/evolutionapi/models/instance.py
View File

@ -1,59 +0,0 @@
from typing import Optional, List, Dict
class WebhookConfig:
def __init__(self, url: str = None, byEvents: bool = False, base64: bool = True,
headers: Dict = None, events: List[str] = None):
self.url = url
self.byEvents = byEvents
self.base64 = base64
self.headers = headers
self.events = events
class EventsConfig:
def __init__(self, enabled: bool = True, events: List[str] = None):
self.enabled = enabled
self.events = events
class ChatwootConfig:
def __init__(self, accountId: str = None, token: str = None, url: str = None,
signMsg: bool = True, 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.chatwootToken = token
self.chatwootUrl = url
self.chatwootSignMsg = signMsg
self.chatwootReopenConversation = reopenConversation
self.chatwootConversationPending = conversationPending
self.chatwootImportContacts = importContacts
self.chatwootNameInbox = nameInbox
self.chatwootMergeBrazilContacts = mergeBrazilContacts
self.chatwootImportMessages = importMessages
self.chatwootDaysLimitImportMessages = daysLimitImportMessages
self.chatwootOrganization = organization
self.chatwootLogo = logo
class InstanceConfig:
def __init__(
self,
instanceName: str,
integration: str = None,
token: str = None,
number: str = None,
qrcode: bool = None,
rejectCall: bool = None,
msgCall: str = None,
groupsIgnore: bool = None,
alwaysOnline: bool = None,
readMessages: bool = None,
readStatus: bool = None,
syncFullHistory: bool = None
):
self.__dict__['instanceName'] = instanceName
for key, value in locals().items():
if key != 'self' and key != 'instanceName' and value is not None:
self.__dict__[key] = value

21
build/lib/evolutionapi/models/label.py
View File

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

260
build/lib/evolutionapi/models/message.py
View File

@ -1,260 +0,0 @@
from enum import Enum
from typing import List, Optional, Union
from dataclasses import dataclass
class MediaType(Enum):
IMAGE = "image"
VIDEO = "video"
DOCUMENT = "document"
class StatusType(Enum):
TEXT = "text"
IMAGE = "image"
VIDEO = "video"
AUDIO = "audio"
class FontType(Enum):
SERIF = 1
NORICAN_REGULAR = 2
BRYNDAN_WRITE = 3
BEBASNEUE_REGULAR = 4
OSWALD_HEAVY = 5
class BaseMessage:
def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class QuotedMessage(BaseMessage):
def __init__(self, key: dict, message: Optional[dict] = None):
super().__init__(key=key, message=message)
class TextMessage(BaseMessage):
def __init__(
self,
number: str,
text: str,
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None,
linkPreview: Optional[bool] = None,
mentionsEveryOne: Optional[bool] = None,
mentioned: Optional[List[str]] = None
):
super().__init__(
number=number,
text=text,
delay=delay,
quoted=quoted.__dict__ if quoted else None,
linkPreview=linkPreview,
mentionsEveryOne=mentionsEveryOne,
mentioned=mentioned
)
class MediaMessage(BaseMessage):
def __init__(
self,
number: str,
media: dict = None,
mediatype: Optional[str] = None,
caption: str = None,
mimetype: str = None,
fileName: str = None,
delay: Optional[Union[int, float, str]] = None,
quoted: Optional[QuotedMessage] = None,
mentionsEveryOne: Optional[bool] = None,
mentioned: Optional[List[str]] = None
):
data = {
'number': number,
'mediatype': mediatype,
'caption': caption,
'mimetype': mimetype,
'fileName': fileName,
'quoted': quoted.__dict__ if quoted else None,
'mentionsEveryOne': mentionsEveryOne,
'mentioned': mentioned
}
if delay is not None:
data['delay'] = delay
if media and media != {}:
data['media'] = media
super().__init__(**{k: v for k, v in data.items() if v is not None})
class StatusMessage(BaseMessage):
def __init__(
self,
type: StatusType,
content: str,
caption: Optional[str] = None,
backgroundColor: Optional[str] = None,
font: Optional[FontType] = None,
allContacts: bool = False,
statusJidList: Optional[List[str]] = None
):
super().__init__(
type=type.value,
content=content,
caption=caption,
backgroundColor=backgroundColor,
font=font.value if font else None,
allContacts=allContacts,
statusJidList=statusJidList
)
class LocationMessage(BaseMessage):
def __init__(
self,
number: str,
name: str,
address: str,
latitude: float,
longitude: float,
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None
):
super().__init__(
number=number,
name=name,
address=address,
latitude=latitude,
longitude=longitude,
delay=delay,
quoted=quoted.__dict__ if quoted else None
)
class Contact(BaseMessage):
def __init__(
self,
fullName: str,
wuid: str,
phoneNumber: str,
organization: Optional[str] = None,
email: Optional[str] = None,
url: Optional[str] = None
):
super().__init__(
fullName=fullName,
wuid=wuid,
phoneNumber=phoneNumber,
organization=organization,
email=email,
url=url
)
class ContactMessage(BaseMessage):
def __init__(self, number: str, contact: List[Contact]):
super().__init__(
number=number,
contact=[c.__dict__ for c in contact]
)
class ReactionMessage(BaseMessage):
def __init__(self, key: dict, reaction: str):
super().__init__(key=key, reaction=reaction)
class PollMessage(BaseMessage):
def __init__(
self,
number: str,
name: str,
selectableCount: int,
values: List[str],
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None
):
super().__init__(
number=number,
name=name,
selectableCount=selectableCount,
values=values,
delay=delay,
quoted=quoted.__dict__ if quoted else None
)
class ListRow(BaseMessage):
def __init__(self, title: str, description: str, rowId: str):
super().__init__(
title=title,
description=description,
rowId=rowId
)
class ListSection(BaseMessage):
def __init__(self, title: str, rows: List[ListRow]):
super().__init__(
title=title,
rows=[r.__dict__ for r in rows]
)
class ListMessage(BaseMessage):
def __init__(
self,
number: str,
title: str,
description: str,
buttonText: str,
footerText: str,
sections: List[ListSection],
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None
):
super().__init__(
number=number,
title=title,
description=description,
buttonText=buttonText,
footerText=footerText,
sections=[s.__dict__ for s in sections],
delay=delay,
quoted=quoted.__dict__ if quoted else None
)
class Button(BaseMessage):
def __init__(
self,
type: str,
displayText: str,
id: Optional[str] = None,
copyCode: Optional[str] = None,
url: Optional[str] = None,
phoneNumber: Optional[str] = None,
currency: Optional[str] = None,
name: Optional[str] = None,
keyType: Optional[str] = None,
key: Optional[str] = None
):
super().__init__(
type=type,
displayText=displayText,
id=id,
copyCode=copyCode,
url=url,
phoneNumber=phoneNumber,
currency=currency,
name=name,
keyType=keyType,
key=key
)
class ButtonMessage(BaseMessage):
def __init__(
self,
number: str,
title: str,
description: str,
footer: str,
buttons: List[Button],
delay: Optional[int] = None,
quoted: Optional[QuotedMessage] = None
):
super().__init__(
number=number,
title=title,
description=description,
footer=footer,
buttons=[b.__dict__ for b in buttons],
delay=delay,
quoted=quoted.__dict__ if quoted else None
)

9
build/lib/evolutionapi/models/presence.py
View File

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

60
build/lib/evolutionapi/models/profile.py
View File

@ -1,60 +0,0 @@
from typing import Literal
class BaseProfile:
def __init__(self, **kwargs):
self.__dict__.update({k: v for k, v in kwargs.items() if v is not None})
class FetchProfile(BaseProfile):
def __init__(
self,
number: str,
):
super().__init__(
number=number,
)
class ProfileName(BaseProfile):
def __init__(
self,
name: str,
):
super().__init__(
name=name,
)
class ProfileStatus(BaseProfile):
def __init__(
self,
status: str,
):
super().__init__(
status=status,
)
class ProfilePicture(BaseProfile):
def __init__(
self,
picture: str,
):
super().__init__(
picture=picture,
)
class PrivacySettings(BaseProfile):
def __init__(
self,
readreceipts: Literal["all", "none"],
profile: Literal["all", "contacts", "contact_blacklist", "none"],
status: Literal["all", "contacts", "contact_blacklist", "none"],
online: Literal["all", "match_last_seen"],
last: Literal["all", "contacts", "contact_blacklist", "none"],
groupadd: Literal["all", "contacts", "contact_blacklist"],
):
super().__init__(
readreceipts=readreceipts,
profile=profile,
status=status,
online=online,
last=last,
groupadd=groupadd,
)

20
build/lib/evolutionapi/models/websocket.py
View File

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

0
build/lib/evolutionapi/services/__init__.py
View File

13
build/lib/evolutionapi/services/call.py
View File

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

122
build/lib/evolutionapi/services/chat.py
View File

@ -1,122 +0,0 @@
from typing import Union, BinaryIO, Optional
from ..models.chat import *
class ChatService:
def __init__(self, client):
self.client = client
def check_is_whatsapp_numbers(self, instance_id: str, data: CheckIsWhatsappNumber, instance_token: str):
return self.client.post(
f'chat/checkIsWhatsappNumber/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def mark_message_as_read(self, instance_id: str, messages: List[ReadMessage], instance_token: str):
return self.client.post(
f'chat/markMessageAsRead/{instance_id}',
data={"readMessages": [m.__dict__ for m in messages]},
instance_token=instance_token
)
def archive_chat(self, instance_id: str, data: ArchiveChat, instance_token: str):
return self.client.post(
f'chat/archiveChat/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def mark_chat_unread(self, instance_id: str, data: UnreadChat, instance_token: str):
return self.client.post(
f'chat/markChatUnread/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def delete_message_for_everyone(self, instance_id: str, data: MessageKey, instance_token: str):
return self.client.delete(
f'chat/deleteMessageForEveryone/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def fetch_profile_picture_url(self, instance_id: str, data: ProfilePicture, instance_token: str):
return self.client.post(
f'chat/fetchProfilePictureUrl/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def get_base64_from_media_message(self, instance_id: str, data: MediaMessage, instance_token: str):
return self.client.post(
f'chat/getBase64FromMediaMessage/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def update_message(self, instance_id: str, data: UpdateMessage, instance_token: str):
return self.client.post(
f'chat/updateMessage/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def send_presence(self, instance_id: str, data: Presence, instance_token: str):
return self.client.post(
f'chat/sendPresence/{instance_id}',
data=data.__dict__,
instance_token=instance_token
)
def get_messages(
self,
instance_id: str,
remote_jid: str,
instance_token: str,
message_id: Optional[str] = None,
whatsapp_message_id: Optional[str] = None,
from_me: Optional[bool] = None,
message_type: Optional[str] = None,
source: Optional[str] = None,
timestamp_start: Optional[str] = None,
timestamp_end: Optional[str] = None,
page: int = 1,
offset: int = 50
):
'''
Obtém mensagens de um chat com filtros opcionais
Args:
timestamp_start: Data inicial no formato ISO (ex: "2025-01-16T00:00:00Z")
timestamp_end: Data final no formato ISO (ex: "2025-01-16T23:59:59Z")
'''
where = {"key": {"remoteJid": remote_jid}}
if message_id:
where["id"] = message_id
if whatsapp_message_id:
where["key"]["id"] = whatsapp_message_id
if from_me is not None:
where["key"]["fromMe"] = from_me
if message_type:
where["messageType"] = message_type
if source:
where["source"] = source
if timestamp_start or timestamp_end:
where["messageTimestamp"] = {}
if timestamp_start:
where["messageTimestamp"]["gte"] = timestamp_start
if timestamp_end:
where["messageTimestamp"]["lte"] = timestamp_end
payload = {
"where": where,
"page": page,
"offset": offset,
}
return self.client.post(
f'chat/findMessages/{instance_id}',
data=payload,
instance_token=instance_token,
)

105
build/lib/evolutionapi/services/group.py
View File

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

9
build/lib/evolutionapi/services/instance.py
View File

@ -1,9 +0,0 @@
class InstanceService:
def __init__(self, client):
self.client = client
def fetch_instances(self):
return self.client.get('instance/fetchInstances')
def create_instance(self, config):
return self.client.post('instance/create', data=config.__dict__)

28
build/lib/evolutionapi/services/instance_operations.py
View File

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

19
build/lib/evolutionapi/services/label.py
View File

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

205
build/lib/evolutionapi/services/message.py
View File

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

60
build/lib/evolutionapi/services/profile.py
View File

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

174
build/lib/evolutionapi/services/websocket.py
View File

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

18
setup.py
View File

@ -1,18 +0,0 @@
from setuptools import setup, find_packages
setup(
name='evolutionapi',
version='0.1.1',
description='Client Python para a API Evolution',
author='Davidson Gomes',
author_email='contato@agenciadgcode.com',
packages=find_packages(),
package_data={'': ['*']},
include_package_data=True,
install_requires=[
'requests>=2.25.1',
'requests_toolbelt>=1.0.0',
'python-socketio>=5.11.1'
],
python_requires='>=3.6',
)