Whatsapp Class Reference

Note: where it isn't specified, a function returns 0 for success, 1 for failure

The main whatsapp handler

Source code in whatsfly/whatsapp.py

class WhatsApp:
    """
    The main whatsapp handler
    """

    c_WhatsAppClientId = None

    def __init__(
        self,
        phone_number: str = "",
        media_path: str = "",
        machine: str = "mac",
        browser: str = "safari",
        database_dir: str = "whatsapp",
        on_event: Callable[[dict], None] = _emptyFunc,
        on_disconnect: Callable[[None], None] = None,
        print_qr_code: bool = True,
    ):
        """
        Import the compiled whatsmeow golang package, and setup basic client and database.
        Auto run based on any database (login and chat info database), hence a user phone number are declared.
        If there is no user login assigned yet, assign a new client.
        Put the database in current file whereever this class instances are imported. database/client.db
        :param phone_number: User phone number. in the Whatsmeow golang are called client.
        :param media_path: A directory to save all the media received
        :param machine: OS login info (showed on the whatsapp app)
        :param browser: Browser login info (showed on the whatsapp app)
        :param database_dir: The directory storing whatsfly's data
        :param on_event: Function to call on event
        :param on_disconnect: Function to call on disconnect
        :param print_qr_code: Setting to true will print the qr code to terminal on connection
        """

        self.phone_number = phone_number
        self.machine = machine
        self.browser = browser
        self.wapi_functions = browser
        self._messageThreadRunner = threading.Thread(target=self._messageThread)
        self._userEventHandlers = [on_event]
        self._methodReturns = {}
        self.print_qr_code = print_qr_code
        self.db_dir = database_dir

        if media_path:
            if not os.path.exists(media_path):
                os.makedirs(media_path)
            for subdir in ["images", "audios", "videos", "documents", "stickers"]:
                full_media_path = media_path + "/" + subdir
                if not os.path.exists(full_media_path):
                    os.makedirs(full_media_path)

        CMPFUNC_NONE_STR = ctypes.CFUNCTYPE(None, ctypes.c_char_p)
        CMPFUNC_NONE = ctypes.CFUNCTYPE(None)

        self.C_ON_EVENT = CMPFUNC_NONE_STR(self._handleMessage)
        self.C_ON_DISCONNECT = (
            CMPFUNC_NONE(on_disconnect)
            if callable(on_disconnect)
            else ctypes.cast(None, CMPFUNC_NONE)
        )

        self.c_WhatsAppClientId = new_whatsapp_client_wrapper(
            phone_number.encode(),
            media_path.encode(),
            self.C_ON_DISCONNECT,
            self.C_ON_EVENT,
        )

        self._messageThreadRunner.start()

    def connect(self):
        """
        Connects the whatsapp client to whatsapp servers. This method SHOULD be called before any other.
        """
        connect_wrapper(
            self.c_WhatsAppClientId, os.path.join(self.db_dir, "wapp.db").encode()
        )

    def disconnect(self):
        """
        Disconnects the whatsapp client to whatsapp servers.
        """
        disconnect_wrapper(self.c_WhatsAppClientId)

    @deprecated
    def runMessageThread(self):
        """
        Legacy method that used to run the message thread, does nothing anymore
        """
        print("This method does nothing anymore, it has been automatised")

    def _messageThread(self):
        """
        New method for runMessageThread
        """
        while True:
            message_thread_wrapper(self.c_WhatsAppClientId)

    def _handleMessage(self, message):
        try:
            message = json.loads(message.decode())
        except Exception as err:
            raise err

        match message["eventType"]:
            case "linkingCode":
                if self.print_qr_code:
                    print(message["content"])
            case "qrCode":
                if self.print_qr_code:
                    print(message)
                    print(message["content"])
                    qr = qrcode.QRCode()
                    qr.add_data(message["content"])
                    qr.print_ascii()
            case "methodReturn":
                self._methodReturns[message["content"]["returnId"]] = message
                return

        if isinstance(message["content"], dict):
            thandler = {**message, **message["content"]}
        else:
            thandler = message

        for handler in self._userEventHandlers:
            handler(self, thandler)

    def loggedIn(self) -> bool:
        """
        Determines if the user is logged into WhatsApp.

        Returns:
            bool: True if the user is logged in, False otherwise.
        """
        if self.c_WhatsAppClientId is None:
            return False
        return logged_in_wrapper(self.c_WhatsAppClientId) == 1

    def isConnected(self) -> bool:
        """

        Checks if the connection is currently established.

        Returns:
            bool: True if the client is connected, otherwise False.
        """
        if self.c_WhatsAppClientId is None:
            return False
        return connected_wrapper(self.c_WhatsAppClientId) == 1

    def sendMessage(
        self,
        phone: str,
        message,
        group: bool = False,
        upload: Upload = None,
        thumbnail_path="",
    ):
        """
        Sends a text message
        :param phone: The phone number or group number to send the message.
        :param message: The message to send. It can be a string with the message, or a protobuf message
        :param group: Is the message sent to a group ?
        :param upload: An optional Upload object to be added to the protobuf before sending.
        """

        # search what is fucked up

        ispb = True

        if isinstance(message, str):
            message1 = WAWebProtobufsE2E_pb2.Message()
            message1.conversation = message
            message = message1
            ispb = False

        if upload is None:
            ret = send_message_protobuf_wrapper(
                self.c_WhatsAppClientId,
                phone.encode(),
                message.SerializeToString(),
                group,
            )
        else:
            ret = send_message_with_upload_wrapper(
                self.c_WhatsAppClientId,
                phone.encode(),
                message.SerializeToString(),  # message.SerializeToString() if ispb else message.encode(),
                group,
                upload._getId().encode(),
                upload._getMimetype().encode(),
                upload._getKind().encode(),
                ispb,
                thumbnail_path.encode(),
            )

        return ret == 0

    def sendReaction(
        self,
        jid: str,
        message_jid: str,
        sender_jid: str,
        reaction: str,
        group: bool = False,
    ):
        """
        Reacts to a message
        :param jid: The phone number or group number the message was sent to.
        :param message_jid: The JID of the message to react to
        :param sender_jid: The JID of the message to react to
        :param reaction: The emoji of the reaction
        :param group: Was the message sent to a group ?
        """

        ret = send_reaction_wrapper(
            self.c_WhatsAppClientId,
            jid.encode(),
            message_jid.encode(),
            sender_jid.encode(),
            reaction.encode(),
            group,
        )

        return ret == 0

    def getGroupInviteLink(self, group: str, reset: bool = False) -> str:
        """
        Get invite link for group.
        Also sends an event to queue for legacy clients
        :param group: Group id
        :param reset: If true, resets the old link before generating the new one
        :return: Invite link
        """
        return_uuid = uuid.uuid1()

        get_group_invite_link_wrapper(
            self.c_WhatsAppClientId, group.encode(), reset, str(return_uuid).encode()
        )

        while str(return_uuid) not in self._methodReturns:
            time.sleep(0.001)

        response = self._methodReturns[str(return_uuid)]["content"]["returnData"]

        return response

    def joinGroupWithInviteLink(self, code: str):
        """
        Joins a group with an invite link
        :param code: The link
        """
        return join_group_with_invite_link_wrapper(
            self.c_WhatsAppClientId,
            code.encode(),
        )

    def setGroupAnnounce(self, group: str, announce: bool = True):
        """
        Set a group's announce mode (only admins can send message)
        :param group: Group id
        :param announce: Enable or not the announcement mode
        """
        return set_group_announce_wrapper(
            self.c_WhatsAppClientId, group.encode(), announce
        )

    def setGroupLocked(self, group: str, locked: bool = True):
        """
        Set a group's lock mode (only admins can change settings)
        :param group: Group id
        :param locked: Enable or not the lock mode
        """
        return set_group_locked_wrapper(self.c_WhatsAppClientId, group.encode(), locked)

    def setGroupName(self, group: str, name: str):
        """
        Set a group's name
        :param group: Group id
        :param name: Name
        """
        return set_group_name_wrapper(
            self.c_WhatsAppClientId, group.encode(), name.encode()
        )

    def setGroupTopic(self, group: str, topic: str):
        """
        Set a group's topic
        :param group: Group id
        :param topic: Topic
        """
        return set_group_topic_wrapper(
            self.c_WhatsAppClientId, group.encode(), topic.encode()
        )

    def getGroupInfo(self, group: str) -> dict:
        """
        Get info for a link
        :param group: Group id
        :return: Group information
        """
        return_uuid = uuid.uuid1()

        get_group_info_wrapper(
            self.c_WhatsAppClientId, group.encode(), str(return_uuid).encode()
        )

        while str(return_uuid) not in self._methodReturns:
            time.sleep(0.001)

        response = json.loads(
            base64.b64decode(
                self._methodReturns[str(return_uuid)]["content"]["returnData"]
            )
        )

        return response

    def uploadFile(self, path: str, kind: str, mimetype: str = None) -> id:
        """
        Uploads a file
        :param path: The filepath
        :param kind: The kind of the upload. One of: image, video, audio, document
        :return: Group information
        """

        if mimetype is None:
            mimetype = mimetypes.guess_type(path)[0]

        if kind not in ["image", "video", "audio", "document"]:
            raise Exception("Invalid kind")

        temporaryDirectory = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)

        tempName = temporaryDirectory.name + "/" + path.split("/")[-1]

        shutil.copyfile(path, tempName)

        return_uuid = uuid.uuid1()

        upload_file_wrapper(
            self.c_WhatsAppClientId,
            path.encode(),
            kind.encode(),
            str(return_uuid).encode(),
        )

        while str(return_uuid) not in self._methodReturns:
            time.sleep(0.001)

        temporaryDirectory.cleanup()

        # response = self._methodReturns[str(return_uuid)]["return"]

        return Upload(str(return_uuid), mimetype, kind)

    def createNewsletter(
        self, name: str, description: str = "", picture_path: str = ""
    ) -> dict:
        """
        Creates a new newsletter (channel)
        :param name: The name of the newsletter
        :param description: The description of the newsletter
        :param picture_path: Path to the picture file
        :return: Newsletter metadata
        """
        return_uuid = uuid.uuid1()

        create_newsletter_wrapper(
            self.c_WhatsAppClientId,
            name.encode(),
            description.encode(),
            picture_path.encode(),
            str(return_uuid).encode(),
        )

        while str(return_uuid) not in self._methodReturns:
            time.sleep(0.001)

        response = json.loads(
            base64.b64decode(
                self._methodReturns[str(return_uuid)]["content"]["returnData"]
            )
        )
        return response

    def getNewsletterInfo(self, jid: str) -> dict:
        """
        Get info for a newsletter
        :param jid: Newsletter JID
        :return: Newsletter metadata
        """
        return_uuid = uuid.uuid1()

        get_newsletter_info_wrapper(
            self.c_WhatsAppClientId, jid.encode(), str(return_uuid).encode()
        )

        while str(return_uuid) not in self._methodReturns:
            time.sleep(0.001)

        response = json.loads(
            base64.b64decode(
                self._methodReturns[str(return_uuid)]["content"]["returnData"]
            )
        )
        return response

    def getNewsletterMessages(self, jid: str, count: int = 10, before: int = 0) -> list:
        """
        Get messages from a newsletter
        :param jid: Newsletter JID
        :param count: Number of messages to retrieve
        :param before: Message server ID to fetch messages before
        :return: List of newsletter messages
        """
        return_uuid = uuid.uuid1()

        get_newsletter_messages_wrapper(
            self.c_WhatsAppClientId,
            jid.encode(),
            count,
            before,
            str(return_uuid).encode(),
        )

        while str(return_uuid) not in self._methodReturns:
            time.sleep(0.001)

        response = json.loads(
            base64.b64decode(
                self._methodReturns[str(return_uuid)]["content"]["returnData"]
            )
        )
        return response

    def getSubscribedNewsletters(self) -> list:
        """
        Get list of subscribed newsletters
        :return: List of newsletter metadata
        """
        return_uuid = uuid.uuid1()

        get_subscribed_newsletters_wrapper(
            self.c_WhatsAppClientId, str(return_uuid).encode()
        )

        while str(return_uuid) not in self._methodReturns:
            time.sleep(0.001)

        response = json.loads(
            base64.b64decode(
                self._methodReturns[str(return_uuid)]["content"]["returnData"]
            )
        )
        return response

    def uploadNewsletter(
        self, path: str, kind: str, mimetype: str = None
    ) -> NewsletterUpload:
        """
        Uploads a file for newsletter
        :param path: The filepath
        :param kind: The kind of the upload. One of: image, video, audio, document
        :return: NewsletterUpload object
        """
        if mimetype is None:
            mimetype = mimetypes.guess_type(path)[0]

        if kind not in ["image", "video", "audio", "document"]:
            raise Exception("Invalid kind")

        return_uuid = uuid.uuid1()

        upload_newsletter_wrapper(
            self.c_WhatsAppClientId,
            path.encode(),
            kind.encode(),
            str(return_uuid).encode(),
        )

        while str(return_uuid) not in self._methodReturns:
            time.sleep(0.001)

        return NewsletterUpload(str(return_uuid), mimetype, kind, "")

    def sendNewsletterMessage(self, jid: str, message, upload: NewsletterUpload = None):
        """
        Sends a message to a newsletter
        :param jid: The newsletter JID
        :param message: The message to send (string or protobuf)
        :param upload: Optional upload for media messages
        """
        if isinstance(message, str):
            message1 = WAWebProtobufsE2E_pb2.Message()
            message1.conversation = message
            message = message1

        upload_id = ""
        if upload:
            upload_id = upload._getId()

        ret = send_newsletter_wrapper(
            self.c_WhatsAppClientId,
            jid.encode(),
            message.SerializeToString(),
            upload_id.encode(),
        )

        return ret == 0

