Source code for slidge_whatsapp.gateway

from logging import getLevelName, getLogger
from pathlib import Path
from typing import TYPE_CHECKING

from slidge import BaseGateway, FormField, GatewayUser, global_config

from . import config
from .generated import whatsapp

if TYPE_CHECKING:
    from .session import Session

[docs]REGISTRATION_INSTRUCTIONS = (
    "Continue and scan the resulting QR codes on your main device, or alternatively, "
    "use the 'pair-phone' command to complete registration. More information at "
    "https://slidge.im/slidge-whatsapp/user.html"
)

[docs]WELCOME_MESSAGE = (
    "Thank you for registering! Please scan the following QR code on your main device "
    "or use the 'pair-phone' command to complete registration, or type 'help' to list "
    "other available commands."
)


[docs]class Gateway(BaseGateway):
[docs]    COMPONENT_NAME = "WhatsApp (slidge)"
[docs]    COMPONENT_TYPE = "whatsapp"
[docs]    COMPONENT_AVATAR = "https://www.whatsapp.com/apple-touch-icon.png"
[docs]    ROSTER_GROUP = "WhatsApp"

[docs]    REGISTRATION_INSTRUCTIONS = REGISTRATION_INSTRUCTIONS
[docs]    WELCOME_MESSAGE = WELCOME_MESSAGE
[docs]    REGISTRATION_FIELDS = []

[docs]    SEARCH_FIELDS = [
        FormField(var="phone", label="Phone number", required=True),
    ]

[docs]    MARK_ALL_MESSAGES = True
[docs]    GROUPS = True
[docs]    PROPER_RECEIPTS = True

    def __init__(self):
        super().__init__()
[docs]        self.whatsapp = whatsapp.NewGateway()
        self.whatsapp.Name = "Slidge on " + str(global_config.JID)
        self.whatsapp.LogLevel = getLevelName(getLogger().level)

        assert config.DB_PATH is not None
        Path(config.DB_PATH.parent).mkdir(exist_ok=True)
        self.whatsapp.DBPath = str(config.DB_PATH)

        (global_config.HOME_DIR / "tmp").mkdir(exist_ok=True)
        self.whatsapp.TempDir = str(global_config.HOME_DIR / "tmp")
        self.whatsapp.Init()

[docs]    async def validate(self, user_jid, registration_form):
        """
        Validate registration form. A no-op for WhatsApp, as actual registration takes place
        after in-band registration commands complete; see :meth:`.Session.login` for more.
        """
        pass

[docs]    async def unregister(self, user: GatewayUser):
        """
        Logout from the active WhatsApp session. This will also force a remote log-out, and thus
        require pairing on next login. For simply disconnecting the active session, look at the
        :meth:`.Session.disconnect` function.
        """
        session: "Session" = self.get_session_from_user(user)  # type:ignore
        session.whatsapp.Logout()
        try:
            device_id = session.user.legacy_module_data["device_id"]
            self.whatsapp.CleanupSession(whatsapp.LinkedDevice(ID=device_id))
        except KeyError:
            pass
        except RuntimeError as err:
            log.error("Failed to clean up WhatsApp session: %s", err)


[docs]log = getLogger(__name__)