mirror of
https://github.com/EvolutionAPI/evolution-client-python.git
synced 2026-02-03 21:36:24 -06:00
adjusts in package
This commit is contained in:
Davidson Gomes
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,81 @@
|
||||
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)
|
||||
@@ -0,0 +1,11 @@
|
||||
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
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,16 @@
|
||||
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
|
||||
)
|
||||
@@ -0,0 +1,93 @@
|
||||
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
|
||||
@@ -0,0 +1,39 @@
|
||||
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
|
||||
@@ -0,0 +1,59 @@
|
||||
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
|
||||
@@ -0,0 +1,21 @@
|
||||
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
|
||||
)
|
||||
@@ -0,0 +1,254 @@
|
||||
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
|
||||
)
|
||||
@@ -0,0 +1,9 @@
|
||||
from enum import Enum
|
||||
|
||||
class PresenceStatus(Enum):
|
||||
AVAILABLE = "available"
|
||||
UNAVAILABLE = "unavailable"
|
||||
|
||||
class PresenceConfig:
|
||||
def __init__(self, presence: PresenceStatus):
|
||||
self.presence = presence.value
|
||||
@@ -0,0 +1,60 @@
|
||||
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,
|
||||
)
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,13 @@
|
||||
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
|
||||
)
|
||||
@@ -0,0 +1,69 @@
|
||||
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
|
||||
)
|
||||
@@ -0,0 +1,117 @@
|
||||
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
|
||||
)
|
||||
@@ -0,0 +1,9 @@
|
||||
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__)
|
||||
@@ -0,0 +1,28 @@
|
||||
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)
|
||||
@@ -0,0 +1,19 @@
|
||||
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
|
||||
)
|
||||
@@ -0,0 +1,111 @@
|
||||
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
|
||||
)
|
||||
@@ -0,0 +1,60 @@
|
||||
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
|
||||
)
|
||||
Reference in New Issue
Block a user