__init__(phone_number='', media_path='', machine='mac', browser='safari', database_dir='whatsapp', on_event=_emptyFunc, on_disconnect=None, print_qr_code=True)

Import the compiled whatsmeow golang package, and setup basic client and database. Auto run based on any database (login and chat info database), hence a user phone number are declared. If there is no user login assigned yet, assign a new client. Put the database in current file whereever this class instances are imported. database/client.db

Parameters:

Name	Type	Description	Default
phone_number	str	User phone number. in the Whatsmeow golang are called client.	''
media_path	str	A directory to save all the media received	''
machine	str	OS login info (showed on the whatsapp app)	'mac'
browser	str	Browser login info (showed on the whatsapp app)	'safari'
database_dir	str	The directory storing whatsfly's data	'whatsapp'
on_event	Callable[[dict], None]	Function to call on event	_emptyFunc
on_disconnect	Callable[[None], None]	Function to call on disconnect	None
print_qr_code	bool	Setting to true will print the qr code to terminal on connection	True

Source code in whatsfly/whatsapp.py

def __init__(
    self,
    phone_number: str = "",
    media_path: str = "",
    machine: str = "mac",
    browser: str = "safari",
    database_dir: str = "whatsapp",
    on_event: Callable[[dict], None] = _emptyFunc,
    on_disconnect: Callable[[None], None] = None,
    print_qr_code: bool = True,
):
    """
    Import the compiled whatsmeow golang package, and setup basic client and database.
    Auto run based on any database (login and chat info database), hence a user phone number are declared.
    If there is no user login assigned yet, assign a new client.
    Put the database in current file whereever this class instances are imported. database/client.db
    :param phone_number: User phone number. in the Whatsmeow golang are called client.
    :param media_path: A directory to save all the media received
    :param machine: OS login info (showed on the whatsapp app)
    :param browser: Browser login info (showed on the whatsapp app)
    :param database_dir: The directory storing whatsfly's data
    :param on_event: Function to call on event
    :param on_disconnect: Function to call on disconnect
    :param print_qr_code: Setting to true will print the qr code to terminal on connection
    """

    self.phone_number = phone_number
    self.machine = machine
    self.browser = browser
    self.wapi_functions = browser
    self._messageThreadRunner = threading.Thread(target=self._messageThread)
    self._userEventHandlers = [on_event]
    self._methodReturns = {}
    self.print_qr_code = print_qr_code
    self.db_dir = database_dir

    if media_path:
        if not os.path.exists(media_path):
            os.makedirs(media_path)
        for subdir in ["images", "audios", "videos", "documents", "stickers"]:
            full_media_path = media_path + "/" + subdir
            if not os.path.exists(full_media_path):
                os.makedirs(full_media_path)

    CMPFUNC_NONE_STR = ctypes.CFUNCTYPE(None, ctypes.c_char_p)
    CMPFUNC_NONE = ctypes.CFUNCTYPE(None)

    self.C_ON_EVENT = CMPFUNC_NONE_STR(self._handleMessage)
    self.C_ON_DISCONNECT = (
        CMPFUNC_NONE(on_disconnect)
        if callable(on_disconnect)
        else ctypes.cast(None, CMPFUNC_NONE)
    )

    self.c_WhatsAppClientId = new_whatsapp_client_wrapper(
        phone_number.encode(),
        media_path.encode(),
        self.C_ON_DISCONNECT,
        self.C_ON_EVENT,
    )

    self._messageThreadRunner.start()

connect()

Connects the whatsapp client to whatsapp servers. This method SHOULD be called before any other.

Source code in whatsfly/whatsapp.py

def connect(self):
    """
    Connects the whatsapp client to whatsapp servers. This method SHOULD be called before any other.
    """
    connect_wrapper(
        self.c_WhatsAppClientId, os.path.join(self.db_dir, "wapp.db").encode()
    )

createNewsletter(name, description='', picture_path='')

Creates a new newsletter (channel)

Parameters:

Name	Type	Description	Default
name	str	The name of the newsletter	required
description	str	The description of the newsletter	''
picture_path	str	Path to the picture file	''

Returns:

Type	Description
dict	Newsletter metadata

Source code in whatsfly/whatsapp.py

def createNewsletter(
    self, name: str, description: str = "", picture_path: str = ""
) -> dict:
    """
    Creates a new newsletter (channel)
    :param name: The name of the newsletter
    :param description: The description of the newsletter
    :param picture_path: Path to the picture file
    :return: Newsletter metadata
    """
    return_uuid = uuid.uuid1()

    create_newsletter_wrapper(
        self.c_WhatsAppClientId,
        name.encode(),
        description.encode(),
        picture_path.encode(),
        str(return_uuid).encode(),
    )

    while str(return_uuid) not in self._methodReturns:
        time.sleep(0.001)

    response = json.loads(
        base64.b64decode(
            self._methodReturns[str(return_uuid)]["content"]["returnData"]
        )
    )
    return response

disconnect()

Disconnects the whatsapp client to whatsapp servers.

Source code in whatsfly/whatsapp.py

def disconnect(self):
    """
    Disconnects the whatsapp client to whatsapp servers.
    """
    disconnect_wrapper(self.c_WhatsAppClientId)

getGroupInfo(group)

Get info for a link

Parameters:

Name	Type	Description	Default
group	str	Group id	required

Returns:

Type	Description
dict	Group information

Source code in whatsfly/whatsapp.py

def getGroupInfo(self, group: str) -> dict:
    """
    Get info for a link
    :param group: Group id
    :return: Group information
    """
    return_uuid = uuid.uuid1()

    get_group_info_wrapper(
        self.c_WhatsAppClientId, group.encode(), str(return_uuid).encode()
    )

    while str(return_uuid) not in self._methodReturns:
        time.sleep(0.001)

    response = json.loads(
        base64.b64decode(
            self._methodReturns[str(return_uuid)]["content"]["returnData"]
        )
    )

    return response

getGroupInviteLink(group, reset=False)

Get invite link for group. Also sends an event to queue for legacy clients

Parameters:

Name	Type	Description	Default
group	str	Group id	required
reset	bool	If true, resets the old link before generating the new one	False

Returns:

Type	Description
str	Invite link

Source code in whatsfly/whatsapp.py

def getGroupInviteLink(self, group: str, reset: bool = False) -> str:
    """
    Get invite link for group.
    Also sends an event to queue for legacy clients
    :param group: Group id
    :param reset: If true, resets the old link before generating the new one
    :return: Invite link
    """
    return_uuid = uuid.uuid1()

    get_group_invite_link_wrapper(
        self.c_WhatsAppClientId, group.encode(), reset, str(return_uuid).encode()
    )

    while str(return_uuid) not in self._methodReturns:
        time.sleep(0.001)

    response = self._methodReturns[str(return_uuid)]["content"]["returnData"]

    return response

getNewsletterInfo(jid)

Get info for a newsletter

Parameters:

Name	Type	Description	Default
jid	str	Newsletter JID	required

Returns:

Type	Description
dict	Newsletter metadata

Source code in whatsfly/whatsapp.py

def getNewsletterInfo(self, jid: str) -> dict:
    """
    Get info for a newsletter
    :param jid: Newsletter JID
    :return: Newsletter metadata
    """
    return_uuid = uuid.uuid1()

    get_newsletter_info_wrapper(
        self.c_WhatsAppClientId, jid.encode(), str(return_uuid).encode()
    )

    while str(return_uuid) not in self._methodReturns:
        time.sleep(0.001)

    response = json.loads(
        base64.b64decode(
            self._methodReturns[str(return_uuid)]["content"]["returnData"]
        )
    )
    return response

getNewsletterMessages(jid, count=10, before=0)

Get messages from a newsletter

Parameters:

Name	Type	Description	Default
jid	str	Newsletter JID	required
count	int	Number of messages to retrieve	10
before	int	Message server ID to fetch messages before	0

Returns:

Type	Description
list	List of newsletter messages

Source code in whatsfly/whatsapp.py

def getNewsletterMessages(self, jid: str, count: int = 10, before: int = 0) -> list:
    """
    Get messages from a newsletter
    :param jid: Newsletter JID
    :param count: Number of messages to retrieve
    :param before: Message server ID to fetch messages before
    :return: List of newsletter messages
    """
    return_uuid = uuid.uuid1()

    get_newsletter_messages_wrapper(
        self.c_WhatsAppClientId,
        jid.encode(),
        count,
        before,
        str(return_uuid).encode(),
    )

    while str(return_uuid) not in self._methodReturns:
        time.sleep(0.001)

    response = json.loads(
        base64.b64decode(
            self._methodReturns[str(return_uuid)]["content"]["returnData"]
        )
    )
    return response

getSubscribedNewsletters()

Get list of subscribed newsletters

Returns:

Type	Description
list	List of newsletter metadata

Source code in whatsfly/whatsapp.py

def getSubscribedNewsletters(self) -> list:
    """
    Get list of subscribed newsletters
    :return: List of newsletter metadata
    """
    return_uuid = uuid.uuid1()

    get_subscribed_newsletters_wrapper(
        self.c_WhatsAppClientId, str(return_uuid).encode()
    )

    while str(return_uuid) not in self._methodReturns:
        time.sleep(0.001)

    response = json.loads(
        base64.b64decode(
            self._methodReturns[str(return_uuid)]["content"]["returnData"]
        )
    )
    return response

isConnected()

Checks if the connection is currently established.

Returns: bool: True if the client is connected, otherwise False.

Source code in whatsfly/whatsapp.py

def isConnected(self) -> bool:
    """

    Checks if the connection is currently established.

    Returns:
        bool: True if the client is connected, otherwise False.
    """
    if self.c_WhatsAppClientId is None:
        return False
    return connected_wrapper(self.c_WhatsAppClientId) == 1

joinGroupWithInviteLink(code)

Joins a group with an invite link

Parameters:

Name	Type	Description	Default
code	str	The link	required

Source code in whatsfly/whatsapp.py

def joinGroupWithInviteLink(self, code: str):
    """
    Joins a group with an invite link
    :param code: The link
    """
    return join_group_with_invite_link_wrapper(
        self.c_WhatsAppClientId,
        code.encode(),
    )

loggedIn()

Determines if the user is logged into WhatsApp.

Returns: bool: True if the user is logged in, False otherwise.

Source code in whatsfly/whatsapp.py

def loggedIn(self) -> bool:
    """
    Determines if the user is logged into WhatsApp.

    Returns:
        bool: True if the user is logged in, False otherwise.
    """
    if self.c_WhatsAppClientId is None:
        return False
    return logged_in_wrapper(self.c_WhatsAppClientId) == 1

runMessageThread()

Legacy method that used to run the message thread, does nothing anymore

Source code in whatsfly/whatsapp.py

@deprecated
def runMessageThread(self):
    """
    Legacy method that used to run the message thread, does nothing anymore
    """
    print("This method does nothing anymore, it has been automatised")

sendMessage(phone, message, group=False, upload=None, thumbnail_path='')

Sends a text message

Parameters:

Name	Type	Description	Default
phone	str	The phone number or group number to send the message.	required
message		The message to send. It can be a string with the message, or a protobuf message	required
group	bool	Is the message sent to a group ?	False
upload	Upload	An optional Upload object to be added to the protobuf before sending.	None

Source code in whatsfly/whatsapp.py

def sendMessage(
    self,
    phone: str,
    message,
    group: bool = False,
    upload: Upload = None,
    thumbnail_path="",
):
    """
    Sends a text message
    :param phone: The phone number or group number to send the message.
    :param message: The message to send. It can be a string with the message, or a protobuf message
    :param group: Is the message sent to a group ?
    :param upload: An optional Upload object to be added to the protobuf before sending.
    """

    # search what is fucked up

    ispb = True

    if isinstance(message, str):
        message1 = WAWebProtobufsE2E_pb2.Message()
        message1.conversation = message
        message = message1
        ispb = False

    if upload is None:
        ret = send_message_protobuf_wrapper(
            self.c_WhatsAppClientId,
            phone.encode(),
            message.SerializeToString(),
            group,
        )
    else:
        ret = send_message_with_upload_wrapper(
            self.c_WhatsAppClientId,
            phone.encode(),
            message.SerializeToString(),  # message.SerializeToString() if ispb else message.encode(),
            group,
            upload._getId().encode(),
            upload._getMimetype().encode(),
            upload._getKind().encode(),
            ispb,
            thumbnail_path.encode(),
        )

    return ret == 0

sendNewsletterMessage(jid, message, upload=None)

Sends a message to a newsletter

Parameters:

Name	Type	Description	Default
jid	str	The newsletter JID	required
message		The message to send (string or protobuf)	required
upload	NewsletterUpload	Optional upload for media messages	None

Source code in whatsfly/whatsapp.py

def sendNewsletterMessage(self, jid: str, message, upload: NewsletterUpload = None):
    """
    Sends a message to a newsletter
    :param jid: The newsletter JID
    :param message: The message to send (string or protobuf)
    :param upload: Optional upload for media messages
    """
    if isinstance(message, str):
        message1 = WAWebProtobufsE2E_pb2.Message()
        message1.conversation = message
        message = message1

    upload_id = ""
    if upload:
        upload_id = upload._getId()

    ret = send_newsletter_wrapper(
        self.c_WhatsAppClientId,
        jid.encode(),
        message.SerializeToString(),
        upload_id.encode(),
    )

    return ret == 0

sendReaction(jid, message_jid, sender_jid, reaction, group=False)

Reacts to a message

Parameters:

Name	Type	Description	Default
jid	str	The phone number or group number the message was sent to.	required
message_jid	str	The JID of the message to react to	required
sender_jid	str	The JID of the message to react to	required
reaction	str	The emoji of the reaction	required
group	bool	Was the message sent to a group ?	False

Source code in whatsfly/whatsapp.py

def sendReaction(
    self,
    jid: str,
    message_jid: str,
    sender_jid: str,
    reaction: str,
    group: bool = False,
):
    """
    Reacts to a message
    :param jid: The phone number or group number the message was sent to.
    :param message_jid: The JID of the message to react to
    :param sender_jid: The JID of the message to react to
    :param reaction: The emoji of the reaction
    :param group: Was the message sent to a group ?
    """

    ret = send_reaction_wrapper(
        self.c_WhatsAppClientId,
        jid.encode(),
        message_jid.encode(),
        sender_jid.encode(),
        reaction.encode(),
        group,
    )

    return ret == 0

setGroupAnnounce(group, announce=True)

Set a group's announce mode (only admins can send message)

Parameters:

Name	Type	Description	Default
group	str	Group id	required
announce	bool	Enable or not the announcement mode	True

Source code in whatsfly/whatsapp.py

def setGroupAnnounce(self, group: str, announce: bool = True):
    """
    Set a group's announce mode (only admins can send message)
    :param group: Group id
    :param announce: Enable or not the announcement mode
    """
    return set_group_announce_wrapper(
        self.c_WhatsAppClientId, group.encode(), announce
    )

setGroupLocked(group, locked=True)

Set a group's lock mode (only admins can change settings)

Parameters:

Name	Type	Description	Default
group	str	Group id	required
locked	bool	Enable or not the lock mode	True

Source code in whatsfly/whatsapp.py

def setGroupLocked(self, group: str, locked: bool = True):
    """
    Set a group's lock mode (only admins can change settings)
    :param group: Group id
    :param locked: Enable or not the lock mode
    """
    return set_group_locked_wrapper(self.c_WhatsAppClientId, group.encode(), locked)

setGroupName(group, name)

Set a group's name

Parameters:

Name	Type	Description	Default
group	str	Group id	required
name	str	Name	required

Source code in whatsfly/whatsapp.py

def setGroupName(self, group: str, name: str):
    """
    Set a group's name
    :param group: Group id
    :param name: Name
    """
    return set_group_name_wrapper(
        self.c_WhatsAppClientId, group.encode(), name.encode()
    )

setGroupTopic(group, topic)

Set a group's topic

Parameters:

Name	Type	Description	Default
group	str	Group id	required
topic	str	Topic	required

Source code in whatsfly/whatsapp.py

def setGroupTopic(self, group: str, topic: str):
    """
    Set a group's topic
    :param group: Group id
    :param topic: Topic
    """
    return set_group_topic_wrapper(
        self.c_WhatsAppClientId, group.encode(), topic.encode()
    )

uploadFile(path, kind, mimetype=None)

Uploads a file

Parameters:

Name	Type	Description	Default
path	str	The filepath	required
kind	str	The kind of the upload. One of: image, video, audio, document	required

Returns:

Type	Description
id	Group information

Source code in whatsfly/whatsapp.py

def uploadFile(self, path: str, kind: str, mimetype: str = None) -> id:
    """
    Uploads a file
    :param path: The filepath
    :param kind: The kind of the upload. One of: image, video, audio, document
    :return: Group information
    """

    if mimetype is None:
        mimetype = mimetypes.guess_type(path)[0]

    if kind not in ["image", "video", "audio", "document"]:
        raise Exception("Invalid kind")

    temporaryDirectory = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)

    tempName = temporaryDirectory.name + "/" + path.split("/")[-1]

    shutil.copyfile(path, tempName)

    return_uuid = uuid.uuid1()

    upload_file_wrapper(
        self.c_WhatsAppClientId,
        path.encode(),
        kind.encode(),
        str(return_uuid).encode(),
    )

    while str(return_uuid) not in self._methodReturns:
        time.sleep(0.001)

    temporaryDirectory.cleanup()

    # response = self._methodReturns[str(return_uuid)]["return"]

    return Upload(str(return_uuid), mimetype, kind)

uploadNewsletter(path, kind, mimetype=None)

Uploads a file for newsletter

Parameters:

Name	Type	Description	Default
path	str	The filepath	required
kind	str	The kind of the upload. One of: image, video, audio, document	required

Returns:

Type	Description
NewsletterUpload	NewsletterUpload object

Source code in whatsfly/whatsapp.py

def uploadNewsletter(
    self, path: str, kind: str, mimetype: str = None
) -> NewsletterUpload:
    """
    Uploads a file for newsletter
    :param path: The filepath
    :param kind: The kind of the upload. One of: image, video, audio, document
    :return: NewsletterUpload object
    """
    if mimetype is None:
        mimetype = mimetypes.guess_type(path)[0]

    if kind not in ["image", "video", "audio", "document"]:
        raise Exception("Invalid kind")

    return_uuid = uuid.uuid1()

    upload_newsletter_wrapper(
        self.c_WhatsAppClientId,
        path.encode(),
        kind.encode(),
        str(return_uuid).encode(),
    )

    while str(return_uuid) not in self._methodReturns:
        time.sleep(0.001)

    return NewsletterUpload(str(return_uuid), mimetype, kind, "")

Comments